adm-ads/common.in/config/config.go

package config

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"time"

	"github.com/fsnotify/fsnotify"
)

var Conf *Configure

var ConfigPath = "conf/common.json"

type KeepaliveConfig struct {
	ClientTime     json.Number `json:"client_time"`
	ClientTimeout  json.Number `json:"client_timeout"`
	ServerTime     json.Number `json:"server_time"`
	ServerTimeout  json.Number `json:"server_timeout"`
	ServerMiniTime json.Number `json:"server_mini_time"`
}

type OssConfig struct {
	Endpoint string `json:"endpoint"`
	Id       string `json:"id"`
	Secret   string `json:"secret"`
	Bucket   string `json:"bucket"`
}

// mysql 配置
type MysqlConfig struct {
	User      string      `json:"user"`
	Password  string      `json:"password"`
	Addr      string      `json:"addr"`
	Db       string      `json:"db"`
	Charset   string      `json:"charset"`
	MaxIdle   json.Number `json:"max_idle"`
	MaxConn   json.Number `json:"max_conn"`
}

type MongoConfig struct {
	User     string `json:"user"`
	Password string `json:"password"`
	Addr     string `json:"addr"`
}

// redis 配置
type RedisConfig struct {
	Addrs        string      `json:"addrs"`
	Password     string      `json:"password"`
	Db    json.Number `json:"db"`
	PoolSize     json.Number `json:"pool_size"`
	MinIdleConns json.Number `json:"min_idle_conns"`
	MaxRetries   json.Number `json:"max_retries"`
	IsCluster    string      `json:"is_cluster"`
}

type ElasticConfig struct {
	Addr  string `json:"addr"`
	Sniff string `json:"sniff"`
}

type LogConfig struct {
	MaxSize    json.Number `json:"max_size"`
	MaxBackups json.Number `json:"max_backups"`
	MaxAge     json.Number `json:"max_age"`
	Level      string      `json:"level"`
	Stacktrace string      `json:"stacktrace"`
	Path       string      `json:"path"`
}

type RPCNode struct {
	ServiceName   string      `json:"service_name"`
	ServicePort   string      `json:"service_port"`
	ServiceIp     string      `json:"service_ip"`
	MysqlDb       string      `json:"mysql_db"`
	RedisDb       json.Number `json:"redis_db"`
	LogLevel      string      `json:"log_level"`
	LogStacktrace string      `json:"log_stacktrace"`
}

type RPCConfig struct {
	Prefix    string
	Keepalive KeepaliveConfig `json:"keepalive"`
	AdmDws    RPCNode `json:"adm_dws"`
	AdmOds    RPCNode `json:"adm_ods"`
	AdmAds    RPCNode `json:"adm_ads"`
	AdmTask   RPCNode `json:"adm_task"`
}

type RabbitmqConfig struct {
	Addr         string `json:"addr"`
	Username     string `json:"username"`
	Passwrod     string `json:"passwrod"`
	Vhost        string `json:"vhost"`
	ExchangeName string `json:"exchange_name"`
	QueueName    string `json:"queue_name"`
	RouteBindKey string `json:"route_bind_key"`
	ConsumerCount json.Number `json:"consumer_count"`
}

type Configure struct {
	RunMode     string         `json:"run_mode"`
	Log         LogConfig      `json:"log"`
	Mysql       MysqlConfig    `json:"mysql"`
	Mongo       MongoConfig    `json:"mongo"`
	Redis       RedisConfig    `json:"redis"`
	Elastic     ElasticConfig  `json:"elastic"`
	Oss         OssConfig      `json:"oss"`
	Rpc         RPCConfig      `json:"rpc"`
	OdsRabbitmq RabbitmqConfig `json:"ods_rabbitmq"`
	DwsRabbitmq RabbitmqConfig `json:"dws_rabbitmq"`
	AdsRabbitmq RabbitmqConfig `json:"ads_rabbitmq"`
}

