feat: rewrite duration in conf

internal/meta/yamlConfigMeta.go

@@ -239,21 +239,11 @@ func GetYamlConf(configOperatorKey, language string) (b []byte, err error) {
 	}
 }
 
-func replaceConfigurations(plgName string, cf YamlConfigurations) YamlConfigurations {
-	switch plgName {
-	case "sql":
-		for key, props := range cf {
-			replaced, newProps := replace.ReplacePropsDBURL(props)
-			if replaced {
-				cf[key] = newProps
-			}
-		}
-	default:
-		for key, props := range cf {
-			replaced, newProps := replace.ReplacePassword(props)
-			if replaced {
-				cf[key] = newProps
-			}
+func replaceConfigurations(plg string, cf YamlConfigurations) YamlConfigurations {
+	for key, props := range cf {
+		replaced, newProps := replace.ReplacePropsWithPlug(plg, props)
+		if replaced {
+			cf[key] = newProps
 		}
 	}
 	return cf

internal/server/bump/bump4.go

@@ -0,0 +1,54 @@
+// 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 (
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/replace"
+)
+
+func bumpFrom3TO4() error {
+	return rewriteReplacedProps()
+}
+
+func rewriteReplacedProps() error {
+	if err := rewritePlugProps("sources"); err != nil {
+		return err
+	}
+	if err := rewritePlugProps("sinks"); err != nil {
+		return err
+	}
+	if err := rewritePlugProps("connections"); err != nil {
+		return err
+	}
+	return nil
+}
+
+func rewritePlugProps(typ string) error {
+	keyProps, err := conf.GetCfgFromKVStorage(typ, "", "")
+	if err != nil {
+		return err
+	}
+	for key, props := range keyProps {
+		_, plug, confKey, valid := extractKey(key)
+		if valid {
+			changed, newProps := replace.ReplacePropsWithPlug(plug, props)
+			if changed {
+				return conf.WriteCfgIntoKVStorage(typ, plug, confKey, newProps)
+			}
+		}
+	}
+	return nil
+}

internal/server/bump/bump4_test.go

@@ -0,0 +1,41 @@
+// 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/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+)
+
+func TestBumpTo4(t *testing.T) {
+	conf.IsTesting = true
+	require.NoError(t, conf.WriteCfgIntoKVStorage("sources", "sql", "sql1", map[string]interface{}{
+		"url":      "123",
+		"cacheTtl": 1000,
+	}))
+	require.NoError(t, bumpFrom3TO4())
+	mm, err := conf.GetCfgFromKVStorage("sources", "sql", "sql1")
+	require.NoError(t, err)
+	for _, props := range mm {
+		require.Equal(t, map[string]interface{}{
+			"dburl":    "123",
+			"cacheTtl": "1s",
+		}, props)
+		break
+	}
+}

internal/server/bump/bump_manager.go

@@ -31,7 +31,7 @@
 )
 
 const (
-	currentVersion = 3
+	currentVersion = 4
 	bumpTable      = "eKuiperMeta_bump_version"
 )
 
@@ -100,6 +100,14 @@
 				return err
 			}
 			GlobalBumpManager.Version = 3
+		case 3:
+			if err := bumpFrom3TO4(); err != nil {
+				return err
+			}
+			if err := storeGlobalVersion(4); err != nil {
+				return err
+			}
+			GlobalBumpManager.Version = 4
 		}
 	}
 	return nil

internal/server/bump/bump_test.go

@@ -40,7 +40,7 @@ func TestBumpVersion(t *testing.T) {
 	}
 	require.NoError(t, InitBumpManager())
 	GlobalBumpManager.Version = 0
