ebpf/config/config.go

package config

import (
	"fmt"
	"github.com/coroot/coroot-node-agent/flags"
	log "github.com/sirupsen/logrus"
	"github.com/spf13/viper"
	"strings"
	"sync"
)

type Config struct {
	m          *sync.RWMutex
	subscribes map[scbType]chan struct{}
	//subscribes     map[scbType]chan<- struct{}
	euspaceCfg     *euspaceCfg
	omniCommCfg    *omniCommonCfg
	watchingConfig bool
}

type scbType string

const (
	scbTypeUndefined          scbType = ""
	ScbTypeLogLevelChg        scbType = "logLevelChange"
	ScbTypeNetDataIntervalChg scbType = "netDataIntervalChange"
	ScbTypeConfigServerChg    scbType = "configServerChg"
	ScbTypeDataServerChg      scbType = "dataServerChg"
	ScbTypeTracesEndpointChg  scbType = "tracesEndpointChg"
)

var Cfg *Config

func init() {

	//fmt.Println("\n")
	//fmt.Printf("------- *flags.LogLevel :%s \n", *flags.LogLevel)
	//fmt.Printf("------- *flags.DataServer :%s \n", *flags.DataServer)
	//fmt.Printf("------- *flags.ConfigServer :%s \n", *flags.ConfigServer)
	//fmt.Println("\n")

	var err error
	Cfg, err = New()
	if err != nil {
		log.Fatalf(err.Error())
	}
	//fmt.Println("\n")
	//fmt.Printf("cfg --- logLevel :%s \n", Cfg.GetString("logLevel"))
	//fmt.Printf("cfg --- netDataInterval :%d  \n", Cfg.GetInt("netDataInterval"))
	//fmt.Printf("cfg --- configServer :%s  \n", Cfg.GetString("configServer"))
	//fmt.Printf("cfg --- dataServer :%s \n", Cfg.GetString("dataServer"))
}

func New() (*Config, error) {
	/*
		1. Viper uses the following precedence order: explicit call to `Set` > flag > env > Config > key/value store > default
		2. 按照现有业务实现，设计为：命令行参数（*flags.xxx）优先级高于配置文件中的项 。
		3. 会将euspace的配置 和 omniagent-common.ini的配置进行合并，然后同步到配置文件。（命令行参数会覆盖上下文的配置，但不会被同步到配置文件）
	*/

	ec, err := initEuspaceCfg()
	if err != nil {
		return nil, fmt.Errorf("init config,load `euspace` config failed : %s", err.Error())
	}

	omc, err := initOmniCommonCfg()
	if err != nil {
		return nil, fmt.Errorf("init config,load `omniagent` config failed : %s", err.Error())
	}

	cfg := &Config{
		m:          new(sync.RWMutex),
		subscribes: make(map[scbType]chan struct{}),
		//subscribes:  make(map[scbType]chan<- struct{}),
		euspaceCfg:  ec,
		omniCommCfg: omc,
	}

	//merge and sync omniagent common.ini Config
	if err = cfg.mergeAndSyncOmniCommConf(); err != nil {
		return nil, fmt.Errorf("init config,%s", err.Error())
	}

	//merge the flags
	cfg.mergeFlags()

	//simple check
	theCfgSvr := cfg.euspaceCfg.runtimeV.GetString("configServer")
	theDataSvr := cfg.euspaceCfg.runtimeV.GetString("dataServer")
	if theCfgSvr == "" || theDataSvr == "" {
		return nil, fmt.Errorf("init config,the config-server[%s] or the data-server[%s] is not set", theCfgSvr, theDataSvr)
	}
	if !strings.HasPrefix(theCfgSvr, "http") {
		cfg.euspaceCfg.runtimeV.Set("configServer", fmt.Sprintf("http://%s", theCfgSvr))
	}
	if !strings.HasPrefix(theDataSvr, "http") {
		cfg.euspaceCfg.runtimeV.Set("dataServer", fmt.Sprintf("http://%s", theDataSvr))
	}

	// set flags.TracesEndpoint
	/*if *flags.TracesEndpoint == nil {
		dataServer, err := url.Parse(theDataSvr)
		if err == nil && dataServer != nil {
			*flags.TracesEndpoint = dataServer.JoinPath(*flags.ServerPrefix + "/api/v2/data/receive")
		}
	}*/

	//start watch
	cfg.watchConfig()

	return cfg, nil
}

