Repository: lf-edge/ekuiper
Pull request #2895 (2024-06-03 03:51:46)

feat: support rewrite url/token conf

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `83.07692%` with `11 lines` in your changes are missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@77afce2`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/server/bump/bump2.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?src=pr&el=tree&filepath=internal%2Fserver%2Fbump%2Fbump2.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL2J1bXAvYnVtcDIuZ28=) | 86.67% | [4 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/bump/bump\_manager.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?src=pr&el=tree&filepath=internal%2Fserver%2Fbump%2Fbump_manager.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL2J1bXAvYnVtcF9tYW5hZ2VyLmdv) | 33.33% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/connection/connection\_pool.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?src=pr&el=tree&filepath=internal%2Fio%2Fconnection%2Fconnection_pool.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vY29ubmVjdGlvbi9jb25uZWN0aW9uX3Bvb2wuZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2895   +/-   ##
============================================
  Coverage             ?   67.84%           
============================================
  Files                ?      294           
  Lines                ?    28179           
  Branches             ?        0           
============================================
  Hits                 ?    19116           
  Misses               ?     7360           
  Partials             ?     1703           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2895?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/conf/yaml_config_storage.go b/internal/conf/yaml_config_storage.go
index c78452a3..4bd0d5a5 100644
--- a/internal/conf/yaml_config_storage.go
+++ b/internal/conf/yaml_config_storage.go
@@ -219,8 +219,8 @@ func DropCfgKeyFromStorage(typ string, plugin string, confKey string) error {
 	return delCfgKeyInStorage(key)
 }
 
-// GotCfgFromKVStorage ...
-func GotCfgFromKVStorage(typ string, plugin string, confKey string) (map[string]map[string]interface{}, error) {
+// GetCfgFromKVStorage ...
+func GetCfgFromKVStorage(typ string, plugin string, confKey string) (map[string]map[string]interface{}, error) {
 	key := buildKey(typ, plugin, confKey)
 	kvStorage, err := getKVStorage()
 	if err != nil {
diff --git a/internal/io/connection/connection_pool.go b/internal/io/connection/connection_pool.go
index 556a233e..a2ff1ce3 100644
--- a/internal/io/connection/connection_pool.go
+++ b/internal/io/connection/connection_pool.go
@@ -210,7 +210,7 @@ func InitConnectionManager() error {
 	if conf.IsTesting {
 		return nil
 	}
-	cfgs, err := conf.GotCfgFromKVStorage("connections", "", "")
+	cfgs, err := conf.GetCfgFromKVStorage("connections", "", "")
 	if err != nil {
 		return err
 	}
diff --git a/internal/server/bump/bump2.go b/internal/server/bump/bump2.go
new file mode 100644
index 00000000..bfd5eb1e
--- /dev/null
+++ b/internal/server/bump/bump2.go
@@ -0,0 +1,127 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed 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 bump
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/pingcap/failpoint"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+)
+
+func bumpFrom1TO2() error {
+	err := rewriteConfiguration("sources")
+	if err != nil {
+		return err
+	}
+	err = rewriteConfiguration("sinks")
+	if err != nil {
+		return err
+	}
+	err = rewriteConfiguration("connections")
+	return err
+}
+
+func rewriteConfiguration(typ string) error {
+	failpoint.Inject("rewriteErr", func(val failpoint.Value) {
+		switch val.(int) {
+		case 1:
+			if typ == "sources" {
+				failpoint.Return(fmt.Errorf("rewriteErr"))
+			}
+		case 2:
+			if typ == "sinks" {
+				failpoint.Return(fmt.Errorf("rewriteErr"))
+			}
+		case 3:
+			if typ == "connections" {
+				failpoint.Return(fmt.Errorf("rewriteErr"))
+			}
+		}
+	})
+
+	props, err := conf.GetCfgFromKVStorage(typ, "", "")
+	if err != nil {
+		return err
+	}
+	for key, prop := range props {
+		t, pluginTyp, id, ok := extractKey(key)
+		if !ok {
+			continue
+		}
+		rewrite := false
+		switch pluginTyp {
+		case "sql":
+			rewrite = true
+			prop = replaceURL(prop)
+		case "influx2":
+			rewrite = true
+			prop = replaceToken(prop)
+		case "kafka":
+			rewrite = true
+			prop = replaceSaslPassword(prop)
+		}
+		if rewrite {
+			err := conf.WriteCfgIntoKVStorage(t, pluginTyp, id, prop)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func extractKey(key string) (string, string, string, bool) {
+	names := strings.Split(key, ".")
+	if len(names) != 3 {
+		return "", "", "", false
+	}
+	return names[0], names[1], names[2], true
+}
+
+func replaceToken(props map[string]interface{}) map[string]interface{} {
+	for key, value := range props {
+		if key == "token" {
+			props["password"] = value
+			delete(props, "token")
+			break
+		}
+	}
+	return props
+}
+
+func replaceURL(props map[string]interface{}) map[string]interface{} {
+	for key, value := range props {
+		if key == "url" {
+			props["dburl"] = value
+			delete(props, "url")
+			break
+		}
+	}
+	return props
+}
+
+func replaceSaslPassword(prop map[string]interface{}) map[string]interface{} {
+	for key, value := range prop {
+		if key == "saslPassword" {
+			prop["password"] = value
+			delete(prop, "saslPassword")
+			break
+		}
+	}
+	return prop
+}
diff --git a/internal/server/bump/bump_manager.go b/internal/server/bump/bump_manager.go
index 2c6a0039..a760aad5 100644
--- a/internal/server/bump/bump_manager.go
+++ b/internal/server/bump/bump_manager.go
@@ -30,7 +30,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/kv"
 )
 
-const currentVersion = 1
+const currentVersion = 2
 
 var GlobalBumpManager *BumpManager
 
@@ -77,6 +77,10 @@ func BumpToCurrentVersion(dataDir string) error {
 			if err := bumpFrom0To1(dataDir); err != nil {
 				return err
 			}
+		case 2:
+			if err := bumpFrom1TO2(); err != nil {
+				return err
+			}
 		}
 	}
 	return nil


Test patch:
diff --git a/internal/server/bump/bump2_test.go b/internal/server/bump/bump2_test.go
new file mode 100644
index 00000000..e04edf6f
--- /dev/null
+++ b/internal/server/bump/bump2_test.go
@@ -0,0 +1,118 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed 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 bump
+
+import (
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+)
+
+func TestBumpFrom1To2(t *testing.T) {
+	conf.IsTesting = true
+	prepareData(t)
+	err := bumpFrom1TO2()
+	require.NoError(t, err)
+	assertData(t)
+}
+
+func prepareData(t *testing.T) {
+	err := conf.WriteCfgIntoKVStorage("sinks", "influx2", "a", map[string]interface{}{
+		"token": "123",
+	})
+	require.NoError(t, err)
+
+	err = conf.WriteCfgIntoKVStorage("sinks", "sql", "a", map[string]interface{}{
+		"url": "123",
+	})
+	require.NoError(t, err)
+
+	err = conf.WriteCfgIntoKVStorage("sources", "sql", "a", map[string]interface{}{
+		"url": "123",
+	})
+	require.NoError(t, err)
+
+	err = conf.WriteCfgIntoKVStorage("sources", "kafka", "a", map[string]interface{}{
+		"saslPassword": "123",
+	})
+	require.NoError(t, err)
+}
+
+func assertData(t *testing.T) {
+	props, err := conf.GetCfgFromKVStorage("sources", "sql", "a")
+	require.NoError(t, err)
+	require.NotNil(t, props)
+	prop, ok := props["sources.sql.a"]
+	require.True(t, ok)
+
+	_, ok = prop["url"]
+	require.False(t, ok)
+	v, ok := prop["dburl"]
+	require.True(t, ok)
+	require.Equal(t, "123", v)
+
+	props, err = conf.GetCfgFromKVStorage("sources", "kafka", "a")
+	require.NoError(t, err)
+	require.NotNil(t, props)
+	prop, ok = props["sources.kafka.a"]
+	require.True(t, ok)
+
+	_, ok = prop["saslPassword"]
+	require.False(t, ok)
+	v, ok = prop["password"]
+	require.True(t, ok)
+	require.Equal(t, "123", v)
+
+	props, err = conf.GetCfgFromKVStorage("sinks", "influx2", "a")
+	require.NoError(t, err)
+	require.NotNil(t, prop)
+	prop, ok = props["sinks.influx2.a"]
+	require.True(t, ok)
+	_, ok = prop["token"]
+	require.False(t, ok)
+	v, ok = prop["password"]
+	require.True(t, ok)
+	require.Equal(t, "123", v)
+
+	props, err = conf.GetCfgFromKVStorage("sinks", "sql", "a")
+	require.NoError(t, err)
+	require.NotNil(t, prop)
+	prop, ok = props["sinks.sql.a"]
+	require.True(t, ok)
+	_, ok = prop["url"]
+	require.False(t, ok)
+	v, ok = prop["dburl"]
+	require.True(t, ok)
+	require.Equal(t, "123", v)
+}
+
+func TestBump2Err(t *testing.T) {
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/rewriteErr", "return(1)")
+	err := bumpFrom1TO2()
+	require.Error(t, err)
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/rewriteErr", "return(2)")
+	err = bumpFrom1TO2()
+	require.Error(t, err)
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/rewriteErr", "return(3)")
+	err = bumpFrom1TO2()
+	require.Error(t, err)
+
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/bump/rewriteErr")
+}
diff --git a/internal/server/bump/bump_test.go b/internal/server/bump/bump_test.go
index 275a8c9b..545d94ea 100644
--- a/internal/server/bump/bump_test.go
+++ b/internal/server/bump/bump_test.go
@@ -63,12 +63,12 @@ func testBumpVersion(t *testing.T) {
 }
 
 func assertBumpVersion1Data(t *testing.T, typ string) {
-	d, err := conf.GotCfgFromKVStorage(typ, "mqtt", "conf1")
+	d, err := conf.GetCfgFromKVStorage(typ, "mqtt", "conf1")
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"p1": 1,
 	}, d[fmt.Sprintf("%s.mqtt.conf1", typ)])
-	d, err = conf.GotCfgFromKVStorage(typ, "mqtt", "conf2")
+	d, err = conf.GetCfgFromKVStorage(typ, "mqtt", "conf2")
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"p2": 2,