container-go/service/CollectorService.go

123 lines
3.3 KiB
Go
Raw Normal View History

2018-05-02 10:59:45 +00:00
package service
import (
"fmt"
"reflect"
"strconv"
2018-05-03 11:00:07 +00:00
"sync"
2018-05-02 10:59:45 +00:00
"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"
ocmsc "git.loafle.net/overflow/commons-go/model/sensorconfig"
ocsp "git.loafle.net/overflow/commons-go/service/probe"
// For annotation
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var CollectorServiceType = reflect.TypeOf((*CollectorService)(nil))
func init() {
cdr.RegisterType(CollectorServiceType)
}
type CollectorService struct {
ocsp.CollectorService
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
2018-05-03 11:00:07 +00:00
ProbeService *ProbeService `annotation:"@Inject()"`
CrawlerService *CrawlerService `annotation:"@Inject()"`
2018-05-02 10:59:45 +00:00
2018-05-03 11:00:07 +00:00
scheduler *cuts.Scheduler
scheduleMap sync.Map
2018-05-02 10:59:45 +00:00
}
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
}
return nil
}
func (s *CollectorService) StopService() {
s.scheduler.Stop()
}
func (s *CollectorService) DestroyService() {
}
2018-05-03 11:00:07 +00:00
func (s *CollectorService) AddSensorConfigs(sensorConfigs []*ocmsc.SensorConfig) error {
2018-05-02 10:59:45 +00:00
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)
2018-05-03 10:43:37 +00:00
logging.Logger().Debugf("scheduler of config[%s] has been added", sensorConfig.ConfigID)
2018-05-02 10:59:45 +00:00
}
return nil
}
2018-05-03 11:00:07 +00:00
func (s *CollectorService) RemoveSensorConfigs(sensorConfigs []*ocmsc.SensorConfig) error {
if nil == sensorConfigs || 0 == len(sensorConfigs) {
return nil
}
for _, sensorConfig := range sensorConfigs {
2018-05-04 11:44:16 +00:00
s.removeSchedule(sensorConfig.ConfigID)
2018-05-03 11:00:07 +00:00
}
return nil
}
2018-05-02 10:59:45 +00:00
func (s *CollectorService) addSchedule(interval int64, sensorConfig *ocmsc.SensorConfig) {
2018-05-04 11:57:16 +00:00
s.removeSchedule(sensorConfig.ConfigID)
2018-05-03 11:00:07 +00:00
scheduleID, err := s.scheduler.RunEvery(time.Duration(interval)*time.Second, s.collectTask, sensorConfig)
if nil != err {
logging.Logger().Errorf("Cannot add schedule for config[%s] %v", sensorConfig.ConfigID, err)
return
}
s.scheduleMap.Store(sensorConfig.ConfigID, scheduleID)
2018-05-02 10:59:45 +00:00
}
2018-05-04 11:44:16 +00:00
func (s *CollectorService) removeSchedule(configID string) {
scheduleID, ok := s.scheduleMap.Load(configID)
if !ok {
return
}
s.scheduleMap.Delete(configID)
s.scheduler.Cancel(scheduleID.(string))
}
2018-05-02 10:59:45 +00:00
func (s *CollectorService) collectTask(sensorConfig *ocmsc.SensorConfig) {
logging.Logger().Debugf("CollectorService.collectTask for sensor config id[%s] of crawler[%s]", sensorConfig.ConfigID, sensorConfig.Crawler.Name)
result, err := s.CrawlerService.Get(sensorConfig.ConfigID)
if nil != err {
logging.Logger().Errorf("Cannot get data from crawler[%s] %v", sensorConfig.Crawler.Name, err)
return
}
2018-05-08 11:04:30 +00:00
if err := s.ProbeService.Send("DataService.Metric", sensorConfig.ConfigID, result); nil != err {
2018-05-02 10:59:45 +00:00
logging.Logger().Errorf("Cannot send data from config id[%s] of crawler[%s] %v", sensorConfig.ConfigID, sensorConfig.Crawler.Name, err)
}
}