Skip to content

fix: add replay job config to replay dry run #580

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Mryashbhardwaj merged 3 commits into from
Jul 23, 2026
Merged

fix: add replay job config to replay dry run #580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7ddebd3
fix: add replay job config to replay dry run
Jul 21, 2026
b827cbb
fix: return task config as part of job replay config
Jul 22, 2026
5ca6645
fix: lint
Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ NAME = "github.com/goto/optimus"
LAST_COMMIT := $(shell git rev-parse --short HEAD)
LAST_TAG := "$(shell git rev-list --tags --max-count=1)"
OPMS_VERSION := "$(shell git describe --tags ${LAST_TAG})-next"
PROTON_COMMIT := "0d8cde8f458ccf46ea3176b0265168091cc77e8b"
PROTON_COMMIT := "bba736bdc93dabd74e2ce2ec155cee99692c0268"


.PHONY: build test test-ci generate-proto unit-test-ci integration-test vet coverage clean install lint
Expand Down
8 changes: 8 additions & 0 deletions core/scheduler/handler/v1beta1/replay.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ type ReplayService interface {
GetReplayByID(ctx context.Context, replayID uuid.UUID) (replay *scheduler.ReplayWithRun, err error)
GetReplayByApprovalID(ctx context.Context, approvalID string) (*scheduler.ReplayWithRun, error)
GetRunsStatus(ctx context.Context, tenant tenant.Tenant, jobName scheduler.JobName, config *scheduler.ReplayConfig) (runs []*scheduler.JobRunStatus, err error)
GetJobConfig(ctx context.Context, tenant tenant.Tenant, jobName scheduler.JobName, config *scheduler.ReplayConfig) (map[string]string, error)
CancelReplay(ctx context.Context, replayWithRun *scheduler.ReplayWithRun) error
}

Expand Down Expand Up @@ -61,8 +62,15 @@ func (h ReplayHandler) ReplayDryRun(ctx context.Context, req *pb.ReplayDryRunReq
return nil, errors.GRPCErr(err, "unable to fetch runs status for "+req.JobName)
}

jobConfig, err := h.service.GetJobConfig(ctx, replayReq.Tenant(), replayReq.JobName(), replayReq.Config())
if err != nil {
h.l.Error("error fetching runs status for replay dry run: %s", err)
return nil, errors.GRPCErr(err, "unable to fetch runs status for "+req.JobName)
}

return &pb.ReplayDryRunResponse{
ReplayRuns: replayRunsToProto(runs),
JobConfig: jobConfig,
}, nil
}

Expand Down
5 changes: 3 additions & 2 deletions core/scheduler/handler/v1beta1/replay_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,7 @@ func TestReplayHandler(t *testing.T) {
}

service.On("GetRunsStatus", ctx, jobTenant, jobName, replayConfig).Return(runs, nil)
service.On("GetJobConfig", ctx, jobTenant, jobName, replayConfig).Return(jobConfig, nil)

result, err := replayHandler.ReplayDryRun(ctx, req)
assert.NoError(t, err)
Expand Down Expand Up @@ -778,8 +779,8 @@ func (_m *mockReplayService) CancelReplay(ctx context.Context, replayWithRun *sc
return r0
}

