diff --git a/internal/controller/bucket/acl.go b/internal/controller/bucket/acl.go index 88558703..06bd6e3b 100644 --- a/internal/controller/bucket/acl.go +++ b/internal/controller/bucket/acl.go @@ -5,66 +5,54 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" - "github.com/crossplane/crossplane-runtime/v2/pkg/errors" "github.com/go-logr/logr" "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" - apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" "github.com/linode/provider-ceph/internal/backendstore" "github.com/linode/provider-ceph/internal/consts" "github.com/linode/provider-ceph/internal/controller/s3clienthandler" "github.com/linode/provider-ceph/internal/otel/traces" "github.com/linode/provider-ceph/internal/rgw" - - "go.opentelemetry.io/otel" ) // ACLClient is the client for API methods and reconciling the ACL type ACLClient struct { - backendStore *backendstore.BackendStore - s3ClientHandler *s3clienthandler.Handler - log logr.Logger + BaseSubresourceClient } // NewACLClient creates the client for ACL func NewACLClient(b *backendstore.BackendStore, h *s3clienthandler.Handler, l logr.Logger) *ACLClient { - return &ACLClient{backendStore: b, s3ClientHandler: h, log: l} + return &ACLClient{BaseSubresourceClient: NewBaseSubresourceClient(b, h, l)} } -func (l *ACLClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { - _, span := otel.Tracer("").Start(ctx, "bucket.ACLClient.Observe") - defer span.End() - - observationChan := make(chan ResourceStatus) - - for _, backendName := range backendNames { - beName := backendName - go func() { - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - // If a backend is marked as unhealthy, we can ignore it for now by returning NoAction. - // The backend may be down for some time and we do not want to block Create/Update/Delete - // calls on other backends. By returning NoAction here, we would never pass the Observe - // phase until the backend becomes Healthy or Disabled. - observationChan <- NoAction - - return - } - observationChan <- l.observeBackend(ctx, bucket, beName) - }() - } +func (a *ACLClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { + return a.BaseSubresourceClient.Observe(ctx, bucket, backendNames, a) +} - for i := 0; i < len(backendNames); i++ { - observation := <-observationChan - if observation == NeedsUpdate || observation == NeedsDeletion { - return observation, nil - } - } +func (a *ACLClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { + return a.BaseSubresourceClient.Handle(ctx, b, backendName, bb, a) +} + +// Implement Subresource interface + +func (a *ACLClient) GetLogger() logr.Logger { + return a.log +} - return Updated, nil +func (a *ACLClient) GetBackendStore() *backendstore.BackendStore { + return a.backendStore } -func (l *ACLClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) ResourceStatus { - _, log := traces.InjectTraceAndLogger(ctx, l.log) +func (a *ACLClient) GetS3ClientHandler() *s3clienthandler.Handler { + return a.s3ClientHandler +} + +func (a *ACLClient) GetObserveErrorMsg() string { + return errObserveAcl +} + +func (a *ACLClient) ObserveBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { + _, log := traces.InjectTraceAndLogger(ctx, a.log) log.V(1).Info("Observing subresource acl on backend", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) @@ -73,7 +61,7 @@ func (l *ACLClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, if s3types.ObjectOwnership(aws.ToString(bucket.Spec.ForProvider.ObjectOwnership)) == s3types.ObjectOwnershipBucketOwnerEnforced { log.V(1).Info("Access control limits are disabled - no action required", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) - return Updated + return Updated, nil } if bucket.Spec.ForProvider.ACL == nil && @@ -85,42 +73,38 @@ func (l *ACLClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, bucket.Spec.ForProvider.GrantReadACP == nil { log.V(1).Info("No acl or access control policy or grants requested - no action required", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) - return Updated + return Updated, nil } - return NeedsUpdate + return NeedsUpdate, nil } -func (l *ACLClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { - ctx, span := otel.Tracer("").Start(ctx, "bucket.ACLClient.Handle") - defer span.End() +// Implement Subresource interface - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - traces.SetAndRecordError(span, errUnhealthyBackend) +func (a *ACLClient) GetHandleErrorMsg() string { + return errHandleAcl +} - return errUnhealthyBackend - } +func (a *ACLClient) GetSubresourceName() string { + return "ACLClient" +} - switch l.observeBackend(ctx, b, backendName) { +func (a *ACLClient) HandleObservation(ctx context.Context, observation ResourceStatus, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { + switch observation { case NoAction, Updated: return nil case NeedsUpdate, NeedsDeletion: - if err := l.createOrUpdate(ctx, b, backendName); err != nil { - err = errors.Wrap(err, errHandleAcl) - traces.SetAndRecordError(span, err) - - return err - } + return a.createOrUpdate(ctx, bucket, backendName) } return nil } -func (l *ACLClient) createOrUpdate(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) +func (a *ACLClient) createOrUpdate(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { + ctx, log := traces.InjectTraceAndLogger(ctx, a.log) log.Info("Updating acl", consts.KeyBucketName, b.Name, consts.KeyBackendName, backendName) - s3Client, err := l.s3ClientHandler.GetS3Client(ctx, b, backendName) + s3Client, err := a.s3ClientHandler.GetS3Client(ctx, b, backendName) if err != nil { return err } diff --git a/internal/controller/bucket/acl_test.go b/internal/controller/bucket/acl_test.go index 65aee34a..33c22289 100644 --- a/internal/controller/bucket/acl_test.go +++ b/internal/controller/bucket/acl_test.go @@ -24,6 +24,7 @@ import ( s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/go-logr/logr" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" @@ -240,7 +241,8 @@ func TestACLObserveBackend(t *testing.T) { s3clienthandler.WithBackendStore(tc.fields.backendStore)), logr.Discard()) - got := c.observeBackend(context.Background(), tc.args.bucket, tc.args.backendName) + got, err := c.ObserveBackend(context.Background(), tc.args.bucket, tc.args.backendName) + require.NoError(t, err, "unexpected error") assert.Equal(t, tc.want.status, got, "unexpected status") }) } diff --git a/internal/controller/bucket/base_subresource.go b/internal/controller/bucket/base_subresource.go new file mode 100644 index 00000000..f4d2e237 --- /dev/null +++ b/internal/controller/bucket/base_subresource.go @@ -0,0 +1,167 @@ +package bucket + +import ( + "context" + + "github.com/crossplane/crossplane-runtime/v2/pkg/errors" + "github.com/go-logr/logr" + + "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" + apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" + "github.com/linode/provider-ceph/internal/backendstore" + "github.com/linode/provider-ceph/internal/consts" + "github.com/linode/provider-ceph/internal/controller/s3clienthandler" + "github.com/linode/provider-ceph/internal/otel/traces" + + "go.opentelemetry.io/otel" +) + +// Subresource provides the plugin interface for observing and handling backend state. +type Subresource interface { + // ObserveBackend observes the resource on a specific backend. + // Should return NoAction/Updated/NeedsUpdate/NeedsDeletion and any error. + ObserveBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) + + // GetObserveErrorMsg returns the error message for observe failures. + GetObserveErrorMsg() string + + // SkipObservation returns true if observation should be skipped for this subresource. + // By default (false), all subresources are observed. Override to implement conditional observation. + SkipObservation(bucket *v1alpha1.Bucket) bool + + // HandleObservation handles the resource on a specific backend based on the observation result. + HandleObservation(ctx context.Context, observation ResourceStatus, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends) error + + // GetHandleErrorMsg returns the error message for handle failures. + GetHandleErrorMsg() string + + // GetLogger returns the logger for this subresource. + GetLogger() logr.Logger + + // GetBackendStore returns the backend store. + GetBackendStore() *backendstore.BackendStore + + // GetS3ClientHandler returns the s3 client handler. + GetS3ClientHandler() *s3clienthandler.Handler + + // GetSubresourceName returns the name of the subresource for tracing. + GetSubresourceName() string +} + +// BaseSubresourceClient provides common logic for all subresource clients. +type BaseSubresourceClient struct { + backendStore *backendstore.BackendStore + s3ClientHandler *s3clienthandler.Handler + log logr.Logger +} + +// NewBaseSubresourceClient creates a new base subresource client. +func NewBaseSubresourceClient(b *backendstore.BackendStore, h *s3clienthandler.Handler, l logr.Logger) BaseSubresourceClient { + return BaseSubresourceClient{ + backendStore: b, + s3ClientHandler: h, + log: l, + } +} + +// SkipObservation provides a default implementation that does not skip observations. +// Subresources can override this to implement conditional observation logic. +func (b *BaseSubresourceClient) SkipObservation(bucket *v1alpha1.Bucket) bool { + return false // By default, observe all subresources +} + +// Observe implements the common observe pattern for all subresources. +// It handles concurrent observation of all backends and consolidates results. +func (b *BaseSubresourceClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string, subresource Subresource) (ResourceStatus, error) { + ctx, span := otel.Tracer("").Start(ctx, "bucket."+subresource.GetSubresourceName()+".Observe") + defer span.End() + ctx, log := traces.InjectTraceAndLogger(ctx, subresource.GetLogger()) + + // Check if this subresource should be skipped + if subresource.SkipObservation(bucket) { + log.V(1).Info(subresource.GetSubresourceName() + " observation skipped") + + return Updated, nil + } + + observationChan := make(chan ResourceStatus, len(backendNames)) + errChan := make(chan error, len(backendNames)) + + for _, backendName := range backendNames { + beName := backendName + go func() { + if subresource.GetBackendStore().GetBackendHealthStatus(beName) == apisv1alpha1.HealthStatusUnhealthy { + // If a backend is marked as unhealthy, we can ignore it for now by returning NoAction. + // The backend may be down for some time and we do not want to block Create/Update/Delete + // calls on other backends. By returning NoAction here, we would never pass the Observe + // phase until the backend becomes Healthy or Disabled. + observationChan <- NoAction + + return + } + + observation, err := subresource.ObserveBackend(ctx, bucket, beName) + if err != nil { + errChan <- err + + return + } + observationChan <- observation + }() + } + + for i := 0; i < len(backendNames); i++ { + select { + case <-ctx.Done(): + log.Info("Context timeout during bucket "+subresource.GetSubresourceName()+" observation", consts.KeyBucketName, bucket.Name) + err := errors.Wrap(ctx.Err(), subresource.GetObserveErrorMsg()) + traces.SetAndRecordError(span, err) + + return NeedsUpdate, err + case observation := <-observationChan: + if observation == NeedsUpdate || observation == NeedsDeletion { + return observation, nil + } + case err := <-errChan: + err = errors.Wrap(err, subresource.GetObserveErrorMsg()) + traces.SetAndRecordError(span, err) + + return NeedsUpdate, err + } + } + + return Updated, nil +} + +// Handle implements the common handle pattern for all subresources. +// It performs health checks and delegates to the handler for observation-specific logic. +func (b *BaseSubresourceClient) Handle(ctx context.Context, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends, subresource Subresource) error { + ctx, span := otel.Tracer("").Start(ctx, "bucket."+subresource.GetSubresourceName()+".Handle") + defer span.End() + + // Check if this subresource should be skipped + if subresource.SkipObservation(bucket) { + return nil + } + + if subresource.GetBackendStore().GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { + traces.SetAndRecordError(span, errUnhealthyBackend) + + return errUnhealthyBackend + } + + observation, err := subresource.ObserveBackend(ctx, bucket, backendName) + if err != nil { + err = errors.Wrap(err, subresource.GetHandleErrorMsg()) + traces.SetAndRecordError(span, err) + + return err + } + + err = subresource.HandleObservation(ctx, observation, bucket, backendName, bb) + if err != nil { + traces.SetAndRecordError(span, err) + } + + return err +} diff --git a/internal/controller/bucket/lifecycleconfiguration.go b/internal/controller/bucket/lifecycleconfiguration.go index 7b5631cd..8623d462 100644 --- a/internal/controller/bucket/lifecycleconfiguration.go +++ b/internal/controller/bucket/lifecycleconfiguration.go @@ -12,7 +12,6 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/errors" "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" - apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" "github.com/linode/provider-ceph/internal/backendstore" "github.com/linode/provider-ceph/internal/consts" "github.com/linode/provider-ceph/internal/controller/s3clienthandler" @@ -21,77 +20,44 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" - - "go.opentelemetry.io/otel" ) // LifecycleConfigurationClient is the client for API methods and reconciling the LifecycleConfiguration type LifecycleConfigurationClient struct { - backendStore *backendstore.BackendStore - s3ClientHandler *s3clienthandler.Handler - log logr.Logger + BaseSubresourceClient } func NewLifecycleConfigurationClient(b *backendstore.BackendStore, h *s3clienthandler.Handler, l logr.Logger) *LifecycleConfigurationClient { - return &LifecycleConfigurationClient{backendStore: b, s3ClientHandler: h, log: l} + return &LifecycleConfigurationClient{BaseSubresourceClient: NewBaseSubresourceClient(b, h, l)} } -//nolint:dupl // LifecycleConfiguration and Policy are different feature. func (l *LifecycleConfigurationClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { - ctx, span := otel.Tracer("").Start(ctx, "bucket.LifecycleConfigurationClient.Observe") - defer span.End() - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) + return l.BaseSubresourceClient.Observe(ctx, bucket, backendNames, l) +} - observationChan := make(chan ResourceStatus) - errChan := make(chan error) - - for _, backendName := range backendNames { - beName := backendName - go func() { - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - // If a backend is marked as unhealthy, we can ignore it for now by returning NoAction. - // The backend may be down for some time and we do not want to block Create/Update/Delete - // calls on other backends. By returning NoAction here, we would never pass the Observe - // phase until the backend becomes Healthy or Disabled. - observationChan <- NoAction - - return - } - - observation, err := l.observeBackend(ctx, bucket, beName) - if err != nil { - errChan <- err - - return - } - observationChan <- observation - }() - } +func (l *LifecycleConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { + return l.BaseSubresourceClient.Handle(ctx, b, backendName, bb, l) +} - for i := 0; i < len(backendNames); i++ { - select { - case <-ctx.Done(): - log.Info("Context timeout during bucket lifecycle configuration observation", consts.KeyBucketName, bucket.Name) - err := errors.Wrap(ctx.Err(), errObserveLifecycleConfig) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - case observation := <-observationChan: - if observation == NeedsUpdate || observation == NeedsDeletion { - return observation, nil - } - case err := <-errChan: - err = errors.Wrap(err, errObserveLifecycleConfig) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - } - } +// Implement Subresource interface - return Updated, nil +func (l *LifecycleConfigurationClient) GetLogger() logr.Logger { + return l.log +} + +func (l *LifecycleConfigurationClient) GetBackendStore() *backendstore.BackendStore { + return l.backendStore +} + +func (l *LifecycleConfigurationClient) GetS3ClientHandler() *s3clienthandler.Handler { + return l.s3ClientHandler +} + +func (l *LifecycleConfigurationClient) GetObserveErrorMsg() string { + return errObserveLifecycleConfig } -func (l *LifecycleConfigurationClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { +func (l *LifecycleConfigurationClient) ObserveBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { ctx, log := traces.InjectTraceAndLogger(ctx, l.log) log.V(1).Info("Observing subresource lifecycle configuration on backend", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) @@ -145,25 +111,18 @@ func (l *LifecycleConfigurationClient) observeBackend(ctx context.Context, bucke return Updated, nil } -//nolint:dupl // LifecycleConfiguration and ServerSideEncryptionConfiguration have similar Handle logic. -func (l *LifecycleConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { - ctx, span := otel.Tracer("").Start(ctx, "bucket.LifecycleConfigurationClient.Handle") - defer span.End() - - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - traces.SetAndRecordError(span, errUnhealthyBackend) - - return errUnhealthyBackend - } +// Implement Subresource interface - observation, err := l.observeBackend(ctx, b, backendName) - if err != nil { - err = errors.Wrap(err, errHandleLifecycleConfig) - traces.SetAndRecordError(span, err) +func (l *LifecycleConfigurationClient) GetHandleErrorMsg() string { + return errHandleLifecycleConfig +} - return err - } +func (l *LifecycleConfigurationClient) GetSubresourceName() string { + return "LifecycleConfigurationClient" +} +//nolint:dupl // Pattern is intentionally shared with other subresources (ServerSideEncryption) +func (l *LifecycleConfigurationClient) HandleObservation(ctx context.Context, observation ResourceStatus, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { switch observation { case NoAction: return nil @@ -171,32 +130,32 @@ func (l *LifecycleConfigurationClient) Handle(ctx context.Context, b *v1alpha1.B // The lifecycle config is updated, so we can consider this // sub resource Available. available := xpv1.Available() - bb.setLifecycleConfigCondition(b.Name, backendName, &available) + bb.setLifecycleConfigCondition(bucket.Name, backendName, &available) + return nil case NeedsDeletion: - if err := l.delete(ctx, b, backendName); err != nil { + if err := l.delete(ctx, bucket, backendName); err != nil { err = errors.Wrap(err, errHandleLifecycleConfig) deleting := xpv1.Deleting().WithMessage(err.Error()) - bb.setLifecycleConfigCondition(b.Name, backendName, &deleting) - - traces.SetAndRecordError(span, err) + bb.setLifecycleConfigCondition(bucket.Name, backendName, &deleting) return err } - bb.setLifecycleConfigCondition(b.Name, backendName, nil) + bb.setLifecycleConfigCondition(bucket.Name, backendName, nil) + return nil case NeedsUpdate: - if err := l.createOrUpdate(ctx, b, backendName); err != nil { + if err := l.createOrUpdate(ctx, bucket, backendName); err != nil { err = errors.Wrap(err, errHandleLifecycleConfig) unavailable := xpv1.Unavailable().WithMessage(err.Error()) - bb.setLifecycleConfigCondition(b.Name, backendName, &unavailable) - - traces.SetAndRecordError(span, err) + bb.setLifecycleConfigCondition(bucket.Name, backendName, &unavailable) return err } available := xpv1.Available() - bb.setLifecycleConfigCondition(b.Name, backendName, &available) + bb.setLifecycleConfigCondition(bucket.Name, backendName, &available) + + return nil } return nil diff --git a/internal/controller/bucket/lifecycleconfiguration_test.go b/internal/controller/bucket/lifecycleconfiguration_test.go index 9860a1c9..f2cbb928 100644 --- a/internal/controller/bucket/lifecycleconfiguration_test.go +++ b/internal/controller/bucket/lifecycleconfiguration_test.go @@ -454,7 +454,7 @@ func TestObserveBackend(t *testing.T) { s3clienthandler.WithBackendStore(tc.fields.backendStore)), logr.Discard()) - got, err := c.observeBackend(context.Background(), tc.args.bucket, tc.args.backendName) + got, err := c.ObserveBackend(context.Background(), tc.args.bucket, tc.args.backendName) require.ErrorIs(t, err, tc.want.err, "unexpected error") assert.Equal(t, tc.want.status, got, "unexpected status") }) diff --git a/internal/controller/bucket/objectlockconfiguration.go b/internal/controller/bucket/objectlockconfiguration.go index be05cc26..9c766a90 100644 --- a/internal/controller/bucket/objectlockconfiguration.go +++ b/internal/controller/bucket/objectlockconfiguration.go @@ -12,7 +12,6 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/errors" "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" - apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" "github.com/linode/provider-ceph/internal/backendstore" "github.com/linode/provider-ceph/internal/consts" "github.com/linode/provider-ceph/internal/controller/s3clienthandler" @@ -21,87 +20,53 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" - - "go.opentelemetry.io/otel" ) // ObjectLockConfigurationClient is the client for API methods and reconciling the ObjectLockConfiguration type ObjectLockConfigurationClient struct { - backendStore *backendstore.BackendStore - s3ClientHandler *s3clienthandler.Handler - log logr.Logger + BaseSubresourceClient } func NewObjectLockConfigurationClient(b *backendstore.BackendStore, h *s3clienthandler.Handler, l logr.Logger) *ObjectLockConfigurationClient { - return &ObjectLockConfigurationClient{backendStore: b, s3ClientHandler: h, log: l} + return &ObjectLockConfigurationClient{BaseSubresourceClient: NewBaseSubresourceClient(b, h, l)} } -func (l *ObjectLockConfigurationClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { - ctx, span := otel.Tracer("").Start(ctx, "bucket.ObjectLockConfigurationClient.Observe") - defer span.End() - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) - - if bucket.Spec.ForProvider.ObjectLockEnabledForBucket == nil || !*bucket.Spec.ForProvider.ObjectLockEnabledForBucket { - log.V(1).Info("Object lock configuration not enabled in Bucket CR", consts.KeyBucketName, bucket.Name) - - return Updated, nil - } - - observationChan := make(chan ResourceStatus) - errChan := make(chan error) +func (o *ObjectLockConfigurationClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { + return o.BaseSubresourceClient.Observe(ctx, bucket, backendNames, o) +} - for _, backendName := range backendNames { - beName := backendName - go func() { - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - // If a backend is marked as unhealthy, we can ignore it for now by returning NoAction. - // The backend may be down for some time and we do not want to block Create/Update/Delete - // calls on other backends. By returning NoAction here, we would never pass the Observe - // phase until the backend becomes Healthy or Disabled. - observationChan <- NoAction +func (o *ObjectLockConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { + return o.BaseSubresourceClient.Handle(ctx, b, backendName, bb, o) +} - return - } +// Implement Subresource interface - observation, err := l.observeBackend(ctx, bucket, beName) - if err != nil { - errChan <- err +func (o *ObjectLockConfigurationClient) SkipObservation(bucket *v1alpha1.Bucket) bool { + return bucket.Spec.ForProvider.ObjectLockEnabledForBucket == nil || !*bucket.Spec.ForProvider.ObjectLockEnabledForBucket +} - return - } - observationChan <- observation - }() - } +func (o *ObjectLockConfigurationClient) GetLogger() logr.Logger { + return o.log +} - for i := 0; i < len(backendNames); i++ { - select { - case <-ctx.Done(): - log.Info("Context timeout during object lock configuration observation", consts.KeyBucketName, bucket.Name) - err := errors.Wrap(ctx.Err(), errObserveObjectLockConfig) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - case observation := <-observationChan: - if observation == NeedsUpdate || observation == NeedsDeletion { - return observation, nil - } - case err := <-errChan: - err = errors.Wrap(err, errObserveObjectLockConfig) - traces.SetAndRecordError(span, err) +func (o *ObjectLockConfigurationClient) GetBackendStore() *backendstore.BackendStore { + return o.backendStore +} - return NeedsUpdate, err - } - } +func (o *ObjectLockConfigurationClient) GetS3ClientHandler() *s3clienthandler.Handler { + return o.s3ClientHandler +} - return Updated, nil +func (o *ObjectLockConfigurationClient) GetObserveErrorMsg() string { + return errObserveObjectLockConfig } -func (l *ObjectLockConfigurationClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) +func (o *ObjectLockConfigurationClient) ObserveBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { + ctx, log := traces.InjectTraceAndLogger(ctx, o.log) log.V(1).Info("Observing subresource object lock configuration on backend", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) - s3Client, err := l.s3ClientHandler.GetS3Client(ctx, bucket, backendName) + s3Client, err := o.s3ClientHandler.GetS3Client(ctx, bucket, backendName) if err != nil { return NeedsUpdate, err } @@ -115,9 +80,9 @@ func (l *ObjectLockConfigurationClient) observeBackend(ctx context.Context, buck external = response.ObjectLockConfiguration } - desiredVersioningConfig := rgw.GenerateObjectLockConfiguration(bucket.Spec.ForProvider.ObjectLockConfiguration) + desiredObjectLockConfig := rgw.GenerateObjectLockConfiguration(bucket.Spec.ForProvider.ObjectLockConfiguration) - if !cmp.Equal(external, desiredVersioningConfig, cmpopts.IgnoreTypes(document.NoSerde{})) { + if !cmp.Equal(external, desiredObjectLockConfig, cmpopts.IgnoreTypes(document.NoSerde{})) { log.Info("Object lock configuration requires update on backend", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) return NeedsUpdate, nil @@ -126,28 +91,17 @@ func (l *ObjectLockConfigurationClient) observeBackend(ctx context.Context, buck return Updated, nil } -func (l *ObjectLockConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { - ctx, span := otel.Tracer("").Start(ctx, "bucket.ObjectLockConfigurationClient.Handle") - defer span.End() - - if b.Spec.ForProvider.ObjectLockEnabledForBucket == nil || !*b.Spec.ForProvider.ObjectLockEnabledForBucket { - return nil - } - - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - traces.SetAndRecordError(span, errUnhealthyBackend) - - return errUnhealthyBackend - } +// Implement Subresource interface - observation, err := l.observeBackend(ctx, b, backendName) - if err != nil { - err = errors.Wrap(err, errHandleVersioningConfig) - traces.SetAndRecordError(span, err) +func (o *ObjectLockConfigurationClient) GetHandleErrorMsg() string { + return errHandleObjectLockConfig +} - return err - } +func (o *ObjectLockConfigurationClient) GetSubresourceName() string { + return "ObjectLockConfigurationClient" +} +func (o *ObjectLockConfigurationClient) HandleObservation(ctx context.Context, observation ResourceStatus, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { switch observation { case NoAction: return nil @@ -155,7 +109,7 @@ func (l *ObjectLockConfigurationClient) Handle(ctx context.Context, b *v1alpha1. // The object lock config is updated, so we can consider this // sub resource Available. available := xpv1.Available() - bb.setObjectLockConfigCondition(b.Name, backendName, &available) + bb.setObjectLockConfigCondition(bucket.Name, backendName, &available) return nil case NeedsDeletion: @@ -166,35 +120,35 @@ func (l *ObjectLockConfigurationClient) Handle(ctx context.Context, b *v1alpha1. // has been enabled for the bucket and no object lock configuration is // specified in the Bucket CR Spec, we should default to a basic "enabled" // object lock configuration. - bucketCopy := b.DeepCopy() + bucketCopy := bucket.DeepCopy() enabled := v1alpha1.ObjectLockEnabledEnabled - if b.Spec.ForProvider.ObjectLockConfiguration == nil { + if bucket.Spec.ForProvider.ObjectLockConfiguration == nil { bucketCopy.Spec.ForProvider.ObjectLockConfiguration = &v1alpha1.ObjectLockConfiguration{ ObjectLockEnabled: &enabled, } } - if err := l.createOrUpdate(ctx, bucketCopy, backendName); err != nil { + if err := o.createOrUpdate(ctx, bucketCopy, backendName); err != nil { err = errors.Wrap(err, errHandleObjectLockConfig) unavailable := xpv1.Unavailable().WithMessage(err.Error()) bb.setObjectLockConfigCondition(bucketCopy.Name, backendName, &unavailable) - traces.SetAndRecordError(span, err) - return err } available := xpv1.Available() bb.setObjectLockConfigCondition(bucketCopy.Name, backendName, &available) + + return nil } return nil } -func (l *ObjectLockConfigurationClient) createOrUpdate(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) +func (o *ObjectLockConfigurationClient) createOrUpdate(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { + ctx, log := traces.InjectTraceAndLogger(ctx, o.log) log.Info("Updating object lock configuration", consts.KeyBucketName, b.Name, consts.KeyBackendName, backendName) - s3Client, err := l.s3ClientHandler.GetS3Client(ctx, b, backendName) + s3Client, err := o.s3ClientHandler.GetS3Client(ctx, b, backendName) if err != nil { return err } diff --git a/internal/controller/bucket/objectlockconfiguration_test.go b/internal/controller/bucket/objectlockconfiguration_test.go index 145f0f39..90f8dbd6 100644 --- a/internal/controller/bucket/objectlockconfiguration_test.go +++ b/internal/controller/bucket/objectlockconfiguration_test.go @@ -200,7 +200,7 @@ func TestObjectLockConfigObserveBackend(t *testing.T) { s3clienthandler.WithBackendStore(tc.fields.backendStore)), logr.Discard()) - got, err := c.observeBackend(context.Background(), tc.args.bucket, tc.args.backendName) + got, err := c.ObserveBackend(context.Background(), tc.args.bucket, tc.args.backendName) require.ErrorIs(t, err, tc.want.err, "unexpected error") assert.Equal(t, tc.want.status, got, "unexpected status") }) diff --git a/internal/controller/bucket/policy.go b/internal/controller/bucket/policy.go index abab7af3..ee61ae73 100644 --- a/internal/controller/bucket/policy.go +++ b/internal/controller/bucket/policy.go @@ -10,82 +10,49 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/errors" "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" - apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" "github.com/linode/provider-ceph/internal/backendstore" "github.com/linode/provider-ceph/internal/consts" "github.com/linode/provider-ceph/internal/controller/s3clienthandler" "github.com/linode/provider-ceph/internal/otel/traces" "github.com/linode/provider-ceph/internal/rgw" - "go.opentelemetry.io/otel" ) // PolicyClient is the client for API methods and reconciling a BucketPolicy type PolicyClient struct { - backendStore *backendstore.BackendStore - s3ClientHandler *s3clienthandler.Handler - log logr.Logger + BaseSubresourceClient } func NewPolicyClient(b *backendstore.BackendStore, h *s3clienthandler.Handler, l logr.Logger) *PolicyClient { - return &PolicyClient{backendStore: b, s3ClientHandler: h, log: l} + return &PolicyClient{BaseSubresourceClient: NewBaseSubresourceClient(b, h, l)} } -//nolint:dupl // LifecycleConfiguration and Policy are different feature. func (p *PolicyClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { - ctx, span := otel.Tracer("").Start(ctx, "bucket.PolicyClient.Observe") - defer span.End() - ctx, log := traces.InjectTraceAndLogger(ctx, p.log) + return p.BaseSubresourceClient.Observe(ctx, bucket, backendNames, p) +} - observationChan := make(chan ResourceStatus) - errChan := make(chan error) - - for _, backendName := range backendNames { - beName := backendName - go func() { - if p.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - // If a backend is marked as unhealthy, we can ignore it for now by returning NoAction. - // The backend may be down for some time and we do not want to block Create/Update/Delete - // calls on other backends. By returning NoAction here, we would never pass the Observe - // phase until the backend becomes Healthy or Disabled. - observationChan <- NoAction - - return - } - - observation, err := p.observeBackend(ctx, bucket, beName) - if err != nil { - errChan <- err - - return - } - observationChan <- observation - }() - } +func (p *PolicyClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { + return p.BaseSubresourceClient.Handle(ctx, b, backendName, bb, p) +} - for i := 0; i < len(backendNames); i++ { - select { - case <-ctx.Done(): - log.Info("Context timeout during bucket policy observation", consts.KeyBucketName, bucket.Name) - err := errors.Wrap(ctx.Err(), errObservePolicy) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - case observation := <-observationChan: - if observation == NeedsUpdate || observation == NeedsDeletion { - return observation, nil - } - case err := <-errChan: - err = errors.Wrap(err, errObservePolicy) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - } - } +// Implement Subresource interface - return Updated, nil +func (p *PolicyClient) GetLogger() logr.Logger { + return p.log } -func (p *PolicyClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { +func (p *PolicyClient) GetBackendStore() *backendstore.BackendStore { + return p.backendStore +} + +func (p *PolicyClient) GetS3ClientHandler() *s3clienthandler.Handler { + return p.s3ClientHandler +} + +func (p *PolicyClient) GetObserveErrorMsg() string { + return errObservePolicy +} + +func (p *PolicyClient) ObserveBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { ctx, log := traces.InjectTraceAndLogger(ctx, p.log) log.V(1).Info("Observing subresource policy on backend", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) @@ -132,43 +99,24 @@ func (p *PolicyClient) observeBackend(ctx context.Context, bucket *v1alpha1.Buck return Updated, nil } -func (p *PolicyClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { - ctx, span := otel.Tracer("").Start(ctx, "bucket.PolicyClient.Handle") - defer span.End() +// Implement Subresource interface - if p.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - traces.SetAndRecordError(span, errUnhealthyBackend) - - return errUnhealthyBackend - } - - observation, err := p.observeBackend(ctx, b, backendName) - if err != nil { - err = errors.Wrap(err, errHandlePolicy) - traces.SetAndRecordError(span, err) +func (p *PolicyClient) GetHandleErrorMsg() string { + return errHandlePolicy +} - return err - } +func (p *PolicyClient) GetSubresourceName() string { + return "PolicyClient" +} +func (p *PolicyClient) HandleObservation(ctx context.Context, observation ResourceStatus, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { switch observation { case NoAction, Updated: return nil case NeedsDeletion: - if err := p.delete(ctx, b, backendName); err != nil { - err = errors.Wrap(err, errHandlePolicy) - - traces.SetAndRecordError(span, err) - - return err - } + return p.delete(ctx, bucket, backendName) case NeedsUpdate: - if err := p.createOrUpdate(ctx, b, backendName); err != nil { - err = errors.Wrap(err, errHandlePolicy) - - traces.SetAndRecordError(span, err) - - return err - } + return p.createOrUpdate(ctx, bucket, backendName) } return nil diff --git a/internal/controller/bucket/policy_test.go b/internal/controller/bucket/policy_test.go index 9d3d4390..c57a561b 100644 --- a/internal/controller/bucket/policy_test.go +++ b/internal/controller/bucket/policy_test.go @@ -299,7 +299,7 @@ func TestPolicyObserveBackend(t *testing.T) { logr.Discard(), ) - got, err := c.observeBackend(context.Background(), tc.args.bucket, tc.args.backendName) + got, err := c.ObserveBackend(context.Background(), tc.args.bucket, tc.args.backendName) require.ErrorIs(t, err, tc.want.err, "unexpected error") assert.Equal(t, tc.want.status, got, "unexpected status") }) diff --git a/internal/controller/bucket/serversideencryptionconfiguration.go b/internal/controller/bucket/serversideencryptionconfiguration.go index bb39bf31..c01252f4 100644 --- a/internal/controller/bucket/serversideencryptionconfiguration.go +++ b/internal/controller/bucket/serversideencryptionconfiguration.go @@ -12,7 +12,6 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/errors" "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" - apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" "github.com/linode/provider-ceph/internal/backendstore" "github.com/linode/provider-ceph/internal/consts" "github.com/linode/provider-ceph/internal/controller/s3clienthandler" @@ -21,83 +20,50 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" - - "go.opentelemetry.io/otel" ) // ServerSideEncryptionConfigurationClient is the client for API methods and reconciling the ServerSideEncryptionConfiguration type ServerSideEncryptionConfigurationClient struct { - backendStore *backendstore.BackendStore - s3ClientHandler *s3clienthandler.Handler - log logr.Logger + BaseSubresourceClient } func NewServerSideEncryptionConfigurationClient(b *backendstore.BackendStore, h *s3clienthandler.Handler, l logr.Logger) *ServerSideEncryptionConfigurationClient { - return &ServerSideEncryptionConfigurationClient{backendStore: b, s3ClientHandler: h, log: l} + return &ServerSideEncryptionConfigurationClient{BaseSubresourceClient: NewBaseSubresourceClient(b, h, l)} } -//nolint:dupl // ServerSideEncryptionConfiguration is similar to other subresource clients. -func (l *ServerSideEncryptionConfigurationClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { - ctx, span := otel.Tracer("").Start(ctx, "bucket.ServerSideEncryptionConfigurationClient.Observe") - defer span.End() - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) - - observationChan := make(chan ResourceStatus) - errChan := make(chan error) - - for _, backendName := range backendNames { - beName := backendName - go func() { - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - // If a backend is marked as unhealthy, we can ignore it for now by returning NoAction. - // The backend may be down for some time and we do not want to block Create/Update/Delete - // calls on other backends. By returning NoAction here, we would never pass the Observe - // phase until the backend becomes Healthy or Disabled. - observationChan <- NoAction - - return - } - - observation, err := l.observeBackend(ctx, bucket, beName) - if err != nil { - errChan <- err - - return - } - observationChan <- observation - }() - } +func (s *ServerSideEncryptionConfigurationClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { + return s.BaseSubresourceClient.Observe(ctx, bucket, backendNames, s) +} - for i := 0; i < len(backendNames); i++ { - select { - case <-ctx.Done(): - log.Info("Context timeout during bucket server side encryption configuration observation", consts.KeyBucketName, bucket.Name) - err := errors.Wrap(ctx.Err(), errObserveSSEConfig) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - case observation := <-observationChan: - if observation == NeedsUpdate || observation == NeedsDeletion { - return observation, nil - } - case err := <-errChan: - err = errors.Wrap(err, errObserveSSEConfig) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - } - } +func (s *ServerSideEncryptionConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { + return s.BaseSubresourceClient.Handle(ctx, b, backendName, bb, s) +} - return Updated, nil +// Implement Subresource interface + +func (s *ServerSideEncryptionConfigurationClient) GetLogger() logr.Logger { + return s.log +} + +func (s *ServerSideEncryptionConfigurationClient) GetBackendStore() *backendstore.BackendStore { + return s.backendStore +} + +func (s *ServerSideEncryptionConfigurationClient) GetS3ClientHandler() *s3clienthandler.Handler { + return s.s3ClientHandler +} + +func (s *ServerSideEncryptionConfigurationClient) GetObserveErrorMsg() string { + return errObserveSSEConfig } //nolint:gocyclo,cyclop // Function requires numerous checks. -func (l *ServerSideEncryptionConfigurationClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) +func (s *ServerSideEncryptionConfigurationClient) ObserveBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { + ctx, log := traces.InjectTraceAndLogger(ctx, s.log) log.V(1).Info("Observing subresource server side encryption configuration on backend", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) - s3Client, err := l.s3ClientHandler.GetS3Client(ctx, bucket, backendName) + s3Client, err := s.s3ClientHandler.GetS3Client(ctx, bucket, backendName) if err != nil { return NeedsUpdate, err } @@ -146,25 +112,18 @@ func (l *ServerSideEncryptionConfigurationClient) observeBackend(ctx context.Con return Updated, nil } -//nolint:dupl // ServerSideEncryptionConfiguration and LifecycleConfiguration have similar Handle logic. -func (l *ServerSideEncryptionConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { - ctx, span := otel.Tracer("").Start(ctx, "bucket.ServerSideEncryptionConfigurationClient.Handle") - defer span.End() +// Implement Subresource interface - if l.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - traces.SetAndRecordError(span, errUnhealthyBackend) - - return errUnhealthyBackend - } - - observation, err := l.observeBackend(ctx, b, backendName) - if err != nil { - err = errors.Wrap(err, errHandleSSEConfig) - traces.SetAndRecordError(span, err) +func (s *ServerSideEncryptionConfigurationClient) GetHandleErrorMsg() string { + return errHandleSSEConfig +} - return err - } +func (s *ServerSideEncryptionConfigurationClient) GetSubresourceName() string { + return "ServerSideEncryptionConfigurationClient" +} +//nolint:dupl // Pattern is intentionally shared with other subresources (LifecycleConfiguration) +func (s *ServerSideEncryptionConfigurationClient) HandleObservation(ctx context.Context, observation ResourceStatus, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { switch observation { case NoAction: return nil @@ -172,42 +131,42 @@ func (l *ServerSideEncryptionConfigurationClient) Handle(ctx context.Context, b // The SSE config is updated, so we can consider this // sub resource Available. available := xpv1.Available() - bb.setSSEConfigCondition(b.Name, backendName, &available) + bb.setSSEConfigCondition(bucket.Name, backendName, &available) + return nil case NeedsDeletion: - if err := l.delete(ctx, b, backendName); err != nil { + if err := s.delete(ctx, bucket, backendName); err != nil { err = errors.Wrap(err, errHandleSSEConfig) deleting := xpv1.Deleting().WithMessage(err.Error()) - bb.setSSEConfigCondition(b.Name, backendName, &deleting) - - traces.SetAndRecordError(span, err) + bb.setSSEConfigCondition(bucket.Name, backendName, &deleting) return err } - bb.setSSEConfigCondition(b.Name, backendName, nil) + bb.setSSEConfigCondition(bucket.Name, backendName, nil) + return nil case NeedsUpdate: - if err := l.createOrUpdate(ctx, b, backendName); err != nil { + if err := s.createOrUpdate(ctx, bucket, backendName); err != nil { err = errors.Wrap(err, errHandleSSEConfig) unavailable := xpv1.Unavailable().WithMessage(err.Error()) - bb.setSSEConfigCondition(b.Name, backendName, &unavailable) - - traces.SetAndRecordError(span, err) + bb.setSSEConfigCondition(bucket.Name, backendName, &unavailable) return err } available := xpv1.Available() - bb.setSSEConfigCondition(b.Name, backendName, &available) + bb.setSSEConfigCondition(bucket.Name, backendName, &available) + + return nil } return nil } -func (l *ServerSideEncryptionConfigurationClient) createOrUpdate(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) +func (s *ServerSideEncryptionConfigurationClient) createOrUpdate(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { + ctx, log := traces.InjectTraceAndLogger(ctx, s.log) log.Info("Updating server side encryption configuration", consts.KeyBucketName, b.Name, consts.KeyBackendName, backendName) - s3Client, err := l.s3ClientHandler.GetS3Client(ctx, b, backendName) + s3Client, err := s.s3ClientHandler.GetS3Client(ctx, b, backendName) if err != nil { return err } @@ -220,11 +179,11 @@ func (l *ServerSideEncryptionConfigurationClient) createOrUpdate(ctx context.Con return nil } -func (l *ServerSideEncryptionConfigurationClient) delete(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { - ctx, log := traces.InjectTraceAndLogger(ctx, l.log) +func (s *ServerSideEncryptionConfigurationClient) delete(ctx context.Context, b *v1alpha1.Bucket, backendName string) error { + ctx, log := traces.InjectTraceAndLogger(ctx, s.log) log.Info("Deleting server side encryption configuration", consts.KeyBucketName, b.Name, consts.KeyBackendName, backendName) - s3Client, err := l.s3ClientHandler.GetS3Client(ctx, b, backendName) + s3Client, err := s.s3ClientHandler.GetS3Client(ctx, b, backendName) if err != nil { return err } diff --git a/internal/controller/bucket/serversideencryptionconfiguration_test.go b/internal/controller/bucket/serversideencryptionconfiguration_test.go index c8a77a35..1a8d75ae 100644 --- a/internal/controller/bucket/serversideencryptionconfiguration_test.go +++ b/internal/controller/bucket/serversideencryptionconfiguration_test.go @@ -459,7 +459,7 @@ func TestSSEConfigObserveBackend(t *testing.T) { s3clienthandler.WithBackendStore(tc.fields.backendStore)), logr.Discard()) - got, err := c.observeBackend(context.Background(), tc.args.bucket, tc.args.backendName) + got, err := c.ObserveBackend(context.Background(), tc.args.bucket, tc.args.backendName) require.ErrorIs(t, err, tc.want.err, "unexpected error") assert.Equal(t, tc.want.status, got, "unexpected status") }) diff --git a/internal/controller/bucket/versioningconfiguration.go b/internal/controller/bucket/versioningconfiguration.go index ccbd13d0..783a087c 100644 --- a/internal/controller/bucket/versioningconfiguration.go +++ b/internal/controller/bucket/versioningconfiguration.go @@ -12,7 +12,6 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/errors" "github.com/linode/provider-ceph/apis/provider-ceph/v1alpha1" - apisv1alpha1 "github.com/linode/provider-ceph/apis/v1alpha1" "github.com/linode/provider-ceph/internal/backendstore" "github.com/linode/provider-ceph/internal/consts" "github.com/linode/provider-ceph/internal/controller/s3clienthandler" @@ -21,77 +20,44 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" - - "go.opentelemetry.io/otel" ) // VersioningConfigurationClient is the client for API methods and reconciling the VersioningConfiguration type VersioningConfigurationClient struct { - backendStore *backendstore.BackendStore - s3ClientHandler *s3clienthandler.Handler - log logr.Logger + BaseSubresourceClient } func NewVersioningConfigurationClient(b *backendstore.BackendStore, h *s3clienthandler.Handler, l logr.Logger) *VersioningConfigurationClient { - return &VersioningConfigurationClient{backendStore: b, s3ClientHandler: h, log: l} + return &VersioningConfigurationClient{BaseSubresourceClient: NewBaseSubresourceClient(b, h, l)} } -//nolint:dupl // VersioningConfiguration and Policy are different feature. func (v *VersioningConfigurationClient) Observe(ctx context.Context, bucket *v1alpha1.Bucket, backendNames []string) (ResourceStatus, error) { - ctx, span := otel.Tracer("").Start(ctx, "bucket.VersioningConfigurationClient.Observe") - defer span.End() - ctx, log := traces.InjectTraceAndLogger(ctx, v.log) - - observationChan := make(chan ResourceStatus) - errChan := make(chan error) - - for _, backendName := range backendNames { - beName := backendName - go func() { - if v.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - // If a backend is marked as unhealthy, we can ignore it for now by returning NoAction. - // The backend may be down for some time and we do not want to block Create/Update/Delete - // calls on other backends. By returning NoAction here, we would never pass the Observe - // phase until the backend becomes Healthy or Disabled. - observationChan <- NoAction + return v.BaseSubresourceClient.Observe(ctx, bucket, backendNames, v) +} - return - } +func (v *VersioningConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { + return v.BaseSubresourceClient.Handle(ctx, b, backendName, bb, v) +} - observation, err := v.observeBackend(ctx, bucket, beName) - if err != nil { - errChan <- err +// Implement Subresource interface - return - } - observationChan <- observation - }() - } +func (v *VersioningConfigurationClient) GetLogger() logr.Logger { + return v.log +} - for i := 0; i < len(backendNames); i++ { - select { - case <-ctx.Done(): - log.Info("Context timeout during bucket versioning configuration observation", consts.KeyBucketName, bucket.Name) - err := errors.Wrap(ctx.Err(), errObserveVersioningConfig) - traces.SetAndRecordError(span, err) - - return NeedsUpdate, err - case observation := <-observationChan: - if observation == NeedsUpdate || observation == NeedsDeletion { - return observation, nil - } - case err := <-errChan: - err = errors.Wrap(err, errObserveVersioningConfig) - traces.SetAndRecordError(span, err) +func (v *VersioningConfigurationClient) GetBackendStore() *backendstore.BackendStore { + return v.backendStore +} - return NeedsUpdate, err - } - } +func (v *VersioningConfigurationClient) GetS3ClientHandler() *s3clienthandler.Handler { + return v.s3ClientHandler +} - return Updated, nil +func (v *VersioningConfigurationClient) GetObserveErrorMsg() string { + return errObserveVersioningConfig } -func (v *VersioningConfigurationClient) observeBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { +func (v *VersioningConfigurationClient) ObserveBackend(ctx context.Context, bucket *v1alpha1.Bucket, backendName string) (ResourceStatus, error) { ctx, log := traces.InjectTraceAndLogger(ctx, v.log) log.V(1).Info("Observing subresource versioning configuration on backend", consts.KeyBucketName, bucket.Name, consts.KeyBackendName, backendName) @@ -146,24 +112,17 @@ func (v *VersioningConfigurationClient) observeBackend(ctx context.Context, buck return Updated, nil } -func (v *VersioningConfigurationClient) Handle(ctx context.Context, b *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { - ctx, span := otel.Tracer("").Start(ctx, "bucket.VersioningConfigurationClient.Handle") - defer span.End() - - if v.backendStore.GetBackendHealthStatus(backendName) == apisv1alpha1.HealthStatusUnhealthy { - traces.SetAndRecordError(span, errUnhealthyBackend) +// Implement Subresource interface - return errUnhealthyBackend - } - - observation, err := v.observeBackend(ctx, b, backendName) - if err != nil { - err = errors.Wrap(err, errHandleVersioningConfig) - traces.SetAndRecordError(span, err) +func (v *VersioningConfigurationClient) GetHandleErrorMsg() string { + return errHandleVersioningConfig +} - return err - } +func (v *VersioningConfigurationClient) GetSubresourceName() string { + return "VersioningConfigurationClient" +} +func (v *VersioningConfigurationClient) HandleObservation(ctx context.Context, observation ResourceStatus, bucket *v1alpha1.Bucket, backendName string, bb *bucketBackends) error { switch observation { case NoAction: return nil @@ -171,14 +130,14 @@ func (v *VersioningConfigurationClient) Handle(ctx context.Context, b *v1alpha1. // The versioning config is updated, so we can consider this // sub resource Available. available := xpv1.Available() - bb.setVersioningConfigCondition(b.Name, backendName, &available) + bb.setVersioningConfigCondition(bucket.Name, backendName, &available) return nil case NeedsDeletion: // Versioning Configurations are not deleted, only suspended, which requires an update. // Create a deep copy of bucket and give it a suspended version config. // This will be used in th PutBucketVersioning request to suspend versioning. - bucketCopy := b.DeepCopy() + bucketCopy := bucket.DeepCopy() disabled := v1alpha1.MFADeleteDisabled suspended := v1alpha1.VersioningStatusSuspended @@ -189,9 +148,7 @@ func (v *VersioningConfigurationClient) Handle(ctx context.Context, b *v1alpha1. if err := v.createOrUpdate(ctx, bucketCopy, backendName); err != nil { err = errors.Wrap(err, errHandleVersioningConfig) unavailable := xpv1.Unavailable().WithMessage(err.Error()) - bb.setVersioningConfigCondition(b.Name, backendName, &unavailable) - - traces.SetAndRecordError(span, err) + bb.setVersioningConfigCondition(bucket.Name, backendName, &unavailable) return err } @@ -199,11 +156,11 @@ func (v *VersioningConfigurationClient) Handle(ctx context.Context, b *v1alpha1. // un-version a bucket, we must not remove its versioningConfigCondition. // Instead, we set it as Available, signifying that the update was a success. available := xpv1.Available() - bb.setVersioningConfigCondition(b.Name, backendName, &available) + bb.setVersioningConfigCondition(bucket.Name, backendName, &available) return nil case NeedsUpdate: - bucketCopy := b.DeepCopy() + bucketCopy := bucket.DeepCopy() // If no versioning configuration was specified, but object lock is enabled // for the bucket, then versioning should be enabled without mfa delete. @@ -212,9 +169,9 @@ func (v *VersioningConfigurationClient) Handle(ctx context.Context, b *v1alpha1. // If objectLockEnabledForBucket was true upon bucket creation, then this // versioning configuration should already exist. But we perform the operation // anyway to make sure, as it is idempotent. - if b.Spec.ForProvider.VersioningConfiguration == nil && - b.Spec.ForProvider.ObjectLockEnabledForBucket != nil && - *b.Spec.ForProvider.ObjectLockEnabledForBucket { + if bucket.Spec.ForProvider.VersioningConfiguration == nil && + bucket.Spec.ForProvider.ObjectLockEnabledForBucket != nil && + *bucket.Spec.ForProvider.ObjectLockEnabledForBucket { enabled := v1alpha1.VersioningStatusEnabled disabled := v1alpha1.MFADeleteDisabled @@ -229,12 +186,12 @@ func (v *VersioningConfigurationClient) Handle(ctx context.Context, b *v1alpha1. unavailable := xpv1.Unavailable().WithMessage(err.Error()) bb.setVersioningConfigCondition(bucketCopy.Name, backendName, &unavailable) - traces.SetAndRecordError(span, err) - return err } available := xpv1.Available() bb.setVersioningConfigCondition(bucketCopy.Name, backendName, &available) + + return nil } return nil diff --git a/internal/controller/bucket/versioningconfiguration_test.go b/internal/controller/bucket/versioningconfiguration_test.go index 60531c68..625bd534 100644 --- a/internal/controller/bucket/versioningconfiguration_test.go +++ b/internal/controller/bucket/versioningconfiguration_test.go @@ -309,7 +309,7 @@ func TestVersioningConfigObserveBackend(t *testing.T) { s3clienthandler.WithBackendStore(tc.fields.backendStore)), logr.Discard()) - got, err := c.observeBackend(context.Background(), tc.args.bucket, tc.args.backendName) + got, err := c.ObserveBackend(context.Background(), tc.args.bucket, tc.args.backendName) require.ErrorIs(t, err, tc.want.err, "unexpected error") assert.Equal(t, tc.want.status, got, "unexpected status") })