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

* fix: add replay job config to replay dry run

* fix: return task config as part of job replay config

---------

Co-authored-by: Yash bhardwaj <yash.bhardwaj@gojek.com>

Author: Mryashbhardwaj

Makefile

@@ -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

core/scheduler/handler/v1beta1/replay.go

@@ -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
 }
 
@@ -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
 }
 

core/scheduler/handler/v1beta1/replay_test.go

@@ -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)
@@ -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)
 	}

core/scheduler/service/executor_input_compiler.go

@@ -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)
@@ -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)

core/scheduler/service/executor_input_compiler_test.go

@@ -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)
@@ -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)
@@ -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)
@@ -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)
@@ -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)
@@ -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)

core/scheduler/service/replay_service.go

@@ -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)
@@ -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(),

protos/gotocompany/optimus/core/v1beta1/job_run.pb.go

@@ -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",