CHASM Scheduler Experimental Flag

common/dynamicconfig/constants.go

@@ -563,6 +563,11 @@ is currently processing a task.
 	)
 
 	// keys for frontend
+	FrontendAllowedExperiments = NewNamespaceTypedSetting(
+		"frontend.allowedExperiments",
+		[]string(nil),
+		`FrontendAllowedExperiments is a list of experiment names that can be enabled via the temporal-experiment header for a specific namespace.`,
+	)
 	FrontendHTTPAllowedHosts = NewGlobalTypedSettingWithConverter(
 		"frontend.httpAllowedHosts",
 		ConvertWildcardStringListToRegexp,

common/headers/headers.go

@@ -2,6 +2,7 @@ package headers
 
 import (
 	"context"
+	"strings"
 
 	"google.golang.org/grpc/metadata"
 )
@@ -18,6 +19,8 @@ const (
 	CallerNameHeaderName = "caller-name"
 	CallerTypeHeaderName = "caller-type"
 	CallOriginHeaderName = "call-initiation"
+
+	ExperimentHeaderName = "temporal-experiment"
 )
 
 var (
@@ -90,3 +93,25 @@ func (h GRPCHeaderGetter) Get(key string) string {
 	}
 	return ""
 }
+
+// IsExperimentRequested checks if a specific experiment is present in the temporal-experiment header.
+// Returns true if the experiment is explicitly listed or if "*" (wildcard) is present.
+// Headers exceeding a length of 100 will be skipped.
+func IsExperimentRequested(ctx context.Context, experiment string) bool {
+	experimentalValues := metadata.ValueFromIncomingContext(ctx, ExperimentHeaderName)
+
+	for _, headerValue := range experimentalValues {
+		// limit value size to prevent misuse
+		if len(headerValue) > 100 {
+			continue
+		}
+		for requested := range strings.SplitSeq(headerValue, ",") {
+			requested = strings.TrimSpace(requested)
+			if requested == "*" || requested == experiment {
+				return true
+			}
+		}
+	}
+
+	return false
+}

common/headers/headers_test.go

@@ -2,29 +2,16 @@ package headers
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
 	"google.golang.org/grpc/metadata"
 )
 
