mirror of
https://github.com/go-kratos/kratos.git
synced 2025-02-15 13:53:35 +02:00
fix(config): apollo get the original configuration file directly (#2100)
This commit is contained in:
包子
GitHub
parent
123fc1e6c8
commit
4b7afe52af
@ -4,7 +4,6 @@ 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"
|
||||
@ -104,61 +103,9 @@ func NewSource(opts ...Option) config.Source {
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
return &apollo{client: client, opt: &op}
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func format(ns string) string {
|
||||
arr := strings.Split(ns, ".")
|
||||
if len(arr) <= 1 {
|
||||
@ -173,26 +120,15 @@ func (e *apollo) load() []*config.KeyValue {
|
||||
namespaces := strings.Split(e.opt.namespace, ",")
|
||||
|
||||
for _, ns := range namespaces {
|
||||
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
|
||||
})
|
||||
|
||||
// serialize the namespace content KeyValue into bytes.
|
||||
f := format(ns)
|
||||
codec := encoding.GetCodec(f)
|
||||
val, err := codec.Marshal(next)
|
||||
value, err := e.client.GetConfigCache(ns).Get("content")
|
||||
if err != nil {
|
||||
log.Warnf("apollo could not handle namespace %s: %v", ns, err)
|
||||
continue
|
||||
log.Warnw("apollo get config failed", "err", err)
|
||||
}
|
||||
|
||||
// serialize the namespace content KeyValue into bytes.
|
||||
kv = append(kv, &config.KeyValue{
|
||||
Key: ns,
|
||||
Value: val,
|
||||
Format: f,
|
||||
Value: []byte(value.(string)),
|
||||
Format: format(ns),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -1,189 +0,0 @@
|
||||
package apollo
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func Test_genKey(t *testing.T) {
|
||||
type args struct {
|
||||
ns string
|
||||
sub string
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "case 1",
|
||||
args: args{
|
||||
ns: "",
|
||||
sub: "has_no_ns",
|
||||
},
|
||||
want: "has_no_ns",
|
||||
},
|
||||
{
|
||||
name: "case 2",
|
||||
args: args{
|
||||
ns: "ns.ext",
|
||||
sub: "sub",
|
||||
},
|
||||
want: "ns.sub",
|
||||
},
|
||||
{
|
||||
name: "case 3",
|
||||
args: args{
|
||||
ns: "",
|
||||
sub: "",
|
||||
},
|
||||
want: "",
|
||||
},
|
||||
{
|
||||
name: "case 4",
|
||||
args: args{
|
||||
ns: "ns.ext",
|
||||
sub: "sub.sub2.sub3",
|
||||
},
|
||||
want: "ns.sub.sub2.sub3",
|
||||
},
|
||||
{
|
||||
name: "case 5",
|
||||
args: args{
|
||||
ns: "ns.more.ext",
|
||||
sub: "sub.sub2.sub3",
|
||||
},
|
||||
want: "ns.more.sub.sub2.sub3",
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := genKey(tt.args.ns, tt.args.sub); got != tt.want {
|
||||
t.Errorf("genKey() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_format(t *testing.T) {
|
||||
type args struct {
|
||||
ns string
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "case 0",
|
||||
args: args{
|
||||
ns: "ns.yaml",
|
||||
},
|
||||
want: "yaml",
|
||||
},
|
||||
{
|
||||
name: "case 1",
|
||||
args: args{
|
||||
ns: "ns",
|
||||
},
|
||||
want: "json",
|
||||
},
|
||||
{
|
||||
name: "case 2",
|
||||
args: args{
|
||||
ns: "ns.more.json",
|
||||
},
|
||||
want: "json",
|
||||
},
|
||||
{
|
||||
name: "case 3",
|
||||
args: args{
|
||||
ns: "",
|
||||
},
|
||||
want: "json",
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := format(tt.args.ns); got != tt.want {
|
||||
t.Errorf("format() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_convertProperties(t *testing.T) {
|
||||
type args struct {
|
||||
key string
|
||||
value interface{}
|
||||
target map[string]interface{}
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want map[string]interface{}
|
||||
}{
|
||||
{
|
||||
name: "case 0",
|
||||
args: args{
|
||||
key: "application.name",
|
||||
value: "app name",
|
||||
target: map[string]interface{}{},
|
||||
},
|
||||
want: map[string]interface{}{
|
||||
"application": map[string]interface{}{
|
||||
"name": "app name",
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "case 1",
|
||||
args: args{
|
||||
key: "application",
|
||||
value: []string{"1", "2", "3"},
|
||||
target: map[string]interface{}{},
|
||||
},
|
||||
want: map[string]interface{}{
|
||||
"application": []string{"1", "2", "3"},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
resolve(tt.args.key, tt.args.value, tt.args.target)
|
||||
if !reflect.DeepEqual(tt.args.target, tt.want) {
|
||||
t.Errorf("convertProperties() = %v, want %v", tt.args.target, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_convertProperties_duplicate(t *testing.T) {
|
||||
target := map[string]interface{}{}
|
||||
resolve("application.name", "name", target)
|
||||
_, ok := target["application"]
|
||||
if !reflect.DeepEqual(ok, true) {
|
||||
t.Errorf("ok = %v, want %v", ok, true)
|
||||
}
|
||||
_, ok = target["application"].(map[string]interface{})["name"]
|
||||
if !reflect.DeepEqual(ok, true) {
|
||||
t.Errorf("ok = %v, want %v", ok, true)
|
||||
}
|
||||
if !reflect.DeepEqual(target["application"].(map[string]interface{})["name"], "name") {
|
||||
t.Errorf("target[\"application\"][\"name\"] = %v, want %v", target["application"].(map[string]interface{})["name"], "name")
|
||||
}
|
||||
|
||||
// cause duplicate, the oldest value will be kept
|
||||
resolve("application.name.first", "first name", target)
|
||||
_, ok = target["application"]
|
||||
if !reflect.DeepEqual(ok, true) {
|
||||
t.Errorf("ok = %v, want %v", ok, true)
|
||||
}
|
||||
_, ok = target["application"].(map[string]interface{})["name"]
|
||||
if !reflect.DeepEqual(ok, true) {
|
||||
t.Errorf("ok = %v, want %v", ok, true)
|
||||
}
|
||||
if !reflect.DeepEqual(target["application"].(map[string]interface{})["name"], "name") {
|
||||
t.Errorf("target[\"application\"][\"name\"] = %v, want %v", target["application"].(map[string]interface{})["name"], "name")
|
||||
}
|
||||
}
|
||||
@ -1,10 +1,6 @@
|
||||
package apollo
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"gopkg.in/yaml.v3"
|
||||
|
||||
"github.com/apolloconfig/agollo/v4/constant"
|
||||
"github.com/apolloconfig/agollo/v4/extension"
|
||||
)
|
||||
@ -12,25 +8,13 @@ import (
|
||||
type jsonExtParser struct{}
|
||||
|
||||
func (parser jsonExtParser) Parse(configContent interface{}) (map[string]interface{}, error) {
|
||||
v, ok := configContent.(string)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
out := make(map[string]interface{}, 4)
|
||||
err := json.Unmarshal([]byte(v), &out)
|
||||
return out, err
|
||||
return map[string]interface{}{"content": configContent}, nil
|
||||
}
|
||||
|
||||
type yamlExtParser struct{}
|
||||
|
||||
func (parser yamlExtParser) Parse(configContent interface{}) (out map[string]interface{}, err error) {
|
||||
v, ok := configContent.(string)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
out = make(map[string]interface{}, 4)
|
||||
err = yaml.Unmarshal([]byte(v), &out)
|
||||
return
|
||||
return map[string]interface{}{"content": configContent}, nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
|
||||
@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
|
||||
"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/storage"
|
||||
@ -16,28 +15,20 @@ type watcher struct {
|
||||
}
|
||||
|
||||
type customChangeListener struct {
|
||||
in chan<- []*config.KeyValue
|
||||
in chan<- []*config.KeyValue
|
||||
apollo *apollo
|
||||
}
|
||||
|
||||
func (c *customChangeListener) onChange(namespace string, changes map[string]*storage.ConfigChange) []*config.KeyValue {
|
||||
kv := make([]*config.KeyValue, 0, 2)
|
||||
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)
|
||||
value, err := c.apollo.client.GetConfigCache(namespace).Get("content")
|
||||
if err != nil {
|
||||
log.Warnf("apollo could not handle namespace %s: %v", namespace, err)
|
||||
return nil
|
||||
log.Warnw("apollo get config failed", "err", err)
|
||||
}
|
||||
kv = append(kv, &config.KeyValue{
|
||||
Key: namespace,
|
||||
Value: val,
|
||||
Format: f,
|
||||
Value: []byte(value.(string)),
|
||||
Format: format(namespace),
|
||||
})
|
||||
|
||||
return kv
|
||||
@ -56,7 +47,7 @@ func (c *customChangeListener) OnNewestChange(changeEvent *storage.FullChangeEve
|
||||
|
||||
func newWatcher(a *apollo) (config.Watcher, error) {
|
||||
changeCh := make(chan []*config.KeyValue)
|
||||
listener := &customChangeListener{in: changeCh}
|
||||
listener := &customChangeListener{in: changeCh, apollo: a}
|
||||
a.client.AddChangeListener(listener)
|
||||
return &watcher{
|
||||
out: changeCh,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user