From c423dc34942cb6a0f66215a5d0f1259bd7b055b6 Mon Sep 17 00:00:00 2001 From: baozhecheng Date: Wed, 10 Aug 2022 13:07:45 +0800 Subject: [PATCH] fix: apollo unable to get and watch to the properties file --- contrib/config/apollo/apollo.go | 89 +++++++++++++++++++++++++++++--- contrib/config/apollo/parser.go | 2 +- contrib/config/apollo/watcher.go | 34 ++++++++++-- 3 files changed, 112 insertions(+), 13 deletions(-) diff --git a/contrib/config/apollo/apollo.go b/contrib/config/apollo/apollo.go index a0a38b023..2901e5af2 100644 --- a/contrib/config/apollo/apollo.go +++ b/contrib/config/apollo/apollo.go @@ -4,6 +4,7 @@ import ( "strings" "github.com/go-kratos/kratos/v2/config" + "github.com/go-kratos/kratos/v2/encoding" "github.com/go-kratos/kratos/v2/log" "github.com/apolloconfig/agollo/v4" @@ -108,10 +109,9 @@ func NewSource(opts ...Option) config.Source { func format(ns string) string { arr := strings.Split(ns, ".") - if len(arr) <= 1 { + if len(arr) <= 1 || arr[len(arr)-1] == "properties" { return "json" } - return arr[len(arr)-1] } @@ -120,18 +120,40 @@ func (e *apollo) load() []*config.KeyValue { namespaces := strings.Split(e.opt.namespace, ",") for _, ns := range namespaces { - value, err := e.client.GetConfigCache(ns).Get("content") + if strings.Contains(ns, ".") && !strings.Contains(ns, "properties") && + (format(ns) == "yaml" || format(ns) == "yaml" || format(ns) == "json") { + value, err := e.client.GetConfigCache(ns).Get("content") + if err != nil { + log.Errorf("apollo get config failed，err:%v", err) + continue + } + // serialize the namespace content KeyValue into bytes. + kv = append(kv, &config.KeyValue{ + Key: ns, + Value: []byte(value.(string)), + Format: format(ns), + }) + continue + } + next := map[string]interface{}{} + e.client.GetConfigCache(ns).Range(func(key, value interface{}) bool { + // all values are out properties format + resolve(genKey(ns, key.(string)), value, next) + return true + }) + f := format(ns) + codec := encoding.GetCodec(f) + val, err := codec.Marshal(next) if err != nil { - log.Warnw("apollo get config failed", "err", err) + log.Warnf("apollo could not handle namespace %s: %v", ns, err) + continue } - // serialize the namespace content KeyValue into bytes. kv = append(kv, &config.KeyValue{ Key: ns, - Value: []byte(value.(string)), - Format: format(ns), + Value: val, + Format: f, }) } - return kv } @@ -146,3 +168,54 @@ func (e *apollo) Watch() (config.Watcher, error) { } return w, nil } + +// resolve convert kv pair into one map[string]interface{} by split key into different +// map level. such as: app.name = "application" => map[app][name] = "application" +func resolve(key string, value interface{}, target map[string]interface{}) { + // expand key "aaa.bbb" into map[aaa]map[bbb]interface{} + keys := strings.Split(key, ".") + last := len(keys) - 1 + cursor := target + + for i, k := range keys { + if i == last { + cursor[k] = value + break + } + + // not the last key, be deeper + v, ok := cursor[k] + if !ok { + // create a new map + deeper := make(map[string]interface{}) + cursor[k] = deeper + cursor = deeper + continue + } + + // current exists, then check existing value type, if it's not map + // that means duplicate keys, and at least one is not map instance. + if cursor, ok = v.(map[string]interface{}); !ok { + log.Warnf("duplicate key: %v\n", strings.Join(keys[:i+1], ".")) + break + } + } +} + +// genKey got the key of config.KeyValue pair. +// eg: namespace.ext with subKey got namespace.subKey +func genKey(ns, sub string) string { + arr := strings.Split(ns, ".") + if len(arr) < 1 { + return sub + } + + if len(arr) == 1 { + if ns == "" { + return sub + } + return ns + "." + sub + } + + return strings.Join(arr[:len(arr)-1], ".") + "." + sub +} diff --git a/contrib/config/apollo/parser.go b/contrib/config/apollo/parser.go index a449256ff..eaee845e9 100644 --- a/contrib/config/apollo/parser.go +++ b/contrib/config/apollo/parser.go @@ -13,7 +13,7 @@ func (parser jsonExtParser) Parse(configContent interface{}) (map[string]interfa type yamlExtParser struct{} -func (parser yamlExtParser) Parse(configContent interface{}) (out map[string]interface{}, err error) { +func (parser yamlExtParser) Parse(configContent interface{}) (map[string]interface{}, error) { return map[string]interface{}{"content": configContent}, nil } diff --git a/contrib/config/apollo/watcher.go b/contrib/config/apollo/watcher.go index 8c039005e..b31dec7a4 100644 --- a/contrib/config/apollo/watcher.go +++ b/contrib/config/apollo/watcher.go @@ -2,6 +2,8 @@ package apollo import ( "context" + "github.com/go-kratos/kratos/v2/encoding" + "strings" "github.com/go-kratos/kratos/v2/config" "github.com/go-kratos/kratos/v2/log" @@ -23,14 +25,38 @@ type customChangeListener struct { func (c *customChangeListener) onChange(namespace string, changes map[string]*storage.ConfigChange) []*config.KeyValue { kv := make([]*config.KeyValue, 0, 2) - value, err := c.apollo.client.GetConfigCache(namespace).Get("content") + if strings.Contains(namespace, ".") && !strings.Contains(namespace, "properties") && + (format(namespace) == "yaml" || format(namespace) == "yaml" || format(namespace) == "json") { + value, err := c.apollo.client.GetConfigCache(namespace).Get("content") + if err != nil { + log.Warnw("apollo get config failed", "err", err) + } + kv = append(kv, &config.KeyValue{ + Key: namespace, + Value: []byte(value.(string)), + Format: format(namespace), + }) + + return kv + } + + next := make(map[string]interface{}) + + for key, change := range changes { + resolve(genKey(namespace, key), change.NewValue, next) + } + + f := format(namespace) + codec := encoding.GetCodec(f) + val, err := codec.Marshal(next) if err != nil { - log.Warnw("apollo get config failed", "err", err) + log.Warnf("apollo could not handle namespace %s: %v", namespace, err) + return nil } kv = append(kv, &config.KeyValue{ Key: namespace, - Value: []byte(value.(string)), - Format: format(namespace), + Value: val, + Format: f, }) return kv