Merge pull request #109 from TencentBlueKing/feat/diff_check_reload

feat: 支持主配置变更重加载

beater/beater.go

@@ -28,22 +28,28 @@ package beater
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/output/gse"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/utils/host"
+	"github.com/elastic/beats/libbeat/cfgfile"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/pkg/errors"
+
 	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
-	// 加载 Filebeat Input插件及配置优化模块
-	_ "github.com/TencentBlueKing/bkunifylogbeat/include"
+	_ "github.com/TencentBlueKing/bkunifylogbeat/include" // 加载 Filebeat Input插件及配置优化模块
 	"github.com/TencentBlueKing/bkunifylogbeat/registrar"
-	"github.com/pkg/errors"
+	"github.com/TencentBlueKing/bkunifylogbeat/utils"
 )
 
 var Registrar *registrar.Registrar
 
+const beatName = "bkunifylogbeat"
+
 // LogBeat  package cadvisor
 type LogBeat struct {
 	Name    string
@@ -53,7 +59,8 @@ type LogBeat struct {
 
 	hostIDWatcher host.Watcher
 
-	isReload bool
+	isReload     bool
+	lastTaskHash string
 }
 
 // New create cadvisor beat
@@ -118,6 +125,8 @@ func (bt *LogBeat) Run() error {
 	}
 
 	reloadTicker := time.NewTicker(10 * time.Second)
+	diffTaskTicker := time.NewTicker(60 * time.Second)
+	defer diffTaskTicker.Stop()
 	defer reloadTicker.Stop()
 	for {
 		select {
@@ -126,10 +135,23 @@ func (bt *LogBeat) Run() error {
 			if bt.isReload {
 				bt.isReload = false
 				config := beat.GetConfig()
+				if bt.config.CheckDiff {
+					config, err = GetRawConfig()
+					if err != nil {
+						logp.L.Error(err)
+					}
+				}
 				if config != nil {
 					bt.Reload(config)
 				}
 			}
+		// 处理采集器主配置是否变更，变更则发送重加载信号
+		case <-diffTaskTicker.C:
+			if bt.config.CheckDiff {
+				if err = bt.checkDiffReload(); err != nil {
+					logp.L.Error(err)
+				}
+			}
 		case <-beat.ReloadChan:
 			bt.isReload = true
 		// 处理采集器框架发送的结束采集器的信号（常由SIGINT引起），关闭采集器
@@ -148,6 +170,53 @@ func (bt *LogBeat) Stop() {
 	close(bt.done)
 }
 
+// GetRawConfig Get raw main config
+func GetRawConfig() (*common.Config, error) {
+	rawConfig, err := cfgfile.Load("", nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if !rawConfig.HasField(beatName) {
+		return nil, errors.New("no beat name field found")
+	}
+
+	beatConfig, err := rawConfig.Child(beatName, -1)
+	if err != nil {
+		return nil, err
+	}
+
+	return beatConfig, nil
+}
+
+// Main config diff check
+func (bt *LogBeat) checkDiffReload() error {
+	beatConfig, err := GetRawConfig()
+	if err != nil {
+		return err
+	}
+	config, err := cfg.Parse(beatConfig)
+	if err != nil {
+		return err
+	}
+	b, err := json.Marshal(config)
+	if err != nil {
+		return err
+	}
+
+	currentTaskHash := utils.Md5(string(b))
+	if len(bt.lastTaskHash) == 0 {
+		bt.lastTaskHash = currentTaskHash
+	}
+	if bt.lastTaskHash != currentTaskHash {
+		bt.lastTaskHash = currentTaskHash
+		bt.Reload(beatConfig)
+		logp.L.Info("Reload main config task.")
+	}
+
+	return nil
+}
+
 // Close cadvisor storage interface
 func (bt *LogBeat) Close() error {
 	return nil

beater/manager.go

@@ -29,12 +29,13 @@ import (
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
 	bkmonitoring "github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/monitoring"
+	"github.com/elastic/beats/filebeat/input/file"
+	"github.com/elastic/beats/libbeat/monitoring"
+
 	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/TencentBlueKing/bkunifylogbeat/registrar"
 	"github.com/TencentBlueKing/bkunifylogbeat/task"
 	"github.com/TencentBlueKing/bkunifylogbeat/utils"
-	"github.com/elastic/beats/filebeat/input/file"
-	"github.com/elastic/beats/libbeat/monitoring"
 )
 
 var (

config/config.go

@@ -55,6 +55,7 @@ type Config struct {
 	CmdbLevelMaxLength int    `config:"cmdb_level_max_length"`
 	IgnoreCmdbLevel    bool   `config:"ignore_cmdb_level"`
 	MustHostIDExist    bool   `config:"must_host_id_exist"`
+	CheckDiff          bool   `config:"check_diff"`
 }
 
 // 从配置目录

config/input/log.go

@@ -27,8 +27,9 @@ import (
 	"time"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
-	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/dustin/go-humanize"
+
+	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 )
 
 // 如果未配置close_inactive则直接默认为5分钟

config/input/log_test.go

@@ -26,10 +26,11 @@ import (
 	"testing"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
-	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/elastic/beats/libbeat/common"
 	libbeatlogp "github.com/elastic/beats/libbeat/logp"
 	"github.com/stretchr/testify/assert"
+
+	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 )
 
 func init() {

config/input/redis.go

@@ -24,6 +24,7 @@ package input
 
 import (
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
+
 	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 )
 

config/input/wineventlog.go

@@ -23,10 +23,12 @@
 package input
 
 import (
+	"time"
+
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
-	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/elastic/beats/libbeat/common"
-	"time"
+
+	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 )
 
 // WinEventLogConfig window event log collect config

config/task.go

@@ -33,9 +33,10 @@ import (
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
-	"github.com/TencentBlueKing/bkunifylogbeat/utils"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/processors"
+
+	"github.com/TencentBlueKing/bkunifylogbeat/utils"
 )
 
 // ConditionConfig : 用于条件表达式，目前支持=、!=、eq、neq、include、exclude、regex、nregex

include/default.go

@@ -23,25 +23,14 @@
 package include
 
 import (
-	// input type : kafka
 	_ "github.com/elastic/beats/filebeat/input/kafka"
-	// input type : log
 	_ "github.com/elastic/beats/filebeat/input/log"
-	// input type : redis
 	_ "github.com/elastic/beats/filebeat/input/redis"
-	// input type : stdin
 	_ "github.com/elastic/beats/filebeat/input/stdin"
-	// input type : syslog
 	_ "github.com/elastic/beats/filebeat/input/syslog"
-	// input type : udp
 	_ "github.com/elastic/beats/filebeat/input/udp"
 
-	// window event log
-	_ "github.com/TencentBlueKing/bkunifylogbeat/task/input/wineventlog"
-
-	// input config
 	_ "github.com/TencentBlueKing/bkunifylogbeat/config/input"
-
-	// formatter
 	_ "github.com/TencentBlueKing/bkunifylogbeat/task/formatter"
+	_ "github.com/TencentBlueKing/bkunifylogbeat/task/input/wineventlog"
 )

main.go

@@ -27,10 +27,11 @@ import (
 	"os"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
-	"github.com/TencentBlueKing/bkunifylogbeat/beater"
-	_ "github.com/TencentBlueKing/bkunifylogbeat/json"
 	"github.com/elastic/beats/libbeat/cmd/instance"
 	"github.com/elastic/beats/libbeat/publisher/processing"
+
+	"github.com/TencentBlueKing/bkunifylogbeat/beater"
+	_ "github.com/TencentBlueKing/bkunifylogbeat/json"
 )
 
 var (

registrar/registrar.go

@@ -33,11 +33,12 @@ import (
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
 	bkmonitoring "github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/monitoring"
 	bkStorage "github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/storage"
-	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
-	"github.com/TencentBlueKing/bkunifylogbeat/task/input/wineventlog"
 	"github.com/elastic/beats/filebeat/input/file"
 	commonFile "github.com/elastic/beats/libbeat/common/file"
 	"github.com/elastic/beats/libbeat/monitoring"
+
+	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
+	"github.com/TencentBlueKing/bkunifylogbeat/task/input/wineventlog"
 )
 
 const (

registrar/registrar_test.go

@@ -30,11 +30,12 @@ import (
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
 	bkStorage "github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/storage"
-	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
-	"github.com/TencentBlueKing/bkunifylogbeat/tests"
 	"github.com/elastic/beats/filebeat/input/file"
 	libbeatlogp "github.com/elastic/beats/libbeat/logp"
 	"github.com/stretchr/testify/assert"
+
+	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
+	"github.com/TencentBlueKing/bkunifylogbeat/tests"
 )
 
 func init() {

task/base/base.go

@@ -24,10 +24,11 @@ package base
 
 import (
 	"fmt"
+	"sync"
+
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
 	bkmonitoring "github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/monitoring"
 	"github.com/elastic/beats/libbeat/monitoring"
-	"sync"
 )
 
 var (

task/filter/filter.go

@@ -28,10 +28,11 @@ import (
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
 	bkmonitoring "github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/monitoring"
+	"github.com/elastic/beats/filebeat/util"
+
 	"github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/TencentBlueKing/bkunifylogbeat/task/base"
 	"github.com/TencentBlueKing/bkunifylogbeat/task/processor"
-	"github.com/elastic/beats/filebeat/util"
 )
 
 var (

task/filter/filter_test.go

@@ -23,11 +23,12 @@
 package filter
 
 import (
+	"testing"
+	"time"
+
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/logp"
 	libbeatlogp "github.com/elastic/beats/libbeat/logp"
-	"testing"
-	"time"
 
 	cfg "github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/TencentBlueKing/bkunifylogbeat/tests"

task/formatter/formatter.go

@@ -24,14 +24,15 @@ package formatter
 
 import (
 	"fmt"
-	"github.com/golang/groupcache/lru"
 	"path/filepath"
 	"strconv"
 	"strings"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
-	"github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/elastic/beats/filebeat/util"
+	"github.com/golang/groupcache/lru"
+
+	"github.com/TencentBlueKing/bkunifylogbeat/config"
 )
 
 // Formatter 采集器事件包格式化接口, 根据任务配置返回相应的格式

task/formatter/tqos.go

@@ -28,10 +28,11 @@ import (
 	"fmt"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
-	"github.com/TencentBlueKing/bkunifylogbeat/config"
-	"github.com/TencentBlueKing/bkunifylogbeat/utils"
 	"github.com/elastic/beats/filebeat/util"
 	"github.com/golang/groupcache/lru"
+
+	"github.com/TencentBlueKing/bkunifylogbeat/config"
+	"github.com/TencentBlueKing/bkunifylogbeat/utils"
 )
 
 // TQOSLogConfig 如果未配置close_inactive则直接默认为5分钟

task/formatter/tqos_test.go

@@ -23,8 +23,9 @@
 package formatter
 
 import (
-	"github.com/TencentBlueKing/bkunifylogbeat/config"
 	"testing"
+
+	"github.com/TencentBlueKing/bkunifylogbeat/config"
 )
 
 func TestNewTQOSFormatter(t *testing.T) {

task/formatter/unifytlogc.go

@@ -28,10 +28,11 @@ import (
 	"fmt"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/libgse/beat"
-	"github.com/TencentBlueKing/bkunifylogbeat/config"
-	"github.com/TencentBlueKing/bkunifylogbeat/utils"
 	"github.com/elastic/beats/filebeat/util"
 	"github.com/golang/groupcache/lru"
+
+	"github.com/TencentBlueKing/bkunifylogbeat/config"
+	"github.com/TencentBlueKing/bkunifylogbeat/utils"
 )
 
 // LogConfig 如果未配置close_inactive则直接默认为5分钟

task/formatter/unifytlogc_test.go

@@ -26,12 +26,13 @@ import (
 	"testing"
 	"time"
 
-	"github.com/TencentBlueKing/bkunifylogbeat/config"
 	"github.com/elastic/beats/filebeat/input/file"
 	"github.com/elastic/beats/filebeat/util"
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/stretchr/testify/assert"
+
+	"github.com/TencentBlueKing/bkunifylogbeat/config"
 )
 
 func BaseFormatter(t *testing.T, taskConfig *config.TaskConfig) {