From 503e3fc1d04ae503b098d9ed5d6af657d0a22cc1 Mon Sep 17 00:00:00 2001 From: Dario Tranchitella Date: Thu, 13 Oct 2022 15:55:21 +0200 Subject: [PATCH] feat: globaltenantresource and tenantresource reconciliation --- api/v1beta2/additional_metadata.go | 9 - api/v1beta2/tenantresource_namespaced.go | 4 +- api/v1beta2/zz_generated.deepcopy.go | 31 +-- controllers/resources/global.go | 186 ++++++++++++++++ controllers/resources/namespaced.go | 144 ++++++++++++ controllers/resources/processor.go | 17 ++ controllers/resources/processor_finalizer.go | 54 +++++ controllers/resources/processor_pruning.go | 67 ++++++ controllers/resources/processor_section.go | 223 +++++++++++++++++++ go.sum | 1 - main.go | 11 + pkg/indexer/indexer.go | 5 +- pkg/indexer/tenant/namespaces.go | 16 +- 13 files changed, 716 insertions(+), 52 deletions(-) delete mode 100644 api/v1beta2/additional_metadata.go create mode 100644 controllers/resources/global.go create mode 100644 controllers/resources/namespaced.go create mode 100644 controllers/resources/processor.go create mode 100644 controllers/resources/processor_finalizer.go create mode 100644 controllers/resources/processor_pruning.go create mode 100644 controllers/resources/processor_section.go diff --git a/api/v1beta2/additional_metadata.go b/api/v1beta2/additional_metadata.go deleted file mode 100644 index 9d708e73..00000000 --- a/api/v1beta2/additional_metadata.go +++ /dev/null @@ -1,9 +0,0 @@ -// Copyright 2020-2021 Clastix Labs -// SPDX-License-Identifier: Apache-2.0 - -package v1beta2 - -type AdditionalMetadataSpec struct { - Labels map[string]string `json:"labels,omitempty"` - Annotations map[string]string `json:"annotations,omitempty"` -} diff --git a/api/v1beta2/tenantresource_namespaced.go b/api/v1beta2/tenantresource_namespaced.go index d0852613..9a0d18d6 100644 --- a/api/v1beta2/tenantresource_namespaced.go +++ b/api/v1beta2/tenantresource_namespaced.go @@ -6,6 +6,8 @@ package v1beta2 import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + + "github.com/clastix/capsule/pkg/api" ) // TenantResourceSpec defines the desired state of TenantResource. @@ -32,7 +34,7 @@ type ResourceSpec struct { RawItems []RawExtension `json:"rawItems,omitempty"` // Besides the Capsule metadata required by TenantResource controller, defines additional metadata that must be // added to the replicated resources. - AdditionalMetadata *AdditionalMetadataSpec `json:"additionalMetadata,omitempty"` + AdditionalMetadata *api.AdditionalMetadataSpec `json:"additionalMetadata,omitempty"` } // +kubebuilder:validation:XEmbeddedResource diff --git a/api/v1beta2/zz_generated.deepcopy.go b/api/v1beta2/zz_generated.deepcopy.go index f0994bb1..b6c0df5d 100644 --- a/api/v1beta2/zz_generated.deepcopy.go +++ b/api/v1beta2/zz_generated.deepcopy.go @@ -15,35 +15,6 @@ import ( "k8s.io/apimachinery/pkg/runtime" ) -// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *AdditionalMetadataSpec) DeepCopyInto(out *AdditionalMetadataSpec) { - *out = *in - if in.Labels != nil { - in, out := &in.Labels, &out.Labels - *out = make(map[string]string, len(*in)) - for key, val := range *in { - (*out)[key] = val - } - } - if in.Annotations != nil { - in, out := &in.Annotations, &out.Annotations - *out = make(map[string]string, len(*in)) - for key, val := range *in { - (*out)[key] = val - } - } -} - -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AdditionalMetadataSpec. -func (in *AdditionalMetadataSpec) DeepCopy() *AdditionalMetadataSpec { - if in == nil { - return nil - } - out := new(AdditionalMetadataSpec) - in.DeepCopyInto(out) - return out -} - // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *AdditionalRoleBindingsSpec) DeepCopyInto(out *AdditionalRoleBindingsSpec) { *out = *in @@ -525,7 +496,7 @@ func (in *ResourceSpec) DeepCopyInto(out *ResourceSpec) { } if in.AdditionalMetadata != nil { in, out := &in.AdditionalMetadata, &out.AdditionalMetadata - *out = new(AdditionalMetadataSpec) + *out = new(api.AdditionalMetadataSpec) (*in).DeepCopyInto(*out) } } diff --git a/controllers/resources/global.go b/controllers/resources/global.go new file mode 100644 index 00000000..be485319 --- /dev/null +++ b/controllers/resources/global.go @@ -0,0 +1,186 @@ +// Copyright 2020-2021 Clastix Labs +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + + "github.com/hashicorp/go-multierror" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + "sigs.k8s.io/controller-runtime/pkg/source" + + capsulev1beta2 "github.com/clastix/capsule/api/v1beta2" +) + +type Global struct { + client client.Client + processor Processor +} + +func (r *Global) enqueueRequestFromTenant(object client.Object) (reqs []reconcile.Request) { + tnt := object.(*capsulev1beta2.Tenant) //nolint:forcetypeassert + + resList := capsulev1beta2.GlobalTenantResourceList{} + if err := r.client.List(context.Background(), &resList); err != nil { + return nil + } + + set := sets.NewString() + + for _, res := range resList.Items { + selector, err := metav1.LabelSelectorAsSelector(&res.Spec.TenantSelector) + if err != nil { + continue + } + + if selector.Matches(labels.Set(tnt.GetLabels())) { + set.Insert(res.GetName()) + } + } + // No need of ordered value here + for res := range set { + reqs = append(reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{ + Name: res, + }, + }) + } + + return reqs +} + +func (r *Global) SetupWithManager(mgr ctrl.Manager) error { + unstructuredCachingClient, err := client.NewDelegatingClient( + client.NewDelegatingClientInput{ + Client: mgr.GetClient(), + CacheReader: mgr.GetCache(), + CacheUnstructured: true, + }, + ) + if err != nil { + return err + } + + r.client = mgr.GetClient() + r.processor = Processor{ + client: r.client, + unstructuredClient: unstructuredCachingClient, + } + + return ctrl.NewControllerManagedBy(mgr). + For(&capsulev1beta2.GlobalTenantResource{}). + Watches(&source.Kind{Type: &capsulev1beta2.Tenant{}}, handler.EnqueueRequestsFromMapFunc(r.enqueueRequestFromTenant)). + Complete(r) +} + +func (r *Global) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) { + log := ctrllog.FromContext(ctx) + + log.Info("start processing") + + tntResource := capsulev1beta2.GlobalTenantResource{} + if err := r.client.Get(ctx, request.NamespacedName, &tntResource); err != nil { + if errors.IsNotFound(err) { + log.Info("Request object not found, could have been deleted after reconcile request") + + return reconcile.Result{}, nil + } + + return reconcile.Result{}, err + } + // Adding the default value for the status + if tntResource.Status.ProcessedItems == nil { + tntResource.Status.ProcessedItems = make([]capsulev1beta2.ObjectReferenceStatus, 0, 0) + } + // Handling the finalizer section for the given GlobalTenantResource + enqueueBack, err := r.processor.HandleFinalizer(ctx, &tntResource, *tntResource.Spec.PruningOnDelete, tntResource.Status.ProcessedItems) + if err != nil || enqueueBack { + return reconcile.Result{}, err + } + // Retrieving the list of the Tenants up to the selector provided by the GlobalTenantResource resource. + tntSelector, err := metav1.LabelSelectorAsSelector(&tntResource.Spec.TenantSelector) + if err != nil { + log.Error(err, "cannot create MatchingLabelsSelector for Global filtering") + + return reconcile.Result{}, err + } + + tntList := capsulev1beta2.TenantList{} + if err = r.client.List(ctx, &tntList, &client.MatchingLabelsSelector{Selector: tntSelector}); err != nil { + log.Error(err, "cannot list Tenants matching the provided selector") + + return reconcile.Result{}, err + } + // This is the list of newer Tenants that are matching the provided GlobalTenantResource Selector: + // upon replication and pruning, this will be updated in the status of the resource. + tntSet := sets.NewString() + + err = new(multierror.Error) + // A TenantResource is made of several Resource sections, each one with specific options: + // the Status can be updated only in case of no errors across all of them to guarantee a valid and coherent status. + processedItems := sets.NewString() + + for index, resource := range tntResource.Spec.Resources { + tenantLabel, labelErr := capsulev1beta2.GetTypeLabel(&capsulev1beta2.Tenant{}) + if labelErr != nil { + log.Error(labelErr, "expected label for selection") + + return reconcile.Result{}, labelErr + } + + for _, tnt := range tntList.Items { + tntSet.Insert(tnt.GetName()) + + items, sectionErr := r.processor.HandleSection(ctx, tnt, true, tenantLabel, index, resource) + if sectionErr != nil { + // Upon a process error storing the last error occurred and continuing to iterate, + // avoid to block the whole processing. + err = multierror.Append(err, sectionErr) + } else { + processedItems.Insert(items...) + } + } + } + + if err.(*multierror.Error).ErrorOrNil() != nil { //nolint:errorlint,forcetypeassert + log.Error(err, "unable to replicate the requested resources") + + return reconcile.Result{}, err + } + + shouldUpdateStatus := !sets.NewString(tntResource.Status.SelectedTenants...).Equal(tntSet) + + if r.processor.HandlePruning(ctx, tntResource.Status.ProcessedItems, processedItems) { + tntResource.Status.ProcessedItems = make([]capsulev1beta2.ObjectReferenceStatus, 0, len(processedItems)) + + for _, item := range processedItems.List() { + if or := (capsulev1beta2.ObjectReferenceStatus{}); or.ParseFromString(item) == nil { + tntResource.Status.ProcessedItems = append(tntResource.Status.ProcessedItems, or) + } + } + + shouldUpdateStatus = true + } + + if shouldUpdateStatus { + tntResource.Status.SelectedTenants = tntSet.List() + + if updateErr := r.client.Status().Update(ctx, &tntResource); updateErr != nil { + log.Error(updateErr, "unable to update TenantResource status") + } + } + + log.Info("processing completed") + + return reconcile.Result{Requeue: true, RequeueAfter: tntResource.Spec.ResyncPeriod.Duration}, nil +} diff --git a/controllers/resources/namespaced.go b/controllers/resources/namespaced.go new file mode 100644 index 00000000..891d3a3d --- /dev/null +++ b/controllers/resources/namespaced.go @@ -0,0 +1,144 @@ +// Copyright 2020-2021 Clastix Labs +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + + "github.com/hashicorp/go-multierror" + apierr "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/client-go/util/retry" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + capsulev1beta2 "github.com/clastix/capsule/api/v1beta2" +) + +type Namespaced struct { + client client.Client + finalizer Processor +} + +func (r *Namespaced) SetupWithManager(mgr ctrl.Manager) error { + unstructuredCachingClient, err := client.NewDelegatingClient( + client.NewDelegatingClientInput{ + Client: mgr.GetClient(), + CacheReader: mgr.GetCache(), + CacheUnstructured: true, + }, + ) + if err != nil { + return err + } + + r.client = mgr.GetClient() + r.finalizer = Processor{ + client: r.client, + unstructuredClient: unstructuredCachingClient, + } + + return ctrl.NewControllerManagedBy(mgr). + For(&capsulev1beta2.TenantResource{}). + Complete(r) +} + +func (r *Namespaced) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) { + log := ctrllog.FromContext(ctx) + + log.Info("start processing") + // Retrieving the TenantResource + tntResource := capsulev1beta2.TenantResource{} + if err := r.client.Get(ctx, request.NamespacedName, &tntResource); err != nil { + if apierr.IsNotFound(err) { + log.Info("Request object not found, could have been deleted after reconcile request") + + return reconcile.Result{}, nil + } + + log.Error(err, "cannot retrieve capsulev1beta2.TenantResource") + + return reconcile.Result{}, err + } + // Adding the default value for the status + if tntResource.Status.ProcessedItems == nil { + tntResource.Status.ProcessedItems = make([]capsulev1beta2.ObjectReferenceStatus, 0, 0) + } + // Handling the finalizer section for the given TenantResource + enqueueBack, err := r.finalizer.HandleFinalizer(ctx, &tntResource, *tntResource.Spec.PruningOnDelete, tntResource.Status.ProcessedItems) + if err != nil || enqueueBack { + return reconcile.Result{}, err + } + // Retrieving the parent of the Global Resource: + // can be owned, or being deployed in one of its Namespace. + tl := &capsulev1beta2.TenantList{} + if err = r.client.List(ctx, tl, client.MatchingFieldsSelector{Selector: fields.OneTermEqualSelector(".status.namespaces", tntResource.GetNamespace())}); err != nil { + log.Error(err, "unable to detect the Global for the given TenantResource") + + return reconcile.Result{}, err + } + + if len(tl.Items) == 0 { + log.Info("skipping sync, the current Namespace is not belonging to any Global") + + return reconcile.Result{}, nil + } + + err = new(multierror.Error) + // A TenantResource is made of several Resource sections, each one with specific options: + // the Status can be updated only in case of no errors across all of them to guarantee a valid and coherent status. + processedItems := sets.NewString() + + tenantLabel, labelErr := capsulev1beta2.GetTypeLabel(&capsulev1beta2.Tenant{}) + if labelErr != nil { + log.Error(labelErr, "expected label for selection") + + return reconcile.Result{}, labelErr + } + + for index, resource := range tntResource.Spec.Resources { + items, sectionErr := r.finalizer.HandleSection(ctx, tl.Items[0], false, tenantLabel, index, resource) + if sectionErr != nil { + // Upon a process error storing the last error occurred and continuing to iterate, + // avoid to block the whole processing. + err = multierror.Append(err, sectionErr) + } else { + processedItems.Insert(items...) + } + } + + if err.(*multierror.Error).ErrorOrNil() != nil { //nolint:errorlint,forcetypeassert + log.Error(err, "unable to replicate the requested resources") + + return reconcile.Result{}, err + } + + if r.finalizer.HandlePruning(ctx, tntResource.Status.ProcessedItems, processedItems) { + statusErr := retry.RetryOnConflict(retry.DefaultRetry, func() (err error) { + if err = r.client.Get(ctx, request.NamespacedName, &tntResource); err != nil { + return err + } + + tntResource.Status.ProcessedItems = make([]capsulev1beta2.ObjectReferenceStatus, 0, len(processedItems)) + + for _, item := range processedItems.List() { + if or := (capsulev1beta2.ObjectReferenceStatus{}); or.ParseFromString(item) == nil { + tntResource.Status.ProcessedItems = append(tntResource.Status.ProcessedItems, or) + } + } + + return r.client.Status().Update(ctx, &tntResource) + }) + if statusErr != nil { + log.Error(statusErr, "unable to update TenantResource status") + } + } + + log.Info("processing completed") + + return reconcile.Result{Requeue: true, RequeueAfter: tntResource.Spec.ResyncPeriod.Duration}, nil +} diff --git a/controllers/resources/processor.go b/controllers/resources/processor.go new file mode 100644 index 00000000..ed71c665 --- /dev/null +++ b/controllers/resources/processor.go @@ -0,0 +1,17 @@ +// Copyright 2020-2021 Clastix Labs +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + finalizer = "capsule.clastix.io/resources" +) + +type Processor struct { + client client.Client + unstructuredClient client.Client +} diff --git a/controllers/resources/processor_finalizer.go b/controllers/resources/processor_finalizer.go new file mode 100644 index 00000000..edc666bb --- /dev/null +++ b/controllers/resources/processor_finalizer.go @@ -0,0 +1,54 @@ +// Copyright 2020-2021 Clastix Labs +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + + "k8s.io/apimachinery/pkg/util/sets" + "sigs.k8s.io/controller-runtime/pkg/client" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" + + capsulev1beta2 "github.com/clastix/capsule/api/v1beta2" +) + +func (r *Processor) HandleFinalizer(ctx context.Context, obj client.Object, shouldPrune bool, items []capsulev1beta2.ObjectReferenceStatus) (enqueueBack bool, err error) { + log := ctrllog.FromContext(ctx) + // If the object has been marked for deletion, + // we have to clean up the created resources before removing the finalizer. + if obj.GetDeletionTimestamp() != nil { + log.Info("pruning prior finalizer removal") + + if shouldPrune { + _ = r.HandlePruning(ctx, items, nil) + } + + obj.SetFinalizers(nil) + + if err = r.client.Update(ctx, obj); err != nil { + log.Error(err, "cannot remove finalizer") + + return true, err + } + + return true, nil + } + // When the pruning for the given resource is enabled, a finalizer is required when the TenantResource is marked + // for deletion: this allows to perform a clean-up of all the underlying resources. + if shouldPrune && !sets.NewString(obj.GetFinalizers()...).Has(finalizer) { + obj.SetFinalizers(append(obj.GetFinalizers(), finalizer)) + + if err = r.client.Update(ctx, obj); err != nil { + log.Error(err, "cannot add finalizer") + + return true, err + } + + log.Info("added finalizer, enqueuing back for processing") + + return true, nil + } + + return false, nil +} diff --git a/controllers/resources/processor_pruning.go b/controllers/resources/processor_pruning.go new file mode 100644 index 00000000..a6a37c83 --- /dev/null +++ b/controllers/resources/processor_pruning.go @@ -0,0 +1,67 @@ +// Copyright 2020-2021 Clastix Labs +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + + apierr "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/sets" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" + + capsulev1beta2 "github.com/clastix/capsule/api/v1beta2" +) + +func (r *Processor) HandlePruning(ctx context.Context, current []capsulev1beta2.ObjectReferenceStatus, desired sets.String) (updateStatus bool) { + log := ctrllog.FromContext(ctx) + // The status items are the actual replicated resources, these must be collected in order to perform the resulting + // diff that will be cleaned-up. + status := sets.NewString() + + for _, item := range current { + status.Insert(item.String()) + } + + diff := status.Difference(desired) + // We don't want to trigger a reconciliation of the Status every time, + // rather, only in case of a difference between the processed and the actual status. + // This can happen upon the first reconciliation, or a removal, or a change, of a resource. + updateStatus = diff.Len() > 0 || status.Len() != desired.Len() + + if diff.Len() > 0 { + log.Info("starting processing pruning", "length", diff.Len()) + } + + // The outer resources must be removed, iterating over these to clean-up + for item := range diff { + or := capsulev1beta2.ObjectReferenceStatus{} + if err := or.ParseFromString(item); err != nil { + log.Error(err, "unable to parse resource to prune", "resource", item) + + continue + } + + obj := unstructured.Unstructured{} + obj.SetNamespace(or.Namespace) + obj.SetName(or.Name) + obj.SetGroupVersionKind(schema.FromAPIVersionAndKind(or.APIVersion, or.Kind)) + + if err := r.unstructuredClient.Delete(ctx, &obj); err != nil { + if apierr.IsNotFound(err) { + // Object may have been already deleted, we can ignore this error + continue + } + + log.Error(err, "unable to prune resource", "resource", item) + + continue + } + + log.Info("resource has been pruned", "resource", item) + } + + return updateStatus +} diff --git a/controllers/resources/processor_section.go b/controllers/resources/processor_section.go new file mode 100644 index 00000000..77c1b662 --- /dev/null +++ b/controllers/resources/processor_section.go @@ -0,0 +1,223 @@ +// Copyright 2020-2021 Clastix Labs +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + "fmt" + + "github.com/hashicorp/go-multierror" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/apimachinery/pkg/selection" + "k8s.io/apimachinery/pkg/util/sets" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" + + capsulev1beta2 "github.com/clastix/capsule/api/v1beta2" +) + +func (r *Processor) HandleSection(ctx context.Context, tnt capsulev1beta2.Tenant, allowCrossNamespaceSelection bool, tenantLabel string, resourceIndex int, spec capsulev1beta2.ResourceSpec) ([]string, error) { + log := ctrllog.FromContext(ctx) + + var err error + // Creating Namespace selector + var selector labels.Selector + + if spec.NamespaceSelector != nil { + selector, err = metav1.LabelSelectorAsSelector(spec.NamespaceSelector) + if err != nil { + log.Error(err, "cannot create Namespace selector for Namespace filtering and resource replication", "index", resourceIndex) + + return nil, err + } + } else { + selector = labels.NewSelector() + } + // Resources can be replicated only on Namespaces belonging to the same Global: + // preventing a boundary cross by enforcing the selection. + tntRequirement, err := labels.NewRequirement(tenantLabel, selection.Equals, []string{tnt.GetName()}) + if err != nil { + log.Error(err, "unable to create requirement for Namespace filtering and resource replication", "index", resourceIndex) + + return nil, err + } + + selector = selector.Add(*tntRequirement) + // Selecting the targeted Namespace according to the TenantResource specification. + namespaces := corev1.NamespaceList{} + if err = r.client.List(ctx, &namespaces, client.MatchingLabelsSelector{Selector: selector}); err != nil { + log.Error(err, "cannot retrieve Namespaces for resource", "index", resourceIndex) + + return nil, err + } + // Generating additional metadata + objAnnotations, objLabels := map[string]string{}, map[string]string{} + + if spec.AdditionalMetadata != nil { + objAnnotations = spec.AdditionalMetadata.Annotations + objLabels = spec.AdditionalMetadata.Labels + } + + objAnnotations[tenantLabel] = tnt.GetName() + + objLabels["capsule.clastix.io/resources"] = fmt.Sprintf("%d", resourceIndex) + objLabels[tenantLabel] = tnt.GetName() + // processed will contain the sets of resources replicated, both for the raw and the Namespaced ones: + // these are required to perform a final pruning once the replication has been occurred. + processed := sets.NewString() + + tntNamespaces := sets.NewString(tnt.Status.Namespaces...) + + syncErr := new(multierror.Error) + + for nsIndex, item := range spec.NamespacedItems { + keysAndValues := []interface{}{"index", nsIndex, "namespace", item.Namespace} + // A TenantResource is created by a TenantOwner, and potentially, they could point to a resource in a non-owned + // Namespace: this must be blocked by checking it this is the case. + if !allowCrossNamespaceSelection && !tntNamespaces.Has(item.Namespace) { + log.Info("skipping processing of namespacedItem, referring a Namespace that is not part of the given Global", keysAndValues...) + + continue + } + // Namespaced Items are relying on selecting resources, rather than specifying a specific name: + // creating it to get used by the client List action. + itemSelector, selectorErr := metav1.LabelSelectorAsSelector(&item.Selector) + if err != nil { + log.Error(selectorErr, "cannot create Selector for namespacedItem", keysAndValues...) + + continue + } + + objs := unstructured.UnstructuredList{} + objs.SetGroupVersionKind(schema.FromAPIVersionAndKind(item.APIVersion, fmt.Sprintf("%sList", item.Kind))) + + if clientErr := r.unstructuredClient.List(ctx, &objs, client.InNamespace(item.Namespace), client.MatchingLabelsSelector{Selector: itemSelector}); clientErr != nil { + log.Error(clientErr, "cannot retrieve object for namespacedItem", keysAndValues...) + + syncErr = multierror.Append(syncErr, clientErr) + + continue + } + + multiErr := new(multierror.Group) + // Iterating over all the retrieved objects from the resource spec to get replicated in all the selected Namespaces: + // in case of error during the create or update function, this will be appended to the list of errors. + for _, o := range objs.Items { + obj := o + + multiErr.Go(func() error { + nsItems, nsErr := r.createOrUpdate(ctx, &obj, objLabels, objAnnotations, namespaces) + if nsErr != nil { + log.Error(err, "unable to sync namespacedItems", keysAndValues...) + + return nsErr + } + + processed.Insert(nsItems...) + + return nil + }) + } + + if objsErr := multiErr.Wait(); objsErr != nil { + syncErr = multierror.Append(syncErr, objsErr) + } + } + + codecFactory := serializer.NewCodecFactory(r.client.Scheme()) + + for rawIndex, item := range spec.RawItems { + obj, keysAndValues := unstructured.Unstructured{}, []interface{}{"index", rawIndex} + + if _, _, decodeErr := codecFactory.UniversalDeserializer().Decode(item.Raw, nil, &obj); decodeErr != nil { + log.Error(decodeErr, "unable to deserialize rawItem", keysAndValues...) + + syncErr = multierror.Append(syncErr, decodeErr) + + continue + } + + syncedRaw, rawErr := r.createOrUpdate(ctx, &obj, objLabels, objAnnotations, namespaces) + if rawErr != nil { + log.Info("unable to sync rawItem", keysAndValues...) + // In case of error processing an item in one of any selected Namespaces, storing it to report it lately + // to the upper call to ensure a partial sync that will be fixed by a subsequent reconciliation. + syncErr = multierror.Append(syncErr, rawErr) + } else { + processed.Insert(syncedRaw...) + } + } + + return processed.List(), syncErr.ErrorOrNil() +} + +// createOrUpdate replicates the provided unstructured object to all the provided Namespaces: +// this function mimics the CreateOrUpdate, by retrieving the object to understand if it must be created or updated, +// along adding the additional metadata, if required. +func (r *Processor) createOrUpdate(ctx context.Context, obj *unstructured.Unstructured, labels map[string]string, annotations map[string]string, namespaces corev1.NamespaceList) ([]string, error) { + log := ctrllog.FromContext(ctx) + + errGroup := new(multierror.Group) + + var items []string + + for _, item := range namespaces.Items { + ns := item.GetName() + + errGroup.Go(func() (err error) { + actual, desired := obj.DeepCopy(), obj.DeepCopy() + // Using a deferred function to properly log the results, and adding the item to the processed set. + defer func() { + keysAndValues := []interface{}{"resource", fmt.Sprintf("%s/%s", ns, desired.GetName())} + + if err != nil { + log.Error(err, "unable to replicate resource", keysAndValues...) + + return + } + + log.Info("resource has been replicated", keysAndValues...) + + replicatedItem := &capsulev1beta2.ObjectReferenceStatus{ + Name: obj.GetName(), + } + replicatedItem.Kind = obj.GetKind() + replicatedItem.Namespace = ns + replicatedItem.APIVersion = obj.GetAPIVersion() + + items = append(items, replicatedItem.String()) + }() + + actual.SetNamespace(ns) + + _, err = controllerutil.CreateOrUpdate(ctx, r.unstructuredClient, actual, func() error { + UID := actual.GetUID() + + actual.SetUnstructuredContent(desired.Object) + actual.SetNamespace(ns) + actual.SetLabels(labels) + actual.SetAnnotations(annotations) + actual.SetResourceVersion("") + actual.SetUID(UID) + + return nil + }) + + return + }) + } + // Wait returns *multierror.Error that implements stdlib error: + // the nil check must be performed down here rather than at the caller level to avoid wrong casting. + if err := errGroup.Wait(); err != nil { + return items, err + } + + return items, nil +} diff --git a/go.sum b/go.sum index 06ade127..1972247b 100644 --- a/go.sum +++ b/go.sum @@ -694,7 +694,6 @@ golang.org/x/sys v0.0.0-20210426230700-d19ff857e887/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I= golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a h1:dGzPydgVsqGcTRVwiLJ1jVbufYwmzD3LfVPLKsKg+0k= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/main.go b/main.go index 7cb64a83..ccf7029a 100644 --- a/main.go +++ b/main.go @@ -29,6 +29,7 @@ import ( capsulev1beta2 "github.com/clastix/capsule/api/v1beta2" configcontroller "github.com/clastix/capsule/controllers/config" rbaccontroller "github.com/clastix/capsule/controllers/rbac" + "github.com/clastix/capsule/controllers/resources" servicelabelscontroller "github.com/clastix/capsule/controllers/servicelabels" tenantcontroller "github.com/clastix/capsule/controllers/tenant" tlscontroller "github.com/clastix/capsule/controllers/tls" @@ -266,6 +267,16 @@ func main() { os.Exit(1) } + if err = (&resources.Global{}).SetupWithManager(manager); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "resources.Global") + os.Exit(1) + } + + if err = (&resources.Namespaced{}).SetupWithManager(manager); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "resources.Namespaced") + os.Exit(1) + } + setupLog.Info("starting manager") if err = manager.Start(ctx); err != nil { diff --git a/pkg/indexer/indexer.go b/pkg/indexer/indexer.go index 2e3e88f2..a80c7497 100644 --- a/pkg/indexer/indexer.go +++ b/pkg/indexer/indexer.go @@ -16,6 +16,8 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/manager" + capsulev1beta1 "github.com/clastix/capsule/api/v1beta1" + capsulev1beta2 "github.com/clastix/capsule/api/v1beta2" "github.com/clastix/capsule/pkg/indexer/ingress" "github.com/clastix/capsule/pkg/indexer/namespace" "github.com/clastix/capsule/pkg/indexer/tenant" @@ -29,7 +31,8 @@ type CustomIndexer interface { func AddToManager(ctx context.Context, log logr.Logger, mgr manager.Manager) error { indexers := []CustomIndexer{ - tenant.NamespacesReference{}, + tenant.NamespacesReference{Obj: &capsulev1beta1.Tenant{}}, + tenant.NamespacesReference{Obj: &capsulev1beta2.Tenant{}}, tenant.OwnerReference{}, namespace.OwnerReference{}, ingress.HostnamePath{Obj: &extensionsv1beta1.Ingress{}}, diff --git a/pkg/indexer/tenant/namespaces.go b/pkg/indexer/tenant/namespaces.go index f292d07b..d4388f92 100644 --- a/pkg/indexer/tenant/namespaces.go +++ b/pkg/indexer/tenant/namespaces.go @@ -6,13 +6,15 @@ package tenant import ( "sigs.k8s.io/controller-runtime/pkg/client" - capsulev1beta1 "github.com/clastix/capsule/api/v1beta1" + "github.com/clastix/capsule/pkg/api" ) -type NamespacesReference struct{} +type NamespacesReference struct { + Obj client.Object +} func (o NamespacesReference) Object() client.Object { - return &capsulev1beta1.Tenant{} + return o.Obj } func (o NamespacesReference) Field() string { @@ -22,12 +24,6 @@ func (o NamespacesReference) Field() string { // nolint:forcetypeassert func (o NamespacesReference) Func() client.IndexerFunc { return func(object client.Object) []string { - namespaces := object.(*capsulev1beta1.Tenant).DeepCopy().Status.Namespaces - - if namespaces == nil { - return []string{} - } - - return namespaces + return object.(api.Tenant).GetNamespaces() } }