DMAC/internal/pkg/reactor/coordinator.go

package reactor

// file describes reactor level coordinator and associated implementation

import (
	pb "FRMS/internal/pkg/grpc"
	"FRMS/internal/pkg/influxdb"
	"FRMS/internal/pkg/logging"
	"FRMS/internal/pkg/manager"
	"FRMS/internal/pkg/system"
	"context"
	"fmt"
	"time"

	"github.com/spf13/viper"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"
)

// basic manager
// I dont think I actually need this interface, package manager has a point
type Manager interface {
	Start() error
	Exit() error
	Timeout() (time.Duration, error)
	HeartBeat(chan struct{}, int, int, time.Duration) // creates a hb
}

func NewManager(max int) Manager {
	return manager.New(max)
}

// db client
type DBClient interface {
	//
	Start() error
}

func NewDBClient(config *viper.Viper) (DBClient, error) {
	return influxdb.NewDBClient(config)
}

type Server struct {
	Ip   string `mapstructure:"ip"`
	Port int    `mapstructure:"port"`
}

type ReactorInfo struct {
	Name  string `mapstructure:"name,omitempty"`
	ID    int    `mapstructure:"id,omitempty"`
	Model string `mapstructure:"model,omitempty"`
	HB    int    `mapstructure:"heartbeat"`
	Bus   int    `mapstructure:"bus"`

	Server
}

type ReactorCoordinator struct {
	Manager              // base manager
	Config  *viper.Viper // config

	ReactorInfo `mapstructure:",squash"`

	Database DBClient

	pb.MonitoringClient // grpc embedding

	*DeviceCoordinator // struct for locking

	Err chan error
}

func NewCoordinator(config *viper.Viper, errCh chan error) *ReactorCoordinator {

	m := NewManager(6) // max 6 attempts
	dc := NewDeviceCoordinator(config)

	c := &ReactorCoordinator{
		Manager:           m,
		Config:            config,
		DeviceCoordinator: dc,
		Err:               errCh,
	}

	// this is going to be scuffed
	//c.DB = &DB{Bucket: "bb", Org: "ForeLight", URL: url, Token: "S1UZssBu6KPfHaQCt34pZFpyc5lzbH9XanYJWCkOI5FqLY7gq205C6FTH-CmugiPH6o2WoKlTkEuPgIfaJjAhw=="}

	return c
}

func (c *ReactorCoordinator) Start() {
	// should discover hwinfo and sensors on its own
	// now setting up sensor managers
	var err error

	if err = c.Manager.Start(); err != nil {
		c.Err <- err
	}
	// load config
	if err = c.LoadConfig(); err != nil { // loads info
		c.Err <- err
	}

	if err = c.DeviceCoordinator.Start(c.ReactorInfo.Bus); err != nil {
		c.Err <- err
	}

	// loading clients
	if c.Database, err = NewDBClient(c.Config); err != nil {
		c.Err <- err
	}

	go c.Discover()
	go c.Database.Start()
}

func (c *ReactorCoordinator) LoadConfig() error {

	var err error

	// get hb
	if !c.Config.IsSet("reactor.heartbeat") {
		// default to 5 seconds
		c.Config.Set("reactor.heartbeat", 5)
	}

	// check id
	if !c.Config.IsSet("reactor.id") {
		// get from hw
		var id int
		if id, err = system.GetId("eth0"); err != nil {
			return err
		}
		c.Config.Set("reactor.id", id)
	}

	// check Model
	if !c.Config.IsSet("reactor.model") {
		// get from hw
		var model string
		if model, err = system.GetModel(); err != nil {
			return err
		}
		c.Config.Set("reactor.model", model)
	}

	// check i2c bus
	if !c.Config.IsSet("reactor.bus") {
		// get from hw
		var bus int
		if bus, err = system.GetBus(); err != nil {
			return err
		}
		c.Config.Set("reactor.bus", bus)
	}

	// all good, unmarhsaling
	c.Config.UnmarshalKey("reactor", c)

	return err
}

func (c *ReactorCoordinator) Monitor() {
	// periodically grabs connected devs and updates list
	ch := make(chan struct{})
	go c.HeartBeat(ch, c.HB, 0, time.Second)

	for range ch {
		// check devs and ping
		logging.Debug(logging.DClient, "RLC Pinging server")
		// ping central server with status
		go c.Ping()
	}
}

func (c *ReactorCoordinator) Discover() {
	// sets up connection to central coordiantor
	conn, err := c.Connect(c.Ip, c.Port)
	if err != nil {
		c.Err <- err
	}
	defer conn.Close()
	client := pb.NewHandshakeClient(conn)
	req := &pb.ClientRequest{ClientId: uint32(c.ID), ClientType: "reactor"}
	resp, err := client.ClientDiscoveryHandler(context.Background(), req)
	if err != nil {
		c.Err <- err
	}
	c.Port = int(resp.GetServerPort()) // updating server port
	logging.Debug(logging.DClient, "RLC Central server reached, supplied port %v", c.Port)
	// connecting to manager now
	clientConn, err := c.Connect(c.Ip, c.Port)
	if err != nil {
		c.Err <- err
	}
	c.MonitoringClient = pb.NewMonitoringClient(clientConn)
	// manager
	go c.Monitor()

}

func (c *ReactorCoordinator) Connect(ip string, port int) (*grpc.ClientConn, error) {
	// function connects to central server and passes hwinfo
	var opts []grpc.DialOption
	opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
	var conn *grpc.ClientConn
	var err error
	for {
		conn, err = grpc.Dial(fmt.Sprintf("%v:%v", ip, port), opts...)
		code := status.Code(err)
		if code != 0 { // != OK
			if code == (5 | 14) { // service temp down
				var to time.Duration
				if to, err = c.Timeout(); err != nil {
					// from manager
					return &grpc.ClientConn{}, err
				}
				logging.Debug(logging.DClient, "Server currently unavailable, retrying in %v", to)
				time.Sleep(to)
			} else {
				return &grpc.ClientConn{}, err
			}
		}
		break
	}
	return conn, nil
}

func (c *ReactorCoordinator) Ping() {
	// send device info to central coordinator
	fmt.Printf("Pinging server\n")

	var devices []*pb.Device
	var err error

	if devices, err = c.GetDeviceInfo(); err != nil {
		c.Err <- err
	}

	// create request
	req := &pb.ReactorStatusPing{
		Id:      int32(c.ID),
		Devices: devices,
	}

	// ping server
	if _, err = c.ReactorStatusHandler(context.Background(), req); err != nil {
		c.Err <- err
	}
}