overflow_cloud/container-go
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-07-04 11:35:52 +09:00
parent f3e1641301
commit 519e96beea
4 changed files with 33 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
client/client.go
View File

@ -27,12 +27,13 @@ func NewConnector(containerType occp.ContainerType, portNumber int) (*csswc.Conn
connector.Name = containerType.String() connector.Name = containerType.String()
connector.ReconnectInterval = 5 connector.ReconnectInterval = 5
connector.ReconnectTryTime = 2 connector.ReconnectTryTime = 2
connector.MaxMessageSize = 4096 connector.MaxMessageSize = 8192
connector.ReadBufferSize = 4096 connector.ReadBufferSize = 8192
connector.WriteBufferSize = 4096 connector.WriteBufferSize = 8192
connector.PongTimeout = 60 connector.PongTimeout = 60
connector.PingTimeout = 10 connector.PingTimeout = 10
connector.PingPeriod = 9 connector.PingPeriod = 9
connector.CompressionThreshold = 4096
connector.RequestHeader = func() http.Header { connector.RequestHeader = func() http.Header {
header := make(map[string][]string) header := make(map[string][]string)

32
service/CollectorService.go
View File

@ -3,13 +3,13 @@ package service
import ( import (
"fmt" "fmt"
"reflect" "reflect"
"strconv"
"sync" "sync"
"time" "time"
cda "git.loafle.net/commons/di-go/annotation" cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry" cdr "git.loafle.net/commons/di-go/registry"
logging "git.loafle.net/commons/logging-go" 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" cuts "git.loafle.net/commons/util-go/time/scheduler"
cutss "git.loafle.net/commons/util-go/time/scheduler/storage" cutss "git.loafle.net/commons/util-go/time/scheduler/storage"
occu "git.loafle.net/overflow/commons-go/core/util" occu "git.loafle.net/overflow/commons-go/core/util"
@ -67,12 +67,12 @@ func (s *CollectorService) AddSensorConfigs(sensorConfigs []*ocmsc.SensorConfig)
} }
for _, sensorConfig := range sensorConfigs { for _, sensorConfig := range sensorConfigs {
interval, err := strconv.ParseInt(sensorConfig.Schedule.Interval, 10, 64) interval, err := cuej.NumberToInt(sensorConfig.Schedule.Interval)
if nil != err { if nil != err {
return fmt.Errorf("Cannot convert interval[%s] %v", sensorConfig.Schedule.Interval, err) return fmt.Errorf("Cannot convert interval[%s] %v", sensorConfig.Schedule.Interval, err)
} }
s.addSchedule(interval, sensorConfig) s.addSchedule(interval, sensorConfig)
logging.Logger().Debugf("scheduler of config[%s] has been added", sensorConfig.ConfigID) logging.Logger().Debugf("scheduler of config[%s] has been added", sensorConfig.SensorID)
} }
return nil return nil
@ -84,47 +84,47 @@ func (s *CollectorService) RemoveSensorConfigs(sensorConfigs []*ocmsc.SensorConf
} }
for _, sensorConfig := range sensorConfigs { for _, sensorConfig := range sensorConfigs {
s.removeSchedule(sensorConfig.ConfigID) s.removeSchedule(sensorConfig.SensorID.String())
} }
return nil return nil
} }
func (s *CollectorService) addSchedule(interval int64, sensorConfig *ocmsc.SensorConfig) { func (s *CollectorService) addSchedule(interval int, sensorConfig *ocmsc.SensorConfig) {
s.removeSchedule(sensorConfig.ConfigID) s.removeSchedule(sensorConfig.SensorID.String())
scheduleID, err := s.scheduler.RunEvery(time.Duration(interval)*time.Second, s.collectTask, sensorConfig) scheduleID, err := s.scheduler.RunEvery(time.Duration(interval)*time.Second, s.collectTask, sensorConfig)
if nil != err { if nil != err {
logging.Logger().Errorf("Cannot add schedule for config[%s] %v", sensorConfig.ConfigID, err) logging.Logger().Errorf("Cannot add schedule for config[%s] %v", sensorConfig.SensorID, err)
return return
} }
s.scheduleMap.Store(sensorConfig.ConfigID, scheduleID) s.scheduleMap.Store(sensorConfig.SensorID, scheduleID)
} }
func (s *CollectorService) removeSchedule(configID string) { func (s *CollectorService) removeSchedule(sensorID string) {
scheduleID, ok := s.scheduleMap.Load(configID) scheduleID, ok := s.scheduleMap.Load(sensorID)
if !ok { if !ok {
return return
} }
s.scheduleMap.Delete(configID) s.scheduleMap.Delete(sensorID)
s.scheduler.Cancel(scheduleID.(string)) s.scheduler.Cancel(scheduleID.(string))
} }
func (s *CollectorService) collectTask(sensorConfig *ocmsc.SensorConfig) { 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) logging.Logger().Debugf("CollectorService.collectTask for sensor config id[%s] of crawler[%s]", sensorConfig.SensorID, sensorConfig.Crawler.MetaCrawlerKey)
result, err := s.CrawlerService.Get(sensorConfig.ConfigID) result, err := s.CrawlerService.Get(sensorConfig.SensorID.String())
if nil != err { if nil != err {
logging.Logger().Errorf("Cannot get data from crawler[%s] %v", sensorConfig.Crawler.Name, err) logging.Logger().Errorf("Cannot get data from crawler[%s] %v", sensorConfig.Crawler.MetaCrawlerKey, err)
return return
} }
m := &ocmd.Metric{ m := &ocmd.Metric{
SensorConfigID: sensorConfig.ConfigID, SensorConfigID: sensorConfig.SensorID.String(),
Data: result, Data: result,
CollectDate: occu.NowPtr(), CollectDate: occu.NowPtr(),
} }
if err := s.ProbeService.Send("MetricService.Send", m); nil != err { 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.ConfigID, sensorConfig.Crawler.Name, err) logging.Logger().Errorf("Cannot send data from config id[%s] of crawler[%s] %v", sensorConfig.SensorID, sensorConfig.Crawler.MetaCrawlerKey, err)
} }
} }

