first commit

.gitignore

@@ -0,0 +1,7 @@
+.idea/
+test/
+logs/
+log/
+config/cfg.json
+dev
+.DS_Store

README.md

@@ -0,0 +1,60 @@
+## Dynamic Application Configuration Management
+
+支持监听阿里云acm、nacos配置变化并更新本地配置文件, 支持同步完成后执行自定义命令
+
+- 自定义执行命令支持配置延时执行、超时时间控制
+- 支持同时监听多个命名空间、多个group、多个data id的配置
+
+配置文件说明
+
+```text
+{
+     "cluster_type": "nacos", // 集群模式支持acm和nacos, acm配置请参考config/acm.json.example
+     "cluster_nodes": [ // nacos 集群节点列表
+         {
+             "ip": "10.1.5.83",
+             "port": 8848
+         }
+     ],
+     "log_level": "debug", // 日志级别
+     "cache_dir": "/data/dacm/cache", //缓存目录
+     "log_dir": "/data/dacm/log", // 日志目录
+     "rotate_time": "1h", // 日志滚动时间
+     "max_age": 3,  // 保留最近3小时的日志
+     "namespaces": [ // 命名空间配置
+         {
+             "id": "02d4cb2d-2833-48a2-a8ea-bdcee4259359", // 命名空间id
+             "name": "sunline-uat", // 命名空间名称
+             "username": "", // 授权用户名
+             "password": "", // 授权用户密码
+             "configs": [
+                 {
+                     "data_id": "uat-sunline-snactiv-core", // 配置data id
+                     "group": "DEFAULT_GROUP", // 配置所在的group
+                     "sync_file": "/data/apps/sunline/snactiv/server.properties", // 需要动态更新的本地配置文件所在路径
+                     "execute": "cd /data/apps/sunline/snactiv/; source /etc/profile ${HOME}/.bash_profile; sh start.sh", // 动态更新本地配置文件之后执行自定义命令，如果为空默认只更新本地配置文件
+                     "execute_delay": 10, // 自定义命令执行延时时间，单位毫秒
+                     "execute_timeout": 10000, // 自定义命令执行超时时间，单位毫秒
+                     "not_load_cache_at_start": true, // 设置为true
+                     "timeout": 5000 // 监听配置超时时间
+                 }
+             ]
+         }
+     ],
+     "max_cpu_rate": 1, // 允许绑定cpu核心数占比(cpu核心数*max_cpu_rate)
+     "max_mem_rate": 1 // 允许使用的内存占比(内存总大小*max_mem_rate)
+}
+```
+
+使用systemd管理服务
+
+```text
+将压缩包解压到 /usr/local/dacm/ 目录, 创建配置文件 /usr/local/dacm/config/app.json，然后执行以下命令
+
+cd /usr/local/dacm/
+cp systemd/dacm.service /usr/lib/systemd/system/dacm.service
+systemctl daemon-reload
+systemctl start dacm
+systemctl enable dacm
+systemctl status dacm
+```

common/g/config.go

