From 4b00568d851f69d2d6fd93927bc2ddcbf6d01d90 Mon Sep 17 00:00:00 2001 From: davidlin20dev Date: Wed, 2 Sep 2026 10:52:45 -0700 Subject: [PATCH 1/3] feat(runs): apply settings to the run spec at run creation Signed-off-by: davidlin20dev --- runs/service/run_service.go | 17 +++++++++++++++++ runs/service/settings_apply.go | 28 ++++++++++++++++++++++++++++ runs/service/settings_apply_test.go | 1 + runs/setup.go | 6 ++++-- runs/test/api/setup_test.go | 7 +++++-- 5 files changed, 55 insertions(+), 4 deletions(-) create mode 100644 runs/service/settings_apply.go create mode 100644 runs/service/settings_apply_test.go diff --git a/runs/service/run_service.go b/runs/service/run_service.go index 0b29c638b1..c0102aa2b9 100644 --- a/runs/service/run_service.go +++ b/runs/service/run_service.go @@ -14,6 +14,7 @@ import ( "connectrpc.com/connect" semver "github.com/Masterminds/semver/v3" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/settings" "golang.org/x/sync/errgroup" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" @@ -40,6 +41,7 @@ import ( // RunService implements the RunServiceHandler interface type RunService struct { repo interfaces.Repository + settingsRepo interfaces.SettingsRepo actionsClient actionsconnect.ActionsServiceClient projectClient projectconnect.ProjectServiceClient storagePrefix string @@ -143,6 +145,7 @@ func (s *RunService) WatchGroups(ctx context.Context, req *connect.Request[workf // NewRunService creates a new RunService instance func NewRunService( repo interfaces.Repository, + settingsRepo interfaces.SettingsRepo, actionsClient actionsconnect.ActionsServiceClient, projectClient projectconnect.ProjectServiceClient, storagePrefix string, @@ -154,6 +157,7 @@ func NewRunService( ) *RunService { return &RunService{ repo: repo, + settingsRepo: settingsRepo, actionsClient: actionsClient, projectClient: projectClient, storagePrefix: storagePrefix, @@ -276,6 +280,19 @@ func (s *RunService) CreateRun( } request.RunSpec = runSpec + // Settings sit between an explicit request value and the static config defaults + // applied below. Org is empty when the caller passed a ProjectId rather than a + // RunId; the storage key encoder normalizes that to the default org. + resolved, err := resolveSettings(ctx, s.settingsRepo, &settings.SettingsKey{ + Org: runId.GetOrg(), + Domain: runId.GetDomain(), + Project: runId.GetProject(), + }) + if err != nil { + return nil, connect.NewError(connect.CodeInternal, err) + } + applyRunSettings(runSpec, resolved) + // Stamp the run start time, but only for SDKs that understand it (>= 2.3.6) — older task // templates have no {{.runStartTime}} placeholder, so leaving it unset keeps the executor from // substituting anything. The scheduler sets CreateRunRequest.run_start_time to a trigger's diff --git a/runs/service/settings_apply.go b/runs/service/settings_apply.go new file mode 100644 index 0000000000..c912b9b7c3 --- /dev/null +++ b/runs/service/settings_apply.go @@ -0,0 +1,28 @@ +package service + +import ( + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/settings" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/task" +) + +// applyRunSettings fills fields the caller left unset with values resolved from +// settings. An explicit value in the request always wins, so this only ever fills +// gaps, and a setting that is INHERIT or UNSET contributes nothing. +func applyRunSettings(spec *task.RunSpec, resolved *settings.Settings) { + if spec == nil { + return + } + + if spec.GetQueue() == "" && resolved.GetRun().GetDefaultQueue().GetState() == + settings.SettingState_SETTING_STATE_VALUE { + spec.Queue = resolved.GetRun().GetDefaultQueue().GetStringValue() + } + + // The proto defines 0 as unset for this field, so there is no explicit request for + // "unlimited" to override. The settings validator bounds the value to 0 or + // [2, MaxUint32], so narrowing to uint32 cannot overflow. + if concurrency := resolved.GetRun().GetMaxActionConcurrency(); spec.GetMaxActionConcurrency() == 0 && + concurrency.GetState() == settings.SettingState_SETTING_STATE_VALUE { + spec.MaxActionConcurrency = uint32(concurrency.GetIntValue()) + } +} diff --git a/runs/service/settings_apply_test.go b/runs/service/settings_apply_test.go new file mode 100644 index 0000000000..6d43c3366c --- /dev/null +++ b/runs/service/settings_apply_test.go @@ -0,0 +1 @@ +package service diff --git a/runs/setup.go b/runs/setup.go index 4ae376d242..d4630b7be2 100644 --- a/runs/setup.go +++ b/runs/setup.go @@ -97,6 +97,8 @@ func Setup(ctx context.Context, sc *app.SetupContext) error { return fmt.Errorf("runs: failed to create repository: %w", err) } + settingsRepo := impl.NewSettingsRepo(sc.DB) + // In unified mode, intra-service calls go through the same mux. actionsServiceCfg := cfg.ActionsService if sc.BaseURL != "" { @@ -129,7 +131,7 @@ func Setup(ctx context.Context, sc *app.SetupContext) error { return abortReconciler.Run(ctx) }) - runsSvc := service.NewRunService(repo, actionsClient, projectClient, cfg.StoragePrefix, sc.DataStore, abortReconciler, cfg.AuthMetadata.ExternalAuthServerBaseURL, cfg.TrustForwardedIdentityHeaders, cfg.IdentityHeaders) + runsSvc := service.NewRunService(repo, settingsRepo, actionsClient, projectClient, cfg.StoragePrefix, sc.DataStore, abortReconciler, cfg.AuthMetadata.ExternalAuthServerBaseURL, cfg.TrustForwardedIdentityHeaders, cfg.IdentityHeaders) taskSvc := service.NewTaskService(repo, projectClient) runsPath, runsHandler := workflowconnect.NewRunServiceHandler(runsSvc, connect.WithInterceptors(interceptors...)) @@ -178,7 +180,7 @@ func Setup(ctx context.Context, sc *app.SetupContext) error { sc.Mux.Handle(projectPath, projectHandler) logger.Infof(ctx, "Mounted ProjectService at %s", projectPath) - settingsSvc := service.NewSettingsService(impl.NewSettingsRepo(sc.DB)) + settingsSvc := service.NewSettingsService(settingsRepo) settingsPath, settingsHandler := settingsconnect.NewSettingsServiceHandler(settingsSvc, connect.WithInterceptors(otelInterceptor)) sc.Mux.Handle(settingsPath, settingsHandler) logger.Infof(ctx, "Mounted SettingsService at %s", settingsPath) diff --git a/runs/test/api/setup_test.go b/runs/test/api/setup_test.go index 326287f983..b453259785 100644 --- a/runs/test/api/setup_test.go +++ b/runs/test/api/setup_test.go @@ -114,6 +114,9 @@ func TestMain(m *testing.M) { exitCode = 1 return } + + settingsRepo := impl.NewSettingsRepo(testDB) + // Services validate project existence through the ProjectService client, so mount a // real ProjectService on the same mux (mirrors production unified mode in setup.go). endpointURL := fmt.Sprintf("http://localhost:%d", testPort) @@ -123,7 +126,7 @@ func TestMain(m *testing.M) { // Create RunService with a no-op actions client (points at test server; not used by watch tests) actionsClient := actionsconnect.NewActionsServiceClient(http.DefaultClient, endpointURL) - runSvc := service.NewRunService(repo, actionsClient, projectClient, "", nil, nil, "", true, config.GetConfig().IdentityHeaders) + runSvc := service.NewRunService(repo, settingsRepo, actionsClient, projectClient, "", nil, nil, "", true, config.GetConfig().IdentityHeaders) // Setup HTTP server mux := http.NewServeMux() @@ -139,7 +142,7 @@ func TestMain(m *testing.M) { internalRunPath, internalRunHandler := workflowconnect.NewInternalRunServiceHandler(runSvc) mux.Handle(internalRunPath, internalRunHandler) - settingsSvc := service.NewSettingsService(impl.NewSettingsRepo(testDB)) + settingsSvc := service.NewSettingsService(settingsRepo) settingsPath, settingsHandler := settingsconnect.NewSettingsServiceHandler(settingsSvc) mux.Handle(settingsPath, settingsHandler) From e7e82fb0cd6a8abbccfa3bf2086d47a7c67de34d Mon Sep 17 00:00:00 2001 From: davidlin20dev Date: Wed, 2 Sep 2026 10:52:50 -0700 Subject: [PATCH 2/3] test(runs): cover settings applied at run creation Signed-off-by: davidlin20dev --- runs/service/run_service_test.go | 94 +++++++++++++++++++++++++++++ runs/service/settings_apply_test.go | 79 ++++++++++++++++++++++++ 2 files changed, 173 insertions(+) diff --git a/runs/service/run_service_test.go b/runs/service/run_service_test.go index 0e16f70f14..ba2074a12d 100644 --- a/runs/service/run_service_test.go +++ b/runs/service/run_service_test.go @@ -13,10 +13,12 @@ import ( "time" "connectrpc.com/connect" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/settings" "github.com/golang/protobuf/proto" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" + "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/types/known/timestamppb" "github.com/flyteorg/flyte/v2/flytestdlib/storage" @@ -36,6 +38,32 @@ import ( "github.com/flyteorg/flyte/v2/runs/repository/models" ) +// noSettings returns a settings repo reporting no stored rows, so run creation +// resolves to empty settings and the applier changes nothing. +func noSettings(t *testing.T) *repoMocks.SettingsRepo { + m := &repoMocks.SettingsRepo{} + m.On("GetSettingsByKeys", mock.Anything, mock.Anything).Return(nil, nil) + return m +} + +// settingsWithQueue returns a settings repo holding one org-level row that sets the +// default queue. The row key must match what fetchLevels asks for, or the lookup +// aligns to nothing. +func settingsWithQueue(t *testing.T, org, queue string) *repoMocks.SettingsRepo { + t.Helper() + data, err := protojson.Marshal(&settings.Settings{ + Run: &settings.RunSettings{ + DefaultQueue: &settings.StringSetting{State: stateValue, StringValue: queue}, + }, + }) + require.NoError(t, err) + + m := &repoMocks.SettingsRepo{} + m.On("GetSettingsByKeys", mock.Anything, mock.Anything). + Return([]*models.Settings{{Key: models.EncodeSettingsKey(org, "", ""), Data: data, Version: 1}}, nil) + return m +} + // newMockProjectClientAlwaysOK returns a mock ProjectServiceClient whose GetProject always succeeds. func newMockProjectClientAlwaysOK(t *testing.T) *projectMocks.ProjectServiceClient { pc := projectMocks.NewProjectServiceClient(t) @@ -546,6 +574,7 @@ func TestCreateRunResponseIncludesMetadataAndStatus(t *testing.T) { svc := &RunService{ repo: repo, + settingsRepo: noSettings(t), actionsClient: actionsClient, projectClient: newMockProjectClientAlwaysOK(t), storagePrefix: "s3://flyte-data", @@ -1068,6 +1097,7 @@ func TestCreateRun_WritesEmptyInputsProto(t *testing.T) { svc := &RunService{ repo: repo, + settingsRepo: noSettings(t), actionsClient: actionsClient, projectClient: newMockProjectClientAlwaysOK(t), storagePrefix: "s3://flyte-data", @@ -1126,6 +1156,7 @@ func TestCreateRun_ResponseUsesRunModel(t *testing.T) { svc := &RunService{ repo: repo, + settingsRepo: noSettings(t), actionsClient: actionsClient, projectClient: newMockProjectClientAlwaysOK(t), storagePrefix: "s3://flyte-data", @@ -1190,6 +1221,7 @@ func TestCreateRun_TriggerFire_CarriesRunSpecEnvVars(t *testing.T) { svc := &RunService{ repo: repo, + settingsRepo: noSettings(t), actionsClient: actionsClient, projectClient: newMockProjectClientAlwaysOK(t), storagePrefix: "s3://flyte-data", @@ -1287,6 +1319,7 @@ func TestCreateRun_ActionIDUsesRunName(t *testing.T) { svc := &RunService{ repo: repo, + settingsRepo: noSettings(t), actionsClient: actionsClient, projectClient: newMockProjectClientAlwaysOK(t), storagePrefix: "s3://flyte-data", @@ -1377,6 +1410,7 @@ func TestCreateRun_PreservesInputContextAndRawDataPath(t *testing.T) { svc := &RunService{ repo: repo, + settingsRepo: noSettings(t), actionsClient: actionsClient, projectClient: newMockProjectClientAlwaysOK(t), storagePrefix: "s3://flyte-data", @@ -1434,6 +1468,65 @@ func TestCreateRun_PreservesInputContextAndRawDataPath(t *testing.T) { require.NoError(t, err) } +// TestCreateRun_AppliesSettingsQueue proves the applier is actually wired into run +// creation: the request names no queue, and the value stored on the run comes from +// the org's settings row. +func TestCreateRun_AppliesSettingsQueue(t *testing.T) { + actionRepo := &repoMocks.ActionRepo{} + taskRepo := &repoMocks.TaskRepo{} + actionsClient := actionsconnectmocks.NewActionsServiceClient(t) + repo := &repoMocks.Repository{} + store := &storageMocks.ComposedProtobufStore{} + dataStore := &storage.DataStore{ComposedProtobufStore: store} + + repo.On("ActionRepo").Return(actionRepo) + repo.On("TaskRepo").Return(taskRepo) + + svc := &RunService{ + repo: repo, + settingsRepo: settingsWithQueue(t, "org", "fast-queue"), + actionsClient: actionsClient, + projectClient: newMockProjectClientAlwaysOK(t), + storagePrefix: "s3://flyte-data", + dataStore: dataStore, + } + + req := &workflow.CreateRunRequest{ + Id: &workflow.CreateRunRequest_RunId{ + RunId: &common.RunIdentifier{ + Org: "org", + Project: "proj", + Domain: "dev", + Name: "rq-123", + }, + }, + InputWrapper: &workflow.CreateRunRequest_Inputs{Inputs: &task.Inputs{}}, + Task: &workflow.CreateRunRequest_TaskSpec{ + TaskSpec: &task.TaskSpec{}, + }, + } + + store.On("WriteProtobuf", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() + taskRepo.On("CreateTaskSpec", mock.Anything, mock.Anything).Return(nil).Once() + + actionRepo.On("CreateAction", mock.Anything, mock.MatchedBy(func(m *models.Run) bool { + var rs task.RunSpec + _ = proto.Unmarshal(m.RunSpec, &rs) + return rs.GetQueue() == "fast-queue" + }), mock.Anything).Return(&models.Run{ + Project: "proj", + Domain: "dev", + Name: "rq-123", + }, nil).Once() + + actionsClient.On("Enqueue", mock.Anything, mock.MatchedBy(func(req *connect.Request[actions.EnqueueRequest]) bool { + return req.Msg.GetRunSpec().GetQueue() == "fast-queue" + })).Return(connect.NewResponse(&actions.EnqueueResponse{}), nil).Once() + + _, err := svc.CreateRun(context.Background(), connect.NewRequest(req)) + require.NoError(t, err) +} + func TestCreateRun_WithOffloadedInputData(t *testing.T) { actionRepo := &repoMocks.ActionRepo{} taskRepo := &repoMocks.TaskRepo{} @@ -1447,6 +1540,7 @@ func TestCreateRun_WithOffloadedInputData(t *testing.T) { svc := &RunService{ repo: repo, + settingsRepo: noSettings(t), actionsClient: actionsClient, projectClient: newMockProjectClientAlwaysOK(t), storagePrefix: "s3://flyte-data", diff --git a/runs/service/settings_apply_test.go b/runs/service/settings_apply_test.go index 6d43c3366c..00f3ece6a6 100644 --- a/runs/service/settings_apply_test.go +++ b/runs/service/settings_apply_test.go @@ -1 +1,80 @@ package service + +import ( + "testing" + + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/settings" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/task" + "github.com/stretchr/testify/assert" +) + +func runSettings(queue *settings.StringSetting, concurrency *settings.Int64Setting) *settings.Settings { + return &settings.Settings{ + Run: &settings.RunSettings{DefaultQueue: queue, MaxActionConcurrency: concurrency}, + } +} + +func TestApplyRunSettings(t *testing.T) { + tests := []struct { + name string + spec *task.RunSpec + resolved *settings.Settings + wantQueue string + wantConcurrency uint32 + }{ + { + name: "empty queue takes the settings value", + spec: &task.RunSpec{}, + resolved: runSettings(&settings.StringSetting{State: stateValue, StringValue: "fast-queue"}, nil), + wantQueue: "fast-queue", + }, + { + name: "an explicit queue wins over settings", + spec: &task.RunSpec{Queue: "user-queue"}, + resolved: runSettings(&settings.StringSetting{State: stateValue, StringValue: "fast-queue"}, nil), + wantQueue: "user-queue", + }, + { + name: "a queue in INHERIT contributes nothing", + spec: &task.RunSpec{}, + resolved: runSettings(&settings.StringSetting{State: stateInherit, StringValue: "fast-queue"}, nil), + wantQueue: "", + }, + { + name: "zero concurrency takes the settings value", + spec: &task.RunSpec{}, + resolved: runSettings(nil, &settings.Int64Setting{State: stateValue, IntValue: 5}), + wantConcurrency: 5, + }, + { + name: "an explicit concurrency wins over settings", + spec: &task.RunSpec{MaxActionConcurrency: 3}, + resolved: runSettings(nil, &settings.Int64Setting{State: stateValue, IntValue: 5}), + wantConcurrency: 3, + }, + { + name: "concurrency in UNSET contributes nothing", + spec: &task.RunSpec{}, + resolved: runSettings(nil, &settings.Int64Setting{State: stateUnset, IntValue: 5}), + wantConcurrency: 0, + }, + { + name: "no settings at all", + spec: &task.RunSpec{}, + resolved: &settings.Settings{}, + }, + { + name: "nil spec does not panic", + spec: nil, + resolved: runSettings(&settings.StringSetting{State: stateValue, StringValue: "fast-queue"}, nil), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + applyRunSettings(tt.spec, tt.resolved) + assert.Equal(t, tt.wantQueue, tt.spec.GetQueue()) + assert.Equal(t, tt.wantConcurrency, tt.spec.GetMaxActionConcurrency()) + }) + } +} From 3ee332f0ad9223c672289c90f13c5cbdd7702ec0 Mon Sep 17 00:00:00 2001 From: davidlin20dev Date: Fri, 4 Sep 2026 11:13:48 -0700 Subject: [PATCH 3/3] docs(runs): drop the org note from the settings lookup comment Signed-off-by: davidlin20dev --- runs/service/run_service.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/runs/service/run_service.go b/runs/service/run_service.go index c0102aa2b9..8fca9e982c 100644 --- a/runs/service/run_service.go +++ b/runs/service/run_service.go @@ -281,8 +281,7 @@ func (s *RunService) CreateRun( request.RunSpec = runSpec // Settings sit between an explicit request value and the static config defaults - // applied below. Org is empty when the caller passed a ProjectId rather than a - // RunId; the storage key encoder normalizes that to the default org. + // applied below. resolved, err := resolveSettings(ctx, s.settingsRepo, &settings.SettingsKey{ Org: runId.GetOrg(), Domain: runId.GetDomain(),