-type (
-	HeadersSuite struct {
-		*require.Assertions
-		suite.Suite
-	}
-)
+func TestPropagate_CreateNewOutgoingContext(t *testing.T) {
+	t.Parallel()
 
-func TestHeadersSuite(t *testing.T) {
-	suite.Run(t, &HeadersSuite{})
-}
-
-func (s *HeadersSuite) SetupTest() {
-	s.Assertions = require.New(s.T())
-}
-
-func (s *HeadersSuite) TestPropagate_CreateNewOutgoingContext() {
 	ctx := context.Background()
 	ctx = metadata.NewIncomingContext(ctx, metadata.New(map[string]string{
 		ClientVersionHeaderName:           "22.08.78",
@@ -36,15 +23,17 @@ func (s *HeadersSuite) TestPropagate_CreateNewOutgoingContext() {
 	ctx = Propagate(ctx)
 
 	md, ok := metadata.FromOutgoingContext(ctx)
-	s.True(ok)
+	require.True(t, ok)
 
-	s.Equal("22.08.78", md.Get(ClientVersionHeaderName)[0])
-	s.Equal(">21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
-	s.Equal("28.08.14", md.Get(ClientNameHeaderName)[0])
-	s.Equal("my-feature", md.Get(SupportedFeaturesHeaderName)[0])
+	require.Equal(t, "22.08.78", md.Get(ClientVersionHeaderName)[0])
+	require.Equal(t, ">21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
+	require.Equal(t, "28.08.14", md.Get(ClientNameHeaderName)[0])
+	require.Equal(t, "my-feature", md.Get(SupportedFeaturesHeaderName)[0])
 }
 
-func (s *HeadersSuite) TestPropagate_CreateNewOutgoingContext_SomeMissing() {
+func TestPropagate_CreateNewOutgoingContext_SomeMissing(t *testing.T) {
+	t.Parallel()
+
 	ctx := context.Background()
 	ctx = metadata.NewIncomingContext(ctx, metadata.New(map[string]string{
 		ClientVersionHeaderName: "22.08.78",
@@ -54,15 +43,17 @@ func (s *HeadersSuite) TestPropagate_CreateNewOutgoingContext_SomeMissing() {
 	ctx = Propagate(ctx)
 
 	md, ok := metadata.FromOutgoingContext(ctx)
-	s.True(ok)
+	require.True(t, ok)
 
-	s.Equal("22.08.78", md.Get(ClientVersionHeaderName)[0])
-	s.Equal(0, len(md.Get(SupportedServerVersionsHeaderName)))
-	s.Equal("28.08.14", md.Get(ClientNameHeaderName)[0])
-	s.Equal(0, len(md.Get(SupportedFeaturesHeaderName)))
+	require.Equal(t, "22.08.78", md.Get(ClientVersionHeaderName)[0])
+	require.Empty(t, md.Get(SupportedServerVersionsHeaderName))
+	require.Equal(t, "28.08.14", md.Get(ClientNameHeaderName)[0])
+	require.Empty(t, md.Get(SupportedFeaturesHeaderName))
 }
 
-func (s *HeadersSuite) TestPropagate_UpdateExistingEmptyOutgoingContext() {
+func TestPropagate_UpdateExistingEmptyOutgoingContext(t *testing.T) {
+	t.Parallel()
+
 	ctx := context.Background()
 	ctx = metadata.NewIncomingContext(ctx, metadata.New(map[string]string{
 		ClientVersionHeaderName:           "22.08.78",
@@ -76,15 +67,17 @@ func (s *HeadersSuite) TestPropagate_UpdateExistingEmptyOutgoingContext() {
 	ctx = Propagate(ctx)
 
 	md, ok := metadata.FromOutgoingContext(ctx)
-	s.True(ok)
+	require.True(t, ok)
 
-	s.Equal("22.08.78", md.Get(ClientVersionHeaderName)[0])
-	s.Equal("<21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
-	s.Equal("28.08.14", md.Get(ClientNameHeaderName)[0])
-	s.Equal("my-feature", md.Get(SupportedFeaturesHeaderName)[0])
+	require.Equal(t, "22.08.78", md.Get(ClientVersionHeaderName)[0])
+	require.Equal(t, "<21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
+	require.Equal(t, "28.08.14", md.Get(ClientNameHeaderName)[0])
+	require.Equal(t, "my-feature", md.Get(SupportedFeaturesHeaderName)[0])
 }
 
-func (s *HeadersSuite) TestPropagate_UpdateExistingNonEmptyOutgoingContext() {
+func TestPropagate_UpdateExistingNonEmptyOutgoingContext(t *testing.T) {
+	t.Parallel()
+
 	ctx := context.Background()
 	ctx = metadata.NewIncomingContext(ctx, metadata.New(map[string]string{
 		ClientVersionHeaderName:           "07.08.78",   // Must be ignored
@@ -101,15 +94,17 @@ func (s *HeadersSuite) TestPropagate_UpdateExistingNonEmptyOutgoingContext() {
 	ctx = Propagate(ctx)
 
 	md, ok := metadata.FromOutgoingContext(ctx)
-	s.True(ok)
+	require.True(t, ok)
 
-	s.Equal("22.08.78", md.Get(ClientVersionHeaderName)[0])
-	s.Equal("<21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
-	s.Equal("28.08.14", md.Get(ClientNameHeaderName)[0])
-	s.Equal("my-feature", md.Get(SupportedFeaturesHeaderName)[0])
+	require.Equal(t, "22.08.78", md.Get(ClientVersionHeaderName)[0])
+	require.Equal(t, "<21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
+	require.Equal(t, "28.08.14", md.Get(ClientNameHeaderName)[0])
+	require.Equal(t, "my-feature", md.Get(SupportedFeaturesHeaderName)[0])
 }
 
-func (s *HeadersSuite) TestPropagate_EmptyIncomingContext() {
+func TestPropagate_EmptyIncomingContext(t *testing.T) {
+	t.Parallel()
+
 	ctx := context.Background()
 
 	ctx = metadata.NewOutgoingContext(ctx, metadata.New(map[string]string{
@@ -121,9 +116,85 @@ func (s *HeadersSuite) TestPropagate_EmptyIncomingContext() {
 	ctx = Propagate(ctx)
 
 	md, ok := metadata.FromOutgoingContext(ctx)
-	s.True(ok)
+	require.True(t, ok)
+
+	require.Equal(t, "22.08.78", md.Get(ClientVersionHeaderName)[0])
+	require.Equal(t, "<21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
+	require.Equal(t, "28.08.14", md.Get(ClientNameHeaderName)[0])
+}
+
+func TestIsExperimentRequested(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name            string
+		headerValues    []string
+		checkExperiment string
+		expected        bool
+	}{
+		{
+			name:            "no header returns false",
+			headerValues:    nil,
+			checkExperiment: "chasm-scheduler",
+			expected:        false,
+		},
+		{
+			name:            "exact match returns true",
+			headerValues:    []string{"chasm-scheduler"},
+			checkExperiment: "chasm-scheduler",
+			expected:        true,
+		},
+		{
+			name:            "comma separated list finds match",
+			headerValues:    []string{"chasm-scheduler, other-exp, third-exp"},
+			checkExperiment: "other-exp",
+			expected:        true,
+		},
+		{
+			name:            "wildcard matches any experiment",
+			headerValues:    []string{"*"},
+			checkExperiment: "any-experiment",
+			expected:        true,
+		},
+		{
+			name:            "wildcard in list matches",
+			headerValues:    []string{"chasm-scheduler,*,other-exp"},
+			checkExperiment: "random-experiment",
+			expected:        true,
+		},
+		{
+			name:            "multiple header values finds match",
+			headerValues:    []string{"chasm-scheduler", "other-exp,third-exp"},
+			checkExperiment: "third-exp",
+			expected:        true,
+		},
+		{
+			name:            "max experiment size limit match",
+			headerValues:    []string{strings.Repeat("a,", 49)},
+			checkExperiment: "a",
+			expected:        true, // 98 chars, under 100 char limit
+		},
+		{
+			name:            "at max experiment size limit no match",
+			headerValues:    []string{strings.Repeat("a,", 51)},
+			checkExperiment: "a",
+			expected:        false, // exceeds 100 char limit, should be skipped
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
 
-	s.Equal("22.08.78", md.Get(ClientVersionHeaderName)[0])
-	s.Equal("<21.04.16", md.Get(SupportedServerVersionsHeaderName)[0])
-	s.Equal("28.08.14", md.Get(ClientNameHeaderName)[0])
+			ctx := context.Background()
+			md := metadata.MD{}
+			for _, val := range tt.headerValues {
+				md.Append(ExperimentHeaderName, val)
+			}
+			ctx = metadata.NewIncomingContext(ctx, md)
+
+			result := IsExperimentRequested(ctx, tt.checkExperiment)
+			require.Equal(t, tt.expected, result)
+		})
+	}
 }

service/frontend/experiments.go

@@ -0,0 +1,6 @@
+package frontend
+
+const (
+	// ChasmSchedulerExperiment is the experiment name for enabling CHASM (V2) scheduler
+	ChasmSchedulerExperiment = "chasm-scheduler"
+)

service/frontend/service.go

@@ -213,7 +213,21 @@ type Config struct {
 	ListWorkersEnabled      dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	WorkerCommandsEnabled   dynamicconfig.BoolPropertyFnWithNamespaceFilter
 
-	HTTPAllowedHosts dynamicconfig.TypedPropertyFn[*regexp.Regexp]
+	HTTPAllowedHosts   dynamicconfig.TypedPropertyFn[*regexp.Regexp]
+	AllowedExperiments dynamicconfig.TypedPropertyFnWithNamespaceFilter[[]string]
+}
+
+// IsExperimentAllowed checks if an experiment is enabled for a given namespace in the dynamic config.
+// Returns true if the experiment is explicitly listed or if "*" (wildcard)
+// is present in the allowed experiments list.
+func (c *Config) IsExperimentAllowed(experiment string, namespace string) bool {
+	allowedExperiments := c.AllowedExperiments(namespace)
+	for _, allowed := range allowedExperiments {
+		if allowed == "*" || allowed == experiment {
+			return true
+		}
+	}
+	return false
 }
 
 // NewConfig returns new service config with default values
@@ -351,7 +365,8 @@ func NewConfig(
 		ListWorkersEnabled:             dynamicconfig.ListWorkersEnabled.Get(dc),
 		WorkerCommandsEnabled:          dynamicconfig.WorkerCommandsEnabled.Get(dc),
 
-		HTTPAllowedHosts: dynamicconfig.FrontendHTTPAllowedHosts.Get(dc),
+		HTTPAllowedHosts:   dynamicconfig.FrontendHTTPAllowedHosts.Get(dc),
+		AllowedExperiments: dynamicconfig.FrontendAllowedExperiments.Get(dc),
 	}
 }
 

service/frontend/service_test.go

@@ -0,0 +1,85 @@
+package frontend
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfig_IsExperimentAllowed(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		namespace  string
+		experiment string
+		setupFunc  func(namespace string) []string
+		expected   bool
+	}{
+		{
+			name:       "no experiments allowed",
+			namespace:  "test-namespace",
+			experiment: "chasm-scheduler",
+			setupFunc:  func(ns string) []string { return []string{} },
+			expected:   false,
+		},
+		{
+			name:       "specific experiment match",
+			namespace:  "test-namespace",
+			experiment: "chasm-scheduler",
+			setupFunc:  func(ns string) []string { return []string{"chasm-scheduler"} },
+			expected:   true,
+		},
+		{
+			name:       "specific experiment no match",
+			namespace:  "test-namespace",
+			experiment: "other-experiment",
+			setupFunc:  func(ns string) []string { return []string{"chasm-scheduler"} },
+			expected:   false,
+		},
+		{
+			name:       "wildcard allows any",
+			namespace:  "test-namespace",
+			experiment: "any-experiment",
+			setupFunc:  func(ns string) []string { return []string{"*"} },
+			expected:   true,
+		},
+		{
+			name:       "namespace specific wildcard",
+			namespace:  "ns-with-wildcard",
+			experiment: "any-experiment",
+			setupFunc: func(ns string) []string {
+				if ns == "ns-with-wildcard" {
+					return []string{"*"}
+				}
+				return nil
+			},
+			expected: true,
+		},
+		{
+			name:       "namespace specific match",
+			namespace:  "ns-with-specific",
+			experiment: "chasm-scheduler",
+			setupFunc: func(ns string) []string {
+				if ns == "ns-with-specific" {
+					return []string{"chasm-scheduler"}
+				}
+				return nil
+			},
+			expected: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			config := &Config{
+				AllowedExperiments: tt.setupFunc,
+			}
+
+			result := config.IsExperimentAllowed(tt.experiment, tt.namespace)
+			require.Equal(t, tt.expected, result)
+		})
+	}
+}