fix: skip update only if the conflict error is related to main output

Fix `qtransform` controller bug which was skipping the reconciliation if the conflict error is returned from the `transformFunc` and is related to the extra outputs. Introduce a way to check if the error was raised by any particular resource type. Signed-off-by: Artem Chernyshev <[email protected]>
cosi-project · Sep 12, 2024 · c0a68e9 · c0a68e9
1 parent 8911486
commit c0a68e9
Show file tree

Hide file tree

Showing 7 changed files with 93 additions and 17 deletions.
diff --git a/pkg/controller/generic/qtransform/qtransform.go b/pkg/controller/generic/qtransform/qtransform.go
@@ -253,7 +253,10 @@ func (ctrl *QController[Input, Output]) reconcileRunning(ctx context.Context, lo
 
 		return transformError
 	}); err != nil {
-		if state.IsConflictError(err) {
+		if state.IsConflictError(err,
+			state.WithResourceNamespace(mappedOut.Metadata().Namespace()),
+			state.WithResourceType(mappedOut.Metadata().Type()),
+		) {
 			// conflict due to wrong phase, skip it
 			return nil
 		}

diff --git a/pkg/controller/generic/transform/controller.go b/pkg/controller/generic/transform/controller.go
@@ -328,7 +328,10 @@ func (ctrl *Controller[Input, Output]) processInputs(
 		if err = safe.WriterModify(ctx, r, mappedOut, func(out Output) error {
 			return ctrl.transformFunc(ctx, r, logger, in, out)
 		}); err != nil {
-			if state.IsConflictError(err) {
+			if state.IsConflictError(err,
+				state.WithResourceNamespace(mappedOut.Metadata().Namespace()),
+				state.WithResourceType(mappedOut.Metadata().Type()),
+			) {
 				// conflict due to wrong phase, skip it
 				continue
 			}

diff --git a/pkg/state/errors.go b/pkg/state/errors.go
@@ -11,6 +11,29 @@ import (
 	"github.com/cosi-project/runtime/pkg/resource"
 )
 
+// ErrcheckOptions defines additional error check options.
+type ErrcheckOptions struct {
+	resourceType      resource.Type
+	resourceNamespace resource.Namespace
+}
+
+// ErrcheckOption defines an additional error check option.
+type ErrcheckOption func(*ErrcheckOptions)
+
+// WithResourceType checks if the error is related to the resource type.
+func WithResourceType(rt resource.Type) ErrcheckOption {
+	return func(eo *ErrcheckOptions) {
+		eo.resourceType = rt
+	}
+}
+
+// WithResourceNamespace checks if the error is related to the resource namespace.
+func WithResourceNamespace(ns resource.Namespace) ErrcheckOption {
+	return func(eo *ErrcheckOptions) {
+		eo.resourceNamespace = ns
+	}
+}
+
 // ErrUnsupported should be implemented by unsupported operation errors.
 type ErrUnsupported interface {
 	UnsupportedError()
@@ -38,13 +61,34 @@ func IsNotFoundError(err error) bool {
 // ErrConflict should be implemented by already exists/update conflict errors.
 type ErrConflict interface {
 	ConflictError()
+	GetResource() resource.Pointer
 }
 
 // IsConflictError checks if err is resource already exists/update conflict.
-func IsConflictError(err error) bool {
+func IsConflictError(err error, opts ...ErrcheckOption) bool {
 	var i ErrConflict
 
-	return errors.As(err, &i)
+	var options ErrcheckOptions
+
+	for _, o := range opts {
+		o(&options)
+	}
+
+	if !errors.As(err, &i) {
+		return false
+	}
+
+	res := i.GetResource()
+
+	if options.resourceNamespace != "" && res.Namespace() != options.resourceNamespace {
+		return false
+	}
+
+	if options.resourceType != "" && res.Type() != options.resourceType {
+		return false
+	}
+
+	return true
 }
 
 // ErrOwnerConflict should be implemented by owner conflict errors.

diff --git a/pkg/state/impl/inmem/errors.go b/pkg/state/impl/inmem/errors.go
@@ -25,10 +25,15 @@ func ErrNotFound(r resource.Pointer) error {
 
 type eConflict struct {
 	error
+	resource resource.Pointer
 }
 
 func (eConflict) ConflictError() {}
 
+func (e eConflict) GetResource() resource.Pointer {
+	return e.resource
+}
+
 type eOwnerConflict struct {
 	eConflict
 }
@@ -44,36 +49,40 @@ func (ePhaseConflict) PhaseConflictError() {}
 // ErrAlreadyExists generates error compatible with state.ErrConflict.
 func ErrAlreadyExists(r resource.Reference) error {
 	return eConflict{
-		fmt.Errorf("resource %s already exists", r),
+		error:    fmt.Errorf("resource %s already exists", r),
+		resource: r,
 	}
 }
 
 // ErrVersionConflict generates error compatible with state.ErrConflict.
 func ErrVersionConflict(r resource.Reference, expected, found resource.Version) error {
 	return eConflict{
-		fmt.Errorf("resource %s update conflict: expected version %q, actual version %q", r, expected, found),
+		error: fmt.Errorf("resource %s update conflict: expected version %q, actual version %q", r, expected, found),
 	}
 }
 
 // ErrUpdateSameVersion generates error compatible with state.ErrConflict.
 func ErrUpdateSameVersion(r resource.Reference, version resource.Version) error {
 	return eConflict{
-		fmt.Errorf("resource %s update conflict: same %q version for new and existing objects", r, version),
+		error:    fmt.Errorf("resource %s update conflict: same %q version for new and existing objects", r, version),
+		resource: r,
 	}
 }
 
 // ErrPendingFinalizers generates error compatible with state.ErrConflict.
 func ErrPendingFinalizers(r resource.Metadata) error {
 	return eConflict{
-		fmt.Errorf("resource %s has pending finalizers %s", r, r.Finalizers()),
+		error:    fmt.Errorf("resource %s has pending finalizers %s", r, r.Finalizers()),
+		resource: r,
 	}
 }
 
 // ErrOwnerConflict generates error compatible with state.ErrConflict.
 func ErrOwnerConflict(r resource.Reference, owner string) error {
 	return eOwnerConflict{
 		eConflict{
-			fmt.Errorf("resource %s is owned by %q", r, owner),
+			error:    fmt.Errorf("resource %s is owned by %q", r, owner),
+			resource: r,
 		},
 	}
 }
@@ -82,7 +91,8 @@ func ErrOwnerConflict(r resource.Reference, owner string) error {
 func ErrPhaseConflict(r resource.Reference, expectedPhase resource.Phase) error {
 	return ePhaseConflict{
 		eConflict{
-			fmt.Errorf("resource %s is not in phase %s", r, expectedPhase),
+			error:    fmt.Errorf("resource %s is not in phase %s", r, expectedPhase),
+			resource: r,
 		},
 	}
 }
diff --git a/pkg/state/impl/inmem/errors_test.go b/pkg/state/impl/inmem/errors_test.go
@@ -23,4 +23,13 @@ func TestErrors(t *testing.T) {
 	assert.True(t, state.IsConflictError(inmem.ErrOwnerConflict(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined), "owner")))
 	assert.True(t, state.IsOwnerConflictError(inmem.ErrOwnerConflict(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined), "owner")))
 	assert.True(t, state.IsPhaseConflictError(inmem.ErrPhaseConflict(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined), resource.PhaseTearingDown)))
+
+	assert.True(t, state.IsConflictError(inmem.ErrAlreadyExists(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined)), state.WithResourceNamespace("ns")))
+	assert.False(t, state.IsConflictError(inmem.ErrAlreadyExists(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined)), state.WithResourceNamespace("a")))
+
+	assert.True(t, state.IsConflictError(inmem.ErrAlreadyExists(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined)), state.WithResourceType("a")))
+	assert.False(t, state.IsConflictError(inmem.ErrAlreadyExists(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined)), state.WithResourceType("z")))
+
+	assert.True(t, state.IsConflictError(inmem.ErrAlreadyExists(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined)), state.WithResourceType("a"), state.WithResourceNamespace("ns")))
+	assert.False(t, state.IsConflictError(inmem.ErrAlreadyExists(resource.NewMetadata("ns", "a", "b", resource.VersionUndefined)), state.WithResourceType("z"), state.WithResourceNamespace("ns")))
 }
diff --git a/pkg/state/protobuf/client/client.go b/pkg/state/protobuf/client/client.go
@@ -172,9 +172,9 @@ func (adapter *Adapter) Create(ctx context.Context, r resource.Resource, opt ...
 		case codes.NotFound:
 			return eNotFound{err}
 		case codes.PermissionDenied:
-			return eOwnerConflict{eConflict{err}}
+			return eOwnerConflict{eConflict{error: err, resource: r.Metadata()}}
 		case codes.AlreadyExists:
-			return eConflict{err}
+			return eConflict{error: err, resource: r.Metadata()}
 		default:
 			return err
 		}
@@ -223,11 +223,11 @@ func (adapter *Adapter) Update(ctx context.Context, newResource resource.Resourc
 		case codes.NotFound:
 			return eNotFound{err}
 		case codes.PermissionDenied:
-			return eOwnerConflict{eConflict{err}}
+			return eOwnerConflict{eConflict{error: err, resource: newResource.Metadata()}}
 		case codes.InvalidArgument:
-			return ePhaseConflict{eConflict{err}}
+			return ePhaseConflict{eConflict{error: err, resource: newResource.Metadata()}}
 		case codes.FailedPrecondition:
-			return eConflict{err}
+			return eConflict{error: err, resource: newResource.Metadata()}
 		default:
 			return err
 		}
@@ -261,9 +261,9 @@ func (adapter *Adapter) Destroy(ctx context.Context, resourcePointer resource.Po
 		case codes.NotFound:
 			return eNotFound{err}
 		case codes.PermissionDenied:
-			return eOwnerConflict{eConflict{err}}
+			return eOwnerConflict{eConflict{error: err, resource: resourcePointer}}
 		case codes.FailedPrecondition:
-			return eConflict{err}
+			return eConflict{error: err, resource: resourcePointer}
 		default:
 			return err
 		}

diff --git a/pkg/state/protobuf/client/errors.go b/pkg/state/protobuf/client/errors.go
@@ -4,6 +4,8 @@
 
 package client
 
+import "github.com/cosi-project/runtime/pkg/resource"
+
 type eNotFound struct {
 	error
 }
@@ -12,10 +14,15 @@ func (eNotFound) NotFoundError() {}
 
 type eConflict struct {
 	error
+	resource resource.Pointer
 }
 
 func (eConflict) ConflictError() {}
 
+func (e eConflict) GetResource() resource.Pointer {
+	return e.resource
+}
+
 type eOwnerConflict struct {
 	eConflict
 }