func (_m *mockReplayService) GetReplayConfig(ctx context.Context, projectName tenant.ProjectName, name scheduler.JobName, scheduledAt time.Time) (map[string]string, error) {
args := _m.Called(ctx, projectName, name, scheduledAt)
func (_m *mockReplayService) GetJobConfig(ctx context.Context, jobTenant tenant.Tenant, jobName scheduler.JobName, config *scheduler.ReplayConfig) (map[string]string, error) {
args := _m.Called(ctx, jobTenant, jobName, config)
if args.Get(0) == nil {
return nil, args.Error(1)
}
Expand Down
5 changes: 5 additions & 0 deletions core/scheduler/service/executor_input_compiler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,10 @@ const (

var invalidLabelCharacterRegex *regexp.Regexp

var jobRunMode ctxContext = "run-mode"

type ctxContext string

type TenantService interface {
GetDetails(ctx context.Context, tnnt tenant.Tenant) (*tenant.WithDetails, error)
GetSecrets(ctx context.Context, tnnt tenant.Tenant) ([]*tenant.PlainTextSecret, error)
Expand Down Expand Up @@ -138,6 +142,7 @@ func (i InputCompiler) Compile(ctx context.Context, job *scheduler.JobWithDetail
)

mergedContext := utils.MergeAnyMaps(taskContext, allTaskConfigs)
ctx = context.WithValue(ctx, jobRunMode, scheduler.DryRun)
fileMap, err := i.assetCompiler.CompileJobRunAssets(ctx, job.Job, systemDefinedVars, interval, mergedContext)
if err != nil {
i.logger.Error("error compiling job run assets: %s", err)
Expand Down
12 changes: 6 additions & 6 deletions core/scheduler/service/executor_input_compiler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,7 @@ func TestExecutorCompiler(t *testing.T) {
templateCompiler.On("Compile", mock.Anything, taskContext).Return(map[string]string{}, nil)
defer templateCompiler.AssertExpectations(t)
assetCompiler := new(mockAssetCompiler)
assetCompiler.On("CompileJobRunAssets", ctx, &job, systemDefinedVars, interval, taskContext).Return(nil, fmt.Errorf("CompileJobRunAssets error"))
assetCompiler.On("CompileJobRunAssets", mock.Anything, &job, systemDefinedVars, interval, taskContext).Return(nil, fmt.Errorf("CompileJobRunAssets error"))
defer assetCompiler.AssertExpectations(t)

inputCompiler := service.NewJobInputCompiler(tenantService, templateCompiler, assetCompiler, logger)
Expand Down Expand Up @@ -263,7 +263,7 @@ func TestExecutorCompiler(t *testing.T) {
Return(map[string]string{"secret.config.compiled": "a.secret.val.compiled"}, nil)
defer templateCompiler.AssertExpectations(t)
assetCompiler := new(mockAssetCompiler)
assetCompiler.On("CompileJobRunAssets", ctx, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
assetCompiler.On("CompileJobRunAssets", mock.Anything, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
defer assetCompiler.AssertExpectations(t)
inputCompiler := service.NewJobInputCompiler(tenantService, templateCompiler, assetCompiler, logger)
inputExecutorResp, err := inputCompiler.Compile(ctx, &details, config, executedAt, nil)
Expand Down Expand Up @@ -313,7 +313,7 @@ func TestExecutorCompiler(t *testing.T) {
}

assetCompilerNew := new(mockAssetCompiler)
assetCompilerNew.On("CompileJobRunAssets", ctx, &jobNew, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
assetCompilerNew.On("CompileJobRunAssets", mock.Anything, &jobNew, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
defer assetCompilerNew.AssertExpectations(t)

inputCompiler := service.NewJobInputCompiler(tenantService, templateCompiler, assetCompilerNew, logger)
Expand Down Expand Up @@ -408,7 +408,7 @@ func TestExecutorCompiler(t *testing.T) {
"someFileName": "fileContents",
}
assetCompiler := new(mockAssetCompiler)
assetCompiler.On("CompileJobRunAssets", ctx, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
assetCompiler.On("CompileJobRunAssets", mock.Anything, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
defer assetCompiler.AssertExpectations(t)

templateCompiler := new(mockTemplateCompiler)
Expand Down Expand Up @@ -507,7 +507,7 @@ func TestExecutorCompiler(t *testing.T) {
"someFileName": "fileContents",
}
assetCompiler := new(mockAssetCompiler)
assetCompiler.On("CompileJobRunAssets", ctx, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
assetCompiler.On("CompileJobRunAssets", mock.Anything, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
defer assetCompiler.AssertExpectations(t)

templateCompiler := new(mockTemplateCompiler)
Expand Down Expand Up @@ -579,7 +579,7 @@ func TestExecutorCompiler(t *testing.T) {
"someFileName": "fileContents",
}
assetCompiler := new(mockAssetCompiler)
assetCompiler.On("CompileJobRunAssets", ctx, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
assetCompiler.On("CompileJobRunAssets", mock.Anything, &job, systemDefinedVars, interval, taskContext).Return(compiledFile, nil)
defer assetCompiler.AssertExpectations(t)

templateCompiler := new(mockTemplateCompiler)
Expand Down
13 changes: 13 additions & 0 deletions core/scheduler/service/replay_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ type ReplayRepository interface {
ScanAbandonedReplayRequests(ctx context.Context, unhandledClassifierDuration time.Duration) ([]*scheduler.Replay, error)
AcquireReplayRequest(ctx context.Context, replayID uuid.UUID, unhandledClassifierDuration time.Duration) error

GetReplayJobConfig(ctx context.Context, jobTenant tenant.Tenant, jobName scheduler.JobName, scheduledAt time.Time) (map[string]string, error)
GetReplayRequestByID(ctx context.Context, replayID uuid.UUID) (*scheduler.Replay, error)

GetReplayByFilters(ctx context.Context, projectName tenant.ProjectName, filters ...filter.FilterOpt) ([]*scheduler.ReplayWithRun, error)
Expand Down Expand Up @@ -192,6 +193,18 @@ func (r *ReplayService) GetReplayByApprovalID(ctx context.Context, approvalID st
return replayWithRun, nil
}

func (r *ReplayService) GetJobConfig(ctx context.Context, tenant tenant.Tenant, jobName scheduler.JobName, config *scheduler.ReplayConfig) (map[string]string, error) {
details, err := r.jobRepo.GetJobDetails(ctx, tenant.ProjectName(), jobName)
if err != nil {
r.logger.Error("error getting job [%s]: %s", jobName, err)
return nil, err
}
for k, v := range config.JobConfig {
details.Job.Task.Config[k] = v
}
return details.Job.Task.Config, nil
}

func (r *ReplayService) GetRunsStatus(ctx context.Context, tenant tenant.Tenant, jobName scheduler.JobName, config *scheduler.ReplayConfig) ([]*scheduler.JobRunStatus, error) {
jobRunCriteria := &scheduler.JobRunsCriteria{
Name: jobName.String(),
Expand Down
64 changes: 32 additions & 32 deletions protos/gotocompany/optimus/core/v1beta1/job_run.pb.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion protos/gotocompany/optimus/core/v1beta1/job_run.swagger.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -750,7 +750,7 @@
"NULL_VALUE"
],
"default": "NULL_VALUE",
"description": "`NullValue` is a singleton enumeration to represent the null value for the\n`Value` type union.\n\nThe JSON representation for `NullValue` is JSON `null`.\n\n - NULL_VALUE: Null value."
"description": "`NullValue` is a singleton enumeration to represent the null value for the\n`Value` type union.\n\n The JSON representation for `NullValue` is JSON `null`.\n\n - NULL_VALUE: Null value."
},
"rpcStatus": {
"type": "object",
Expand Down
Loading
Loading