Commit c9ec1358 authored by Lei Li's avatar Lei Li
Browse files

feat: 插件管理器支持增量处理

parent 175d0ee6
......@@ -4,7 +4,7 @@ go 1.23.2
replace linkfog.com/public => web.lueluesay.top/git/lil/public v0.0.0-20241015034013-544ad0ec734d
replace linkfog.com/pluginx => web.lueluesay.top/git/lil/pluginx v0.0.0-20241014064823-a6e286ccb5cd
replace linkfog.com/pluginx => web.lueluesay.top/git/lil/pluginx v0.0.0-20241017031415-3651ccc20390
require (
github.com/beevik/ntp v1.4.3
......
......@@ -2,6 +2,7 @@ package plugin
import (
"errors"
"path/filepath"
"time"
"agent/cmd/agent/config"
......@@ -9,8 +10,8 @@ import (
// 插件名
const (
HadesPlugin = "agent"
DobermanPlugin = "doberman"
PushStreamingPlugin = "pushStreaming"
ReportDCSInfoPlugin = "reportDCSInfo"
)
// 插件函数
......@@ -25,9 +26,11 @@ const (
)
var (
HadesPluginSocket = config.PluginDir + "/" + HadesPlugin + ".sock"
DobermanPluginSocket = config.PluginDir + "/" + DobermanPlugin + ".sock"
DefaultCallTimeout = 5 * time.Second
PushStreamingPluginSocket = PushStreamingPlugin + ".sock"
ReportDCSInfoPluginSocket = ReportDCSInfoPlugin + ".sock"
defaultPluginMgrConfFile = "plugin.conf"
DefaultCallTimeout = 5 * time.Second
ErrDisconnected = errors.New("doberman plugin disconnected")
)
......@@ -40,3 +43,9 @@ type Msg struct {
type CommMsg struct {
MsgType string
}
func setPluginSocketPath() {
PushStreamingPluginSocket = filepath.Join(config.PluginDir, PushStreamingPluginSocket)
ReportDCSInfoPluginSocket = filepath.Join(config.PluginDir, ReportDCSInfoPluginSocket)
defaultPluginMgrConfFile = filepath.Join(config.PluginDir, defaultPluginMgrConfFile)
}
package plugin
import (
"agent/cmd/agent/global"
"fmt"
"linkfog.com/public/lib/common"
"sync"
"time"
"agent/cmd/agent/global"
"agent/cmd/agent/option"
"linkfog.com/pluginx/pluginmgr"
"linkfog.com/pluginx/pluginrpc"
"linkfog.com/public/lib/common"
"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,
}
var defaultPluginCliConf = make(map[string]string)
type Plugin struct {
signal chan *global.Message
......@@ -48,6 +36,7 @@ func New(opts ...PluginOpt) *Plugin {
for _, opt := range opts {
opt(&p)
}
setPluginSocketPath()
p.signal = make(chan *global.Message, p.signalSize)
return &p
......@@ -70,11 +59,19 @@ func (p *Plugin) Start() error {
// 启动插件管理器,运行各插件进程
pluginmgr.EnableValidatePluginMD5 = option.Opt.EnableValidatePluginMD5
pluginmgr.InitDur = option.Opt.PluginStatusCheckDur
plgCfgMap, err := pluginmgr.LoadPluginConfigWithData([]byte(defaultPluginMgrConf))
plgCfgMap, err := pluginmgr.LoadPluginConfigWithFile(defaultPluginMgrConfFile)
if err != nil {
return fmt.Errorf("LoadPluginConfigWithData err: %v", err)
l.Warnf("load plugin config file failed, %v", err)
plgCfgMap = make(map[string]*pluginmgr.PluginProcessConf)
}
for name, plg := range plgCfgMap {
if name == PushStreamingPlugin {
defaultPluginCliConf[name] = PushStreamingPluginSocket
} else if name == ReportDCSInfoPlugin {
defaultPluginCliConf[name] = ReportDCSInfoPluginSocket
}
l.Info("plugin config:", name, plg)
}
p.plgConf = plgCfgMap
......@@ -82,10 +79,17 @@ func (p *Plugin) Start() error {
p.processMgr.Start()
// 建立与各插件进程的grpc连接
p.grpcClient, err = pluginrpc.NewPluginGrpcClient(defaultPluginCliConf)
p.grpcClient, err = pluginrpc.New()
if err != nil {
return fmt.Errorf("NewGrpcClientHelper err: %v", err)
}
if len(defaultPluginCliConf) > 0 {
err := p.grpcClient.NewPluginClient(defaultPluginCliConf)
if err != nil {
l.Error(err)
return err
}
}
// 等待所有插件运行正常
allIsRunning := false
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment