Skip to content

Commit

Permalink
fix: allow jobs submitted with result paths and no publisher (#4496)
Browse files Browse the repository at this point in the history 
- closes #4391

---------

Co-authored-by: frrist <[email protected]>
Co-authored-by: Walid Baruni <[email protected]>
  • Loading branch information
@frrist @wdbaruni
3 people authored Sep 24, 2024
1 parent 9d6ef49 commit e4ccb07
Show file tree
Hide file tree
Showing 3 changed files with 104 additions and 31 deletions.
16 changes: 10 additions & 6 deletions pkg/models/task.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -106,12 +106,13 @@ func (t *Task) Validate() error {
var mErr error
mErr = errors.Join(mErr, t.ValidateSubmission())

if len(t.ResultPaths) > 0 && t.Publisher.IsEmpty() {
mErr = errors.Join(mErr, errors.New("publisher must be set if result paths are set"))
}

if err := t.Timeouts.Validate(); err != nil {
mErr = errors.Join(mErr, fmt.Errorf("task timeouts validation failed: %v", err))
}
if err := t.ResourcesConfig.Validate(); err != nil {
mErr = errors.Join(mErr, fmt.Errorf("task resources validation failed: %v", err))
}
return mErr
}

Expand All @@ -129,15 +130,18 @@ func (t *Task) ValidateSubmission() error {
if err := t.Publisher.ValidateAllowBlank(); err != nil {
mErr = errors.Join(mErr, fmt.Errorf("publisher validation failed: %v", err))
}
if err := t.Timeouts.ValidateSubmission(); err != nil {
mErr = errors.Join(mErr, fmt.Errorf("task timeouts validation failed: %v", err))
}
if err := t.ResourcesConfig.Validate(); err != nil {
mErr = errors.Join(mErr, fmt.Errorf("task resources validation failed: %v", err))
}
if err := ValidateSlice(t.InputSources); err != nil {
mErr = errors.Join(mErr, fmt.Errorf("artifact validation failed: %v", err))
}
if err := ValidateSlice(t.ResultPaths); err != nil {
mErr = errors.Join(mErr, fmt.Errorf("output validation failed: %v", err))
}
if len(t.ResultPaths) > 0 && t.Publisher.IsEmpty() {
mErr = errors.Join(mErr, errors.New("publisher must be set if result paths are set"))
}

// Check for collisions in input sources
seenInputAliases := make(map[string]bool)
Expand Down
94 changes: 76 additions & 18 deletions pkg/models/task_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,11 +46,19 @@ func (suite *TaskTestSuite) TestTaskNormalization() {
}

func (suite *TaskTestSuite) TestTaskValidation() {
type validationMode int

const (
noError validationMode = iota
submissionError
postSubmissionError
)

tests := []struct {
name string
task *Task
wantErr bool
errMsg string
name string
task *Task
validationMode validationMode
errMsg string
}{
{
name: "Valid task",
Expand All @@ -67,16 +75,16 @@ func (suite *TaskTestSuite) TestTaskValidation() {
},
Publisher: &SpecConfig{Type: "s3"},
},
wantErr: false,
validationMode: noError,
},
{
name: "Empty task name",
task: &Task{
Name: "",
Engine: &SpecConfig{Type: "docker"},
},
wantErr: true,
errMsg: "missing task name",
validationMode: submissionError,
errMsg: "missing task name",
},
{
name: "Duplicate input source alias",
Expand All @@ -88,8 +96,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
{Alias: "input1", Target: "/input2", Source: &SpecConfig{Type: "http"}},
},
},
wantErr: true,
errMsg: "input source with alias 'input1' already exists",
validationMode: submissionError,
errMsg: "input source with alias 'input1' already exists",
},
{
name: "Duplicate input source target",
Expand All @@ -101,8 +109,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
{Alias: "input2", Target: "/input", Source: &SpecConfig{Type: "http"}},
},
},
wantErr: true,
errMsg: "input source with target '/input' already exists",
validationMode: submissionError,
errMsg: "input source with target '/input' already exists",
},
{
name: "Duplicate result path name",
Expand All @@ -115,8 +123,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
},
Publisher: &SpecConfig{Type: "s3"},
},
wantErr: true,
errMsg: "result path with name 'output' already exists",
validationMode: submissionError,
errMsg: "result path with name 'output' already exists",
},
{
name: "Duplicate result path",
Expand All @@ -129,8 +137,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
},
Publisher: &SpecConfig{Type: "s3"},
},
wantErr: true,
errMsg: "result path '/output' already exists",
validationMode: submissionError,
errMsg: "result path '/output' already exists",
},
{
name: "Result paths without publisher",
Expand All @@ -141,15 +149,65 @@ func (suite *TaskTestSuite) TestTaskValidation() {
{Name: "output", Path: "/output"},
},
},
wantErr: true,
errMsg: "publisher must be set if result paths are set",
validationMode: postSubmissionError,
errMsg: "publisher must be set if result paths are set",
},
{
name: "Misconfigured timeouts",
task: &Task{
Name: "invalid-timeouts",
Engine: &SpecConfig{Type: "docker"},
Timeouts: &TimeoutConfig{
ExecutionTimeout: 100,
TotalTimeout: 10,
},
},
validationMode: postSubmissionError,
errMsg: "should be less than total timeout",
},
{
name: "Invalid timeouts",
task: &Task{
Name: "invalid-timeouts",
Engine: &SpecConfig{Type: "docker"},
Timeouts: &TimeoutConfig{
ExecutionTimeout: -1,
},
},
validationMode: submissionError,
errMsg: "task timeouts validation failed",
},
{
name: "Invalid resources",
task: &Task{
Name: "invalid-resources",
Engine: &SpecConfig{Type: "docker"},
ResourcesConfig: &ResourcesConfig{
CPU: "-1",
},
},
validationMode: submissionError,
errMsg: "task resources validation failed",
},
}

