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 ClientTimeout json.Number ServerTime json.Number ServerTimeout json.Number ServerMiniTime json.Number } type OssConfig struct { AccessKey string `json:"access_key"` AccessSecret string `json:"access_secret"` Bucket string `json:"bucket"` EndPoint string `json:"end_point"` DefaultBrandImage string `json:"default_brand_image"` BrandImage string `json:"brand_image"` DefaultSeriesImage string `json:"default_series_image"` SeriesImage string `json:"series_image"` } // 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 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"` Md5Rabbitmq RabbitmqConfig `json:"md5_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 } } } }