package plugin import ( "agent/cmd/agent/global" "fmt" "linkfog.com/public/lib/common" "strings" "sync" "time" "agent/cmd/agent/option" "linkfog.com/pluginx/pluginmgr" "linkfog.com/pluginx/pluginrpc" "linkfog.com/public/lib/l" ) var defaultPluginMgrConf string = `{ "doberman": { "name": "doberman", "path": "/dosec/plugin/doberman", "enable": true, "mem": 629145600, "md5": "uninitialized" } }` var defaultPluginCliConf = map[string]string{ DobermanPlugin: DobermanPluginSocket, } type Plugin struct { signal chan string signalSize int isRunning bool plgServer *pluginServer plgConf map[string]*pluginmgr.PluginProcessConf processMgr *pluginmgr.PluginProcessMgr grpcServer *pluginrpc.PluginGrpcServer grpcClient *pluginrpc.PluginGrpcClient sync.Mutex } type PluginOpt func(*Plugin) func New(opts ...PluginOpt) *Plugin { p := Plugin{ signalSize: 10, } for _, opt := range opts { opt(&p) } p.signal = make(chan string, p.signalSize) return &p } func (p *Plugin) Start() error { if p.IsRunning() { return nil } l.Info("init plugin module") //// 启动自身grpc服务 //p.plgServer = newPluginServer() //var err error //p.grpcServer, err = pluginrpc.NewPluginGrpcServer(HadesPluginSocket, p.plgServer) //if err != nil { // return fmt.Errorf("NewPluginGrpcServer err: %v", err) //} // 启动插件管理器,运行各插件进程 pluginmgr.EnableValidatePluginMD5 = option.Opt.EnableValidatePluginMD5 pluginmgr.InitDur = option.Opt.PluginStatusCheckDur plgCfgMap, err := pluginmgr.LoadPluginConfigWithData([]byte(defaultPluginMgrConf)) if err != nil { return fmt.Errorf("LoadPluginConfigWithData err: %v", err) } for name, plg := range plgCfgMap { l.Info("plugin config:", name, plg) } p.plgConf = plgCfgMap p.processMgr = pluginmgr.NewProcessMgr(p.plgConf) p.processMgr.Start() // 建立与各插件进程的grpc连接 p.grpcClient, err = pluginrpc.NewPluginGrpcClient(defaultPluginCliConf) if err != nil { return fmt.Errorf("NewGrpcClientHelper err: %v", err) } // 等待所有插件运行正常 allIsRunning := false for i := 0; i < option.Opt.PluginStatusCheckTimes; i++ { if p.processMgr.AllPluginProcessIsRunning() { allIsRunning = true break } l.Info("all plugin process are init ...") time.Sleep(option.Opt.PluginStatusCheckDur) } if !allIsRunning { return fmt.Errorf("plugin processes init failed") } go func() { var sig string var ok bool for { select { case sig, ok = <-p.signal: if !ok { p.Stop() l.Info("module resources exit") return } else { if len(sig) < 120 { l.Info("module resources receive signal:", sig) } else { l.Info("module resources receive signal:", sig[:120]) } p.dealWithSig(sig) } } } }() p.isRunning = true l.Info("init plugin module success") return nil } func (p *Plugin) Name() string { return global.PluginModuleName } func (p *Plugin) IsRunning() bool { return p.isRunning } func (p *Plugin) Stop() { if p.IsRunning() { if p.processMgr != nil { p.processMgr.Stop() } if p.grpcClient != nil { p.grpcClient.Close() } if p.grpcServer != nil { p.grpcServer.Close() } close(p.signal) p.isRunning = false } } func (p *Plugin) Receive(msg string) error { if len(p.signal) > (p.signalSize - 2) { return l.WrapError("signal chan is full, drop msg:", msg) } p.signal <- msg return nil } func (p *Plugin) dealWithSig(msgStr string) { splits := strings.SplitN(msgStr, ":", 3) if len(splits) != 3 { l.Warn("bad msg fmt from server", msgStr) return } msgType := splits[1] defer common.TimeCost("deal sig finished, msgType:" + msgType)() l.Info("ignore sig:", msgStr) }