Add support for flag overwriting

Signed-off-by: Valery Piashchynski <[email protected]>

6 files changed, 220 insertions, 2 deletions

diff --git a/pkg/pool/supervisor_test.go b/pkg/pool/supervisor_test.go
index cbe9f5cb..85af4672 100644
--- a/pkg/pool/supervisor_test.go
+++ b/pkg/pool/supervisor_test.go
@@ -53,7 +53,7 @@ func TestSupervisedPool_Exec(t *testing.T) {
 					assert.NotNil(t, s)
 					// since this is soft limit, double max memory limit watch
 					if (s.MemoryUsage / MB) > cfgSupervised.Supervisor.MaxWorkerMemory*2 {
-						assert.Fail(t, "max memory reached")
+						assert.Fail(t, "max memory reached, worker still alive")
 					}
 				}
 			}
@@ -61,7 +61,7 @@ func TestSupervisedPool_Exec(t *testing.T) {
 	}()
 
 	for i := 0; i < 100; i++ {
-		time.Sleep(time.Millisecond * 50)
+		time.Sleep(time.Millisecond * 100)
 		_, err = p.Exec(payload.Payload{
 			Context: []byte(""),
 			Body:    []byte("foo"),
diff --git a/plugins/config/plugin.go b/plugins/config/plugin.go
index ce2baa85..ddb7fe88 100755
--- a/plugins/config/plugin.go
+++ b/plugins/config/plugin.go
@@ -2,6 +2,7 @@ package config
 
 import (
 	"bytes"
+	"fmt"
 	"strings"
 
 	"github.com/spf13/viper"
@@ -14,6 +15,9 @@ type Viper struct {
 	Prefix    string
 	Type      string
 	ReadInCfg []byte
+	// user defined Flags in the form of <option>.<key> = <value>
+	// which overwrites initial config key
+	Flags []string
 }
 
 // Inits config provider.
@@ -40,6 +44,23 @@ func (v *Viper) Init() error {
 	v.viper.SetConfigFile(v.Path)
 	v.viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
 
+	err := v.viper.ReadInConfig()
+	if err != nil {
+		return errors.E(op, err)
+	}
+
+	// override config Flags
+	if len(v.Flags) > 0 {
+		for _, f := range v.Flags {
+			key, val, err := parseFlag(f)
+			if err != nil {
+				return errors.E(op, err)
+			}
+
+			v.viper.Set(key, val)
+		}
+	}
+
 	return v.viper.ReadInConfig()
 }
 
@@ -82,3 +103,25 @@ func (v *Viper) Get(name string) interface{} {
 func (v *Viper) Has(name string) bool {
 	return v.viper.IsSet(name)
 }
+
+func parseFlag(flag string) (string, string, error) {
+	const op = errors.Op("parse_flag")
+	if !strings.Contains(flag, "=") {
+		return "", "", errors.E(op, errors.Errorf("invalid flag `%s`", flag))
+	}
+
+	parts := strings.SplitN(strings.TrimLeft(flag, " \"'`"), "=", 2)
+
+	return strings.Trim(parts[0], " \n\t"), parseValue(strings.Trim(parts[1], " \n\t")), nil
+}
+
+func parseValue(value string) string {
+	escape := []rune(value)[0]
+
+	if escape == '"' || escape == '\'' || escape == '`' {
+		value = strings.Trim(value, string(escape))
+		value = strings.ReplaceAll(value, fmt.Sprintf("\\%s", string(escape)), string(escape))
+	}
+
+	return value
+}
diff --git a/tests/docker-compose-full.yaml b/tests/docker-compose-full.yaml
new file mode 100644
index 00000000..1ccc34f6
--- /dev/null
+++ b/tests/docker-compose-full.yaml
@@ -0,0 +1,49 @@
+version: '3.5'
+
+services:
+  postgresql:
+    container_name: temporal-postgresql
+    image: postgres:13.1
+    environment:
+      POSTGRES_PASSWORD: temporal
+      POSTGRES_USER: temporal
+    ports:
+      - 5432:5432
+
+  temporal:
+    container_name: temporal
+    image: temporalio/auto-setup:1.6.3
+    depends_on:
+      - postgresql
+    environment:
+      - DB=postgresql
+      - DB_PORT=5432
+      - POSTGRES_USER=temporal
+      - POSTGRES_PWD=temporal
+      - POSTGRES_SEEDS=postgresql
+      - DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/development.yaml
+    ports:
+      - 7233:7233
+    volumes:
+      - ./dynamicconfig:/etc/temporal/config/dynamicconfig
+
+  temporal-admin-tools:
+    container_name: temporal-admin-tools
+    image: temporalio/admin-tools:1.6.3
+    depends_on:
+      - temporal
+    environment:
+      - TEMPORAL_CLI_ADDRESS=temporal:7233
+    stdin_open: true
+    tty: true
+
+  temporal-web:
+    container_name: temporal-web
+    image: temporalio/web:1.6.2
+    depends_on:
+      - temporal
+    environment:
+      - TEMPORAL_GRPC_ENDPOINT=temporal:7233
+      - TEMPORAL_PERMIT_WRITE_API=true
+    ports:
+      - 8088:8088
diff --git a/tests/plugins/config/.rr.yaml b/tests/plugins/config/.rr.yaml
index a6e80921..f449dcf3 100755
--- a/tests/plugins/config/.rr.yaml
+++ b/tests/plugins/config/.rr.yaml
@@ -1,6 +1,10 @@
 rpc:
   listen: tcp://localhost:6060
 
+logs:
+  mode: development
+  level: error
+
 reload:
   interval: 1s
   patterns: [".php"]
diff --git a/tests/plugins/config/config_test.go b/tests/plugins/config/config_test.go
index 364960db..609b62ef 100755
--- a/tests/plugins/config/config_test.go
+++ b/tests/plugins/config/config_test.go
@@ -8,6 +8,8 @@ import (
 
 	endure "github.com/spiral/endure/pkg/container"
 	"github.com/spiral/roadrunner/v2/plugins/config"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+	"github.com/spiral/roadrunner/v2/plugins/rpc"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -62,3 +64,73 @@ func TestViperProvider_Init(t *testing.T) {
 		}
 	}
 }
+
+func TestConfigOverwriteFail(t *testing.T) {
+	container, err := endure.NewContainer(nil, endure.RetryOnFail(false), endure.SetLogLevel(endure.ErrorLevel))
+	if err != nil {
+		t.Fatal(err)
+	}
+	vp := &config.Viper{}
+	vp.Path = ".rr.yaml"
+	vp.Prefix = "rr"
+	vp.Flags = []string{"rpc.listen=tcp//not_exist"}
+
+	err = container.RegisterAll(
+		&logger.ZapLogger{},
+		&rpc.Plugin{},
+		vp,
+		&Foo2{},
+	)
+	assert.NoError(t, err)
+
+	err = container.Init()
+	assert.Error(t, err)
+}
+
+func TestConfigOverwriteValid(t *testing.T) {
+	container, err := endure.NewContainer(nil, endure.RetryOnFail(false), endure.SetLogLevel(endure.ErrorLevel))
+	if err != nil {
+		t.Fatal(err)
+	}
+	vp := &config.Viper{}
+	vp.Path = ".rr.yaml"
+	vp.Prefix = "rr"
+	vp.Flags = []string{"rpc.listen=tcp://localhost:6061"}
+
+	err = container.RegisterAll(
+		&logger.ZapLogger{},
+		&rpc.Plugin{},
+		vp,
+		&Foo2{},
+	)
+	assert.NoError(t, err)
+
+	err = container.Init()
+	assert.NoError(t, err)
+
+	errCh, err := container.Serve()
+	assert.NoError(t, err)
+
+	// stop by CTRL+C
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt)
+
+	tt := time.NewTicker(time.Second * 3)
+	defer tt.Stop()
+
+	for {
+		select {
+		case e := <-errCh:
+			assert.NoError(t, e.Error)
+			assert.NoError(t, container.Stop())
+			return
+		case <-c:
+			er := container.Stop()
+			assert.NoError(t, er)
+			return
+		case <-tt.C:
+			assert.NoError(t, container.Stop())
+			return
+		}
+	}
+}
diff --git a/tests/plugins/config/plugin2.go b/tests/plugins/config/plugin2.go
new file mode 100755
index 00000000..0fea9007
--- /dev/null
+++ b/tests/plugins/config/plugin2.go
@@ -0,0 +1,50 @@
+package config
+
+import (
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/plugins/config"
+)
+
+type Foo2 struct {
+	configProvider config.Configurer
+}
+
+// Depends on S2 and DB (S3 in the current case)
+func (f *Foo2) Init(p config.Configurer) error {
+	f.configProvider = p
+	return nil
+}
+
+func (f *Foo2) Serve() chan error {
+	const op = errors.Op("foo_plugin_serve")
+	errCh := make(chan error, 1)
+
+	r := &ReloadConfig{}
+	err := f.configProvider.UnmarshalKey("reload", r)
+	if err != nil {
+		errCh <- err
+	}
+
+	if len(r.Patterns) == 0 {
+		errCh <- errors.E(op, errors.Str("should be at least one pattern, but got 0"))
+		return errCh
+	}
+
+	var allCfg AllConfig
+	err = f.configProvider.Unmarshal(&allCfg)
+	if err != nil {
+		errCh <- errors.E(op, errors.Str("should be at least one pattern, but got 0"))
+		return errCh
+	}
+
+	if allCfg.RPC.Listen != "tcp://localhost:6061" {
+		errCh <- errors.E(op, errors.Str("RPC.Listen should be overwritten"))
+		return errCh
+	}
+
+	return errCh
+}
+
+func (f *Foo2) Stop() error {
+	return nil
+}