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

return run details for list run

This commit is contained in:
IronPan 2019-04-05 00:48:48 -07:00
parent c246ca195e
commit f6b8139e19
7 changed files with 173 additions and 279 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
backend/api/run.proto
View File

@ -150,7 +150,8 @@ message TerminateRunRequest {
}
message ListRunsResponse {
repeated Run runs = 1;
repeated Run runs = 1 [deprecated=true];
repeated RunDetail run_details = 4;
int32 total_size = 3;
string next_page_token = 2;
}

2
backend/src/apiserver/resource/resource_manager.go
View File

@ -230,7 +230,7 @@ func (r *ResourceManager) GetRun(runId string) (*model.RunDetail, error) {
}
func (r *ResourceManager) ListRuns(filterContext *common.FilterContext,
opts *list.Options) (runs []*model.Run, total_size int, nextPageToken string, err error) {
opts *list.Options) (runs []*model.RunDetail, total_size int, nextPageToken string, err error) {
return r.runStore.ListRuns(filterContext, opts)
}

12
backend/src/apiserver/server/api_converter.go
View File

@ -131,10 +131,18 @@ func toApiRun(run *model.Run) *api.Run {
}
}
func ToApiRuns(runs []*model.Run) []*api.Run {
func ToApiRuns(runs []*model.RunDetail) []*api.Run {
apiRuns := make([]*api.Run, 0)
for _, run := range runs {
apiRuns = append(apiRuns, toApiRun(run))
apiRuns = append(apiRuns, toApiRun(&run.Run))
}
return apiRuns
}
func ToApiRunDetails(runs []*model.RunDetail) []*api.RunDetail {
apiRuns := make([]*api.RunDetail, 0)
for _, run := range runs {
apiRuns = append(apiRuns, ToApiRunDetail(run))
}
return apiRuns
}

13
backend/src/apiserver/server/api_converter_test.go
View File

@ -124,7 +124,8 @@ func TestToApiRuns(t *testing.T) {
Value: &api.RunMetric_NumberValue{NumberValue: metric2.NumberValue},
Format: api.RunMetric_PERCENTAGE,
}
modelRun1 := model.Run{
modelRun1 := model.RunDetail{
Run: model.Run{
UUID: "run1",
Name: "name1",
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
@ -141,8 +142,12 @@ func TestToApiRuns(t *testing.T) {
ReferenceUUID: "job1", ReferenceType: common.Job, Relationship: common.Creator},
},
Metrics: []*model.RunMetric{metric1, metric2},
},
PipelineRuntime: model.PipelineRuntime{WorkflowRuntimeManifest: "workflow123"},
}
modelRun2 := model.Run{
modelRun2 := model.RunDetail{
Run: model.Run{
UUID: "run2",
Name: "name2",
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
@ -159,8 +164,10 @@ func TestToApiRuns(t *testing.T) {
ReferenceUUID: "job2", ReferenceType: common.Job, Relationship: common.Creator},
},
Metrics: []*model.RunMetric{metric2},
},
PipelineRuntime: model.PipelineRuntime{WorkflowRuntimeManifest: "workflow123"},
}
apiRuns := ToApiRuns([]*model.Run{&modelRun1, &modelRun2})
apiRuns := ToApiRuns([]*model.RunDetail{&modelRun1, &modelRun2})
expectedApiRun := []*api.Run{
{
Id: "run1",

2
backend/src/apiserver/server/run_server.go
View File

@ -63,7 +63,7 @@ func (s *RunServer) ListRuns(ctx context.Context, request *api.ListRunsRequest)
if err != nil {
return nil, util.Wrap(err, "Failed to list runs.")
}
return &api.ListRunsResponse{Runs: ToApiRuns(runs), TotalSize: int32(total_size), NextPageToken: nextPageToken}, nil
return &api.ListRunsResponse{Runs: ToApiRuns(runs), RunDetails: ToApiRunDetails(runs),TotalSize: int32(total_size), NextPageToken: nextPageToken}, nil
}
func (s *RunServer) ArchiveRun(ctx context.Context, request *api.ArchiveRunRequest) (*empty.Empty, error) {

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

@ -34,7 +34,7 @@ import (
type RunStoreInterface interface {
GetRun(runId string) (*model.RunDetail, error)
ListRuns(filterContext *common.FilterContext, opts *list.Options) ([]*model.Run, int, string, error)
ListRuns(filterContext *common.FilterContext, opts *list.Options) ([]*model.RunDetail, int, string, error)
// Create a run entry in the database
CreateRun(run *model.RunDetail) (*model.RunDetail, error)
@ -72,8 +72,8 @@ type RunStore struct {
// total_size. The total_size does not reflect the page size, but it does reflect the number of runs
// matching the supplied filters and resource references.
func (s *RunStore) ListRuns(
filterContext *common.FilterContext, opts *list.Options) ([]*model.Run, int, string, error) {
errorF := func(err error) ([]*model.Run, int, string, error) {
filterContext *common.FilterContext, opts *list.Options) ([]*model.RunDetail, int, string, error) {
errorF := func(err error) ([]*model.RunDetail, int, string, error) {
return nil, 0, "", util.NewInternalServerError(err, "Failed to list runs: %v", err)
}
@ -123,18 +123,12 @@ func (s *RunStore) ListRuns(
return errorF(err)
}
var runs []*model.Run
for _, rd := range runDetails {
r := rd.Run
runs = append(runs, &r)
if len(runDetails) <= opts.PageSize {
return runDetails, total_size, "", nil
}
if len(runs) <= opts.PageSize {
return runs, total_size, "", nil
}
npt, err := opts.NextPageToken(runs[opts.PageSize])
return runs[:opts.PageSize], total_size, npt, err
npt, err := opts.NextPageToken(runDetails[opts.PageSize])
return runDetails[:opts.PageSize], total_size, npt, err
}
func (s *RunStore) buildSelectRunsQuery(selectCount bool, opts *list.Options,

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

@ -27,15 +27,7 @@ import (
"google.golang.org/grpc/codes"
)
func initializeRunStore() (*DB, *RunStore) {
db := NewFakeDbOrFatal()
expStore := NewExperimentStore(db, util.NewFakeTimeForEpoch(), util.NewFakeUUIDGeneratorOrFatal(defaultFakeExpId, nil))
expStore.CreateExperiment(&model.Experiment{Name: "exp1"})
expStore = NewExperimentStore(db, util.NewFakeTimeForEpoch(), util.NewFakeUUIDGeneratorOrFatal(defaultFakeExpIdTwo, nil))
expStore.CreateExperiment(&model.Experiment{Name: "exp2"})
runStore := NewRunStore(db, util.NewFakeTimeForEpoch(), nil)
run1 := &model.RunDetail{
var run1 = &model.RunDetail{
Run: model.Run{
UUID: "1",
Name: "run1",
@ -56,8 +48,9 @@ func initializeRunStore() (*DB, *RunStore) {
PipelineRuntime: model.PipelineRuntime{
WorkflowRuntimeManifest: "workflow1",
},
}
run2 := &model.RunDetail{
}
var run2 = &model.RunDetail{
Run: model.Run{
UUID: "2",
Name: "run2",
@ -78,8 +71,9 @@ func initializeRunStore() (*DB, *RunStore) {
PipelineRuntime: model.PipelineRuntime{
WorkflowRuntimeManifest: "workflow1",
},
}
run3 := &model.RunDetail{
}
var run3 = &model.RunDetail{
Run: model.Run{
UUID: "3",
Name: "run3",
@ -100,7 +94,16 @@ func initializeRunStore() (*DB, *RunStore) {
PipelineRuntime: model.PipelineRuntime{
WorkflowRuntimeManifest: "workflow3",
},
}
}
func initializeRunStore() (*DB, *RunStore) {
db := NewFakeDbOrFatal()
expStore := NewExperimentStore(db, util.NewFakeTimeForEpoch(), util.NewFakeUUIDGeneratorOrFatal(defaultFakeExpId, nil))
expStore.CreateExperiment(&model.Experiment{Name: "exp1"})
expStore = NewExperimentStore(db, util.NewFakeTimeForEpoch(), util.NewFakeUUIDGeneratorOrFatal(defaultFakeExpIdTwo, nil))
expStore.CreateExperiment(&model.Experiment{Name: "exp2"})
runStore := NewRunStore(db, util.NewFakeTimeForEpoch(), nil)
runStore.CreateRun(run1)
runStore.CreateRun(run2)
runStore.CreateRun(run3)
@ -111,43 +114,6 @@ func TestListRuns_Pagination(t *testing.T) {
db, runStore := initializeRunStore()
defer db.Close()
expectedFirstPageRuns := []*model.Run{
{
UUID: "1",
Name: "run1",
DisplayName: "run1",
Namespace: "n1",
CreatedAtInSec: 1,
ScheduledAtInSec: 1,
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
Conditions: "Running",
ResourceReferences: []*model.ResourceReference{
{
ResourceUUID: "1", ResourceType: common.Run,
ReferenceUUID: defaultFakeExpId, ReferenceType: common.Experiment,
Relationship: common.Creator,
},
},
}}
expectedSecondPageRuns := []*model.Run{
{
UUID: "2",
Name: "run2",
DisplayName: "run2",
Namespace: "n2",
CreatedAtInSec: 2,
ScheduledAtInSec: 2,
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
Conditions: "done",
ResourceReferences: []*model.ResourceReference{
{
ResourceUUID: "2", ResourceType: common.Run,
ReferenceUUID: defaultFakeExpId, ReferenceType: common.Experiment,
Relationship: common.Creator,
},
},
}}
opts, err := list.NewOptions(&model.Run{}, 1, "", nil)
assert.Nil(t, err)
@ -155,7 +121,7 @@ func TestListRuns_Pagination(t *testing.T) {
&common.FilterContext{ReferenceKey: &common.ReferenceKey{Type: common.Experiment, ID: defaultFakeExpId}}, opts)
assert.Nil(t, err)
assert.Equal(t, 2, total_size)
assert.Equal(t, expectedFirstPageRuns, runs, "Unexpected Run listed.")
assert.Equal(t, []*model.RunDetail{run1}, runs, "Unexpected Run listed.")
assert.NotEmpty(t, nextPageToken)
opts, err = list.NewOptionsFromToken(nextPageToken, 1)
@ -164,7 +130,7 @@ func TestListRuns_Pagination(t *testing.T) {
&common.FilterContext{ReferenceKey: &common.ReferenceKey{Type: common.Experiment, ID: defaultFakeExpId}}, opts)
assert.Nil(t, err)
assert.Equal(t, 2, total_size)
assert.Equal(t, expectedSecondPageRuns, runs, "Unexpected Run listed.")
assert.Equal(t, []*model.RunDetail{run2}, runs, "Unexpected Run listed.")
assert.Empty(t, nextPageToken)
}
@ -209,43 +175,6 @@ func TestListRuns_Pagination_Descend(t *testing.T) {
db, runStore := initializeRunStore()
defer db.Close()
expectedFirstPageRuns := []*model.Run{
{
UUID: "2",
Name: "run2",
DisplayName: "run2",
Namespace: "n2",
CreatedAtInSec: 2,
ScheduledAtInSec: 2,
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
Conditions: "done",
ResourceReferences: []*model.ResourceReference{
{
ResourceUUID: "2", ResourceType: common.Run,
ReferenceUUID: defaultFakeExpId, ReferenceType: common.Experiment,
Relationship: common.Creator,
},
},
}}
expectedSecondPageRuns := []*model.Run{
{
UUID: "1",
Name: "run1",
DisplayName: "run1",
Namespace: "n1",
CreatedAtInSec: 1,
ScheduledAtInSec: 1,
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
Conditions: "Running",
ResourceReferences: []*model.ResourceReference{
{
ResourceUUID: "1", ResourceType: common.Run,
ReferenceUUID: defaultFakeExpId, ReferenceType: common.Experiment,
Relationship: common.Creator,
},
},
}}
opts, err := list.NewOptions(&model.Run{}, 1, "id desc", nil)
assert.Nil(t, err)
runs, total_size, nextPageToken, err := runStore.ListRuns(
@ -253,7 +182,7 @@ func TestListRuns_Pagination_Descend(t *testing.T) {
assert.Nil(t, err)
assert.Equal(t, 2, total_size)
assert.Equal(t, expectedFirstPageRuns, runs, "Unexpected Run listed.")
assert.Equal(t, []*model.RunDetail{run2}, runs, "Unexpected Run listed.")
assert.NotEmpty(t, nextPageToken)
opts, err = list.NewOptionsFromToken(nextPageToken, 1)
@ -262,7 +191,7 @@ func TestListRuns_Pagination_Descend(t *testing.T) {
&common.FilterContext{ReferenceKey: &common.ReferenceKey{Type: common.Experiment, ID: defaultFakeExpId}}, opts)
assert.Nil(t, err)
assert.Equal(t, 2, total_size)
assert.Equal(t, expectedSecondPageRuns, runs, "Unexpected Run listed.")
assert.Equal(t, []*model.RunDetail{run1}, runs, "Unexpected Run listed.")
assert.Empty(t, nextPageToken)
}
@ -270,49 +199,13 @@ func TestListRuns_Pagination_LessThanPageSize(t *testing.T) {
db, runStore := initializeRunStore()
defer db.Close()
expectedRuns := []*model.Run{
{
UUID: "1",
Name: "run1",
DisplayName: "run1",
Namespace: "n1",
CreatedAtInSec: 1,
ScheduledAtInSec: 1,
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
Conditions: "Running",
ResourceReferences: []*model.ResourceReference{
{
ResourceUUID: "1", ResourceType: common.Run,
ReferenceUUID: defaultFakeExpId, ReferenceType: common.Experiment,
Relationship: common.Creator,
},
},
},
{
UUID: "2",
Name: "run2",
DisplayName: "run2",
Namespace: "n2",
CreatedAtInSec: 2,
ScheduledAtInSec: 2,
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
Conditions: "done",
ResourceReferences: []*model.ResourceReference{
{
ResourceUUID: "2", ResourceType: common.Run,
ReferenceUUID: defaultFakeExpId, ReferenceType: common.Experiment,
Relationship: common.Creator,
},
},
}}
opts, err := list.NewOptions(&model.Run{}, 10, "", nil)
assert.Nil(t, err)
runs, total_size, nextPageToken, err := runStore.ListRuns(
&common.FilterContext{ReferenceKey: &common.ReferenceKey{Type: common.Experiment, ID: defaultFakeExpId}}, opts)
assert.Nil(t, err)
assert.Equal(t, 2, total_size)
assert.Equal(t, expectedRuns, runs, "Unexpected Run listed.")
assert.Equal(t, []*model.RunDetail{run1, run2}, runs, "Unexpected Run listed.")
assert.Empty(t, nextPageToken)
}
@ -331,30 +224,9 @@ func TestGetRun(t *testing.T) {
db, runStore := initializeRunStore()
defer db.Close()
expectedRun := &model.RunDetail{
Run: model.Run{
UUID: "1",
Name: "run1",
DisplayName: "run1",
Namespace: "n1",
CreatedAtInSec: 1,
ScheduledAtInSec: 1,
StorageState: api.Run_STORAGESTATE_AVAILABLE.String(),
Conditions: "Running",
ResourceReferences: []*model.ResourceReference{
{
ResourceUUID: "1", ResourceType: common.Run,
ReferenceUUID: defaultFakeExpId, ReferenceType: common.Experiment,
Relationship: common.Creator,
},
},
},
PipelineRuntime: model.PipelineRuntime{WorkflowRuntimeManifest: "workflow1"},
}
runDetail, err := runStore.GetRun("1")
assert.Nil(t, err)
assert.Equal(t, expectedRun, runDetail)
assert.Equal(t, run1, runDetail)
}
func TestGetRun_NotFoundError(t *testing.T) {
@ -725,8 +597,8 @@ func TestListRuns_WithMetrics(t *testing.T) {
runStore.ReportMetric(metric2)
runStore.ReportMetric(metric3)
expectedRuns := []*model.Run{
{
expectedRuns := []*model.RunDetail{
{Run: model.Run{
UUID: "1",
Name: "run1",
DisplayName: "run1",
@ -744,7 +616,11 @@ func TestListRuns_WithMetrics(t *testing.T) {
},
Metrics: []*model.RunMetric{metric1, metric2},
},
{
PipelineRuntime: model.PipelineRuntime{
WorkflowRuntimeManifest: "workflow1",
},
},
{Run: model.Run{
UUID: "2",
Name: "run2",
DisplayName: "run2",
@ -762,6 +638,10 @@ func TestListRuns_WithMetrics(t *testing.T) {
},
Metrics: []*model.RunMetric{metric3},
},
PipelineRuntime: model.PipelineRuntime{
WorkflowRuntimeManifest: "workflow1",
},
},
}
opts, err := list.NewOptions(&model.Run{}, 2, "", nil)
@ -854,8 +734,8 @@ func TestArchiveRun_IncludedInRunList(t *testing.T) {
assert.Nil(t, getRunErr)
assert.Equal(t, run.Run.StorageState, api.Run_STORAGESTATE_ARCHIVED.String())
expectedRuns := []*model.Run{
{
expectedRuns := []*model.RunDetail{
{Run: model.Run{
UUID: "1",
Name: "run1",
DisplayName: "run1",
@ -871,6 +751,10 @@ func TestArchiveRun_IncludedInRunList(t *testing.T) {
Relationship: common.Creator,
},
},
},
PipelineRuntime: model.PipelineRuntime{
WorkflowRuntimeManifest: "workflow1",
},
}}
opts, err := list.NewOptions(&model.Run{}, 1, "", nil)
runs, total_size, nextPageToken, err := runStore.ListRuns(