collector_go/collector.go

package collector_go

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	s "loafle.com/overflow/collector_go/scheduler"
	conf "loafle.com/overflow/crawler_go/config"
	g "loafle.com/overflow/crawler_go/grpc"
	crm "loafle.com/overflow/crawler_manager_go"
	"log"
	"os"
	"path/filepath"
	"strings"
	"time"
)

const CONFIG_ROOT = "/config/container"

type Collector struct {
	scheduler s.Scheduler
	//configs   []*conf.Config
	configs map[string]*conf.Config
}

func (c *Collector) Start() {
	go func() {
		c.configs = make(map[string]*conf.Config, 0)

		if err := c.readAllConfig(); err != nil {
			log.Println(err)
		}

		c.scheduler = s.Scheduler{}
		c.scheduler.Init()

		for _, conf := range c.configs {
			if err := c.addSensor(conf); err != nil {
				log.Println(err)
			}
		}
	}()
}

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

	conf := c.configs[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 := g.NewDataClient(conn)
	in := &g.Input{}

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

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

	if err != nil {
		log.Println(err)
	}
	log.Println(out)
}

//func (c *Collector) AddSensor(path string) {
//	config := c.readConfig(CONFIG_ROOT + path)
//	if config != nil {
//		if err := c.addSensor(config); err != nil {
//			log.Println(err)
//		}
//	}
//}

func (c *Collector) AddSensor(conf *conf.Config) {

	if c.checkExist(conf.Id) {
		log.Println("The Same Id already exists.")
		return
	}
	c.configs[conf.Id] = conf

	if conf != nil {
		if err := c.addSensor(conf); err != nil {
			log.Println(err)
		}
	}
}

func (c *Collector) RemoveSensor(id string) {
	if err := c.scheduler.RemoveSchedule(id); err != nil {
		log.Println(err)
	}
}

func (c *Collector) UpdateSensor(newConf *conf.Config) {
	if newConf != nil {
		if !c.checkExist(newConf.Id) {
			log.Println("Cannot update Sensor : ID not exist [" + newConf.Id + "]")
			return
		}

		exConf := c.configs[newConf.Id]

		if exConf.Schedule.Interval != newConf.Schedule.Interval {
			c.scheduler.UpdateSchedule(newConf.Id, newConf.Schedule.Interval)
		}

		c.configs[newConf.Id] = newConf
	}
}

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

func (c *Collector) addSensor(conf *conf.Config) error {
	return c.scheduler.NewSchedule(conf.Id, conf.Schedule.Interval, c.collect)
}

func (c *Collector) readAllConfig() error {

	err := filepath.Walk(CONFIG_ROOT, func(path string, f os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if !f.IsDir() && strings.HasSuffix(f.Name(), ".conf") {
			c.readConfig(path)
		}
		return nil
	})

	if err != nil {
		return err
	}

	if len(c.configs) <= 0 {
		return errors.New("No configuration file found.")
	}

	return nil
}

func (c *Collector) readConfig(path string) *conf.Config {
	bytes, err := ioutil.ReadFile(path)
	if err != nil {
		fmt.Println(err)
		return nil
	}
	conf := conf.Config{}
	json.Unmarshal(bytes, &conf)

	if err := c.validateConfig(&conf, path); err != nil {
		fmt.Println(err)
		return nil
	}
	c.configs[conf.Id] = &conf
	return &conf
}

func (c *Collector) validateConfig(conf *conf.Config, configPath string) error {

	//todo : some validations
	if c.checkExist(conf.Id) {
		return errors.New("The Same Id already exists. " + configPath)
	}
	return nil
}

func (c *Collector) checkExist(id string) bool {
	if _, exists := c.configs[id]; exists {
		return true
	}
	return false
}