[OSPP]Support Kubernetes ConfigMap for Apollo java, golang client #79

README.md

@@ -19,7 +19,7 @@ Agollo - Go Client for Apollo
 * 实时同步配置
 * 灰度配置
 * 延迟加载（运行时）namespace
-* 客户端，配置文件容灾
+* 客户端，多种配置文件容灾（本地文件，k8s中configmap）
 * 自定义日志，缓存组件
 * 支持配置访问秘钥
 

client.go

@@ -49,7 +49,7 @@ const separator = ","
 func init() {
 	extension.SetCacheFactory(&memory.DefaultCacheFactory{})
 	extension.SetLoadBalance(&roundrobin.RoundRobin{})
-	extension.SetFileHandler(&jsonFile.FileHandler{})
+	extension.AddFileHandler(&jsonFile.FileHandler{}, 10)
 	extension.SetHTTPAuth(&sign.AuthSignature{})
 
 	// file parser

client_test.go

@@ -247,7 +247,7 @@ func createApolloConfigWithJSON(b []byte) (o interface{}, err error) {
 }
 
 func checkBackupFile(client *internalClient, t *testing.T) {
-	newConfig, e := extension.GetFileHandler().LoadConfigFile(client.appConfig.GetBackupConfigPath(), client.appConfig.AppID, testDefaultNamespace)
+	newConfig, e := extension.GetFileHandlers()[0].LoadConfigFile(client.appConfig.GetBackupConfigPath(), client.appConfig.AppID, testDefaultNamespace, client.appConfig.Cluster)
 	Assert(t, newConfig, NotNilVal())
 	Assert(t, e, NilVal())
 	Assert(t, newConfig.Configurations, NotNilVal())

component/remote/async.go

@@ -147,7 +147,8 @@ func toApolloConfig(resBody []byte) ([]*config.Notification, error) {
 func loadBackupConfig(namespace string, appConfig config.AppConfig) []*config.ApolloConfig {
 	apolloConfigs := make([]*config.ApolloConfig, 0)
 	config.SplitNamespaces(namespace, func(namespace string) {
-		c, err := extension.GetFileHandler().LoadConfigFile(appConfig.BackupConfigPath, appConfig.AppID, namespace)
+		c, err := loadBackupConfiguration(appConfig, namespace, appConfig.Cluster)
+
 		if err != nil {
 			log.Errorf("LoadConfigFile error, error: %v", err)
 			return
@@ -160,6 +161,27 @@ func loadBackupConfig(namespace string, appConfig config.AppConfig) []*config.Ap
 	return apolloConfigs
 }
 
+// 从按优先级排好序的所有备份文件来源中尝试加载配置
+func loadBackupConfiguration(appConfig config.AppConfig, namespace string, cluster string) (*config.ApolloConfig, error) {
+	log.Debugf("Loading backup config")
+
+	for _, handler := range extension.GetFileHandlers() {
+		c, err := handler.LoadConfigFile(appConfig.BackupConfigPath, appConfig.AppID, namespace, cluster)
+		if err == nil && c != nil {
+			log.Infof("The backup file was successfully loaded. config: %s", c)
+			return c, nil
+		}
+		// 改进日志便于排查
+		if err != nil {
+			log.Warnf("Failed to load config with handler %T: %v", handler, err)
+		} else {
+			log.Warnf("Handler %T returned nil config", handler)
+		}
+	}
+
+	return nil, fmt.Errorf("failed to load backup configuration for namespace %s", namespace)
+}
+
 func createApolloConfigWithJSON(b []byte, callback http.CallBack) (o interface{}, err error) {
 	apolloConfig := &config.ApolloConfig{}
 	err = json.Unmarshal(b, apolloConfig)

configmap/handler.go

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package configmap
+
+import (
+	"github.com/apolloconfig/agollo/v4/component/log"
+	"github.com/apolloconfig/agollo/v4/env/config"
+)
+
+var ApolloConfigCache = "apollo-configcache-"
+
+type ConfigMapHandler struct {
+	k8sManager *K8sManager
+}
+
+// NewConfigMapHandler 是 ConfigMapHandler 的构造函数
+func NewConfigMapHandler(k8sManager *K8sManager) *ConfigMapHandler {
+	return &ConfigMapHandler{
+		k8sManager: k8sManager,
+	}
+}
+
+// WriteConfigFile write apollo config to configmap
+func (c *ConfigMapHandler) WriteConfigFile(config *config.ApolloConfig, configPath string) error {
+	configMapName := ApolloConfigCache + config.AppID
+	key := config.Cluster + "-" + config.NamespaceName
+	err := c.k8sManager.SetConfigMapWithRetry(configMapName, key, config)
+	if err != nil {
+		log.Errorf("Failed to write ConfigMap %s : %v", configMapName, err)
+		return err
+	}
+	return nil
+}
+
+func (c *ConfigMapHandler) GetConfigFile(configDir string, appID string, namespace string) string {
+	return ""
+}
+
+// LoadConfigFile load ApolloConfig from configmap
+func (c *ConfigMapHandler) LoadConfigFile(configPath string, appID string, namespace string, cluster string) (*config.ApolloConfig, error) {
+	var apolloConfig = &config.ApolloConfig{}
+	var err error
+
+	if cluster == "" {
+		cluster = "default"
+		log.Infof("cluster is empty, use default cluster")
+	}
+	configMapName := ApolloConfigCache + appID
+	key := cluster + "-" + namespace
+	// 这里把json转回ApolloConfig, ReleaseKey字段会丢失
+	apolloConfig.Configurations, err = c.k8sManager.GetConfigMap(configMapName, key)
+
+	apolloConfig.AppID = appID
+	apolloConfig.Cluster = cluster
+	apolloConfig.NamespaceName = namespace
+	return apolloConfig, err
+}

configmap/handler_test.go

@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package configmap
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/apolloconfig/agollo/v4/env/config"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+	"testing"
+)
+
+const (
+	appId        = "testAppId"
+	k8sNamespace = "testConfigMapNamespace"
+	cluster      = "testCluster"
+	namespace    = "testNamespace"
+)
+
+var testData = map[string]interface{}{
+	"stringKey": "stringValue",
+	"intKey":    123,
+	"boolKey":   true,
+	"sliceKey":  []interface{}{1, 2, 3},
+	"mapKey": map[string]interface{}{
+		"nestedStringKey": "nestedStringValue",
+		"nestedIntKey":    456,
+	},
+}
+
+var appConfig = config.AppConfig{
+	AppID:         appId,
+	NamespaceName: namespace,
+	Cluster:       cluster,
+}
+
+func TestStore_LoadConfigMap(t *testing.T) {
+	// 初始化fake clientset
+	clientset := fake.NewSimpleClientset()
+	jsonData, err := json.Marshal(testData)
+	if err != nil {
+		fmt.Println("Error marshalling map to JSON:", err)
+		return
+	}
+
+	// 创建一个ConfigMap对象
+	configMap := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "apollo-configcache-" + appId,
+			Namespace: k8sNamespace,
+		},
+		Data: map[string]string{
+			cluster + "-" + namespace: string(jsonData),
+		},
+	}
+
+	// 使用fake clientset创建ConfigMap
+	_, err = clientset.CoreV1().ConfigMaps(k8sNamespace).Create(context.TODO(), configMap, metav1.CreateOptions{})
+	assert.NoError(t, err)
+
+	// 初始化handler，注入fake clientset
+	handler := NewConfigMapHandler(&K8sManager{
+		clientSet:    clientset,
+		k8sNamespace: k8sNamespace,
+	})
+
+	// 执行
+	loadedConfig, err := handler.LoadConfigFile("", appConfig.AppID, namespace, cluster)
+	assert.NoError(t, err)
+
+	// 测试LoadConfigMap方法
+	loadedJson, _ := json.Marshal(loadedConfig.Configurations)
+	assert.NotNil(t, loadedConfig)
+	assert.Equal(t, jsonData, loadedJson)
+}
+
+func TestStore_WriteConfigMap(t *testing.T) {
+	// 初始化fake clientset
+	clientset := fake.NewSimpleClientset()
+
+	var err error
+	var key = cluster + "-" + namespace
+	jsonData, err := json.Marshal(testData)
+	if err != nil {
+		fmt.Println("Error marshalling map to JSON:", err)
+		return
+	}
+
+	// 初始化handler，注入fake clientset
+	handler := NewConfigMapHandler(&K8sManager{
+		clientSet:    clientset,
+		k8sNamespace: k8sNamespace,
+	})
+
+	// 反序列化到ApolloConfig
+	apolloConfig := &config.ApolloConfig{}
+	apolloConfig.Configurations = testData
+	apolloConfig.AppID = appId
+	apolloConfig.Cluster = cluster
+	apolloConfig.NamespaceName = namespace
+
+	// 测试WriteConfigMap方法
+	err = handler.WriteConfigFile(apolloConfig, "")
+	assert.NoError(t, err)
+
+	// 验证ConfigMap是否被正确创建或更新
+	configMap, err := clientset.CoreV1().ConfigMaps(k8sNamespace).Get(context.TODO(), "apollo-configcache-"+appId, metav1.GetOptions{})
+	loadedJson, _ := configMap.Data[key]
+
+	var configurations map[string]interface{}
+	err = json.Unmarshal([]byte(loadedJson), &configurations)
+
+	assert.NoError(t, err)
+	assert.NotNil(t, configMap)
+	assert.Equal(t, jsonData, []byte(loadedJson))
+}
+
+func TestStore_LoadConfigMap_EmptyConfigMap(t *testing.T) {
+	// 初始化fake clientset
+	clientset := fake.NewSimpleClientset()
+
+	// 初始化handler，注入fake clientset
+	handler := NewConfigMapHandler(&K8sManager{
+		clientSet:    clientset,
+		k8sNamespace: k8sNamespace,
+	})
+
+	// 执行
+	loadedConfig, err := handler.LoadConfigFile("", appId, namespace, cluster)
+	assert.Nil(t, loadedConfig.Configurations)
+	assert.Error(t, err)
+}
+
+func TestStore_LoadConfigMap_InvalidJSON(t *testing.T) {
+	// 初始化fake clientset
+	clientset := fake.NewSimpleClientset()
+
+	// 初始化handler，注入fake clientset
+	handler := NewConfigMapHandler(&K8sManager{
+		clientSet:    clientset,
+		k8sNamespace: k8sNamespace,
+	})
+	// 创建一个ConfigMap对象
+	configMap := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "apollo-configcache-" + appId,
+			Namespace: k8sNamespace,
+		},
+		Data: map[string]string{
+			cluster + "-" + namespace: "invalid json",
+		},
+	}
+	_, err := clientset.CoreV1().ConfigMaps(k8sNamespace).Create(context.TODO(), configMap, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	loadedConfig, err := handler.LoadConfigFile("", appId, namespace, cluster)
+	assert.Nil(t, loadedConfig.Configurations)
+	assert.Error(t, err)
+}