for _, tt := range tests {
suite.Run(tt.name, func() {
tt.task.Normalize()

// Test ValidateSubmission()
err := tt.task.ValidateSubmission()
if tt.wantErr {
if tt.validationMode == submissionError {
suite.Error(err)
suite.Contains(err.Error(), tt.errMsg)
} else {
suite.NoError(err)
}

// Test Validate()
// Should always fail if ValidateSubmission() failed
err = tt.task.Validate()
if tt.validationMode != noError {
suite.Error(err)
suite.Contains(err.Error(), tt.errMsg)
} else {
Expand Down
25 changes: 18 additions & 7 deletions pkg/models/timeout.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,25 @@ func (t *TimeoutConfig) Copy() *TimeoutConfig {
}
}

// Validate is used to check a timeout config for reasonable configuration.
// This is called after server side defaults are applied.
func (t *TimeoutConfig) Validate() error {
if t == nil {
return errors.New("missing timeout config")
}
mErr := t.ValidateSubmission()
if t.TotalTimeout > 0 {
if (t.ExecutionTimeout + t.QueueTimeout) > t.TotalTimeout {
mErr = errors.Join(mErr, fmt.Errorf(
"execution timeout %s and queue timeout %s should be less than total timeout %s",
t.GetExecutionTimeout(), t.GetQueueTimeout(), t.GetTotalTimeout()))
}
}
return mErr
}

// ValidateSubmission is used to check a timeout config for reasonable configuration when it is submitted.
func (t *TimeoutConfig) ValidateSubmission() error {
if t == nil {
return errors.New("missing timeout config")
}
Expand All @@ -68,12 +86,5 @@ func (t *TimeoutConfig) Validate() error {
if t.TotalTimeout < 0 {
mErr = errors.Join(mErr, fmt.Errorf("invalid total timeout value: %s", t.GetTotalTimeout()))
}
if t.TotalTimeout > 0 {
if (t.ExecutionTimeout + t.QueueTimeout) > t.TotalTimeout {
mErr = errors.Join(mErr, fmt.Errorf(
"execution timeout %s and queue timeout %s should be less than total timeout %s",
t.GetExecutionTimeout(), t.GetQueueTimeout(), t.GetTotalTimeout()))
}
}
return mErr
}

0 comments on commit e4ccb07

Please sign in to comment.