Set dynamic config values and default search attribute cache as disabled

cmd/temporalite/main.go

@@ -5,6 +5,7 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
 	goLog "log"
 	"net"
@@ -13,6 +14,7 @@ import (
 
 	"github.com/urfave/cli/v2"
 	"go.temporal.io/server/common/config"
+	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/headers"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/temporal"
@@ -35,18 +37,19 @@ var (
 )
 
 const (
-	ephemeralFlag   = "ephemeral"
-	dbPathFlag      = "filename"
-	portFlag        = "port"
-	metricsPortFlag = "metrics-port"
-	uiPortFlag      = "ui-port"
-	headlessFlag    = "headless"
-	ipFlag          = "ip"
-	logFormatFlag   = "log-format"
-	logLevelFlag    = "log-level"
-	namespaceFlag   = "namespace"
-	pragmaFlag      = "sqlite-pragma"
-	configFlag      = "config"
+	ephemeralFlag          = "ephemeral"
+	dbPathFlag             = "filename"
+	portFlag               = "port"
+	metricsPortFlag        = "metrics-port"
+	uiPortFlag             = "ui-port"
+	headlessFlag           = "headless"
+	ipFlag                 = "ip"
+	logFormatFlag          = "log-format"
+	logLevelFlag           = "log-level"
+	namespaceFlag          = "namespace"
+	pragmaFlag             = "sqlite-pragma"
+	configFlag             = "config"
+	dynamicConfigValueFlag = "dynamic-config-value"
 )
 
 func init() {
@@ -146,6 +149,10 @@ func buildCLI() *cli.App {
 					EnvVars: []string{config.EnvKeyConfigDir},
 					Value:   "",
 				},
+				&cli.StringSliceFlag{
+					Name:  dynamicConfigValueFlag,
+					Usage: `dynamic config value, as KEY=JSON_VALUE (meaning strings need quotes)`,
+				},
 			},
 			Before: func(c *cli.Context) error {
 				if c.Args().Len() > 0 {
@@ -262,6 +269,21 @@ func buildCLI() *cli.App {
 				}
 				opts = append(opts, temporalite.WithLogger(logger))
 
+				configVals, err := getDynamicConfigValues(c.StringSlice(dynamicConfigValueFlag))
+				if err != nil {
+					return err
+				}
+				// If there is no config value for search attribute cache disabling,
+				// default it to true
+				if len(configVals[dynamicconfig.ForceSearchAttributesCacheRefreshOnRead]) == 0 {
+					configVals[dynamicconfig.ForceSearchAttributesCacheRefreshOnRead] = []dynamicconfig.ConstrainedValue{
+						{Value: true},
+					}
+				}
+				for k, v := range configVals {
+					opts = append(opts, temporalite.WithDynamicConfigValue(k, v))
+				}
+
 				s, err := temporalite.NewServer(opts...)
 				if err != nil {
 					return err
@@ -289,3 +311,21 @@ func getPragmaMap(input []string) (map[string]string, error) {
 	}
 	return result, nil
 }
+
+func getDynamicConfigValues(input []string) (map[dynamicconfig.Key][]dynamicconfig.ConstrainedValue, error) {
+	ret := make(map[dynamicconfig.Key][]dynamicconfig.ConstrainedValue, len(input))
+	for _, keyValStr := range input {
+		keyVal := strings.SplitN(keyValStr, "=", 2)
+		if len(keyVal) != 2 {
+			return nil, fmt.Errorf("dynamic config value not in KEY=JSON_VAL format")
+		}
+		key := dynamicconfig.Key(keyVal[0])
+		// We don't support constraints currently
+		var val dynamicconfig.ConstrainedValue
+		if err := json.Unmarshal([]byte(keyVal[1]), &val.Value); err != nil {
+			return nil, fmt.Errorf("invalid JSON value for key %q: %w", key, err)
+		}
+		ret[key] = append(ret[key], val)
+	}
+	return ret, nil
+}

cmd/temporalite/main_test.go

@@ -0,0 +1,45 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the MIT License.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021 Datadog, Inc.
+
+package main
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestGetDynamicConfigValues(t *testing.T) {
+	assertBadVal := func(v string) {
+		if _, err := getDynamicConfigValues([]string{v}); err == nil {
+			t.Fatalf("expected error for %v", v)
+		}
+	}
+	type v map[string][]interface{}
+	assertGoodVals := func(expected v, in ...string) {
+		actualVals, err := getDynamicConfigValues(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		actual := make(v, len(actualVals))
+		for k, vals := range actualVals {
+			for _, val := range vals {
+				actual[string(k)] = append(actual[string(k)], val.Value)
+			}
+		}
+		if !reflect.DeepEqual(expected, actual) {
+			t.Fatalf("not equal, expected - actual: %v - %v", expected, actual)
+		}
+	}
+
+	assertBadVal("foo")
+	assertBadVal("foo=")
+	assertBadVal("foo=bar")
+	assertBadVal("foo=123a")
+
+	assertGoodVals(v{"foo": {123.0}}, "foo=123")
+	assertGoodVals(
+		v{"foo": {123.0, []interface{}{"123", false}}, "bar": {"baz"}, "qux": {true}},
+		"foo=123", `bar="baz"`, "qux=true", `foo=["123", false]`,
+	)
+}

internal/liteconfig/config.go

@@ -14,6 +14,7 @@ import (
 
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/config"
+	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite"
@@ -59,6 +60,7 @@ type Config struct {
 	FrontendIP       string
 	UIServer         UIServer
 	BaseConfig       *config.Config
+	DynamicConfig    dynamicconfig.StaticClient
 }
 
 var SupportedPragmas = map[string]struct{}{

options.go

@@ -6,6 +6,7 @@ package temporalite
 
 import (
 	"go.temporal.io/server/common/config"
+	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/temporal"
 
@@ -119,6 +120,26 @@ func WithBaseConfig(base *config.Config) ServerOption {
 	})
 }
 
+// WithDynamicConfigValue sets the given dynamic config key with the given set
+// of values. This will overwrite the key if already set.
+func WithDynamicConfigValue(key dynamicconfig.Key, value []dynamicconfig.ConstrainedValue) ServerOption {
+	return newApplyFuncContainer(func(cfg *liteconfig.Config) {
+		if cfg.DynamicConfig == nil {
+			cfg.DynamicConfig = dynamicconfig.StaticClient{}
+		}
+		cfg.DynamicConfig[key] = value
+	})
+}
+
+// WithSearchAttributeCacheDisabled disables search attribute caching. This
+// delegates to WithDynamicConfigValue.
+func WithSearchAttributeCacheDisabled() ServerOption {
+	return WithDynamicConfigValue(
+		dynamicconfig.ForceSearchAttributesCacheRefreshOnRead,
+		[]dynamicconfig.ConstrainedValue{{Value: true}},
+	)
+}
+
 type applyFuncContainer struct {
 	applyInternal func(*liteconfig.Config)
 }

server.go

@@ -15,7 +15,6 @@ import (
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/server/common/authorization"
 	"go.temporal.io/server/common/config"
-	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/schema/sqlite"
 	"go.temporal.io/server/temporal"
 
@@ -94,7 +93,15 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 		temporal.WithClaimMapper(func(cfg *config.Config) authorization.ClaimMapper {
 			return claimMapper
 		}),
-		temporal.WithDynamicConfigClient(dynamicconfig.NewNoopClient()),
+	}
+
+	if len(c.DynamicConfig) > 0 {
+		// To prevent having to code fall-through semantics right now, we currently
+		// eagerly fail if dynamic config is being configured in two ways
+		if cfg.DynamicConfigClient != nil {
+			return nil, fmt.Errorf("unable to have file-based dynamic config and individual dynamic config values")
+		}
+		serverOpts = append(serverOpts, temporal.WithDynamicConfigClient(c.DynamicConfig))
 	}
 
 	if len(c.UpstreamOptions) > 0 {

temporaltest/server.go

@@ -146,6 +146,7 @@ func NewServer(opts ...TestServerOption) *TestServer {
 		temporalite.WithPersistenceDisabled(),
 		temporalite.WithDynamicPorts(),
 		temporalite.WithLogger(log.NewNoopLogger()),
+		temporalite.WithSearchAttributeCacheDisabled(),
 	)
 
 	s, err := temporalite.NewServer(ts.serverOptions...)

temporaltest/server_test.go

@@ -10,6 +10,8 @@ import (
 	"testing"
 	"time"
 
+	"go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/operatorservice/v1"
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/worker"
 
@@ -134,7 +136,6 @@ func TestDefaultWorkerOptions(t *testing.T) {
 	ts.NewWorker("hello_world", func(registry worker.Registry) {
 		helloworld.RegisterWorkflowsAndActivities(registry)
 	})
-
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
@@ -196,6 +197,31 @@ func TestClientWithDefaultInterceptor(t *testing.T) {
 	}
 }
 
+func TestSearchAttributeCacheDisabled(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	ts := temporaltest.NewServer(temporaltest.WithT(t))
+
+	// Create a search attribute
+	_, err := ts.DefaultClient().OperatorService().AddSearchAttributes(ctx, &operatorservice.AddSearchAttributesRequest{
+		SearchAttributes: map[string]enums.IndexedValueType{
+			"my-search-attr": enums.INDEXED_VALUE_TYPE_TEXT,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Confirm it exists immediately
+	resp, err := ts.DefaultClient().GetSearchAttributes(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.Keys["my-search-attr"] != enums.INDEXED_VALUE_TYPE_TEXT {
+		t.Fatal("search attribute not found")
+	}
+}
+
 func BenchmarkRunWorkflow(b *testing.B) {
 	ts := temporaltest.NewServer()
 	defer ts.Stop()