Silence NotFound when get task resource

Signed-off-by: Hongxin Liang <[email protected]>

flyteadmin/pkg/errors/errors.go

@@ -140,3 +140,8 @@ func NewWorkflowExistsIdenticalStructureError(ctx context.Context, request *admi
 	}
 	return statusErr
 }
+
+func IsDoesNotExistError(err error) bool {
+	adminError, ok := err.(FlyteAdminError)
+	return ok && adminError.Code() == codes.NotFound
+}

flyteadmin/pkg/errors/errors_test.go

@@ -2,6 +2,7 @@ package errors
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -90,3 +91,15 @@ func TestNewWorkflowExistsIdenticalStructureError(t *testing.T) {
 	_, ok = details.GetReason().(*admin.CreateWorkflowFailureReason_ExistsIdenticalStructure)
 	assert.True(t, ok)
 }
+
+func TestIsDoesNotExistError(t *testing.T) {
+	assert.True(t, IsDoesNotExistError(NewFlyteAdminError(codes.NotFound, "foo")))
+}
+
+func TestIsNotDoesNotExistError(t *testing.T) {
+	assert.False(t, IsDoesNotExistError(NewFlyteAdminError(codes.Canceled, "foo")))
+}
+
+func TestIsNotDoesNotExistErrorBecauseOfNoneAdminError(t *testing.T) {
+	assert.False(t, IsDoesNotExistError(errors.New("foo")))
+}

flyteadmin/pkg/manager/impl/util/resources.go

@@ -3,7 +3,7 @@ package util
 import (
 	"context"
 	"fmt"
-
+	"github.com/flyteorg/flyte/flyteadmin/pkg/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/flyteorg/flyte/flyteadmin/pkg/manager/interfaces"
@@ -100,7 +100,7 @@ func GetTaskResources(ctx context.Context, id *core.Identifier, resourceManager
 	}
 
 	resource, err := resourceManager.GetResource(ctx, request)
-	if err != nil {
+	if err != nil && !errors.IsDoesNotExistError(err) {
 		logger.Infof(ctx, "Failed to fetch override values when assigning task resource default values for [%+v]: %v",
 			id, err)
 	}