134 lines
3.7 KiB
Go
134 lines
3.7 KiB
Go
package service
|
|
|
|
import (
|
|
"fmt"
|
|
"reflect"
|
|
"sync"
|
|
"time"
|
|
|
|
cda "git.loafle.net/commons/di-go/annotation"
|
|
cdr "git.loafle.net/commons/di-go/registry"
|
|
logging "git.loafle.net/commons/logging-go"
|
|
cuej "git.loafle.net/commons/util-go/encoding/json"
|
|
cuts "git.loafle.net/commons/util-go/time/scheduler"
|
|
cutss "git.loafle.net/commons/util-go/time/scheduler/storage"
|
|
occu "git.loafle.net/overflow/commons-go/core/util"
|
|
ocmd "git.loafle.net/overflow/commons-go/model/data"
|
|
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()"`
|
|
|
|
ProbeService *ProbeService `annotation:"@Inject()"`
|
|
CrawlerService *CrawlerService `annotation:"@Inject()"`
|
|
|
|
scheduler *cuts.Scheduler
|
|
scheduleMap sync.Map
|
|
}
|
|
|
|
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() {
|
|
|
|
}
|
|
|
|
func (s *CollectorService) AddSensorConfigs(sensorConfigs []*ocmsc.SensorConfig) error {
|
|
if nil == sensorConfigs || 0 == len(sensorConfigs) {
|
|
return nil
|
|
}
|
|
|
|
for _, sensorConfig := range sensorConfigs {
|
|
interval, err := cuej.NumberToInt(sensorConfig.Schedule.Interval)
|
|
if nil != err {
|
|
return fmt.Errorf("Cannot convert interval[%s] %v", sensorConfig.Schedule.Interval, err)
|
|
}
|
|
s.addSchedule(interval, sensorConfig)
|
|
logging.Logger().Debugf("scheduler of config[%s] has been added", sensorConfig.SensorID)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *CollectorService) RemoveSensorConfigs(sensorConfigs []*ocmsc.SensorConfig) error {
|
|
if nil == sensorConfigs || 0 == len(sensorConfigs) {
|
|
return nil
|
|
}
|
|
|
|
for _, sensorConfig := range sensorConfigs {
|
|
s.removeSchedule(sensorConfig.SensorID.String())
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *CollectorService) addSchedule(interval int, sensorConfig *ocmsc.SensorConfig) {
|
|
s.removeSchedule(sensorConfig.SensorID.String())
|
|
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.SensorID, err)
|
|
return
|
|
}
|
|
s.scheduleMap.Store(sensorConfig.SensorID, scheduleID)
|
|
}
|
|
|
|
func (s *CollectorService) removeSchedule(sensorID string) {
|
|
scheduleID, ok := s.scheduleMap.Load(sensorID)
|
|
if !ok {
|
|
return
|
|
}
|
|
s.scheduleMap.Delete(sensorID)
|
|
s.scheduler.Cancel(scheduleID.(string))
|
|
}
|
|
|
|
func (s *CollectorService) collectTask(sensorConfig *ocmsc.SensorConfig) {
|
|
logging.Logger().Debugf("CollectorService.collectTask for sensor config id[%s] of crawler[%s]", sensorConfig.SensorID, sensorConfig.Crawler.MetaCrawlerKey)
|
|
|
|
m := &ocmd.Metric{
|
|
SensorID: sensorConfig.SensorID.String(),
|
|
StartDate: occu.NowPtr(),
|
|
}
|
|
|
|
result, err := s.CrawlerService.Get(sensorConfig.SensorID.String())
|
|
if nil != err {
|
|
logging.Logger().Errorf("Cannot get data from crawler[%s] %v", sensorConfig.Crawler.MetaCrawlerKey, err)
|
|
return
|
|
}
|
|
|
|
m.Data = result
|
|
m.EndDate = occu.NowPtr()
|
|
m.Success = true
|
|
|
|
if err := s.ProbeService.Send("MetricService.Send", m); nil != err {
|
|
logging.Logger().Errorf("Cannot send data from config id[%s] of crawler[%s] %v", sensorConfig.SensorID, sensorConfig.Crawler.MetaCrawlerKey, err)
|
|
}
|
|
}
|