feta/manager.go

package feta

import "sync"
import "time"
import "runtime"

//import "github.com/gin-gonic/gin"
import "github.com/rs/zerolog/log"

const HOST_DISCOVERY_PARALLELISM = 1

type InstanceBackend interface {
	//FIXME
}

type InstanceManager struct {
	mu                       sync.Mutex
	instances                map[InstanceHostname]*instance
	newInstanceNotifications chan InstanceHostname
	startup                  time.Time
	hostAdderSemaphore       chan bool
}

func NewInstanceManager() *InstanceManager {
	i := new(InstanceManager)
	i.hostAdderSemaphore = make(chan bool, HOST_DISCOVERY_PARALLELISM)
	i.instances = make(map[InstanceHostname]*instance)
	return i
}

func (self *InstanceManager) logCaller(msg string) {
	fpcs := make([]uintptr, 1)
	// Skip 2 levels to get the caller
	n := runtime.Callers(3, fpcs)
	if n == 0 {
		log.Debug().Msg("MSG: NO CALLER")
	}

	caller := runtime.FuncForPC(fpcs[0] - 1)
	if caller == nil {
		log.Debug().Msg("MSG CALLER WAS NIL")
	}

	// Print the file name and line number
	filename, line := caller.FileLine(fpcs[0] - 1)
	function := caller.Name()

	log.Debug().
		Str("filename", filename).
		Int("linenum", line).
		Str("function", function).
		Msg(msg)
}

func (self *InstanceManager) Lock() {
	//self.logCaller("instancemanager attempting to lock")
	self.mu.Lock()
	//self.logCaller("instancemanager locked")
}

func (self *InstanceManager) Unlock() {
	self.mu.Unlock()
	//self.logCaller("instancemanager unlocked")
}

func (self *InstanceManager) AddInstanceNotificationChannel(via chan InstanceHostname) {
	self.Lock()
	defer self.Unlock()
	self.newInstanceNotifications = via
}

func (self *InstanceManager) Manage() {
	log.Info().Msg("InstanceManager starting")
	go func() {
		self.receiveNewInstanceHostnames()
	}()
	self.startup = time.Now()
	x := self.startup
	for {
		log.Info().Msg("InstanceManager tick")
		self.managerLoop()
		time.Sleep(1 * time.Second)
		if time.Now().After(x.Add(LOG_REPORT_INTERVAL)) {
			x = time.Now()
			self.logInstanceReport()
		}
	}
}

func (self *InstanceManager) managerLoop() {
	self.Lock()
	il := make([]*instance, 0)
	for _, v := range self.instances {
		il = append(il, v)
	}
	self.Unlock()

	for _, v := range il {
		go func(i *instance) {
			i.Tick()
		}(v)
	}
}

func (self *InstanceManager) hostnameExists(newhn InstanceHostname) bool {
	self.Lock()
	defer self.Unlock()
	for k, _ := range self.instances {
		if newhn == k {
			return true
		}
	}
	return false
}

func (self *InstanceManager) addInstanceByHostname(newhn InstanceHostname) {
	if self.hostnameExists(newhn) {
		return
	}

	// this blocks on the channel size, limiting concurrency
	self.hostAdderSemaphore <- true

	i := NewInstance(func(x *instance) {
		x.hostname = string(newhn)
	})
	// we do node detection under the addLock to avoid thundering
	// on startup
	i.detectNodeTypeIfNecessary()

	// pop an item from the buffered channel
	<-self.hostAdderSemaphore

	// lock the map to insert
	self.Lock()
	self.instances[newhn] = i
	self.Unlock()

}

func (self *InstanceManager) receiveNewInstanceHostnames() {
	var newhn InstanceHostname
	for {
		newhn = <-self.newInstanceNotifications
		// receive them fast out of the channel, let the adding function lock to add
		// them one at a time
		go self.addInstanceByHostname(newhn)
	}
}

func (self *InstanceManager) logInstanceReport() {
	r := self.instanceSummaryReport()

	sublogger := log.With().Logger()

	for k, v := range r {
		sublogger = sublogger.With().Uint(k, v).Logger()
	}

	sublogger.Info().
		Msg("instance report")
}

func (self *InstanceManager) listInstances() []*instance {
	var out []*instance
	self.Lock()
	defer self.Unlock()
	for _, v := range self.instances {
		out = append(out, v)
	}
	return out
}

func (im *InstanceManager) instanceSummaryReport() map[string]uint {
	r := make(map[string]uint)
	for _, v := range im.listInstances() {
		v.Lock()
		r[v.Status()]++
		v.Unlock()
	}
	return r
}