/*
func watchEtcd(keysAPI client.KeysAPI, index uint64, runmode, key string, cli client.Client) {
	basePath := fmt.Sprintf("/%s/config", runmode)
	watcherOptions := &client.WatcherOptions{
		AfterIndex: index,
		Recursive:  true,
	}
	watcher := keysAPI.Watcher(basePath, watcherOptions)
	for {
		r, err := watcher.Next(context.Background())
		if err != nil || r == nil {
			break
		}
		if r.Node != nil && r.PrevNode != nil {
			if r.Node.Key == r.PrevNode.Key && (r.Node.Value != r.PrevNode.Value) {
				reloadEtcd(runmode, key, cli)
			}
		}
		if r.Node != nil && r.PrevNode == nil {
			reloadEtcd(runmode, key, cli)
		}
	}
}

func reloadEtcd(runmode, key string, cli client.Client) {
	keysAPI := client.NewKeysAPI(cli)
	basePath := fmt.Sprintf("/%s/config", runmode)
	if resp, err := keysAPI.Get(context.Background(), basePath, &client.GetOptions{
		Recursive: true,
	}); err == nil && resp != nil && resp.Node != nil {
		conf := &Configure{}
		value := getNodeData(key, resp.Node)
		if jsonStr, err := json.Marshal(value); err == nil {
			if err := json.Unmarshal(jsonStr, &conf); err == nil {
				*Conf = *conf
				return
			} else {
				fmt.Printf("json Unmarshal failed. error:%s", err)
			}
		} else {
			fmt.Printf("json Marshal failed. error:%s", err)
		}
	} else {
		fmt.Printf("get %s failed. error:%s", basePath, err)
	}
}

// key 为加密密钥
func GetConfig(runmode, key string, cli client.Client) *Configure {
	keysAPI := client.NewKeysAPI(cli)
	basePath := fmt.Sprintf("/%s/config", runmode)
	if resp, err := keysAPI.Get(context.Background(), basePath, &client.GetOptions{
		Recursive: true,
	}); err == nil && resp != nil && resp.Node != nil {
		Conf = &Configure{}
		value := getNodeData(key, resp.Node)
		if jsonStr, err := json.Marshal(value); err == nil {
			if err := json.Unmarshal(jsonStr, &Conf); err == nil {
				go watchEtcd(keysAPI, resp.Index, runmode, key, cli)
				return Conf
			} else {
				fmt.Printf("json Unmarshal failed. error:%s", err)
			}
		} else {
			fmt.Printf("json Marshal failed. error:%s", err)
		}
	} else {
		fmt.Printf("get %s failed. error:%s", basePath, err)
		os.Exit(1)
	}

	return nil
}

// 递归取出node的叶子节点值
func getNodeData(key string, head *client.Node) (value interface{}) {
	s0 := strings.Split(head.Key, "/")
	len0 := len(s0)
	if len0 == 0 {
		return
	}

	if head.Dir {
		mapData := map[string]interface{}{}
		for _, node := range head.Nodes {
			s1 := strings.Split(node.Key, "/")
			len1 := len(s1)
			if len1 == 0 {
				break
			}

			mapData[s1[len1-1]] = getNodeData(key, node)
		}

		value = mapData
	} else {
		if key != "" && head.Value != "" {
			if bytesData, err := Base64URLDecode(head.Value); err != nil {
				fmt.Printf("Base64URLDecode(%s) failed. error:%s", head.Value, err)
				os.Exit(1)
			} else {
				if data, err := AesDecrypt(bytesData, []byte(key)); err != nil {
					fmt.Printf("AesDecrypt failed. error:%s", err)
					os.Exit(1)
				} else {
					value = string(data)
				}
			}
		} else {
			// 无加密，直接取值
			value = head.Value
		}
	}

	return
}
*/
// 适配k8s 方式
// 公共配置会以configmap的方式映射到容器的conf/common.json中
func SetConfigFile(configPath string) {
	ConfigPath = configPath
}
func GetConfigForK8s() *Configure {
	buffer, err := ioutil.ReadFile(ConfigPath)
	if err != nil {
		fmt.Printf("get %s failed. error:%s", ConfigPath, err)
		return nil
	}
	Conf = &Configure{}
	if err := json.Unmarshal(buffer, Conf); err != nil {
		fmt.Printf("json Unmarshal failed. error:%s", err)
		return nil
	}
	go watchConfigFileForK8s()
	return Conf
}

func ReloadConfigForK8s() {
	buffer, err := ioutil.ReadFile(ConfigPath)
	if err != nil {
		fmt.Printf("get %s failed. error:%s", ConfigPath, err)
	}
	confTmp := &Configure{}
	if err := json.Unmarshal(buffer, confTmp); err != nil {
		fmt.Printf("json Unmarshal failed. error:%s", err)
	}
	Conf = confTmp
}

// 判断路径文件/文件夹是否存在
func Exists(path string) bool {
	_, err := os.Stat(path)
	if err != nil {
		if os.IsExist(err) {
			return true
		}
		return false
	}
	return true
}

func watchConfigFileForK8s() {
	fileExist := true
	watch, err := fsnotify.NewWatcher()
	if err != nil {
		fmt.Printf("new file watcher failed\n\n")
		os.Exit(1)
	}

	defer watch.Close()

	/*err = watch.Add(ConfigPath)
	if err != nil {
		fmt.Printf("add file watcher failed\n\n")
		os.Exit(1)
	}*/

	for {
		// 判断文件是否存在
		if !Exists(ConfigPath) {
			time.Sleep(10 * time.Second)
			fileExist = false
			continue
		} else {
			watch.Remove(ConfigPath)
			watch.Add(ConfigPath)
			if !fileExist { // 文件重新创建
				ReloadConfigForK8s()
			}
			fileExist = true
		}

		select {
		case ev := <-watch.Events:
			{
				fmt.Println("op : ", ev.Op)
				ReloadConfigForK8s()

			}
		case err := <-watch.Errors:
			{
				fmt.Println("error : ", err)
				continue
			}
		}
	}
}