kubeflow
/
pipelines
mirror of https://github.com/kubeflow/pipelines.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(backend): Add checking for state change in state history (#8903) 

* Add ListAllRuns endpoint

* Add state history change check. Closes #8902

* Fix unit test
This commit is contained in:
gkcalat 2023-02-28 12:22:04 -08:00 committed by GitHub
parent df0c6c6bc9
commit fde6b944b5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 18 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
backend/src/apiserver/storage/run_store.go
View File

@ -479,10 +479,12 @@ func (s *RunStore) CreateRun(r *model.Run) (*model.Run, error) {
return nil, util.NewInvalidInputError("Invalid value for StorageState field: %q", r.StorageState)
}
r.RunDetails.StateHistory = append(r.RunDetails.StateHistory, &model.RuntimeStatus{
UpdateTimeInSec: s.time.Now().Unix(),
State: r.RunDetails.State,
})
if len(r.RunDetails.StateHistory) == 0 || r.RunDetails.StateHistory[len(r.RunDetails.StateHistory)-1].State != r.RunDetails.State {
r.RunDetails.StateHistory = append(r.RunDetails.StateHistory, &model.RuntimeStatus{
UpdateTimeInSec: s.time.Now().Unix(),
State: r.RunDetails.State,
})
}
stateHistoryString := ""
if history, err := json.Marshal(r.RunDetails.StateHistory); err == nil {
@ -556,10 +558,12 @@ func (s *RunStore) UpdateRun(run *model.Run) error {
if err != nil {
return util.NewInternalServerError(err, "transaction creation failed")
}
run.RunDetails.StateHistory = append(run.RunDetails.StateHistory, &model.RuntimeStatus{
UpdateTimeInSec: s.time.Now().Unix(),
State: run.RunDetails.State,
})
if len(run.RunDetails.StateHistory) == 0 || run.RunDetails.StateHistory[len(run.RunDetails.StateHistory)-1].State != run.RunDetails.State {
run.RunDetails.StateHistory = append(run.RunDetails.StateHistory, &model.RuntimeStatus{
UpdateTimeInSec: s.time.Now().Unix(),
State: run.RunDetails.State,
})
}
stateHistoryString := ""
if historyString, err := json.Marshal(run.RunDetails.StateHistory); err == nil {
stateHistoryString = string(historyString)

4
backend/src/apiserver/storage/run_store_test.go
View File

@ -836,10 +836,6 @@ func TestCreateAndUpdateRun_CreateSuccess(t *testing.T) {
UpdateTimeInSec: 4,
State: model.RuntimeStateRunning,
},
{
UpdateTimeInSec: 5,
State: model.RuntimeStateRunning,
},
},
},
PipelineSpec: model.PipelineSpec{

6
backend/src/apiserver/storage/task_store.go
View File

@ -441,6 +441,12 @@ func (s *TaskStore) CreateOrUpdateTasks(tasks []*model.Task) ([]*model.Task, err
task.CreatedTimestamp = now
}
}
if len(task.StateHistory) == 0 || task.StateHistory[len(task.StateHistory)-1].State != task.State {
task.StateHistory = append(task.StateHistory, &model.RuntimeStatus{
UpdateTimeInSec: s.time.Now().Unix(),
State: task.State,
})
}
}
// Execute the query