app/system/system.go

package system

import (
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
	"github.com/spf13/viper"
	"kumoly.io/kumoly/app/history"
	"kumoly.io/kumoly/app/util"
)

var l zerolog.Logger

func init() {
	l = log.With().Str("mod", "system").Logger()
	viper.SetDefault("system.terminate_timeout", 10)
}

type Service interface {
	Init() error
	Load() error
	Main() error
	Del()
	Health() error
	GetName() string
	GetDependencies() []string
	IsService() bool
}

type state int

const (
	sys_init = iota
	sys_load
	sys_main
	sys_wait
	sys_term
	sys_exit
	sys_restart
)

type System struct {
	services   []Service
	isService  bool
	status     state
	lock       sync.Mutex
	started    chan struct{}
	term       chan struct{}
	done       chan struct{}
	notifyDone chan struct{}
	quit       chan os.Signal
}

func (sys *System) Append(srv ...Service) {
	if sys.services == nil {
		sys.services = make([]Service, 0)
	}
	sys.services = append(sys.services, srv...)
}

func (sys *System) order() {
	nodes := []*node{}
	for i := range sys.services {
		nodes = append(nodes, &node{
			srv:  &sys.services[i],
			name: sys.services[i].GetName(),
			edge: sys.services[i].GetDependencies(),
		})
	}
	nodes = topo(nodes)
	sys.services = make([]Service, 0)
	for i := range nodes {
		sys.services = append(sys.services, *nodes[i].srv)
	}
	if !util.PROD {
		seq := ""
		for i := range sys.services {
			seq += sys.services[i].GetName()
			if i < len(sys.services)-1 {
				seq += ", "
			}
		}
		l.Debug().Msg(seq)
	}
}

func New() *System {
	sys := &System{}
	sys.done = make(chan struct{}, 1)
	sys.started = make(chan struct{}, 1)
	sys.term = make(chan struct{}, 1)
	sys.notifyDone = make(chan struct{}, 1)
	sys.quit = make(chan os.Signal, 1)
	signal.Notify(sys.quit, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
	return sys
}

func (sys *System) Start() {
	Setup()
	sys.order()
	sys.status = sys_init
	go func() {
		l.Info().Msg("Initiating...")
		if err := sys.init(); err != nil {
			sys.quit <- syscall.SIGTERM
			return
		}
		sys.status = sys_load

		l.Info().Msg("Loading...")
		if err := sys.load(); err != nil {
			sys.quit <- syscall.SIGTERM
			return
		}

		l.Info().Msg("Starting...")
		// if e := l.Trace(); e.Enabled() {
		// 	config, err := ConfigShow("")
		// 	if err == nil {
		// 		e.Str("config", config).Msg("config")
		// 	}
		// }

		sys.status = sys_main
		sys.main()
		sys.status = sys_wait
		sys.started <- struct{}{}
		if !sys.isService {
			sys.quit <- syscall.SIGTERM
		}
	}()
	go func() {
		<-sys.started
		history.Send(history.Info().Nm("SysUP").Msg("system started"))
	}()
	<-sys.quit
	isRestart := false
	if sys.status == sys_restart {
		isRestart = true
	}
	// only log if system has started
	if sys.status == sys_wait {
		history.Send(history.Info().Nm("SysDown").Msg("system stoped"))
	}
	l.Info().Msg("Terminating...")
	sys.status = sys_term
	go func() {
		time.Sleep(time.Second * time.Duration(viper.GetInt("system.terminate_timeout")))
		if sys.status == sys_term {
			l.Warn().Msg("System Termination Timeout reached! Force Terminating.")
			sys.done <- struct{}{}
		}
	}()
	go func() {
		sys.del()
		sys.done <- struct{}{}
	}()

	<-sys.done
	sys.status = sys_exit
	l.Info().Msg("Service terminated.")
	if isRestart {
		l.Info().Msg("Restarting...")
		sys.term <- struct{}{}
	} else {
		sys.notifyDone <- struct{}{}
	}
}

// Stop stops the system
func (sys *System) Stop() {
	sys.quit <- syscall.SIGTERM
}

func (sys *System) Restart() {
	sys.lock.Lock()
	defer sys.lock.Unlock()
	sys.status = sys_restart
	sys.Stop()
	<-sys.term
	go sys.Start()
}

// Done listens to system termination
func (sys *System) Done() chan struct{} {
	return sys.notifyDone
}

func (sys *System) init() error {
	for i := range sys.services {
		l.Debug().Int("step", i).Str("srvname", sys.services[i].GetName()).Msg("")
		if err := sys.services[i].Init(); err != nil {
			l.Error().Err(err).Int("step", i).Str("srvname", sys.services[i].GetName()).Msg("")
			return err
		}
	}
	return nil
}
func (sys *System) load() error {
	for i := range sys.services {
		l.Debug().Int("step", i).Str("srvname", sys.services[i].GetName()).Msg("")
		if err := sys.services[i].Load(); err != nil {
			l.Error().Err(err).Int("step", i).Str("srvname", sys.services[i].GetName()).Msg("")
			return err
		}
	}
	return nil
}
func (sys *System) main() error {
	for i := range sys.services {
		if sys.services[i].IsService() {
			sys.isService = true
		}
		l.Debug().Int("step", i).Str("srvname", sys.services[i].GetName()).Msg("")
		if err := sys.services[i].Main(); err != nil {
			l.Error().Err(err).Int("step", i).Str("srvname", sys.services[i].GetName()).Msg("")
			return err
		}
	}
	return nil
}
func (sys *System) del() error {
	var wg sync.WaitGroup
	for i := range sys.services {
		wg.Add(1)
		go func(index int, srv Service) {
			l.Debug().Int("step", index).Str("srvname", sys.services[index].GetName()).Msg("")
			srv.Del()
			wg.Done()
		}(i, sys.services[i])
	}
	wg.Wait()
	return nil
}