overflow_cloud/probe
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader
2018-05-03 20:24:07 +09:00
parent c2dd51b071
commit ca823a2e69
36 changed files with 1176 additions and 599 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
service/CollectorService.go
View File

@@ -8,10 +8,8 @@ import (
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"
@@ -45,7 +43,7 @@ func (s *CollectorService) InitService() error {
func (s *CollectorService) StartService() error {
if err := s.scheduler.Start(); nil != err {
return err
return fmt.Errorf("CollectorService: StartService failed %v", err)
}
if err := s.addScheduleAll(); nil != err {
@@ -84,16 +82,16 @@ func (s *CollectorService) addSchedule(interval int64, sensorConfig *ocmsc.Senso
}
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)
// 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)
// //cs.CentralService.Send(oocmp.HTTPEntry_Data, "CrawlerService.Data", result)
// }(sensorConfig)
}

365
service/ContainerService.go
View File

@@ -1,24 +1,16 @@
package service
import (
"context"
"fmt"
"io/ioutil"
"os"
"os/exec"
"reflect"
"strconv"
"time"
"sync"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
"git.loafle.net/commons/logging-go"
crc "git.loafle.net/commons/rpc-go/client"
csc "git.loafle.net/commons/server-go/client"
logging "git.loafle.net/commons/logging-go"
crp "git.loafle.net/commons/rpc-go/protocol"
occp "git.loafle.net/overflow/commons-go/config/probe"
ocsp "git.loafle.net/overflow/commons-go/service/probe"
"git.loafle.net/overflow/probe/client/container"
"git.loafle.net/overflow/probe/config"
"git.loafle.net/overflow/probe/container"
// For annotation
_ "git.loafle.net/overflow/commons-go/core/annotation"
@@ -34,320 +26,107 @@ type ContainerService struct {
ocsp.ContainerService
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
DiscoveryService *DiscoveryService `annotation:"@Inject()"`
DiscoveryService *DiscoveryService `annotation:"@Inject()"`
SensorConfigService *SensorConfigService `annotation:"@Inject()"`
rpcServiceMap map[occp.ContainerType][]interface{}
containerStates map[occp.ContainerType]*containerState
connectorMap map[csc.Connector]*containerState
OnConnectChan <-chan *container.OnConnectInfo `annotation:"@Resource(name='PROBE_ONCONNECT_CHAN')"`
OnDisconnectChan <-chan occp.ContainerType `annotation:"@Resource(name='PROBE_ONDISCONNECT_CHAN')"`
RPCServerCodec crp.ServerCodec `annotation:"@Resource(name='PROBE_RPC_SERVER_CODEC')"`
containerSessions sync.Map
stopChan chan struct{}
stopWg sync.WaitGroup
}
func (s *ContainerService) InitService() error {
s.containerStates = make(map[occp.ContainerType]*containerState)
s.rpcServiceMap = make(map[occp.ContainerType][]interface{})
s.connectorMap = make(map[csc.Connector]*containerState)
return nil
}
func (s *ContainerService) StartService() error {
s.rpcServiceMap[occp.ContainerDiscovery] = []interface{}{
s.DiscoveryService,
}
s.rpcServiceMap[occp.ContainerNetwork] = []interface{}{}
s.rpcServiceMap[occp.ContainerGenernal] = []interface{}{}
s.stopChan = make(chan struct{})
s.stopWg.Add(1)
go s.handleCotainerConnection()
return nil
}
func (s *ContainerService) StopService() {
for containerType := range s.containerStates {
s.removeContainerState(containerType)
}
close(s.stopChan)
s.stopWg.Wait()
s.containerSessions.Range(func(_containerType, _containerSession interface{}) bool {
containerSession := _containerSession.(*container.ContainerSession)
containerSession.Stop()
return true
})
}
func (s *ContainerService) DestroyService() {
}
func (s *ContainerService) Call(containerType occp.ContainerType, result interface{}, method string, params ...interface{}) error {
client, err := s.getClient(containerType)
if nil != err {
return err
func (s *ContainerService) handleCotainerConnection() {
for {
select {
case onConnectInfo := <-s.OnConnectChan:
_containerSession, ok := s.containerSessions.Load(onConnectInfo.ContainerType)
if !ok {
// var err error
// _containerSession, err = s.newContainerSession(onConnectInfo.ContainerType)
// if nil != err {
// logging.Logger().Errorf("ContainerSession is not exist for %s", onConnectInfo.ContainerType.String())
// }
logging.Logger().Errorf("ContainerSession is not exist for %s", onConnectInfo.ContainerType.String())
break
}
containerSession := _containerSession.(*container.ContainerSession)
containerSession.Connected(onConnectInfo.WriteChan)
s.SensorConfigService.SendInitConfig(onConnectInfo.ContainerType)
case containerType := <-s.OnDisconnectChan:
_containerSession, ok := s.containerSessions.Load(containerType)
if ok {
containerSession := _containerSession.(*container.ContainerSession)
containerSession.Disconnected()
}
case <-s.stopChan:
return
}
}
return client.Call(result, method, params...)
}
func (s *ContainerService) Send(containerType occp.ContainerType, method string, params ...interface{}) error {
client, err := s.getClient(containerType)
if nil != err {
return err
}
return client.Send(method, params...)
}
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)
var err error
_containerSession, ok := s.containerSessions.Load(containerType)
if !ok {
_containerSession, err = s.newContainerSession(containerType)
if nil != err {
return nil, err
}
cs = _cs
s.containerStates[containerType] = cs
}
if nil == cs.client {
client, err := container.NewClient(containerType, cs.port, s.rpcServiceMap[containerType])
if nil != err {
s.removeContainerState(containerType)
return nil, err
}
if err := client.Start(); nil != err {
s.removeContainerState(containerType)
return nil, err
}
cs.client = client
cs.client.Connector.SetOnDisconnected(s.onDisconnected)
s.connectorMap[cs.client.Connector] = cs
}
return cs.client, nil
}
func (s *ContainerService) onDisconnected(connector csc.Connector) {
cs, ok := s.connectorMap[connector]
if !ok || nil == cs {
return
}
s.refreshContainer(cs.containerType)
}
func (s *ContainerService) runContainer(containerType occp.ContainerType) (*containerState, error) {
if cs := s.checkContainer(containerType); nil != cs {
return cs, nil
}
cmd, pidFilePath := cotainerCommand(containerType)
removePidFile(pidFilePath)
if err := cmd.Start(); nil != err {
logging.Logger().Errorf("to run Container[%s] failed err %v", containerType.String(), err)
return nil, err
}
port, err := watchPidFileCreate(pidFilePath, time.Duration(time.Second*2))
if nil != err {
return nil, err
}
// go func(containerType occp.ContainerType, cmd *exec.Cmd) {
// if err := cmd.Wait(); nil != err {
// logging.Logger().Error(err)
// }
// logging.Logger().Infof("Container[%s] has been stopped", containerType.String())
// s.refreshContainer(containerType)
// }(containerType, cmd)
cs := &containerState{
containerType: containerType,
cmd: cmd,
port: port,
}
return cs, nil
}
func (s *ContainerService) refreshContainer(containerType occp.ContainerType) {
cs := s.checkContainer(containerType)
if nil == cs {
if _, err := s.getClient(containerType); nil != err {
logging.Logger().Error(err)
}
return
}
cs.cmd.Process.Kill()
cs.cmd = nil
delete(s.connectorMap, cs.client.Connector)
err := cs.client.Stop(context.Background())
if nil != err {
logging.Logger().Error(err)
}
cs.client = nil
delete(s.containerStates, containerType)
if _, err := s.getClient(containerType); nil != err {
logging.Logger().Error(err)
}
//
//if cs, ok := s.checkContainer(containerType); nil == cs {
// cs, ok := s.containerStates[containerType]
// if !ok {
// return
// }
// delete(s.connectorMap, cs.client.Connector)
//
// logging.Logger().Debugf("Client[%s]11 has been disconnected", cs.containerType.String())
// err := cs.client.Stop(context.Background())
// if nil != err {
// logging.Logger().Error(err)
// }
//
// client, err := container.NewClient(cs.containerType, cs.port, s.rpcServiceMap[cs.containerType])
// if nil != err {
// s.removeContainerState(cs.containerType)
// logging.Logger().Error(err)
// return
// }
// if err := client.Start(); nil != err {
// s.removeContainerState(cs.containerType)
// logging.Logger().Error(err)
// return
// }
// cs.client = client
// cs.client.Connector.SetOnDisconnected(s.onDisconnected)
// s.connectorMap[cs.client.Connector] = cs
//} else {
// cs, ok := s.containerStates[containerType]
// if !ok {
// return
// }
// delete(s.connectorMap, cs.client.Connector)
// err := cs.client.Stop(context.Background())
// if nil != err {
// logging.Logger().Error(err)
// }
// delete(s.containerStates, containerType)
// _, err = s.getClient(containerType)
// if nil != err {
// logging.Logger().Error(err)
// }
//}
}
func (s *ContainerService) debugContainer(containerType occp.ContainerType) (*containerState, error) {
cs := &containerState{
containerType: containerType,
cmd: nil,
port: 60000,
}
return cs, nil
}
func (s *ContainerService) checkContainer(containerType occp.ContainerType) *containerState {
cs, ok := s.containerStates[containerType]
if !ok || nil == cs {
return nil
}
// p, err := os.FindProcess(cs.cmd.Process.Pid)
// if nil != err {
// s.removeContainerState(containerType)
// return nil
// }
// if nil == p {
// s.removeContainerState(containerType)
// return nil
// }
// if nil != cs.cmd.ProcessState && cs.cmd.ProcessState.Exited() {
// s.removeContainerState(containerType)
// return nil
// }
return cs
}
func (s *ContainerService) killContainer(containerType occp.ContainerType) error {
cs, ok := s.containerStates[containerType]
if !ok || nil == cs {
return nil
}
if nil == cs.cmd.ProcessState || !cs.cmd.ProcessState.Exited() {
if err := cs.cmd.Process.Kill(); nil != err {
return err
}
}
return nil
}
containerSession := _containerSession.(*container.ContainerSession)
func (s *ContainerService) removeContainerState(containerType occp.ContainerType) {
cs, ok := s.containerStates[containerType]
if !ok || nil == cs {
return
buff, err := s.RPCServerCodec.NewNotification(method, params)
if nil != err {
return err
}
if nil != cs.client {
delete(s.connectorMap, cs.client.Connector)
cs.client.Stop(context.Background())
}
s.killContainer(containerType)
delete(s.containerStates, containerType)
return containerSession.Send(buff)
}
func cotainerCommand(containerType occp.ContainerType) (cmd *exec.Cmd, pidFilePath string) {
pidFilePath = config.ContainerPIDFilePath(containerType)
loggingConfigFilePath := config.ContainerLoggingConfigFilePath(containerType)
binFilePath := config.ContainerBinFilePath(containerType)
switch containerType {
case occp.ContainerDiscovery, occp.ContainerNetwork:
args := []string{
fmt.Sprintf("-%s=%s", occp.FlagPidFilePathName, pidFilePath),
fmt.Sprintf("-%s=%s", occp.FlagLoggingConfigFilePathName, loggingConfigFilePath),
}
cmd = exec.Command(binFilePath, args...)
case occp.ContainerGenernal:
args := []string{
"-jar",
binFilePath,
pidFilePath,
loggingConfigFilePath,
}
cmd = exec.Command(config.JavaBinPath(), args...)
}
return
}
func removePidFile(pidFilePath string) {
if _, err := os.Stat(pidFilePath); err == nil {
if err := os.Remove(pidFilePath); nil != err {
logging.Logger().Errorf("removing pid file has been failed [%v]", err)
func (s *ContainerService) newContainerSession(containerType occp.ContainerType) (*container.ContainerSession, error) {
_containerSession, ok := s.containerSessions.Load(containerType)
if !ok {
_containerSession = container.NewContainerSession(containerType)
s.containerSessions.Store(containerType, _containerSession)
if err := _containerSession.(*container.ContainerSession).Start(); nil != err {
return nil, err
}
}
}
func watchPidFileCreate(pidFilePath string, waitTime time.Duration) (int, error) {
startTime := time.Now()
for {
if _, err := os.Stat(pidFilePath); err == nil {
buf, err := ioutil.ReadFile(pidFilePath)
if nil != err {
return 0, err
}
portNumber, err := strconv.ParseInt(string(buf), 10, 32)
if nil != err {
return 0, err
}
return int(portNumber), nil
}
if time.Since(startTime) > waitTime {
return 0, fmt.Errorf("pid file not exist")
}
time.Sleep(time.Duration(time.Millisecond * 100))
}
}
type containerState struct {
containerType occp.ContainerType
cmd *exec.Cmd
port int
client *crc.Client
return _containerSession.(*container.ContainerSession), nil
}

13
service/DataClientService.go
View File

@@ -1,12 +1,13 @@
package service
import (
"fmt"
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
ocsp "git.loafle.net/overflow/commons-go/service/probe"
"git.loafle.net/overflow/probe/client/central"
"git.loafle.net/overflow/probe/client/data"
// For annotation
_ "git.loafle.net/overflow/commons-go/core/annotation"
@@ -27,7 +28,7 @@ type DataClientService struct {
func (s *DataClientService) InitService() error {
if err := s.RPCClientService.InitService(); nil != err {
return err
return fmt.Errorf("DataClientService: InitService failed %v", err)
}
return nil
@@ -35,17 +36,17 @@ func (s *DataClientService) InitService() error {
func (s *DataClientService) StartService() error {
if err := s.RPCClientService.StartService(); nil != err {
return err
return fmt.Errorf("DataClientService: StartService failed %v", err)
}
client, err := central.NewData()
client, err := data.New()
if nil != err {
return err
return fmt.Errorf("ProbeClientService: StartService failed %v", err)
}
s.client = client
if err := s.client.Start(); nil != err {
return err
return fmt.Errorf("DataClientService: StartService failed %v", err)
}
return nil

48
service/DataService.go Normal file
View File

@@ -0,0 +1,48 @@
package service
import (
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
logging "git.loafle.net/commons/logging-go"
// For annotation
_ "git.loafle.net/overflow/commons-go/core/annotation"
)
var DataServiceType = reflect.TypeOf((*DataService)(nil))
func init() {
cdr.RegisterType(DataServiceType)
}
type DataService struct {
cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
DataClientService *DataClientService `annotation:"@Inject()"`
}
func (s *DataService) InitService() error {
return nil
}
func (s *DataService) StartService() error {
return nil
}
func (s *DataService) StopService() {
}
func (s *DataService) DestroyService() {
}
func (s *DataService) Metric(metric map[string]string) error {
logging.Logger().Debugf("Metric: %v", metric)
return nil
}

13
service/ProbeClientService.go
View File

@@ -1,6 +1,7 @@
package service
import (
"fmt"
"reflect"
cda "git.loafle.net/commons/di-go/annotation"
@@ -8,7 +9,7 @@ import (
// For annotation
_ "git.loafle.net/overflow/commons-go/core/annotation"
"git.loafle.net/overflow/probe/client/central"
"git.loafle.net/overflow/probe/client/probe"
)
var ProbeClientServiceType = reflect.TypeOf((*ProbeClientService)(nil))
@@ -29,7 +30,7 @@ type ProbeClientService struct {
func (s *ProbeClientService) InitService() error {
if err := s.RPCClientService.InitService(); nil != err {
return err
return fmt.Errorf("ProbeClientService: InitService failed %v", err)
}
return nil
@@ -37,17 +38,17 @@ func (s *ProbeClientService) InitService() error {
func (s *ProbeClientService) StartService() error {
if err := s.RPCClientService.StartService(); nil != err {
return err
return fmt.Errorf("ProbeClientService: StartService failed %v", err)
}
client, err := central.NewProbe(s.HandleEncryptionKey, s.DiscoveryService)
client, err := probe.New(s.HandleEncryptionKey, s.DiscoveryService)
if nil != err {
return err
return fmt.Errorf("ProbeClientService: StartService failed %v", err)
}
s.client = client
if err := s.client.Start(); nil != err {
return err
return fmt.Errorf("ProbeClientService: StartService failed %v", err)
}
return nil

82
service/SensorConfigService.go
View File

@@ -10,6 +10,7 @@ import (
cda "git.loafle.net/commons/di-go/annotation"
cdr "git.loafle.net/commons/di-go/registry"
logging "git.loafle.net/commons/logging-go"
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"
@@ -31,7 +32,8 @@ type SensorConfigService struct {
ContainerService *ContainerService `annotation:"@Inject()"`
sensorConfigs map[string]*ocmsc.SensorConfig
sensorConfigs map[string]*ocmsc.SensorConfig
sensorConfigsPerContainer map[occp.ContainerType][]*ocmsc.SensorConfig
}
func (s *SensorConfigService) InitService() error {
@@ -42,15 +44,14 @@ func (s *SensorConfigService) InitService() error {
func (s *SensorConfigService) StartService() error {
if err := s.loadConfigAll(); nil != err {
return err
return fmt.Errorf("SensorConfigService: StartService failed %v", 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
}
s.sortSensorConfigPerContainer()
for containerType := range s.sensorConfigsPerContainer {
s.SendInitConfig(containerType)
}
}
@@ -65,6 +66,19 @@ func (s *SensorConfigService) DestroyService() {
}
func (s *SensorConfigService) SendInitConfig(containerType occp.ContainerType) error {
configs, ok := s.sensorConfigsPerContainer[containerType]
if !ok {
return nil
}
if err := s.ContainerService.Send(containerType, "SensorConfigService.InitConfig", configs); nil != err {
return fmt.Errorf("SensorConfigService: StartService failed %v", err)
}
return nil
}
func (s *SensorConfigService) AddConfig(tempFilePath string) error {
sc, buf, err := s.loadConfigFile(tempFilePath)
if nil != err {
@@ -81,6 +95,37 @@ func (s *SensorConfigService) AddConfig(tempFilePath string) error {
}
s.sensorConfigs[sc.ID.String()] = sc
s.sortSensorConfigPerContainer()
if err := s.ContainerService.Send(occp.ToContainerType(sc.Crawler.Container), "SensorConfigService.AddConfig", sc); nil != err {
return err
}
return nil
}
func (s *SensorConfigService) UpdateConfig(tempFilePath string) error {
sc, buf, err := s.loadConfigFile(tempFilePath)
if nil != err {
return err
}
targetPath := config.SensorConfigFilePath(sc)
ioutil.WriteFile(targetPath, buf, 0644)
// tempfile remove
err = os.Remove(tempFilePath)
if nil != err {
return err
}
delete(s.sensorConfigs, sc.ConfigID)
s.sensorConfigs[sc.ID.String()] = sc
s.sortSensorConfigPerContainer()
if err := s.ContainerService.Send(occp.ToContainerType(sc.Crawler.Container), "SensorConfigService.UpdateConfig", sc); nil != err {
return err
}
return nil
}
@@ -98,6 +143,11 @@ func (s *SensorConfigService) RemoveConfig(sensorConfigID string) error {
}
delete(s.sensorConfigs, sensorConfigID)
s.sortSensorConfigPerContainer()
if err := s.ContainerService.Send(occp.ToContainerType(sc.Crawler.Container), "SensorConfigService.RemoveConfig", sensorConfigID); nil != err {
return err
}
return nil
}
@@ -146,6 +196,7 @@ func (s *SensorConfigService) loadConfigDir(dirPath string) error {
}
func (s *SensorConfigService) loadConfigFile(filePath string) (*ocmsc.SensorConfig, []byte, error) {
logging.Logger().Debugf("filePath: %s", filePath)
buf, err := ioutil.ReadFile(filePath)
if nil != err {
return nil, nil, err
@@ -159,17 +210,16 @@ 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
func (s *SensorConfigService) sortSensorConfigPerContainer() {
if nil == s.sensorConfigs || 0 == len(s.sensorConfigs) {
return
}
m := make(map[string][]*ocmsc.SensorConfig)
s.sensorConfigsPerContainer = nil
s.sensorConfigsPerContainer = make(map[occp.ContainerType][]*ocmsc.SensorConfig)
for _, sensorConfig := range sensorConfigMap {
containerName := sensorConfig.Crawler.Container
m[containerName] = append(m[containerName], sensorConfig)
for _, sensorConfig := range s.sensorConfigs {
containerType := occp.ToContainerType(sensorConfig.Crawler.Container)
s.sensorConfigsPerContainer[containerType] = append(s.sensorConfigsPerContainer[containerType], sensorConfig)
}
return m
}