plugin.go

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)
}