This repository has been archived by the owner on Oct 9, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 59
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Move logic to track terminated workflows to new TerminatedTrackingSto…
…re workflowstore
- Loading branch information
1 parent
c3dd588
commit 0490a53
Showing
6 changed files
with
203 additions
and
37 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package workflowstore | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/flyteorg/flytepropeller/pkg/apis/flyteworkflow/v1alpha1" | ||
| "github.com/flyteorg/flytestdlib/fastcheck" | ||
|
|
||
| "github.com/flyteorg/flytestdlib/promutils" | ||
| ) | ||
|
|
||
| func workflowKey(namespace, name string) string { | ||
| return fmt.Sprintf("%s/%s", namespace, name) | ||
| } | ||
|
|
||
| // A specialized store that stores a LRU cache of all the workflows that are in a terminal phase. | ||
| // Terminated workflows are ignored (Get returns a nil). | ||
| // Processing terminated FlyteWorkflows can occur when workflow updates are reported after a workflow has already completed. | ||
| type terminatedTracking struct { | ||
| w FlyteWorkflow | ||
| terminatedFilter fastcheck.Filter | ||
| } | ||
|
|
||
| func (t *terminatedTracking) Get(ctx context.Context, namespace, name string) (*v1alpha1.FlyteWorkflow, error) { | ||
| if t.terminatedFilter.Contains(ctx, []byte(workflowKey(namespace, name))) { | ||
| return nil, ErrWorkflowTerminated | ||
| } | ||
|
|
||
| return t.w.Get(ctx, namespace, name) | ||
| } | ||
|
|
||
| func (r *terminatedTracking) UpdateStatus(ctx context.Context, workflow *v1alpha1.FlyteWorkflow, priorityClass PriorityClass) ( | ||
| newWF *v1alpha1.FlyteWorkflow, err error) { | ||
| newWF, err = r.w.UpdateStatus(ctx, workflow, priorityClass) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| if newWF != nil { | ||
| if newWF.GetExecutionStatus().IsTerminated() { | ||
| r.terminatedFilter.Add(ctx, []byte(workflowKey(workflow.Namespace, workflow.Name))) | ||
| } | ||
| } | ||
|
|
||
| return newWF, nil | ||
| } | ||
|
|
||
| func (r *terminatedTracking) Update(ctx context.Context, workflow *v1alpha1.FlyteWorkflow, priorityClass PriorityClass) ( | ||
| newWF *v1alpha1.FlyteWorkflow, err error) { | ||
| newWF, err = r.w.Update(ctx, workflow, priorityClass) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| if newWF != nil { | ||
| if newWF.GetExecutionStatus().IsTerminated() { | ||
| r.terminatedFilter.Add(ctx, []byte(workflowKey(workflow.Namespace, workflow.Name))) | ||
| } | ||
| } | ||
|
|
||
| return newWF, nil | ||
| } | ||
|
|
||
| func NewTerminatedTrackingStore(_ context.Context, scope promutils.Scope, workflowStore FlyteWorkflow) (FlyteWorkflow, error) { | ||
| filter, err := fastcheck.NewLRUCacheFilter(1000, scope.NewSubScope("terminated_filter")) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &terminatedTracking{ | ||
| w: workflowStore, | ||
| terminatedFilter: filter, | ||
| }, nil | ||
| } |
102 changes: 102 additions & 0 deletions
102
pkg/controller/workflowstore/terminated_tracking_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| package workflowstore | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| v1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
|
|
||
| "github.com/flyteorg/flytepropeller/pkg/apis/flyteworkflow/v1alpha1" | ||
| "github.com/flyteorg/flytepropeller/pkg/client/clientset/versioned/fake" | ||
| "github.com/flyteorg/flytestdlib/promutils" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestTerminatedTrackingStore_Update(t *testing.T) { | ||
| ctx := context.TODO() | ||
|
|
||
| mockClient := fake.NewSimpleClientset().FlyteworkflowV1alpha1() | ||
|
|
||
| scope := promutils.NewTestScope() | ||
| l := &mockWFNamespaceLister{} | ||
| passthroughWfStore := NewPassthroughWorkflowStore(ctx, scope, mockClient, &mockWFLister{V: l}) | ||
| wfStore, err := NewTerminatedTrackingStore(ctx, scope, passthroughWfStore) | ||
| assert.NoError(t, err) | ||
|
|
||
| t.Run("Succeeding", func(t *testing.T) { | ||
| namespace := "test-ns" | ||
| name := "succeeding" | ||
|
|
||
| wf := dummyWf(namespace, name) | ||
| wf.Status.Phase = v1alpha1.WorkflowPhaseSucceeding | ||
|
|
||
| _, err := mockClient.FlyteWorkflows(wf.GetNamespace()).Create(ctx, wf, v1.CreateOptions{}) | ||
| assert.NoError(t, err) | ||
|
|
||
| _, err = wfStore.Update(ctx, wf, PriorityClassCritical) | ||
| assert.NoError(t, err) | ||
|
|
||
| l.GetCb = func(name string) (*v1alpha1.FlyteWorkflow, error) { | ||
| return wf, nil | ||
| } | ||
| succeedingWf, err := wfStore.Get(ctx, namespace, name) | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, succeedingWf, wf) | ||
| }) | ||
|
|
||
| t.Run("Terminated", func(t *testing.T) { | ||
| namespace := "test-ns" | ||
| name := "terminated" | ||
|
|
||
| wf := dummyWf(namespace, name) | ||
| wf.Status.Phase = v1alpha1.WorkflowPhaseAborted | ||
|
|
||
| _, err := mockClient.FlyteWorkflows(wf.GetNamespace()).Create(ctx, wf, v1.CreateOptions{}) | ||
| assert.NoError(t, err) | ||
|
|
||
| _, err = wfStore.Update(ctx, wf, PriorityClassCritical) | ||
| assert.NoError(t, err) | ||
|
|
||
| l.GetCb = func(name string) (*v1alpha1.FlyteWorkflow, error) { | ||
| return wf, nil | ||
| } | ||
| terminatedWf, err := wfStore.Get(ctx, namespace, name) | ||
| assert.Nil(t, terminatedWf) | ||
| assert.True(t, IsWorkflowTerminated(err)) | ||
| }) | ||
| } | ||
|
|
||
| func TestTerminatedTrackingStore_UpdateStatus(t *testing.T) { | ||
| ctx := context.TODO() | ||
|
|
||
| mockClient := fake.NewSimpleClientset().FlyteworkflowV1alpha1() | ||
|
|
||
| scope := promutils.NewTestScope() | ||
| l := &mockWFNamespaceLister{} | ||
| passthroughWfStore := NewPassthroughWorkflowStore(ctx, scope, mockClient, &mockWFLister{V: l}) | ||
| wfStore, err := NewTerminatedTrackingStore(ctx, scope, passthroughWfStore) | ||
| assert.NoError(t, err) | ||
|
|
||
| namespace := "test-ns" | ||
| name := "name" | ||
|
|
||
| wf := dummyWf(namespace, name) | ||
| wf.Status.Phase = v1alpha1.WorkflowPhaseSucceeding | ||
|
|
||
| _, err = mockClient.FlyteWorkflows(wf.GetNamespace()).Create(ctx, wf, v1.CreateOptions{}) | ||
| assert.NoError(t, err) | ||
|
|
||
| _, err = wfStore.Update(ctx, wf, PriorityClassCritical) | ||
| assert.NoError(t, err) | ||
|
|
||
| wf.Status.Phase = v1alpha1.WorkflowPhaseAborted | ||
| _, err = wfStore.UpdateStatus(ctx, wf, PriorityClassCritical) | ||
| assert.NoError(t, err) | ||
|
|
||
| l.GetCb = func(name string) (*v1alpha1.FlyteWorkflow, error) { | ||
| return wf, nil | ||
| } | ||
| terminatedWf, err := wfStore.Get(ctx, namespace, name) | ||
| assert.Nil(t, terminatedWf) | ||
| assert.True(t, IsWorkflowTerminated(err)) | ||
| } |