-	testBumpVersion(t, 3)
+	testBumpVersion(t, currentVersion)
 }
 
 func prepareBumpVersion(t *testing.T, dir string) {

internal/server/yaml_import_export.go

@@ -575,18 +575,9 @@ func replaceConfigurations(key string, props map[string]any) map[string]any {
 	if err != nil {
 		return props
 	}
-	switch plgName {
-	case "sql":
-		changed, newProps := replace.ReplacePropsDBURL(props)
-		if changed {
-			return newProps
-		}
-		return props
-	default:
-		changed, newProps := replace.ReplacePassword(props)
-		if changed {
-			return newProps
-		}
-		return props
+	changed, newProps := replace.ReplacePropsWithPlug(plgName, props)
+	if changed {
+		return newProps
 	}
+	return props
 }

pkg/replace/replace.go

@@ -2,11 +2,13 @@
 
 import (
 	"encoding/json"
+	"time"
 )
 
 var (
 	replaceURL      = []string{"url"}
 	replacePassword = []string{"saslPassword"}
+	replaceDuration = []string{"cacheTtl", "timeout", "expiration", "interval"}
 	replaceAction   = map[string]struct{}{
 		"kafka": {},
 		"sql":   {},
@@ -35,11 +37,10 @@
 			_, ok1 := replaceAction[actionTyp]
 			actionPropsMap, ok2 := actionProps.(map[string]interface{})
 			if ok1 && ok2 {
-				changed1, m1 := ReplacePassword(actionPropsMap)
-				changed2, m2 := ReplacePropsDBURL(m1)
-				if changed1 || changed2 {
+				replaced, newProps := ReplacePropsWithPlug(actionTyp, actionPropsMap)
+				if replaced {
 					changed = true
-					actionPropsMap = m2
+					actionPropsMap = newProps
 					actionMap[actionTyp] = actionPropsMap
 					actions[index] = actionMap
 				}
@@ -59,6 +60,18 @@
 	return string(got)
 }
 
+func WithDisableReplaceDburl() ReplacePropsOption {
+	return func(c *ReplacePropsConfig) {
+		c.DisableReplaceDbUrl = true
+	}
+}
+
+func WithDisableReplacePassword() ReplacePropsOption {
+	return func(c *ReplacePropsConfig) {
+		c.DisableReplacePassword = true
+	}
+}
+
 func ReplacePropsDBURL(props map[string]interface{}) (bool, map[string]interface{}) {
 	changed := false
 	for _, replaceWord := range replaceURL {
@@ -86,3 +99,64 @@
 	}
 	return changed, props
 }
+
+func ReplaceDuration(props map[string]interface{}) (bool, map[string]interface{}) {
+	changed := false
+	for _, replaceWord := range replaceDuration {
+		v, ok := props[replaceWord]
+		if ok {
+			intRaw, ok := v.(int)
+			if ok {
+				props[replaceWord] = (time.Duration(intRaw) * time.Millisecond).String()
+				changed = true
+				continue
+			}
+			int64Raw, ok := v.(int64)
+			if ok {
+				props[replaceWord] = (time.Duration(int64Raw) * time.Millisecond).String()
+				changed = true
+				continue
+			}
+		}
+	}
+	return changed, props
+}
+
+func ReplacePropsWithPlug(plug string, props map[string]interface{}) (bool, map[string]interface{}) {
+	switch plug {
+	case "sql":
+		return ReplacePropsWithOption(props)
+	default:
+		return ReplacePropsWithOption(props, WithDisableReplaceDburl())
+	}
+}
+
+func ReplacePropsWithOption(props map[string]interface{}, opts ...ReplacePropsOption) (bool, map[string]interface{}) {
+	ReplaceConfig := &ReplacePropsConfig{}
+	for _, opt := range opts {
+		opt(ReplaceConfig)
+	}
+	replaced := false
+	var changed bool
+	if !ReplaceConfig.DisableReplacePassword {
+		changed, props = ReplacePassword(props)
+		replaced = replaced || changed
+	}
+	if !ReplaceConfig.DisableReplaceDbUrl {
+		changed, props = ReplacePropsDBURL(props)
+		replaced = replaced || changed
+	}
+	if !ReplaceConfig.DisableReplaceDuration {
+		changed, props = ReplaceDuration(props)
+		replaced = replaced || changed
+	}
+	return replaced, props
+}
+
+type ReplacePropsOption func(c *ReplacePropsConfig)
+
+type ReplacePropsConfig struct {
+	DisableReplaceDbUrl    bool
+	DisableReplacePassword bool
+	DisableReplaceDuration bool
+}

pkg/replace/replace_test.go

@@ -117,3 +117,45 @@ func TestReplaceRuleJson(t *testing.T) {
 	got = ReplaceRuleJson(data, false)
 	require.Equal(t, data, got)
 }
+
+func TestReplaceDuration(t *testing.T) {
+	props := map[string]interface{}{
+		"cacheTtl": 1000,
+	}
+	changed, newProps := ReplaceDuration(props)
+	require.True(t, changed)
+	require.Equal(t, map[string]interface{}{
+		"cacheTtl": "1s",
+	}, newProps)
+	props = map[string]interface{}{
+		"cacheTtl": int64(1000),
+	}
+	changed, newProps = ReplaceDuration(props)
+	require.True(t, changed)
+	require.Equal(t, map[string]interface{}{
+		"cacheTtl": "1s",
+	}, newProps)
+}
+
+func TestRelacePropsPlug(t *testing.T) {
+	props := map[string]interface{}{
+		"cacheTtl": 1000,
+		"url":      "123",
+	}
+	changed, newProps := ReplacePropsWithPlug("", props)
+	require.True(t, changed)
+	require.Equal(t, map[string]interface{}{
+		"cacheTtl": "1s",
+		"url":      "123",
+	}, newProps)
+	props = map[string]interface{}{
+		"cacheTtl": 1000,
+		"url":      "123",
+	}
+	changed, newProps = ReplacePropsWithPlug("sql", props)
+	require.True(t, changed)
+	require.Equal(t, map[string]interface{}{
+		"cacheTtl": "1s",
+		"dburl":    "123",
+	}, newProps)
+}