Fix: scheduler BC testing issues (#722)

* fix: handel parallel log relay

* fix: upload all paralel return error

* fix: do not fail if dependent hooks not declared in job

* fix: add error logs

* fix: test cases

* fix: error reporitngs, and migration to fix old job run states

* fix: tests

* fix: move run-input cmd client to job folder

client/cmd/job/job.go

@@ -23,6 +23,7 @@ func NewJobCommand() *cobra.Command {
 		NewInspectCommand(),
 		NewReplaceAllCommand(),
 		NewExportCommand(),
+		NewJobRunInputCommand(),
 	)
 	return cmd
 }

client/cmd/scheduler/run_input.go → client/cmd/job/run_input.go

@@ -1,4 +1,4 @@
-package scheduler
+package job
 
 import (
 	"fmt"

client/cmd/scheduler/scheduler.go

@@ -16,7 +16,6 @@ func NewSchedulerCommand() *cobra.Command {
 
 	cmd.AddCommand(
 		UploadCommand(),
-		NewJobRunInputCommand(),
 	)
 	return cmd
 }

core/scheduler/status.go

@@ -40,7 +40,7 @@ func StateFromString(state string) (State, error) {
 	case string(StateFailed):
 		return StateFailed, nil
 	default:
-		return "", errors.InvalidArgument(EntityJobRun, "invalid state for job run "+state)
+		return "", errors.InvalidArgument(EntityJobRun, "invalid state for run "+state)
 	}
 }
 

core/scheduler/status_test.go

@@ -22,7 +22,7 @@ func TestStatus(t *testing.T) {
 
 		expectedJobRunStatus, err = scheduler.JobRunStatusFrom(currentTime, "unregisteredState")
 		assert.NotNil(t, err)
-		assert.EqualError(t, err, "invalid argument for entity jobRun: invalid state for job run unregisteredState")
+		assert.EqualError(t, err, "invalid argument for entity jobRun: invalid state for run unregisteredState")
 		assert.Equal(t, scheduler.JobRunStatus{}, expectedJobRunStatus)
 	})
 	t.Run("ExecutionStart", func(t *testing.T) {
@@ -106,7 +106,7 @@ func TestStatus(t *testing.T) {
 
 		respState, err := scheduler.StateFromString("unregisteredState")
 		assert.NotNil(t, err)
-		assert.EqualError(t, err, "invalid argument for entity jobRun: invalid state for job run unregisteredState")
+		assert.EqualError(t, err, "invalid argument for entity jobRun: invalid state for run unregisteredState")
 		assert.Equal(t, scheduler.State(""), respState)
 	})
 }

entrypoint_init_container.sh

@@ -18,7 +18,7 @@ echo "OPTIMUS_HOST:$OPTIMUS_HOST"
 echo ""
 
 echo "-- initializing optimus assets"
-optimus scheduler run-input "$JOB_NAME" --project-name \
+optimus job run-input "$JOB_NAME" --project-name \
 	"$PROJECT" --output-dir "$JOB_DIR" \
 	--type "$INSTANCE_TYPE" --name "$INSTANCE_NAME" \
 	--scheduled-at "$SCHEDULED_AT" --host "$OPTIMUS_HOST"

internal/compiler/engine.go

@@ -39,13 +39,13 @@ func (e *Engine) Compile(templateMap map[string]string, context map[string]any)
 	for name, content := range templateMap {
 		tmpl, err := e.baseTemplate.New(name).Parse(content)
 		if err != nil {
-			return nil, errors.InvalidArgument(EntityCompiler, "unable to parse content for "+name)
+			return nil, errors.AddErrContext(err, EntityCompiler, "unable to parse content for "+name)
 		}
 
 		var buf bytes.Buffer
 		err = tmpl.Execute(&buf, context)
 		if err != nil {
-			return nil, errors.InvalidArgument(EntityCompiler, "unable to render content for "+name)
+			return nil, errors.AddErrContext(err, EntityCompiler, "unable to render content for "+name)
 		}
 		rendered[name] = strings.TrimSpace(buf.String())
 	}

internal/compiler/engine_test.go

@@ -91,7 +91,7 @@ func TestEngine(t *testing.T) {
 			_, err := comp.Compile(map[string]string{"query": input}, context)
 
 			assert.NotNil(t, err)
-			assert.EqualError(t, err, "invalid argument for entity compiler: unable to parse content for query")
+			assert.EqualError(t, err, "internal error for entity compiler: unable to parse content for query")
 		})
 		t.Run("returns error when rendering fails", func(t *testing.T) {
 			input := `event_timestamp > "{{.DSTART | Date }}"`
@@ -103,7 +103,7 @@ func TestEngine(t *testing.T) {
 			_, err := comp.Compile(map[string]string{"query": input}, context)
 
 			assert.NotNil(t, err)
-			assert.EqualError(t, err, "invalid argument for entity compiler: unable to render content for query")
+			assert.EqualError(t, err, "internal error for entity compiler: unable to render content for query")
 		})
 		t.Run("returns rendered string with values of macros for template map", func(t *testing.T) {
 			testCases := []struct {

internal/store/postgres/migrations/000048_run_state_migration.down.sql

@@ -0,0 +1 @@
+update job_run set status = 'STARTED' where status = 'running';

internal/store/postgres/migrations/000048_run_state_migration.up.sql

@@ -0,0 +1 @@
+update job_run set status = 'running' where status = 'STARTED';

internal/store/postgres/scheduler/job_operator_repository.go

@@ -60,7 +60,7 @@ func operatorTypeToTableName(operatorType scheduler.OperatorType) (string, error
 func (o operatorRun) toOperatorRun() (*scheduler.OperatorRun, error) {
 	status, err := scheduler.StateFromString(o.Status)
 	if err != nil {
-		return nil, errors.NewError(scheduler.EntityJobRun, "invalid job run state in database", err.Error())
+		return nil, errors.NewError(scheduler.EntityJobRun, "invalid operator run state in database", err.Error())
 	}
 	return &scheduler.OperatorRun{
 		ID:           o.ID,

internal/store/postgres/scheduler/job_run_repository.go

@@ -49,7 +49,7 @@ func (j jobRun) toJobRun() (*scheduler.JobRun, error) {
 	}
 	state, err := scheduler.StateFromString(j.Status)
 	if err != nil {
-		return nil, err
+		return nil, errors.AddErrContext(err, scheduler.EntityJobRun, "invalid job run state in database")
 	}
 	return &scheduler.JobRun{
 		ID:        j.ID,