func (c *Config) mergeAndSyncOmniCommConf() error {
	hasChange := false
	if c.omniCommCfg != nil {
		cfgSrv := c.euspaceCfg.runtimeV.GetString("configServer")
		omniCfgSvr := c.omniCommCfg.getConfigServer()
		if omniCfgSvr != "" && omniCfgSvr != cfgSrv {
			hasChange = true
			c.euspaceCfg.runtimeV.Set("configServer", omniCfgSvr)
		}

		dataSvr := c.euspaceCfg.runtimeV.GetString("dataServer")
		omniDataSvr := c.omniCommCfg.getDataServer()
		if omniDataSvr != "" && omniDataSvr != dataSvr {
			hasChange = true
			c.euspaceCfg.runtimeV.Set("dataServer", omniDataSvr)
		}
		// set cfgRecordLast
		if err := c.omniCommCfg.commonSecV.Unmarshal(c.omniCommCfg.cfgRecordLast); err != nil {
			return fmt.Errorf("`omniCommCfg.commonSecV` unmarshal `cfgRecordLast` failed : %s", err.Error())
		}
		//fmt.Printf("------ c.omniCommCfg.cfgRecordLast:%#v \n", c.omniCommCfg.cfgRecordLast)
	}

	// set cfgRecordLast
	if err := c.euspaceCfg.runtimeV.Unmarshal(c.euspaceCfg.cfgRecordLast); err != nil {
		return fmt.Errorf("`euspaceCfg.runtimeV` unmarshal `cfgRecordLast` failed : %s", err.Error())
	}
	//fmt.Printf("------ c.euspaceCfg.cfgRecordLast:%#v \n", c.euspaceCfg.cfgRecordLast)

	//sync Config
	if hasChange {
		if err := c.euspaceCfg.syncConfToFile(c.euspaceCfg.cfgRecordLast); err != nil {
			return err
		}
	}
	return nil
}

func (c *Config) mergeFlags() {
	if *flags.LogLevel != "" {
		c.euspaceCfg.runtimeV.Set("logLevel", *flags.LogLevel)
	}
	if *flags.ConfigServer != "" {
		c.euspaceCfg.runtimeV.Set("configServer", *flags.ConfigServer)
	}
	if *flags.DataServer != "" {
		c.euspaceCfg.runtimeV.Set("dataServer", *flags.DataServer)
	}
}

func (c *Config) watchConfig() {
	c.m.Lock()
	defer c.m.Unlock()

	if c.watchingConfig {
		return
	}
	c.watchingConfig = true

	//watch euspace Config
	c.euspaceCfg.watchConfig(c)

	//watch omniagent-common.ini
	if c.omniCommCfg != nil && c.omniCommCfg.omniCommonV != nil {
		c.omniCommCfg.watchConfig(c)
	}
}

func (c *Config) Get(key string) any {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.Get(key)
}

func (c *Config) Sub(key string) *viper.Viper {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.Sub(key)
}

func (c *Config) GetString(key string) string {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.GetString(key)
}

func (c *Config) GetBool(key string) bool {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.GetBool(key)
}

func (c *Config) GetInt(key string) int {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.GetInt(key)
}

func (c *Config) GetInt32(key string) int32 {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.GetInt32(key)
}

func (c *Config) GetInt64(key string) int64 {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.GetInt64(key)
}

func (c *Config) GetFloat64(key string) float64 {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.GetFloat64(key)
}

func (c *Config) GetConfigFileUsed() string {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.euspaceCfg.runtimeV.ConfigFileUsed()
}

func (c *Config) SubscribeConfigChange(typ scbType) <-chan struct{} {
	c.m.Lock()
	defer c.m.Unlock()

	if typ == scbTypeUndefined {
		return nil
	}
	ch, ok := c.subscribes[typ]
	if !ok {
		ch = make(chan struct{})
		c.subscribes[typ] = ch
	}
	return ch
}

func (c *Config) noticeSubscribers(typ scbType) {
	c.m.RLock()
	defer c.m.RUnlock()
	if ch, ok := c.subscribes[typ]; ok {
		ch <- struct{}{}
	}
}

func (c *Config) Close() {
	c.m.Lock()
	defer c.m.Unlock()

	for typ, ch := range c.subscribes {
		close(ch)
		delete(c.subscribes, typ)
	}
}