mirror of
https://github.com/projectcapsule/capsule.git
synced 2026-08-25 16:07:24 +00:00
refactor: optimizing processing of tenant resources per namespace
This commit is contained in:
Dario Tranchitella
+109
-128
@@ -130,81 +130,105 @@ func (r *Processor) HandleSection(ctx context.Context, tnt capsulev1beta2.Tenant
|
||||
|
||||
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.client.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}
|
||||
for _, ns := range namespaces.Items {
|
||||
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 Tenant", keysAndValues...)
|
||||
|
||||
if _, _, decodeErr := codecFactory.UniversalDeserializer().Decode(item.Raw, nil, &obj); decodeErr != nil {
|
||||
log.Error(decodeErr, "unable to deserialize rawItem", 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 selectorErr != nil {
|
||||
log.Error(selectorErr, "cannot create Selector for namespacedItem", keysAndValues...)
|
||||
|
||||
syncErr = multierror.Append(syncErr, decodeErr)
|
||||
syncErr = multierror.Append(syncErr, selectorErr)
|
||||
|
||||
continue
|
||||
continue
|
||||
}
|
||||
|
||||
objs := unstructured.UnstructuredList{}
|
||||
objs.SetGroupVersionKind(schema.FromAPIVersionAndKind(item.APIVersion, fmt.Sprintf("%sList", item.Kind)))
|
||||
|
||||
if clientErr := r.client.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
|
||||
obj.SetNamespace(ns.Name)
|
||||
|
||||
multiErr.Go(func() error {
|
||||
kv := keysAndValues
|
||||
kv = append(kv, []interface{}{"resource", fmt.Sprintf("%s/%s", obj.GetNamespace(), obj.GetNamespace())})
|
||||
|
||||
if opErr := r.createOrUpdate(ctx, &obj, objLabels, objAnnotations); opErr != nil {
|
||||
log.Error(opErr, "unable to sync namespacedItems", kv...)
|
||||
|
||||
return opErr
|
||||
}
|
||||
|
||||
log.Info("resource has been replicated", kv...)
|
||||
|
||||
replicatedItem := &capsulev1beta2.ObjectReferenceStatus{}
|
||||
replicatedItem.Name = obj.GetName()
|
||||
replicatedItem.Kind = obj.GetKind()
|
||||
replicatedItem.Namespace = ns.Name
|
||||
replicatedItem.APIVersion = obj.GetAPIVersion()
|
||||
|
||||
processed.Insert(replicatedItem.String())
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
if objsErr := multiErr.Wait(); objsErr != nil {
|
||||
syncErr = multierror.Append(syncErr, objsErr)
|
||||
}
|
||||
}
|
||||
|
||||
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...)
|
||||
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
|
||||
}
|
||||
|
||||
obj.SetNamespace(ns.Name)
|
||||
|
||||
if rawErr := r.createOrUpdate(ctx, &obj, objLabels, objAnnotations); 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 {
|
||||
log.Info("resource has been replicated", keysAndValues...)
|
||||
|
||||
replicatedItem := &capsulev1beta2.ObjectReferenceStatus{}
|
||||
replicatedItem.Name = obj.GetName()
|
||||
replicatedItem.Kind = obj.GetKind()
|
||||
replicatedItem.Namespace = ns.Name
|
||||
replicatedItem.APIVersion = obj.GetAPIVersion()
|
||||
|
||||
processed.Insert(replicatedItem.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -214,69 +238,26 @@ func (r *Processor) HandleSection(ctx context.Context, tnt capsulev1beta2.Tenant
|
||||
// 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)
|
||||
func (r *Processor) createOrUpdate(ctx context.Context, obj *unstructured.Unstructured, labels map[string]string, annotations map[string]string) (err error) {
|
||||
actual, desired := &unstructured.Unstructured{}, obj.DeepCopy()
|
||||
|
||||
errGroup := new(multierror.Group)
|
||||
actual.SetAPIVersion(desired.GetAPIVersion())
|
||||
actual.SetKind(desired.GetKind())
|
||||
actual.SetNamespace(desired.GetNamespace())
|
||||
actual.SetName(desired.GetName())
|
||||
|
||||
var items []string
|
||||
_, err = controllerutil.CreateOrUpdate(ctx, r.client, actual, func() error {
|
||||
UID := actual.GetUID()
|
||||
rv := actual.GetResourceVersion()
|
||||
|
||||
for _, item := range namespaces.Items {
|
||||
ns := item.GetName()
|
||||
actual.SetUnstructuredContent(desired.Object)
|
||||
actual.SetLabels(labels)
|
||||
actual.SetAnnotations(annotations)
|
||||
actual.SetResourceVersion(rv)
|
||||
actual.SetUID(UID)
|
||||
|
||||
errGroup.Go(func() (err error) {
|
||||
actual, desired := &unstructured.Unstructured{}, obj.DeepCopy()
|
||||
return nil
|
||||
})
|
||||
|
||||
actual.SetAPIVersion(desired.GetAPIVersion())
|
||||
actual.SetKind(desired.GetKind())
|
||||
actual.SetNamespace(desired.GetNamespace())
|
||||
actual.SetName(desired.GetName())
|
||||
// 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.client, actual, func() error {
|
||||
UID := actual.GetUID()
|
||||
rv := actual.GetResourceVersion()
|
||||
|
||||
actual.SetUnstructuredContent(desired.Object)
|
||||
actual.SetNamespace(ns)
|
||||
actual.SetLabels(labels)
|
||||
actual.SetAnnotations(annotations)
|
||||
actual.SetResourceVersion(rv)
|
||||
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
|
||||
return err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user