fix: concurrent map writes when replicating namespaced objects (#1264)

Signed-off-by: Dario Tranchitella <dario@tranchitella.eu>
This commit is contained in:
Dario Tranchitella
2024-12-04 12:41:29 +01:00
committed by GitHub
parent da66f40462
commit 8d498bb925
+17 -6
View File
@@ -79,6 +79,7 @@ func (r *Processor) HandlePruning(ctx context.Context, current, desired sets.Set
return updateStatus
}
//nolint:gocognit
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)
@@ -125,7 +126,6 @@ func (r *Processor) HandleSection(ctx context.Context, tnt capsulev1beta2.Tenant
objLabels[Label] = 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()
@@ -173,17 +173,20 @@ func (r *Processor) HandleSection(ctx context.Context, tnt capsulev1beta2.Tenant
var wg sync.WaitGroup
errorsChan := make(chan error, len(objs.Items))
// processedRaw is used to avoid concurrent map writes during iteration of namespaced items:
// the objects will be then added to processed variable if the resulting string is not empty,
// meaning it has been processed correctly.
processedRaw := make([]string, len(objs.Items))
// 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 {
for i, o := range objs.Items {
obj := o
obj.SetNamespace(ns.Name)
obj.SetOwnerReferences(nil)
wg.Add(1)
go func(obj unstructured.Unstructured) {
go func(index int, obj unstructured.Unstructured) {
defer wg.Done()
kv := keysAndValues
@@ -204,8 +207,8 @@ func (r *Processor) HandleSection(ctx context.Context, tnt capsulev1beta2.Tenant
replicatedItem.Namespace = ns.Name
replicatedItem.APIVersion = obj.GetAPIVersion()
processed.Insert(replicatedItem.String())
}(obj)
processedRaw[index] = replicatedItem.String()
}(i, obj)
}
wg.Wait()
@@ -216,6 +219,14 @@ func (r *Processor) HandleSection(ctx context.Context, tnt capsulev1beta2.Tenant
syncErr = errors.Join(syncErr, err)
}
}
for _, p := range processedRaw {
if p == "" {
continue
}
processed.Insert(p)
}
}
for rawIndex, item := range spec.RawItems {