@@ -0,0 +1,275 @@
+package g
+
+import (
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"github.com/fanghongbo/dacm/utils"
+	"io/ioutil"
+	"log"
+	"os"
+	"runtime"
+	"strings"
+	"sync"
+)
+
+var (
+	cfg            = flag.String("c", "./config/cfg.json", "specify config file")
+	v              = flag.Bool("v", false, "show version")
+	vv             = flag.Bool("vv", false, "show version detail")
+	ConfigFile     string
+	configFileLock = new(sync.RWMutex)
+	config         *GlobalConfig
+)
+
+type NacosNode struct {
+	Ip   string `json:"ip"`
+	Port int64  `json:"port"`
+}
+
+type Namespace struct {
+	Id       string         `json:"id"`
+	Name     string         `json:"name"`
+	Username string         `json:"username"`
+	Password string         `json:"password"`
+	Configs  []*NacosConfig `json:"configs"`
+}
+
+type NacosConfig struct {
+	DataId              string `json:"data_id"`
+	Group               string `json:"group"`
+	SyncFile            string `json:"sync_file"`
+	Execute             string `json:"execute"`
+	ExecuteDelay        int64  `json:"execute_delay"`
+	ExecuteTimeout      int64  `json:"execute_timeout"`
+	NotLoadCacheAtStart bool   `json:"not_load_cache_at_start"`
+	Timeout             int64  `json:"timeout"`
+}
+
+type GlobalConfig struct {
+	ClusterType  string       `json:"cluster_type"` // nacos | acm
+	ClusterNodes []*NacosNode `json:"cluster_nodes"`
+	Endpoint     string       `json:"endpoint"`
+	RegionId     string       `json:"region_id"`
+	AccessKey    string       `json:"access_key"`
+	SecretKey    string       `json:"secret_key"`
+	OpenKms      bool         `json:"open_kms"`
+	Namespaces   []*Namespace `json:"namespaces"`
+	MaxCPURate   float64      `json:"max_cpu_rate"`
+	MaxMemRate   float64      `json:"max_mem_rate"`
+	CacheDir     string       `json:"cache_dir"`
+	LogDir       string       `json:"log_dir"`
+	LogLevel     string       `json:"log_level"`
+	RotateTime   string       `json:"rotate_time"`
+	MaxAge       int64        `json:"max_age"`
+}
+
+func Conf() *GlobalConfig {
+	configFileLock.RLock()
+	defer configFileLock.RUnlock()
+
+	return config
+}
+
+func InitConfig() error {
+	var (
+		cfgFile   string
+		bs        []byte
+		err       error
+		maxMemMB  int
+		maxCPUNum int
+	)
+
+	flag.Parse()
+
+	if *v {
+		fmt.Println(VersionInfo())
+		os.Exit(0)
+	}
+
+	if *vv {
+		fmt.Println(VersionDetailInfo())
+		os.Exit(0)
+	}
+
+	cfgFile = *cfg
+	ConfigFile = cfgFile
+
+	if cfgFile == "" {
+		return errors.New("config file not specified: use -c $filename")
+	}
+
+	if _, err = os.Stat(cfgFile); os.IsNotExist(err) {
+		return fmt.Errorf("config file specified not found: %s", cfgFile)
+	} else {
+		log.Printf("[INFO] use config file: %s", ConfigFile)
+	}
+
+	if bs, err = ioutil.ReadFile(cfgFile); err != nil {
+		return fmt.Errorf("read config file failed: %s", err.Error())
+	} else {
+		if err = json.Unmarshal(bs, &config); err != nil {
+			return fmt.Errorf("decode config file failed: %s", err.Error())
+		} else {
+			log.Printf("[INFO] load config success from %s", cfgFile)
+		}
+	}
+
+	if err = Validator(); err != nil {
+		return fmt.Errorf("validator config file fail: %s", err)
+	}
+
+	// 最大使用内存
+	maxMemMB = utils.CalculateMemLimit(config.MaxMemRate)
+
+	// 最大cpu核数
+	maxCPUNum = utils.GetCPULimitNum(config.MaxCPURate)
+
+	log.Printf("[INFO] bind [%d] cpu core", maxCPUNum)
+	runtime.GOMAXPROCS(maxCPUNum)
+
+	log.Printf("[INFO] memory limit: %d MB", maxMemMB)
+
+	return nil
+}
+
+func ReloadConfig() error {
+	var (
+		bs  []byte
+		err error
+	)
+
+	if _, err = os.Stat(ConfigFile); os.IsNotExist(err) {
+		return fmt.Errorf("config file specified not found: %s", ConfigFile)
+	} else {
+		log.Printf("[INFO] reload config file: %s", ConfigFile)
+	}
+
+	if bs, err = ioutil.ReadFile(ConfigFile); err != nil {
+		return fmt.Errorf("reload config file failed: %s", err)
+	} else {
+		configFileLock.RLock()
+		defer configFileLock.RUnlock()
+
+		if err = json.Unmarshal(bs, &config); err != nil {
+			return fmt.Errorf("decode config file failed: %s", err)
+		} else {
+			log.Printf("[INFO] reload config success from %s", ConfigFile)
+		}
+	}
+
+	if err = Validator(); err != nil {
+		return fmt.Errorf("validator config file fail: %s", err)
+	}
+
+	return nil
+}
+
+func Validator() error {
+
+	if !utils.InStringArray([]string{"acm", "nacos"}, config.ClusterType) {
+		return fmt.Errorf("cluster type not support: %s", config.ClusterType)
+	}
+
+	if config.RotateTime == "" {
+		config.RotateTime = "1h"
+	}
+
+	if config.MaxAge <= 0 {
+		config.MaxAge = 3
+	}
+
+	if config.LogLevel == "" {
+		config.LogLevel = "info"
+	} else {
+		if !utils.InStringArray([]string{"info", "warn", "error", "debug"}, config.LogLevel) {
+			return fmt.Errorf("log level must be debug, info, warn, error")
+		}
+	}
+
+	if config.ClusterType == "acm" {
+		if config.Endpoint == "" {
+			return fmt.Errorf("acm service endpoint is empty")
+		}
+
+		if config.RegionId == "" {
+			return fmt.Errorf("acm service region id is empty")
+		}
+
+		if config.AccessKey == "" {
+			return fmt.Errorf("acm service access key is empty")
+		}
+
+		if config.SecretKey == "" {
+			return fmt.Errorf("acm service secret key is empty")
+		}
+	} else {
+		var exist bool
+		for _, server := range config.ClusterNodes {
+			if server.Ip == "" {
+				return fmt.Errorf("nacos server host is empty")
+			}
+
+			if server.Port <= 0 {
+				return fmt.Errorf("nacos server port is empty")
+			}
+
+			exist = true
+		}
+
+		if !exist {
+			return fmt.Errorf("nacos cluster nodes is empty")
+		}
+	}
+
+	for _, namespace := range config.Namespaces {
+		if namespace.Id == "" {
+			return fmt.Errorf("namespace id is empty")
+		}
+
+		if namespace.Name == "" {
+			return fmt.Errorf("namespace name is empty")
+		}
+
+		for _, item := range namespace.Configs {
+			if item.DataId != "" {
+				if item.Group == "" {
+					item.Group = "DEFAULT_GROUP"
+				}
+
+				if item.ExecuteDelay < 0 {
+					return fmt.Errorf("execute delay must be ge 0")
+				}
+
+				if item.ExecuteTimeout <= 0 {
+					return fmt.Errorf("execute timeout must be gt 0")
+				}
+
+				if item.Timeout <= 0 {
+					return fmt.Errorf("listen config timeout must be gt 0")
+				}
+
+				if item.ExecuteDelay < 0 {
+					return fmt.Errorf("get nacos config timeout must be ge 0")
+				}
+			}
+		}
+	}
+
+	// MaxCPURate
+	if config.MaxCPURate < 0 || config.MaxCPURate > 1 {
+		return errors.New("max_cpu_rate is range 0 to 1")
+	}
+
+	// MaxMemRate
+	if config.MaxMemRate < 0 || config.MaxMemRate > 1 {
+		return errors.New("max_mem_rate is range 0 to 1")
+	}
+
+	return nil
+}
+
+func ReformatConfigValue(value string) string {
+	return strings.TrimSpace(value)
+}

common/g/init.go

@@ -0,0 +1,40 @@
+package g
+
+import (
+	"context"
+)
+
+func InitAll() error {
+	var err error
+
+	// 初始化运行时环境
+	if err = InitRuntime(); err != nil {
+		return err
+	}
+
+	// 初始化配置文件
+	if err = InitConfig(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func Shutdown(ctx context.Context) error {
+	var (
+		ch  chan struct{}
+	)
+
+	ch = make(chan struct{}, 0)
+
+	go func() {
+		ch <- struct{}{}
+	}()
+
+	select {
+	case <-ch:
+		return nil
+	case <-ctx.Done():
+		return nil
+	}
+}