Silence NotFound when get task resource

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/executioncluster/impl/random_cluster_selector.go

@@ -6,8 +6,6 @@
 	"hash/fnv"
 	"math/rand"
 
-	"google.golang.org/grpc/codes"
-
 	"github.com/flyteorg/flyte/flyteadmin/pkg/errors"
 	"github.com/flyteorg/flyte/flyteadmin/pkg/executioncluster"
 	"github.com/flyteorg/flyte/flyteadmin/pkg/executioncluster/interfaces"
@@ -102,10 +100,8 @@
 		LaunchPlan:   spec.LaunchPlan,
 		ResourceType: admin.MatchableResource_EXECUTION_CLUSTER_LABEL,
 	})
-	if err != nil {
-		if flyteAdminError, ok := err.(errors.FlyteAdminError); !ok || flyteAdminError.Code() != codes.NotFound {
-			return nil, err
-		}
+	if err != nil && !errors.IsDoesNotExistError(err) {
+		return nil, err
 	}
 
 	var weightedRandomList random.WeightedRandomList

flyteadmin/pkg/manager/impl/execution_manager.go

@@ -164,11 +164,8 @@
 		LaunchPlan:   launchPlanName,
 		ResourceType: admin.MatchableResource_PLUGIN_OVERRIDE,
 	})
-	if err != nil {
-		ec, ok := err.(errors.FlyteAdminError)
-		if !ok || ec.Code() != codes.NotFound {
-			return nil, err
-		}
+	if err != nil && !errors.IsDoesNotExistError(err) {
+		return nil, err
 	}
 	if override != nil && override.Attributes != nil && override.Attributes.GetPluginOverrides() != nil {
 		return override.Attributes.GetPluginOverrides().Overrides, nil
@@ -427,11 +424,9 @@
 		Domain:       request.Domain,
 		ResourceType: admin.MatchableResource_CLUSTER_ASSIGNMENT,
 	})
-	if err != nil {
-		if flyteAdminError, ok := err.(errors.FlyteAdminError); !ok || flyteAdminError.Code() != codes.NotFound {
-			logger.Errorf(ctx, "Failed to get cluster assignment overrides with error: %v", err)
-			return nil, err
-		}
+	if err != nil && !errors.IsDoesNotExistError(err) {
+		logger.Errorf(ctx, "Failed to get cluster assignment overrides with error: %v", err)
+		return nil, err
 	}
 	if resource != nil && resource.Attributes.GetClusterAssignment() != nil {
 		return resource.Attributes.GetClusterAssignment(), nil

flyteadmin/pkg/manager/impl/executions/quality_of_service.go

@@ -43,7 +43,7 @@
 		ResourceType: admin.MatchableResource_QUALITY_OF_SERVICE_SPECIFICATION,
 	})
 	if err != nil {
-		if _, ok := err.(errors.FlyteAdminError); !ok || err.(errors.FlyteAdminError).Code() != codes.NotFound {
+		if !errors.IsDoesNotExistError(err) {
 			logger.Warningf(ctx,
 				"Failed to fetch override values when assigning quality of service values for [%+v] with err: %v",
 				workflowIdentifier, err)

flyteadmin/pkg/manager/impl/executions/queues.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"math/rand"
 
+	"github.com/flyteorg/flyte/flyteadmin/pkg/errors"
+
 	"github.com/flyteorg/flyte/flyteadmin/pkg/manager/impl/resources"
 	"github.com/flyteorg/flyte/flyteadmin/pkg/manager/interfaces"
 	repoInterfaces "github.com/flyteorg/flyte/flyteadmin/pkg/repositories/interfaces"
@@ -64,7 +66,7 @@ func (q *queueAllocatorImpl) GetQueue(ctx context.Context, identifier core.Ident
 		ResourceType: admin.MatchableResource_EXECUTION_QUEUE,
 	})
 
-	if err != nil {
+	if err != nil && !errors.IsDoesNotExistError(err) {
 		logger.Warningf(ctx, "Failed to fetch override values when assigning execution queue for [%+v] with err: %v",
 			identifier, err)
 	}

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)
 	}

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

@@ -275,12 +275,10 @@ func GetMatchableResource(ctx context.Context, resourceManager interfaces.Resour
 		Workflow:     workflowName,
 		ResourceType: resourceType,
 	})
-	if err != nil {
-		if flyteAdminError, ok := err.(errors.FlyteAdminError); !ok || flyteAdminError.Code() != codes.NotFound {
-			logger.Errorf(ctx, "Failed to get %v overrides in %s project %s domain %s workflow with error: %v", resourceType,
-				project, domain, workflowName, err)
-			return nil, err
-		}
+	if err != nil && !errors.IsDoesNotExistError(err) {
+		logger.Errorf(ctx, "Failed to get %v overrides in %s project %s domain %s workflow with error: %v", resourceType,
+			project, domain, workflowName, err)
+		return nil, err
 	}
 	return matchableResource, nil
 }