This is an automated email from the ASF dual-hosted git repository. mani pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push: new 27385715 [YUNIKORN-2211] Replace Allocation uuid with allocationID (#740) 27385715 is described below commit 27385715af9e8e4e66dfc9ff072efac5db8adffb Author: Manikandan R <maniraj...@gmail.com> AuthorDate: Tue Dec 5 12:28:41 2023 +0530 [YUNIKORN-2211] Replace Allocation uuid with allocationID (#740) Closes: #740 Signed-off-by: Manikandan R <maniraj...@gmail.com> --- go.mod | 4 +- go.sum | 8 +-- pkg/cache/application.go | 6 +-- pkg/cache/application_state.go | 12 ++--- pkg/cache/application_state_test.go | 60 +++++++++++----------- pkg/cache/application_test.go | 26 +++++----- pkg/cache/appmgmt.go | 2 +- pkg/cache/appmgmt_test.go | 2 +- pkg/cache/context_test.go | 12 ++--- pkg/cache/node_test.go | 4 +- pkg/cache/scheduler_callback.go | 8 +-- pkg/cache/task.go | 30 +++++------ pkg/cache/task_state.go | 30 +++++------ pkg/cache/task_test.go | 18 +++---- pkg/cache/utils_test.go | 6 +-- pkg/common/si_helper.go | 4 +- pkg/common/si_helper_test.go | 2 +- pkg/common/test/recoverable_apps_mock.go | 2 +- test/e2e/basic_scheduling/basic_scheduling_test.go | 2 +- .../recovery_and_restart_test.go | 2 +- .../state_aware_app_scheduling/fallback_test.go | 2 +- 21 files changed, 121 insertions(+), 121 deletions(-) diff --git a/go.mod b/go.mod index c6226485..18ccf08d 100644 --- a/go.mod +++ b/go.mod @@ -21,8 +21,8 @@ module github.com/apache/yunikorn-k8shim go 1.20 require ( - github.com/apache/yunikorn-core v0.0.0-20231127054725-3b9c96615796 - github.com/apache/yunikorn-scheduler-interface v0.0.0-20231020041412-6f80d179257c + github.com/apache/yunikorn-core v0.0.0-20231203141034-3ae625bcfc07 + github.com/apache/yunikorn-scheduler-interface v0.0.0-20231201001639-c81397b31653 github.com/google/go-cmp v0.6.0 github.com/google/uuid v1.3.1 github.com/looplab/fsm v1.0.1 diff --git a/go.sum b/go.sum index 819feafd..9f5619fb 100644 --- a/go.sum +++ b/go.sum @@ -49,10 +49,10 @@ github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= github.com/antlr/antlr4/runtime/Go/antlr v1.4.10 h1:yL7+Jz0jTC6yykIK/Wh74gnTJnrGr5AyrNMXuA0gves= github.com/antlr/antlr4/runtime/Go/antlr v1.4.10/go.mod h1:F7bn7fEU90QkQ3tnmaTx3LTKLEDqnwWODIYppRQ5hnY= -github.com/apache/yunikorn-core v0.0.0-20231127054725-3b9c96615796 h1:3OiAqfOtLrldegMNOq6kcq/vDMo4mqhqQkOS96uI0Ik= -github.com/apache/yunikorn-core v0.0.0-20231127054725-3b9c96615796/go.mod h1:nZRI1fm9wa3bhdD4tpDtrEh7ll/Ft/z+NG/gi8l8M14= -github.com/apache/yunikorn-scheduler-interface v0.0.0-20231020041412-6f80d179257c h1:KTIC3f+3aQdAo42YRxs27VpDWY6y73bxXpWcAii2IlQ= -github.com/apache/yunikorn-scheduler-interface v0.0.0-20231020041412-6f80d179257c/go.mod h1:3NQfrhroMqU++kDTroBrTyCRKAczwwX//Fkj/ag/rsY= +github.com/apache/yunikorn-core v0.0.0-20231203141034-3ae625bcfc07 h1:DNhQrQJYmPpujUBzLtSxFyV4Y1L69hVXuNiE0+EitYA= +github.com/apache/yunikorn-core v0.0.0-20231203141034-3ae625bcfc07/go.mod h1:JG66N3TskSNVAMoAUbAVagS14ZrOgcjGpRXbcpAMMvI= +github.com/apache/yunikorn-scheduler-interface v0.0.0-20231201001639-c81397b31653 h1:pUbVmmR+LWuy0L8dGCZNue9UNpWKsY7yFYcCtPtWAic= +github.com/apache/yunikorn-scheduler-interface v0.0.0-20231201001639-c81397b31653/go.mod h1:zDWV5y9Zh9DM1C65RCVXT1nhNNO8kykVW7bzPFamNYw= github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio= github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a h1:idn718Q4B6AGu/h5Sxe66HYVdqdGu2l9Iebqhi/AEoA= github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY= diff --git a/pkg/cache/application.go b/pkg/cache/application.go index 7e318533..421e5d80 100644 --- a/pkg/cache/application.go +++ b/pkg/cache/application.go @@ -619,14 +619,14 @@ func (app *Application) handleFailApplicationEvent(errMsg string) { } } -func (app *Application) handleReleaseAppAllocationEvent(allocUUID string, terminationType string) { +func (app *Application) handleReleaseAppAllocationEvent(allocationID string, terminationType string) { log.Log(log.ShimCacheApplication).Info("try to release pod from application", zap.String("appID", app.applicationID), - zap.String("allocationUUID", allocUUID), + zap.String("allocationID", allocationID), zap.String("terminationType", terminationType)) for _, task := range app.taskMap { - if task.allocationUUID == allocUUID { + if task.allocationID == allocationID { task.setTaskTerminationType(terminationType) err := task.DeleteTaskPod(task.pod) if err != nil { diff --git a/pkg/cache/application_state.go b/pkg/cache/application_state.go index 4afe33d2..940b2be9 100644 --- a/pkg/cache/application_state.go +++ b/pkg/cache/application_state.go @@ -268,15 +268,15 @@ func (ue UpdateApplicationReservationEvent) GetApplicationID() string { // ------------------------ type ReleaseAppAllocationEvent struct { applicationID string - allocationUUID string + allocationID string terminationType string event ApplicationEventType } -func NewReleaseAppAllocationEvent(appID string, allocTermination si.TerminationType, uuid string) ReleaseAppAllocationEvent { +func NewReleaseAppAllocationEvent(appID string, allocTermination si.TerminationType, allocationID string) ReleaseAppAllocationEvent { return ReleaseAppAllocationEvent{ applicationID: appID, - allocationUUID: uuid, + allocationID: allocationID, terminationType: si.TerminationType_name[int32(allocTermination)], event: ReleaseAppAllocation, } @@ -288,7 +288,7 @@ func (re ReleaseAppAllocationEvent) GetApplicationID() string { func (re ReleaseAppAllocationEvent) GetArgs() []interface{} { args := make([]interface{}, 2) - args[0] = re.allocationUUID + args[0] = re.allocationID args[1] = re.terminationType return args } @@ -558,9 +558,9 @@ func newAppState() *fsm.FSM { //nolint:funlen log.Log(log.ShimFSM).Error("fail to parse event arg", zap.Error(err)) return } - allocUUID := eventArgs[0] + allocationID := eventArgs[0] terminationType := eventArgs[1] - app.handleReleaseAppAllocationEvent(allocUUID, terminationType) + app.handleReleaseAppAllocationEvent(allocationID, terminationType) }, ReleaseAppAllocationAsk.String(): func(_ context.Context, event *fsm.Event) { app := event.Args[0].(*Application) //nolint:errcheck diff --git a/pkg/cache/application_state_test.go b/pkg/cache/application_state_test.go index 7bdc572d..c2f94e93 100644 --- a/pkg/cache/application_state_test.go +++ b/pkg/cache/application_state_test.go @@ -671,22 +671,22 @@ func TestUpdateApplicationReservationEventGetApplicationID(t *testing.T) { func TestNewReleaseAppAllocationEvent(t *testing.T) { tests := []struct { - name string - appID, allocationUUID string - terminationType si.TerminationType - wantID, wantUUID, wantType string - wantEvent ApplicationEventType + name string + appID, allocationID string + terminationType si.TerminationType + wantID, wantAllocationID, wantType string + wantEvent ApplicationEventType }{ - {TestCreateName, "testAppId001", "testUUID001", si.TerminationType_TIMEOUT, "testAppId001", "testUUID001", "TIMEOUT", ReleaseAppAllocation}, + {TestCreateName, "testAppId001", "testAllocationID001", si.TerminationType_TIMEOUT, "testAppId001", "testAllocationID001", "TIMEOUT", ReleaseAppAllocation}, } for _, tt := range tests { - instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationUUID) + instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationID) t.Run(tt.name, func(t *testing.T) { - if instance.applicationID != tt.wantID || instance.allocationUUID != tt.wantUUID || instance.terminationType != tt.wantType || instance.event != tt.wantEvent { + if instance.applicationID != tt.wantID || instance.allocationID != tt.wantAllocationID || instance.terminationType != tt.wantType || instance.event != tt.wantEvent { t.Errorf("want %s %s %s %s, got %s %s %s %s", - tt.wantID, tt.wantUUID, tt.wantType, tt.wantEvent, - instance.applicationID, instance.allocationUUID, instance.terminationType, instance.event) + tt.wantID, tt.wantAllocationID, tt.wantType, tt.wantEvent, + instance.applicationID, instance.allocationID, instance.terminationType, instance.event) } }) } @@ -694,16 +694,16 @@ func TestNewReleaseAppAllocationEvent(t *testing.T) { func TestReleaseAppAllocationEventGetEvent(t *testing.T) { tests := []struct { - name string - appID, allocationUUID string - terminationType si.TerminationType - wantEvent ApplicationEventType + name string + appID, allocationID string + terminationType si.TerminationType + wantEvent ApplicationEventType }{ - {TestEventName, "testAppId001", "testUUID001", si.TerminationType_TIMEOUT, ReleaseAppAllocation}, + {TestEventName, "testAppId001", "testAllocationID001", si.TerminationType_TIMEOUT, ReleaseAppAllocation}, } for _, tt := range tests { - instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationUUID) + instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationID) event := instance.GetEvent() t.Run(tt.name, func(t *testing.T) { if event != tt.wantEvent.String() { @@ -715,18 +715,18 @@ func TestReleaseAppAllocationEventGetEvent(t *testing.T) { func TestReleaseAppAllocationEventGetArgs(t *testing.T) { tests := []struct { - name string - appID, allocationUUID string - terminationType si.TerminationType - wantLen int - castOk []bool - wantArg []string + name string + appID, allocationID string + terminationType si.TerminationType + wantLen int + castOk []bool + wantArg []string }{ - {TestArgsName, "testAppId001", "testUUID001", si.TerminationType_TIMEOUT, 2, []bool{true, true}, []string{"testUUID001", "TIMEOUT"}}, + {TestArgsName, "testAppId001", "testAllocationID001", si.TerminationType_TIMEOUT, 2, []bool{true, true}, []string{"testAllocationID001", "TIMEOUT"}}, } for _, tt := range tests { - instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationUUID) + instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationID) args := instance.GetArgs() t.Run(tt.name, func(t *testing.T) { if len(args) != tt.wantLen { @@ -748,16 +748,16 @@ func TestReleaseAppAllocationEventGetArgs(t *testing.T) { func TestReleaseAppAllocationEventGetApplicationID(t *testing.T) { tests := []struct { - name string - appID, allocationUUID string - terminationType si.TerminationType - wantID string + name string + appID, allocationID string + terminationType si.TerminationType + wantID string }{ - {TestAppIDName, "testAppId001", "testUUID001", si.TerminationType_TIMEOUT, "testAppId001"}, + {TestAppIDName, "testAppId001", "testAllocationID001", si.TerminationType_TIMEOUT, "testAppId001"}, } for _, tt := range tests { - instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationUUID) + instance := NewReleaseAppAllocationEvent(tt.appID, tt.terminationType, tt.allocationID) appID := instance.GetApplicationID() t.Run(tt.name, func(t *testing.T) { if appID != tt.wantID { diff --git a/pkg/cache/application_test.go b/pkg/cache/application_test.go index 092d5d9e..b896f4cd 100644 --- a/pkg/cache/application_test.go +++ b/pkg/cache/application_test.go @@ -442,9 +442,9 @@ func TestReleaseAppAllocation(t *testing.T) { app := NewApplication(appID, "root.abc", "testuser", testGroups, map[string]string{}, ms) task := NewTask("task01", app, context, pod) app.addTask(task) - task.allocationUUID = taskUUID + task.allocationID = taskAllocationID // app must be running states - err := app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskUUID)) + err := app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskAllocationID)) if err == nil { // this should give an error t.Error("expecting error got 'nil'") @@ -452,7 +452,7 @@ func TestReleaseAppAllocation(t *testing.T) { // set app states to running, let event can be trigger app.SetState(ApplicationStates().Running) assertAppState(t, app, ApplicationStates().Running, 3*time.Second) - err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskUUID)) + err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskAllocationID)) assert.NilError(t, err) // after handle release event the states of app must be running assertAppState(t, app, ApplicationStates().Running, 3*time.Second) @@ -816,7 +816,7 @@ func TestTryReservePostRestart(t *testing.T) { Containers: containers, }, }) - task0.allocationUUID = string(task0.pod.UID) + task0.allocationID = string(task0.pod.UID) task0.nodeName = "fake-host" task0.sm.SetState(TaskStates().Allocated) @@ -997,9 +997,9 @@ func TestReleaseAppAllocationInFailingState(t *testing.T) { app := NewApplication(appID, "root.abc", "testuser", testGroups, map[string]string{}, ms) task := NewTask("task01", app, context, pod) app.addTask(task) - task.allocationUUID = taskUUID + task.allocationID = taskAllocationID // app must be running states - err := app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskUUID)) + err := app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskAllocationID)) if err == nil { // this should give an error t.Error("expecting error got 'nil'") @@ -1007,12 +1007,12 @@ func TestReleaseAppAllocationInFailingState(t *testing.T) { // set app states to running, let event can be trigger app.SetState(ApplicationStates().Running) assertAppState(t, app, ApplicationStates().Running, 3*time.Second) - err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskUUID)) + err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskAllocationID)) assert.NilError(t, err) // after handle release event the states of app must be running assertAppState(t, app, ApplicationStates().Running, 3*time.Second) app.SetState(ApplicationStates().Failing) - err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskUUID)) + err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, taskAllocationID)) assert.NilError(t, err) // after handle release event the states of app must be failing assertAppState(t, app, ApplicationStates().Failing, 3*time.Second) @@ -1052,7 +1052,7 @@ func TestResumingStateTransitions(t *testing.T) { // Add tasks app.addTask(task1) app.addTask(task2) - task1.allocationUUID = taskUUID + task1.allocationID = taskAllocationID context.addApplication(app) // Set app state to "reserving" @@ -1167,7 +1167,7 @@ func TestPlaceholderTimeoutEvents(t *testing.T) { assert.Equal(t, len(app.GetNewTasks()), 1) appID := "app00001" - UUID := "UID-POD-00002" + allocationID := "UID-POD-00002" context.addApplication(app) task1 := context.AddTask(&AddTaskRequest{ @@ -1184,16 +1184,16 @@ func TestPlaceholderTimeoutEvents(t *testing.T) { _, taskErr := app.GetTask("task02") assert.NilError(t, taskErr, "Task should exist") - task1.allocationUUID = UUID + task1.allocationID = allocationID // app must be running states - err := app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, UUID)) + err := app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, allocationID)) assert.Error(t, err, "event ReleaseAppAllocation inappropriate in current state New") // set app states to running, let event can be trigger app.SetState(ApplicationStates().Running) assertAppState(t, app, ApplicationStates().Running, 3*time.Second) - err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, UUID)) + err = app.handle(NewReleaseAppAllocationEvent(appID, si.TerminationType_TIMEOUT, allocationID)) assert.NilError(t, err) // after handle release event the states of app must be running assertAppState(t, app, ApplicationStates().Running, 3*time.Second) diff --git a/pkg/cache/appmgmt.go b/pkg/cache/appmgmt.go index 25a67fda..a0992ef6 100644 --- a/pkg/cache/appmgmt.go +++ b/pkg/cache/appmgmt.go @@ -119,7 +119,7 @@ func (svc *AppManagementService) GetExistingAllocation(pod *v1.Pod) *si.Allocati return &si.Allocation{ AllocationKey: string(pod.UID), AllocationTags: meta.Tags, - UUID: string(pod.UID), + AllocationID: string(pod.UID), ResourcePerAlloc: common.GetPodResource(pod), NodeID: pod.Spec.NodeName, ApplicationID: meta.ApplicationID, diff --git a/pkg/cache/appmgmt_test.go b/pkg/cache/appmgmt_test.go index bff38703..cbeb9ac6 100644 --- a/pkg/cache/appmgmt_test.go +++ b/pkg/cache/appmgmt_test.go @@ -425,7 +425,7 @@ func TestAMSvcGetExistingAllocation(t *testing.T) { alloc := am.GetExistingAllocation(pod) assert.Equal(t, alloc.ApplicationID, "app00001") assert.Equal(t, alloc.AllocationKey, string(pod.UID)) - assert.Equal(t, alloc.UUID, string(pod.UID)) + assert.Equal(t, alloc.AllocationID, string(pod.UID)) assert.Equal(t, alloc.NodeID, "allocated-node") } diff --git a/pkg/cache/context_test.go b/pkg/cache/context_test.go index 9ebb59c3..4090d2e9 100644 --- a/pkg/cache/context_test.go +++ b/pkg/cache/context_test.go @@ -846,11 +846,11 @@ func TestRecoverTask(t *testing.T) { assert.Equal(t, len(app.getTasks(TaskStates().New)), 1) taskInfoVerifiers := []struct { - taskID string - expectedState string - expectedAllocationUUID string - expectedPodName string - expectedNodeName string + taskID string + expectedState string + expectedAllocationID string + expectedPodName string + expectedNodeName string }{ {taskUID1, TaskStates().Bound, taskUID1, "pod1", fakeNodeName}, {taskUID2, TaskStates().Completed, taskUID2, "pod2", fakeNodeName}, @@ -864,7 +864,7 @@ func TestRecoverTask(t *testing.T) { rt, err := app.GetTask(tt.taskID) assert.NilError(t, err) assert.Equal(t, rt.GetTaskState(), tt.expectedState) - assert.Equal(t, rt.allocationUUID, tt.expectedAllocationUUID) + assert.Equal(t, rt.allocationID, tt.expectedAllocationID) assert.Equal(t, rt.pod.Name, tt.expectedPodName) assert.Equal(t, rt.alias, fmt.Sprintf("%s/%s", podNamespace, tt.expectedPodName)) }) diff --git a/pkg/cache/node_test.go b/pkg/cache/node_test.go index e244c2cb..f8dbcd9d 100644 --- a/pkg/cache/node_test.go +++ b/pkg/cache/node_test.go @@ -36,7 +36,7 @@ func TestAddExistingAllocation(t *testing.T) { alloc01 := si.Allocation{ AllocationKey: "pod001", AllocationTags: nil, - UUID: "podUID001", + AllocationID: "podUID001", ResourcePerAlloc: nil, Priority: 0, NodeID: "host001", @@ -47,7 +47,7 @@ func TestAddExistingAllocation(t *testing.T) { assert.Equal(t, len(node.existingAllocations), 1) alloc02 := node.existingAllocations[0] assert.Equal(t, alloc02.AllocationKey, alloc01.AllocationKey) - assert.Equal(t, alloc02.UUID, alloc01.UUID) + assert.Equal(t, alloc02.AllocationID, alloc01.AllocationID) assert.Equal(t, alloc02.NodeID, alloc01.NodeID) assert.Equal(t, alloc02.PartitionName, alloc01.PartitionName) } diff --git a/pkg/cache/scheduler_callback.go b/pkg/cache/scheduler_callback.go index 5dad3db0..992c4817 100644 --- a/pkg/cache/scheduler_callback.go +++ b/pkg/cache/scheduler_callback.go @@ -51,7 +51,7 @@ func (callback *AsyncRMCallback) UpdateAllocation(response *si.AllocationRespons // got allocation for pod, bind pod to the scheduled node log.Log(log.ShimRMCallback).Debug("callback: response to new allocation", zap.String("allocationKey", alloc.AllocationKey), - zap.String("UUID", alloc.UUID), + zap.String("allocationID", alloc.AllocationID), zap.String("applicationID", alloc.ApplicationID), zap.String("nodeID", alloc.NodeID)) @@ -60,7 +60,7 @@ func (callback *AsyncRMCallback) UpdateAllocation(response *si.AllocationRespons return err } if app := callback.context.GetApplication(alloc.ApplicationID); app != nil { - ev := NewAllocateTaskEvent(app.GetApplicationID(), alloc.AllocationKey, alloc.UUID, alloc.NodeID) + ev := NewAllocateTaskEvent(app.GetApplicationID(), alloc.AllocationKey, alloc.AllocationID, alloc.NodeID) dispatcher.Dispatch(ev) } } @@ -78,7 +78,7 @@ func (callback *AsyncRMCallback) UpdateAllocation(response *si.AllocationRespons for _, release := range response.Released { log.Log(log.ShimRMCallback).Debug("callback: response to released allocations", - zap.String("UUID", release.UUID)) + zap.String("AllocationID", release.AllocationID)) // update cache callback.context.ForgetPod(release.GetAllocationKey()) @@ -86,7 +86,7 @@ func (callback *AsyncRMCallback) UpdateAllocation(response *si.AllocationRespons // TerminationType 0 mean STOPPED_BY_RM if release.TerminationType != si.TerminationType_STOPPED_BY_RM { // send release app allocation to application states machine - ev := NewReleaseAppAllocationEvent(release.ApplicationID, release.TerminationType, release.UUID) + ev := NewReleaseAppAllocationEvent(release.ApplicationID, release.TerminationType, release.AllocationID) dispatcher.Dispatch(ev) } } diff --git a/pkg/cache/task.go b/pkg/cache/task.go index 8c58695b..0d20cd52 100644 --- a/pkg/cache/task.go +++ b/pkg/cache/task.go @@ -44,7 +44,7 @@ type Task struct { alias string applicationID string application *Application - allocationUUID string + allocationID string resource *si.Resource pod *v1.Pod podStatus v1.PodStatus // pod status, maintained separately for efficiency reasons @@ -169,10 +169,10 @@ func (task *Task) getTaskGroupName() string { return task.taskGroupName } -func (task *Task) getTaskAllocationUUID() string { +func (task *Task) getTaskAllocationID() string { task.lock.RLock() defer task.lock.RUnlock() - return task.allocationUUID + return task.allocationID } func (task *Task) DeleteTaskPod(pod *v1.Pod) error { @@ -209,13 +209,13 @@ func (task *Task) initialize() { // scheduled by us with an allocation, instead of starting // from New, directly set the task to Bound. if utils.NeedRecovery(task.pod) { - task.allocationUUID = string(task.pod.UID) + task.allocationID = string(task.pod.UID) task.nodeName = task.pod.Spec.NodeName task.sm.SetState(TaskStates().Bound) log.Log(log.ShimCacheTask).Info("set task as Bound", zap.String("appID", task.applicationID), zap.String("taskID", task.taskID), - zap.String("allocationUUID", task.allocationUUID), + zap.String("allocationID", task.allocationID), zap.String("nodeName", task.nodeName)) } @@ -224,13 +224,13 @@ func (task *Task) initialize() { // the resources were already released, instead of starting // from New, directly set the task to Completed if utils.IsPodTerminated(task.pod) { - task.allocationUUID = string(task.pod.UID) + task.allocationID = string(task.pod.UID) task.nodeName = task.pod.Spec.NodeName task.sm.SetState(TaskStates().Completed) log.Log(log.ShimCacheTask).Info("set task as Completed", zap.String("appID", task.applicationID), zap.String("taskID", task.taskID), - zap.String("allocationUUID", task.allocationUUID), + zap.String("allocationID", task.allocationID), zap.String("nodeName", task.nodeName)) } } @@ -399,9 +399,9 @@ func (task *Task) postTaskAllocated() { // If we find the task is already in Completed state while handling TaskAllocated // event, we need to explicitly release this allocation because it is no // longer valid. -func (task *Task) beforeTaskAllocated(eventSrc string, allocUUID string, nodeID string) { - // task is allocated on a node with a UUID set the details in the task here to allow referencing later. - task.allocationUUID = allocUUID +func (task *Task) beforeTaskAllocated(eventSrc string, allocationID string, nodeID string) { + // task is allocated on a node with a allocationID set the details in the task here to allow referencing later. + task.allocationID = allocationID task.nodeName = nodeID // If the task is Completed the pod was deleted on K8s but the core was not aware yet. // Notify the core to release this allocation to avoid resource leak. @@ -409,7 +409,7 @@ func (task *Task) beforeTaskAllocated(eventSrc string, allocUUID string, nodeID if eventSrc == TaskStates().Completed { log.Log(log.ShimCacheTask).Info("task is already completed, invalidate the allocation", zap.String("currentTaskState", eventSrc), - zap.String("allocUUID", allocUUID), + zap.String("allocationID", allocationID), zap.String("allocatedNode", nodeID)) task.releaseAllocation() } @@ -491,7 +491,7 @@ func (task *Task) releaseAllocation() { zap.String("applicationID", task.applicationID), zap.String("taskID", task.taskID), zap.String("taskAlias", task.alias), - zap.String("allocationUUID", task.allocationUUID), + zap.String("allocationID", task.allocationID), zap.String("task", task.GetTaskState()), zap.String("terminationType", task.terminationType)) @@ -505,8 +505,8 @@ func (task *Task) releaseAllocation() { releaseRequest = common.CreateReleaseAskRequestForTask( task.applicationID, task.taskID, task.application.partition) default: - if task.allocationUUID == "" { - log.Log(log.ShimCacheTask).Warn("BUG: task allocation UUID is empty on release", + if task.allocationID == "" { + log.Log(log.ShimCacheTask).Warn("BUG: task allocation allocationID is empty on release", zap.String("applicationID", task.applicationID), zap.String("taskID", task.taskID), zap.String("taskAlias", task.alias), @@ -514,7 +514,7 @@ func (task *Task) releaseAllocation() { return } releaseRequest = common.CreateReleaseAllocationRequestForTask( - task.applicationID, task.allocationUUID, task.application.partition, task.terminationType) + task.applicationID, task.allocationID, task.application.partition, task.terminationType) } if releaseRequest.Releases != nil { diff --git a/pkg/cache/task_state.go b/pkg/cache/task_state.go index b628533b..808df8a0 100644 --- a/pkg/cache/task_state.go +++ b/pkg/cache/task_state.go @@ -123,20 +123,20 @@ func (st SubmitTaskEvent) GetApplicationID() string { // Allocate Event // ------------------------ type AllocatedTaskEvent struct { - applicationID string - taskID string - event TaskEventType - nodeID string - allocationUUID string + applicationID string + taskID string + event TaskEventType + nodeID string + allocationID string } -func NewAllocateTaskEvent(appID string, taskID string, allocUUID string, nid string) AllocatedTaskEvent { +func NewAllocateTaskEvent(appID string, taskID string, allocationID string, nid string) AllocatedTaskEvent { return AllocatedTaskEvent{ - applicationID: appID, - taskID: taskID, - event: TaskAllocated, - allocationUUID: allocUUID, - nodeID: nid, + applicationID: appID, + taskID: taskID, + event: TaskAllocated, + allocationID: allocationID, + nodeID: nid, } } @@ -146,7 +146,7 @@ func (ae AllocatedTaskEvent) GetEvent() string { func (ae AllocatedTaskEvent) GetArgs() []interface{} { args := make([]interface{}, 2) - args[0] = ae.allocationUUID + args[0] = ae.allocationID args[1] = ae.nodeID return args } @@ -419,15 +419,15 @@ func newTaskState() *fsm.FSM { }, beforeHook(TaskAllocated): func(_ context.Context, event *fsm.Event) { task := event.Args[0].(*Task) //nolint:errcheck - // All allocation events must include the allocUUID and nodeID passed from the core + // All allocation events must include the allocationID and nodeID passed from the core eventArgs := make([]string, 2) if err := events.GetEventArgsAsStrings(eventArgs, event.Args[1].([]interface{})); err != nil { log.Log(log.ShimFSM).Error("failed to parse event arg", zap.Error(err)) return } - allocUUID := eventArgs[0] + allocationID := eventArgs[0] nodeID := eventArgs[1] - task.beforeTaskAllocated(event.Src, allocUUID, nodeID) + task.beforeTaskAllocated(event.Src, allocationID, nodeID) }, beforeHook(CompleteTask): func(_ context.Context, event *fsm.Event) { task := event.Args[0].(*Task) //nolint:errcheck diff --git a/pkg/cache/task_test.go b/pkg/cache/task_test.go index f9b9afcb..d31339c9 100644 --- a/pkg/cache/task_test.go +++ b/pkg/cache/task_test.go @@ -200,9 +200,9 @@ func TestReleaseTaskAllocation(t *testing.T) { assert.Equal(t, task.GetTaskState(), TaskStates().Allocated) // bind a task is a async process, wait for it to happen err = common.WaitFor(100*time.Millisecond, 3*time.Second, func() bool { - return task.getTaskAllocationUUID() == string(pod.UID) + return task.getTaskAllocationID() == string(pod.UID) }) - assert.NilError(t, err, "failed to wait for allocation UUID being set for task") + assert.NilError(t, err, "failed to wait for allocation allocationID being set for task") // bound event3 := NewBindTaskEvent(app.applicationID, task.taskID) @@ -576,10 +576,10 @@ func TestHandleSubmitTaskEvent(t *testing.T) { func TestSimultaneousTaskCompleteAndAllocate(t *testing.T) { const ( - podUID = "UID-00001" - appID = "app-test-001" - queueName = "root.abc" - allocationUUID = "uuid-xyz" + podUID = "UID-00001" + appID = "app-test-001" + queueName = "root.abc" + allocationID = "allocationid-xyz" ) mockedContext := initContextForTest() mockedAPIProvider, ok := mockedContext.apiProvider.(*client.MockedAPIProvider) @@ -637,7 +637,7 @@ func TestSimultaneousTaskCompleteAndAllocate(t *testing.T) { // can be released from the core to avoid resource leak alloc := &si.Allocation{ AllocationKey: string(pod1.UID), - UUID: allocationUUID, + AllocationID: allocationID, NodeID: "fake-node", ApplicationID: appID, PartitionName: "default", @@ -649,10 +649,10 @@ func TestSimultaneousTaskCompleteAndAllocate(t *testing.T) { "allocationsToRelease is not in the expected length") allocToRelease := request.Releases.AllocationsToRelease[0] assert.Equal(t, allocToRelease.ApplicationID, alloc.ApplicationID) - assert.Equal(t, allocToRelease.UUID, alloc.UUID) + assert.Equal(t, allocToRelease.AllocationID, alloc.AllocationID) return nil }) - ev1 := NewAllocateTaskEvent(app.GetApplicationID(), alloc.AllocationKey, alloc.UUID, alloc.NodeID) + ev1 := NewAllocateTaskEvent(app.GetApplicationID(), alloc.AllocationKey, alloc.AllocationID, alloc.NodeID) err = task1.handle(ev1) assert.NilError(t, err, "failed to handle AllocateTask event") assert.Equal(t, task1.GetTaskState(), TaskStates().Completed) diff --git a/pkg/cache/utils_test.go b/pkg/cache/utils_test.go index cf6d34436..5c79f8cc 100644 --- a/pkg/cache/utils_test.go +++ b/pkg/cache/utils_test.go @@ -30,9 +30,9 @@ import ( ) const ( - appID = "app01" - app2ID = "app02" - taskUUID = "UUID01" + appID = "app01" + app2ID = "app02" + taskAllocationID = "ALLOCATIONID01" ) //nolint:funlen diff --git a/pkg/common/si_helper.go b/pkg/common/si_helper.go index a0638aa6..7c3eadd9 100644 --- a/pkg/common/si_helper.go +++ b/pkg/common/si_helper.go @@ -118,11 +118,11 @@ func GetTerminationTypeFromString(terminationTypeStr string) si.TerminationType return si.TerminationType_STOPPED_BY_RM } -func CreateReleaseAllocationRequestForTask(appID, allocUUID, partition, terminationType string) *si.AllocationRequest { +func CreateReleaseAllocationRequestForTask(appID, allocationID, partition, terminationType string) *si.AllocationRequest { toReleases := make([]*si.AllocationRelease, 0) toReleases = append(toReleases, &si.AllocationRelease{ ApplicationID: appID, - UUID: allocUUID, + AllocationID: allocationID, PartitionName: partition, TerminationType: GetTerminationTypeFromString(terminationType), Message: "task completed", diff --git a/pkg/common/si_helper_test.go b/pkg/common/si_helper_test.go index 4956d7a5..e49a90ef 100644 --- a/pkg/common/si_helper_test.go +++ b/pkg/common/si_helper_test.go @@ -40,7 +40,7 @@ func TestCreateReleaseAllocationRequest(t *testing.T) { assert.Equal(t, len(request.Releases.AllocationsToRelease), 1) assert.Equal(t, len(request.Releases.AllocationAsksToRelease), 0) assert.Equal(t, request.Releases.AllocationsToRelease[0].ApplicationID, "app01") - assert.Equal(t, request.Releases.AllocationsToRelease[0].UUID, "alloc01") + assert.Equal(t, request.Releases.AllocationsToRelease[0].AllocationID, "alloc01") assert.Equal(t, request.Releases.AllocationsToRelease[0].PartitionName, "default") } diff --git a/pkg/common/test/recoverable_apps_mock.go b/pkg/common/test/recoverable_apps_mock.go index 82a9dff4..3ebb1b12 100644 --- a/pkg/common/test/recoverable_apps_mock.go +++ b/pkg/common/test/recoverable_apps_mock.go @@ -40,7 +40,7 @@ func (m *MockedRecoverableAppManager) GetExistingAllocation(pod *v1.Pod) *si.All return &si.Allocation{ AllocationKey: pod.Name, AllocationTags: nil, - UUID: string(pod.UID), + AllocationID: string(pod.UID), ResourcePerAlloc: nil, Priority: 0, NodeID: pod.Spec.NodeName, diff --git a/test/e2e/basic_scheduling/basic_scheduling_test.go b/test/e2e/basic_scheduling/basic_scheduling_test.go index c05405b2..67765360 100644 --- a/test/e2e/basic_scheduling/basic_scheduling_test.go +++ b/test/e2e/basic_scheduling/basic_scheduling_test.go @@ -108,7 +108,7 @@ var _ = ginkgo.Describe("", func() { gomega.Ω(allocation.AllocationKey).NotTo(gomega.BeNil()) gomega.Ω(allocation.NodeID).NotTo(gomega.BeNil()) gomega.Ω(allocation.Partition).NotTo(gomega.BeNil()) - gomega.Ω(allocation.UUID).NotTo(gomega.BeNil()) + gomega.Ω(allocation.AllocationID).NotTo(gomega.BeNil()) gomega.Ω(allocation.ApplicationID).To(gomega.Equal(sleepRespPod.ObjectMeta.Labels["applicationId"])) core := sleepRespPod.Spec.Containers[0].Resources.Requests.Cpu().MilliValue() mem := sleepRespPod.Spec.Containers[0].Resources.Requests.Memory().Value() diff --git a/test/e2e/recovery_and_restart/recovery_and_restart_test.go b/test/e2e/recovery_and_restart/recovery_and_restart_test.go index b31d50a3..1aaa8d1f 100644 --- a/test/e2e/recovery_and_restart/recovery_and_restart_test.go +++ b/test/e2e/recovery_and_restart/recovery_and_restart_test.go @@ -137,7 +137,7 @@ var _ = ginkgo.Describe("", func() { gomega.Ω(allocations.AllocationKey).NotTo(gomega.BeNil()) gomega.Ω(allocations.NodeID).NotTo(gomega.BeNil()) gomega.Ω(allocations.Partition).NotTo(gomega.BeNil()) - gomega.Ω(allocations.UUID).NotTo(gomega.BeNil()) + gomega.Ω(allocations.AllocationID).NotTo(gomega.BeNil()) gomega.Ω(allocations.ApplicationID).To(gomega.Equal(sleepRespPod.ObjectMeta.Labels["applicationId"])) core := sleepRespPod.Spec.Containers[0].Resources.Requests.Cpu().MilliValue() mem := sleepRespPod.Spec.Containers[0].Resources.Requests.Memory().Value() diff --git a/test/e2e/state_aware_app_scheduling/fallback_test.go b/test/e2e/state_aware_app_scheduling/fallback_test.go index aaec7bda..d8cd3216 100644 --- a/test/e2e/state_aware_app_scheduling/fallback_test.go +++ b/test/e2e/state_aware_app_scheduling/fallback_test.go @@ -94,7 +94,7 @@ var _ = Describe("FallbackTest:", func() { Ω(allocation.AllocationKey).NotTo(BeNil()) Ω(allocation.NodeID).NotTo(BeNil()) Ω(allocation.Partition).NotTo(BeNil()) - Ω(allocation.UUID).NotTo(BeNil()) + Ω(allocation.AllocationID).NotTo(BeNil()) Ω(allocation.ApplicationID).To(Equal(sleepRespPod.ObjectMeta.Labels["applicationId"])) core := sleepRespPod.Spec.Containers[0].Resources.Requests.Cpu().MilliValue() mem := sleepRespPod.Spec.Containers[0].Resources.Requests.Memory().Value() --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@yunikorn.apache.org For additional commands, e-mail: issues-h...@yunikorn.apache.org