2
service/CrawlerService.go
View File

@ -69,7 +69,7 @@ func (s *CrawlerService) Get(sensorConfigID string) (map[string]string, error) {
if !ok { if !ok {
return nil, fmt.Errorf("There is no sensor config for id[%s]", sensorConfigID) return nil, fmt.Errorf("There is no sensor config for id[%s]", sensorConfigID)
} }
_crawlerName := sensorConfig.Crawler.Name _crawlerName := sensorConfig.Crawler.MetaCrawlerKey
_crawler, ok := s.Crawlers[_crawlerName] _crawler, ok := s.Crawlers[_crawlerName]
if !ok { if !ok {
return nil, fmt.Errorf("There is no crawler[%s] for id[%s]", _crawlerName, sensorConfigID) return nil, fmt.Errorf("There is no crawler[%s] for id[%s]", _crawlerName, sensorConfigID)

24
service/SensorConfigService.go
View File

@ -52,7 +52,7 @@ func (s *SensorConfigService) InitConfig(sensorConfigs []*ocmsc.SensorConfig) er
} }
for _, sensorConfig := range sensorConfigs { for _, sensorConfig := range sensorConfigs {
s.sensorConfigs[sensorConfig.ConfigID] = sensorConfig s.sensorConfigs[sensorConfig.SensorID.String()] = sensorConfig
} }
if err := s.CollectorService.AddSensorConfigs(sensorConfigs); nil != err { if err := s.CollectorService.AddSensorConfigs(sensorConfigs); nil != err {
@ -69,17 +69,17 @@ func (s *SensorConfigService) AddConfig(sensorConfig *ocmsc.SensorConfig) error
return fmt.Errorf("Sensor config is not valid") return fmt.Errorf("Sensor config is not valid")
} }
configID := sensorConfig.ConfigID sensorID := sensorConfig.SensorID.String()
if _, ok := s.sensorConfigs[configID]; ok { if _, ok := s.sensorConfigs[sensorID]; ok {
return fmt.Errorf("Sensor config[%s] is exist already", configID) return fmt.Errorf("Sensor config[%s] is exist already", sensorID)
} }
s.sensorConfigs[configID] = sensorConfig s.sensorConfigs[sensorID] = sensorConfig
if err := s.CollectorService.AddSensorConfigs([]*ocmsc.SensorConfig{sensorConfig}); nil != err { if err := s.CollectorService.AddSensorConfigs([]*ocmsc.SensorConfig{sensorConfig}); nil != err {
return err return err
} }
logging.Logger().Debugf("Sensor config[%s] was added", configID) logging.Logger().Debugf("Sensor config[%s] was added", sensorID)
return nil return nil
} }
@ -89,12 +89,12 @@ func (s *SensorConfigService) UpdateConfig(sensorConfig *ocmsc.SensorConfig) err
return fmt.Errorf("Sensor config is not valid") return fmt.Errorf("Sensor config is not valid")
} }
configID := sensorConfig.ConfigID sensorID := sensorConfig.SensorID.String()
if _, ok := s.sensorConfigs[configID]; !ok { if _, ok := s.sensorConfigs[sensorID]; !ok {
return fmt.Errorf("Sensor config[%s] is not exist", configID) return fmt.Errorf("Sensor config[%s] is not exist", sensorID)
} }
delete(s.sensorConfigs, configID) delete(s.sensorConfigs, sensorID)
s.sensorConfigs[configID] = sensorConfig s.sensorConfigs[sensorID] = sensorConfig
if err := s.CollectorService.RemoveSensorConfigs([]*ocmsc.SensorConfig{sensorConfig}); nil != err { if err := s.CollectorService.RemoveSensorConfigs([]*ocmsc.SensorConfig{sensorConfig}); nil != err {
return err return err
@ -104,7 +104,7 @@ func (s *SensorConfigService) UpdateConfig(sensorConfig *ocmsc.SensorConfig) err
return err return err
} }
logging.Logger().Debugf("Sensor config[%s] was updated", configID) logging.Logger().Debugf("Sensor config[%s] was updated", sensorID)
return nil return nil
} }