fix(backend): persistence agent - workflow not found error should be a permanent error (#4486)

* fix(backend): workflow not found error should be permanent

* failing test case

* Fix logic

* fix another case

* Switched to not found error

* not found error should be permanent

Author: Bobgy

backend/src/agent/persistence/client/pipeline_client.go

@@ -85,8 +85,10 @@ func (p *PipelineClient) ReportWorkflow(workflow *util.Workflow) error {
 
 	if err != nil {
 		statusCode, _ := status.FromError(err)
-		if statusCode.Code() == codes.InvalidArgument {
-			// Do not retry if there is something wrong with the workflow
+		if statusCode.Code() == codes.InvalidArgument || statusCode.Code() == codes.NotFound {
+			// Do not retry if either:
+			// * there is something wrong with the workflow
+			// * the workflow has been deleted by someone else
 			return util.NewCustomError(err, util.CUSTOM_CODE_PERMANENT,
 				"Error while reporting workflow resource (code: %v, message: %v): %v, %+v",
 				statusCode.Code(),

backend/src/apiserver/client/workflow_fake.go

@@ -23,7 +23,9 @@ import (
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
+	k8errors "k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8schema "k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 )
@@ -55,7 +57,7 @@ func (c *FakeWorkflowClient) Get(name string, options v1.GetOptions) (*v1alpha1.
 	if ok {
 		return workflow, nil
 	}
-	return nil, errors.New("not found")
+	return nil, k8errors.NewNotFound(k8schema.ParseGroupResource("workflows.argoproj.io"), name)
 }
 
 func (c *FakeWorkflowClient) List(opts v1.ListOptions) (*v1alpha1.WorkflowList, error) {
@@ -69,11 +71,20 @@ func (c *FakeWorkflowClient) Watch(opts v1.ListOptions) (watch.Interface, error)
 }
 
 func (c *FakeWorkflowClient) Update(workflow *v1alpha1.Workflow) (*v1alpha1.Workflow, error) {
-	return nil, nil
+	name := workflow.GetObjectMeta().GetName()
+	_, ok := c.workflows[name]
+	if ok {
+		return workflow, nil
+	}
+	return nil, k8errors.NewNotFound(k8schema.ParseGroupResource("workflows.argoproj.io"), name)
 }
 
 func (c *FakeWorkflowClient) Delete(name string, options *v1.DeleteOptions) error {
-	return nil
+	_, ok := c.workflows[name]
+	if ok {
+		return nil
+	}
+	return k8errors.NewNotFound(k8schema.ParseGroupResource("workflows.argoproj.io"), name)
 }
 
 func (c *FakeWorkflowClient) DeleteCollection(options *v1.DeleteOptions,
@@ -85,6 +96,11 @@ func (c *FakeWorkflowClient) DeleteCollection(options *v1.DeleteOptions,
 func (c *FakeWorkflowClient) Patch(name string, pt types.PatchType, data []byte,
 	subresources ...string) (*v1alpha1.Workflow, error) {
 
+	_, ok := c.workflows[name]
+	if !ok {
+		return nil, k8errors.NewNotFound(k8schema.ParseGroupResource("workflows.argoproj.io"), name)
+	}
+
 	var dat map[string]interface{}
 	json.Unmarshal(data, &dat)
 

backend/src/apiserver/resource/resource_manager.go

@@ -708,7 +708,14 @@ func (r *ResourceManager) ReportWorkflowResource(workflow *util.Workflow) error
 		// If workflow's final state has being persisted, the workflow should be garbage collected.
 		err := r.getWorkflowClient(workflow.Namespace).Delete(workflow.Name, &v1.DeleteOptions{})
 		if err != nil {
-			return util.NewInternalServerError(err, "Failed to delete the completed workflow for run %s", runId)
+			// A fix for kubeflow/pipelines#4484, persistence agent might have an outdated item in its workqueue, so it will
+			// report workflows that no longer exist. It's important to return a not found error, so that persistence
+			// agent won't retry again.
+			if util.IsNotFound(err) {
+				return util.NewNotFoundError(err, "Failed to delete the completed workflow for run %s", runId)
+			} else {
+				return util.NewInternalServerError(err, "Failed to delete the completed workflow for run %s", runId)
+			}
 		}
 	}
 
@@ -777,7 +784,15 @@ func (r *ResourceManager) ReportWorkflowResource(workflow *util.Workflow) error
 	if workflow.IsInFinalState() {
 		err := AddWorkflowLabel(r.getWorkflowClient(workflow.Namespace), workflow.Name, util.LabelKeyWorkflowPersistedFinalState, "true")
 		if err != nil {
-			return util.Wrap(err, "Failed to add PersistedFinalState label to workflow")
+			message := fmt.Sprintf("Failed to add PersistedFinalState label to workflow %s", workflow.GetName())
+			// A fix for kubeflow/pipelines#4484, persistence agent might have an outdated item in its workqueue, so it will
+			// report workflows that no longer exist. It's important to return a not found error, so that persistence
+			// agent won't retry again.
+			if util.IsNotFound(err) {
+				return util.NewNotFoundError(err, message)
+			} else {
+				return util.Wrapf(err, message)
+			}
 		}
 	}
 

backend/src/apiserver/resource/resource_manager_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -670,7 +671,7 @@ func TestCreateRun_NullWorkflowSpec(t *testing.T) {
 	apiRun := &api.Run{
 		Name: "run1",
 		PipelineSpec: &api.PipelineSpec{
-			WorkflowManifest: "null",  // this situation occurs for real when the manifest file disappears from object store in some way due to retention policy or manual deletion.
+			WorkflowManifest: "null", // this situation occurs for real when the manifest file disappears from object store in some way due to retention policy or manual deletion.
 			Parameters: []*api.Parameter{
 				{Name: "param1", Value: "world"},
 			},
@@ -1489,7 +1490,7 @@ func TestReportWorkflowResource_WorkflowMissingRunID(t *testing.T) {
 	defer store.Close()
 	workflow := util.NewWorkflow(&v1alpha1.Workflow{
 		ObjectMeta: v1.ObjectMeta{
-			Name:      run.Name,
+			Name: run.Name,
 		},
 	})
 	err := manager.ReportWorkflowResource(workflow)
@@ -1519,6 +1520,24 @@ func TestReportWorkflowResource_WorkflowCompleted(t *testing.T) {
 	assert.Equal(t, wf.Labels[util.LabelKeyWorkflowPersistedFinalState], "true")
 }
 
+func TestReportWorkflowResource_WorkflowCompleted_WorkflowNotFound(t *testing.T) {
+	store, manager, run := initWithOneTimeRun(t)
+	defer store.Close()
+	workflow := util.NewWorkflow(&v1alpha1.Workflow{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "non-existent-workflow",
+			Namespace: "kubeflow",
+			UID:       types.UID(run.UUID),
+			Labels:    map[string]string{util.LabelKeyWorkflowRunId: run.UUID},
+		},
+		Status: v1alpha1.WorkflowStatus{Phase: v1alpha1.NodeFailed},
+	})
+	err := manager.ReportWorkflowResource(workflow)
+	require.NotNil(t, err)
+	assert.Equalf(t, codes.NotFound, err.(*util.UserError).ExternalStatusCode(), "Expected not found error, but got %s", err.Error())
+	assert.Contains(t, err.Error(), "Failed to add PersistedFinalState label")
+}
+
 func TestReportWorkflowResource_WorkflowCompleted_FinalStatePersisted(t *testing.T) {
 	store, manager, run := initWithOneTimeRun(t)
 	defer store.Close()
@@ -1536,6 +1555,24 @@ func TestReportWorkflowResource_WorkflowCompleted_FinalStatePersisted(t *testing
 	assert.Nil(t, err)
 }
 
+func TestReportWorkflowResource_WorkflowCompleted_FinalStatePersisted_WorkflowNotFound(t *testing.T) {
+	store, manager, run := initWithOneTimeRun(t)
+	defer store.Close()
+	workflow := util.NewWorkflow(&v1alpha1.Workflow{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "non-existent-workflow",
+			Namespace: "kubeflow",
+			UID:       types.UID(run.UUID),
+			Labels:    map[string]string{util.LabelKeyWorkflowRunId: run.UUID, util.LabelKeyWorkflowPersistedFinalState: "true"},
+		},
+		Status: v1alpha1.WorkflowStatus{Phase: v1alpha1.NodeFailed},
+	})
+	err := manager.ReportWorkflowResource(workflow)
+	require.NotNil(t, err)
+	assert.Equalf(t, codes.NotFound, err.(*util.UserError).ExternalStatusCode(), "Expected not found error, but got %s", err.Error())
+	assert.Contains(t, err.Error(), "Failed to delete the completed workflow")
+}
+
 func TestReportWorkflowResource_WorkflowCompleted_FinalStatePersisted_DeleteFailed(t *testing.T) {
 	store, manager, run := initWithOneTimeRun(t)
 	manager.argoClient = client.NewFakeArgoClientWithBadWorkflow()

backend/src/common/util/error.go

@@ -144,6 +144,15 @@ func NewInternalServerError(err error, internalMessageFormat string,
 		codes.Internal)
 }
 
+func NewNotFoundError(err error, externalMessageFormat string,
+	a ...interface{}) *UserError {
+	externalMessage := fmt.Sprintf(externalMessageFormat, a...)
+	return newUserError(
+		errors.Wrapf(err, fmt.Sprintf("NotFoundError: %v", externalMessage)),
+		externalMessage,
+		codes.NotFound)
+}
+
 func NewResourceNotFoundError(resourceType string, resourceName string) *UserError {
 	externalMessage := fmt.Sprintf("%s %s not found.", resourceType, resourceName)
 	return newUserError(