ing

service/CollectorService.go

@@ -1,10 +1,18 @@
 package service
 
 import (
+	"fmt"
 	"reflect"
+	"strconv"
+	"time"
 
 	cda "git.loafle.net/commons/di-go/annotation"
 	cdr "git.loafle.net/commons/di-go/registry"
+	logging "git.loafle.net/commons/logging-go"
+	cuts "git.loafle.net/commons/util-go/time/scheduler"
+	cutss "git.loafle.net/commons/util-go/time/scheduler/storage"
+	occp "git.loafle.net/overflow/commons-go/config/probe"
+	ocmsc "git.loafle.net/overflow/commons-go/model/sensorconfig"
 	ocsp "git.loafle.net/overflow/commons-go/service/probe"
 
 	// For annotation
@@ -21,22 +29,71 @@ type CollectorService struct {
 	ocsp.CollectorService
 	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
 
-	DataClientService *DataClientService `annotation:"@Inject()"`
+	DataClientService   *DataClientService   `annotation:"@Inject()"`
+	ContainerService    *ContainerService    `annotation:"@Inject()"`
+	SensorConfigService *SensorConfigService `annotation:"@Inject()"`
+
+	scheduler *cuts.Scheduler
 }
 
 func (s *CollectorService) InitService() error {
+	_storage := cutss.NewMemoryStorage()
+	s.scheduler = cuts.New(_storage)
+
 	return nil
 }
 
 func (s *CollectorService) StartService() error {
+	if err := s.scheduler.Start(); nil != err {
+		return err
+	}
 
+	if err := s.addScheduleAll(); nil != err {
+		return err
+	}
 	return nil
 }
 
 func (s *CollectorService) StopService() {
-
+	s.scheduler.Stop()
 }
 
 func (s *CollectorService) DestroyService() {
 
 }
+
+func (s *CollectorService) addScheduleAll() error {
+	sensorConfigs := s.SensorConfigService.sensorConfigs
+	if nil == sensorConfigs || 0 == len(sensorConfigs) {
+		return nil
+	}
+
+	for _, sensorConfig := range sensorConfigs {
+		interval, err := strconv.ParseInt(sensorConfig.Schedule.Interval, 10, 64)
+		if nil != err {
+			return fmt.Errorf("Cannot convert interval[%s] %v", sensorConfig.Schedule.Interval, err)
+		}
+		s.addSchedule(interval, sensorConfig)
+	}
+
+	return nil
+}
+
+func (s *CollectorService) addSchedule(interval int64, sensorConfig *ocmsc.SensorConfig) {
+	s.scheduler.RunEvery(time.Duration(interval)*time.Second, s.collectTask, sensorConfig)
+}
+
+func (s *CollectorService) collectTask(sensorConfig *ocmsc.SensorConfig) {
+	go func(_sensorConfig *ocmsc.SensorConfig) {
+		logging.Logger().Debugf("CollectorService.collectTask for sensor config id[%s] container[%s] crawler[%s]", _sensorConfig.ConfigID, _sensorConfig.Crawler.Container, _sensorConfig.Crawler.Name)
+		var result map[string]string
+		err := s.ContainerService.Call(occp.ToContainerType(_sensorConfig.Crawler.Container), &result, "CrawlerService.Get", _sensorConfig.ConfigID)
+		if nil != err {
+			logging.Logger().Errorf("Cannot get data from crawler[%s] of container[%s] %v", _sensorConfig.Crawler.Name, _sensorConfig.Crawler.Container, err)
+			return
+		}
+		logging.Logger().Debugf("Data[%v] received from crawler[%s] of container[%s]", result, _sensorConfig.Crawler.Name, _sensorConfig.Crawler.Container)
+
+		//cs.CentralService.Send(oocmp.HTTPEntry_Data, "CrawlerService.Data", result)
+	}(sensorConfig)
+}

service/ContainerService.go

@@ -88,8 +88,8 @@ func (s *ContainerService) Send(containerType occp.ContainerType, method string,
 func (s *ContainerService) getClient(containerType occp.ContainerType) (*crc.Client, error) {
 	cs := s.checkContainer(containerType)
 	if nil == cs {
-		_cs, err := s.runContainer(containerType)
-		//_cs, err := s.debugContainer(containerType)
+		// _cs, err := s.runContainer(containerType)
+		_cs, err := s.debugContainer(containerType)
 		if nil != err {
 			return nil, err
 		}

service/CrawlerService.go

@@ -20,6 +20,9 @@ func init() {
 type CrawlerService struct {
 	ocsp.CrawlerService
 	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
+
+	SensorConfigService *SensorConfigService `annotation:"@Inject()"`
+	ContainerService    *ContainerService    `annotation:"@Inject()"`
 }
 
 func (s *CrawlerService) InitService() error {

service/SensorConfigService.go

@@ -10,6 +10,7 @@ import (
 
 	cda "git.loafle.net/commons/di-go/annotation"
 	cdr "git.loafle.net/commons/di-go/registry"
+	occp "git.loafle.net/overflow/commons-go/config/probe"
 	ocmsc "git.loafle.net/overflow/commons-go/model/sensorconfig"
 	ocsp "git.loafle.net/overflow/commons-go/service/probe"
 	"git.loafle.net/overflow/probe/config"
@@ -28,6 +29,8 @@ type SensorConfigService struct {
 	ocsp.SensorConfigService
 	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
 
+	ContainerService *ContainerService `annotation:"@Inject()"`
+
 	sensorConfigs map[string]*ocmsc.SensorConfig
 }
 
@@ -42,6 +45,15 @@ func (s *SensorConfigService) StartService() error {
 		return err
 	}
 
+	if nil != s.sensorConfigs || 0 < len(s.sensorConfigs) {
+		sortedMap := sortSensorConfigPerContainer(s.sensorConfigs)
+		for containerName, configs := range sortedMap {
+			if err := s.ContainerService.Send(occp.ToContainerType(containerName), "SensorConfigService.InitConfig", configs); nil != err {
+				return err
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -146,3 +158,18 @@ func (s *SensorConfigService) loadConfigFile(filePath string) (*ocmsc.SensorConf
 
 	return m, buf, nil
 }
+
+func sortSensorConfigPerContainer(sensorConfigMap map[string]*ocmsc.SensorConfig) map[string][]*ocmsc.SensorConfig {
+	if nil == sensorConfigMap || 0 == len(sensorConfigMap) {
+		return nil
+	}
+
+	m := make(map[string][]*ocmsc.SensorConfig)
+
+	for _, sensorConfig := range sensorConfigMap {
+		containerName := sensorConfig.Crawler.Container
+		m[containerName] = append(m[containerName], sensorConfig)
+	}
+
+	return m
+}

service/SensorService.go

@@ -0,0 +1,62 @@
+package service
+
+import (
+	"reflect"
+
+	cda "git.loafle.net/commons/di-go/annotation"
+	cdr "git.loafle.net/commons/di-go/registry"
+
+	// For annotation
+	_ "git.loafle.net/overflow/commons-go/core/annotation"
+)
+
+var SensorServiceType = reflect.TypeOf((*SensorService)(nil))
+
+func init() {
+	cdr.RegisterType(SensorServiceType)
+}
+
+type SensorService struct {
+	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
+
+	ContainerService *ContainerService `annotation:"@Inject()"`
+}
+
+func (s *SensorService) InitService() error {
+
+	return nil
+}
+
+func (s *SensorService) StartService() error {
+
+	return nil
+}
+
+func (s *SensorService) StopService() {
+
+}
+
+func (s *SensorService) DestroyService() {
+
+}
+
+func (s *SensorService) StartSensor(id int64) error {
+	return nil
+}
+
+func (s *SensorService) StopSensor(id int64) error {
+	return nil
+}
+
+func (s *SensorService) AddSensor(sensorConfigBase64 string) error {
+
+	return nil
+}
+
+func (s *SensorService) RemoveSensor(sensorConfigID string) error {
+	return nil
+}
+
+func (s *SensorService) UpdateSensor(id int64) error {
+	return nil
+}

service/service.go

@@ -6,11 +6,12 @@ var (
 	OrderedServices = []reflect.Type{
 		ProbeClientServiceType,
 		DataClientServiceType,
-		SensorConfigServiceType,
 		ContainerServiceType,
+		SensorConfigServiceType,
 		CrawlerServiceType,
 		CollectorServiceType,
 		DiscoveryServiceType,
+		SensorServiceType,
 	}
 )