Don't error when attempting to trigger schedules for inactive projects (

flyteorg#5649)

* Don't error when attempting to trigger schedules for inactive projects

Signed-off-by: Katrina Rogan <[email protected]>

* regen

Signed-off-by: Katrina Rogan <[email protected]>

---------

Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Vladyslav Libov <[email protected]>

flyteadmin/pkg/errors/errors.go

@@ -202,3 +202,15 @@ func IsDoesNotExistError(err error) bool {
 	adminError, ok := err.(FlyteAdminError)
 	return ok && adminError.Code() == codes.NotFound
 }
+
+func NewInactiveProjectError(ctx context.Context, id string) FlyteAdminError {
+	errMsg := fmt.Sprintf("project [%s] is not active", id)
+	statusErr, transformationErr := NewFlyteAdminError(codes.InvalidArgument, errMsg).WithDetails(&admin.InactiveProject{
+		Id: id,
+	})
+	if transformationErr != nil {
+		logger.Errorf(ctx, "failed to wrap grpc status in type 'Error': %v", transformationErr)
+		return NewFlyteAdminErrorf(codes.InvalidArgument, errMsg)
+	}
+	return statusErr
+}

flyteadmin/pkg/errors/errors_test.go

@@ -310,3 +310,15 @@ func TestIsNotDoesNotExistError(t *testing.T) {
 func TestIsNotDoesNotExistErrorBecauseOfNoneAdminError(t *testing.T) {
 	assert.False(t, IsDoesNotExistError(errors.New("foo")))
 }
+
+func TestNewInactiveProjectError(t *testing.T) {
+	err := NewInactiveProjectError(context.TODO(), identifier.GetProject())
+	statusErr, ok := status.FromError(err)
+
+	assert.True(t, ok)
+
+	details, ok := statusErr.Details()[0].(*admin.InactiveProject)
+
+	assert.True(t, ok)
+	assert.Equal(t, identifier.GetProject(), details.Id)
+}

flyteadmin/pkg/manager/impl/validation/project_validator.go

@@ -71,8 +71,7 @@ func ValidateProjectAndDomain(
 			projectID, domainID, err)
 	}
 	if *project.State != int32(admin.Project_ACTIVE) {
-		return errors.NewFlyteAdminErrorf(codes.InvalidArgument,
-			"project [%s] is not active", projectID)
+		return errors.NewInactiveProjectError(ctx, projectID)
 	}
 	var validDomain bool
 	domains := config.GetDomainsConfig()

flyteadmin/scheduler/executor/executor_impl.go

@@ -114,6 +114,10 @@ func (w *executor) Execute(ctx context.Context, scheduledTime time.Time, s model
 		},
 		func() error {
 			_, execErr := w.adminServiceClient.CreateExecution(context.Background(), executionRequest)
+			if isInactiveProjectError(execErr) {
+				logger.Debugf(ctx, "project %+v is inactive, ignoring schedule create failure for %+v", s.Project, s)
+				return nil
+			}
 			return execErr
 		},
 	)
@@ -144,3 +148,18 @@ func getExecutorMetrics(scope promutils.Scope) executorMetrics {
 			"count of successful attempts to fire execution for a schedules"),
 	}
 }
+
+func isInactiveProjectError(err error) bool {
+	statusErr, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+	if len(statusErr.Details()) > 0 {
+		for _, detail := range statusErr.Details() {
+			if _, ok := detail.(*admin.InactiveProject); ok {
+				return true
+			}
+		}
+	}
+	return false
+}

flyteadmin/scheduler/executor/executor_impl_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/flyteorg/flyte/flyteadmin/pkg/errors"
 	"github.com/flyteorg/flyte/flyteadmin/scheduler/repositories/models"
@@ -98,3 +99,14 @@ func TestExecutorInactiveSchedule(t *testing.T) {
 	err := executor.Execute(context.Background(), time.Now(), schedule)
 	assert.Nil(t, err)
 }
+
+func TestIsInactiveProjectError(t *testing.T) {
+	statusErr := status.New(codes.InvalidArgument, "foo")
+	var transformationErr error
+	statusErr, transformationErr = statusErr.WithDetails(&admin.InactiveProject{
+		Id: "project",
+	})
+	assert.NoError(t, transformationErr)
+
+	assert.True(t, isInactiveProjectError(statusErr.Err()))
+}