collector_go/collector.go

package collector_go

import (
	"context"
	cm "loafle.com/overflow/agent_api/config_manager"
	"loafle.com/overflow/crawler_go/grpc"
	crm "loafle.com/overflow/crawler_manager_go"
	s "loafle.com/overflow/scheduler_go"
	"log"
	"strconv"
	"sync"
	"time"
)

var (
	instance *Collector
	once     sync.Once
)

type Collector struct {
	scheduler *s.Scheduler
	cm        cm.ConfigManager
	dataCh    chan interface{}
}

func Start(started chan bool, dataCh chan interface{}, conf cm.ConfigManager) {
	c := GetInstance()
	c.dataCh = dataCh
	c.start(started, conf)
}

func Stop(stopped chan bool) {
	c := GetInstance()
	c.stop()
	stopped <- true
}

func GetInstance() *Collector {
	once.Do(func() {
		instance = &Collector{}
	})
	return instance
}

func (c *Collector) start(started chan bool, conf cm.ConfigManager) {
	go func() {
		c.cm = conf
		c.scheduler = &s.Scheduler{}
		c.scheduler.Start()

		for _, conf := range c.cm.GetSensors() {
			if err := c.addSensor(conf.Id); err != nil {
				log.Println(err)
			}
		}
		started <- true
	}()
}

func (c *Collector) stop() {
	c.scheduler.RemoveAllSchedule()
	c.scheduler.Stop()
}

func (c *Collector) collect(id string) {

	conf := c.cm.GetSensorById(id)
	log.Printf("COLLECT %s - [ID: %s] [Crawler : %s]", time.Now(), conf.Id, conf.Crawler.Name)
	conn, err := crm.GetInstance().GetClient(conf.Crawler.Container)
	if err != nil {
		log.Println(err)
	}
	defer conn.Close()

	dc := grpc.NewDataClient(conn)
	in := &grpc.Input{}

	in.Id = id
	in.Name = grpc.Crawlers(grpc.Crawlers_value[conf.Crawler.Name])

	out, err := dc.Get(context.Background(), in)

	if err != nil {
		log.Println(err)
	}
	log.Println("collector get result : ", out)
	c.dataCh <- out

}

func (c *Collector) addSensor(sensorId string) error {
	sensor := c.cm.GetSensorById(sensorId)
	interval, err := strconv.Atoi(sensor.Schedule.Interval)
	if err != nil {
		return err
	}
	return c.scheduler.NewSchedule(sensorId, uint64(interval), c.collect)
}

func (c *Collector) removeSensor(id string) error {
	if err := c.scheduler.RemoveSchedule(id); err != nil {
		return err
	}
	return nil
}

func (c *Collector) updateSensor(id string) error {
	err := c.removeSensor(id)
	if err != nil {
		return err
	}
	return c.addSensor(id)
}

func AddSensor(id string) error {
	return GetInstance().addSensor(id)
}

func RemSensor(id string) error {
	return GetInstance().removeSensor(id)
}

func UpdateSensor(id string) error {
	return GetInstance().updateSensor(id)
}

func StartSensor(id string) error {
	return GetInstance().scheduler.StartSchedule(id)
}

func StopSensor(id string) error {
	return GetInstance().scheduler.StopSchedule(id)
}