fix(kuma-cp): avoid concurrent access on resource meta (#11997)

There were some places where labels on meta weren't not cloned when
duplicating the meta. Walked the code for odd usages of `GetLabels()`.

See also #11886

Signed-off-by: Charly Molter <[email protected]>
Signed-off-by: Mike Beaumont <[email protected]>
Co-authored-by: Mike Beaumont <[email protected]>

pkg/kds/util/meta.go

@@ -28,13 +28,77 @@ func CloneResourceMetaWithNewName(meta model.ResourceMeta, name string) model.Re
 	}
 }
 
+<<<<<<< HEAD
 func kumaResourceMetaToResourceMeta(meta *mesh_proto.KumaResource_Meta) model.ResourceMeta {
 	return &resourceMeta{
 		name: meta.Name,
 		mesh: meta.Mesh,
 	}
 }
 
+=======
+// PopulateNamespaceLabelFromNameExtension on Kubernetes zones adds 'k8s.kuma.io/namespace' label to the resources
+// before syncing them to Global.
+//
+// In 2.7.x method 'GetMeta().GetLabels()' on Kubernetes returned a label map with 'k8s.kuma.io/namespace' added
+// dynamically. This behavior was changed in 2.9.x by https://github.com/kumahq/kuma/pull/11020, the namespace label is now
+// supposed to be set in ComputeLabels function. But this functions is called only on Create/Update of the resources.
+// This means policies that were created on 2.7.x won't have 'k8s.kuma.io/namespace' label when synced to Global.
+// Even though the lack of namespace labels affects only how resource looks in GUI on Global it's still worth setting it.
+func PopulateNamespaceLabelFromNameExtension() CloneResourceMetaOpt {
+	return func(m *resourceMeta) {
+		namespace := m.nameExtensions[model.K8sNamespaceComponent]
+		if _, ok := m.labels[mesh_proto.KubeNamespaceTag]; !ok && namespace != "" {
+			m.labels[mesh_proto.KubeNamespaceTag] = namespace
+		}
+	}
+}
+
+func WithoutLabel(key string) CloneResourceMetaOpt {
+	return func(m *resourceMeta) {
+		delete(m.labels, key)
+	}
+}
+
+func If(condition func(resource model.ResourceMeta) bool, fn CloneResourceMetaOpt) CloneResourceMetaOpt {
+	return func(meta *resourceMeta) {
+		if condition(meta) {
+			fn(meta)
+		}
+	}
+}
+
+func IsKubernetes(storeType config_store.StoreType) func(model.ResourceMeta) bool {
+	return func(_ model.ResourceMeta) bool {
+		return storeType == config_store.KubernetesStore
+	}
+}
+
+func CloneResourceMeta(m model.ResourceMeta, fs ...CloneResourceMetaOpt) model.ResourceMeta {
+	labels := maps.Clone(m.GetLabels())
+	if labels == nil {
+		labels = map[string]string{}
+	}
+	ne := maps.Clone(m.GetNameExtensions())
+	if ne == nil {
+		ne = model.ResourceNameExtensions{}
+	}
+	meta := &resourceMeta{
+		name:           m.GetName(),
+		mesh:           m.GetMesh(),
+		labels:         labels,
+		nameExtensions: ne,
+	}
+	for _, f := range fs {
+		f(meta)
+	}
+	if len(meta.labels) == 0 {
+		meta.labels = nil
+	}
+	return meta
+}
+
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 func (r *resourceMeta) GetName() string {
 	return r.name
 }

pkg/kds/util/resource.go

@@ -2,6 +2,7 @@ package util
 
 import (
 	"fmt"
+	"maps"
 	"strings"
 
 	envoy_sd "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
@@ -53,6 +54,10 @@ func ToEnvoyResources(rlist model.ResourceList) ([]envoy_types.Resource, error)
 			Meta: &mesh_proto.KumaResource_Meta{
 				Name:    r.GetMeta().GetName(),
 				Mesh:    r.GetMeta().GetMesh(),
+<<<<<<< HEAD
+=======
+				Labels:  maps.Clone(r.GetMeta().GetLabels()),
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 				Version: "",
 			},
 			Spec: pbany,
@@ -140,7 +145,20 @@ func toResources(resourceType model.ResourceType, krs []*mesh_proto.KumaResource
 		if err = model.FromAny(kr.Spec, obj.GetSpec()); err != nil {
 			return nil, err
 		}
+<<<<<<< HEAD
 		obj.SetMeta(kumaResourceMetaToResourceMeta(kr.Meta))
+=======
+		if obj.Descriptor().HasStatus && kr.Status != nil {
+			if err = model.FromAny(kr.Status, obj.GetStatus()); err != nil {
+				return nil, err
+			}
+		}
+		obj.SetMeta(&resourceMeta{
+			name:   kr.GetMeta().GetName(),
+			mesh:   kr.GetMeta().GetMesh(),
+			labels: maps.Clone(kr.GetMeta().GetLabels()),
+		})
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 		if err := list.AddItem(obj); err != nil {
 			return nil, err
 		}

pkg/plugins/resources/memory/store.go

@@ -3,6 +3,7 @@ package memory
 import (
 	"context"
 	"fmt"
+	"maps"
 	"strconv"
 	"strings"
 	"sync"
@@ -112,6 +113,10 @@ func (c *memoryStore) Create(_ context.Context, r core_model.Resource, fs ...sto
 		Version:          initialVersion(),
 		CreationTime:     opts.CreationTime,
 		ModificationTime: opts.CreationTime,
+<<<<<<< HEAD
+=======
+		Labels:           maps.Clone(opts.Labels),
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 	}
 
 	// fill the meta
@@ -167,6 +172,12 @@ func (c *memoryStore) Update(_ context.Context, r core_model.Resource, fs ...sto
 	}
 	meta.Version = meta.Version.Next()
 	meta.ModificationTime = opts.ModificationTime
+<<<<<<< HEAD
+=======
+	if opts.ModifyLabels {
+		meta.Labels = maps.Clone(opts.Labels)
+	}
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 	r.SetMeta(meta)
 
 	record.Version = meta.Version

pkg/plugins/resources/remote/store.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"io"
+	"maps"
 	"net/http"
 	"strconv"
 
@@ -38,9 +39,16 @@ type remoteStore struct {
 func (s *remoteStore) Create(ctx context.Context, res model.Resource, fs ...store.CreateOptionsFunc) error {
 	opts := store.NewCreateOptions(fs...)
 	meta := rest_v1alpha1.ResourceMeta{
+<<<<<<< HEAD
 		Type: string(res.Descriptor().Name),
 		Name: opts.Name,
 		Mesh: opts.Mesh,
+=======
+		Type:   string(res.Descriptor().Name),
+		Name:   opts.Name,
+		Mesh:   opts.Mesh,
+		Labels: maps.Clone(opts.Labels),
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 	}
 	if err := s.upsert(ctx, res, meta); err != nil {
 		return err
@@ -50,9 +58,16 @@ func (s *remoteStore) Create(ctx context.Context, res model.Resource, fs ...stor
 
 func (s *remoteStore) Update(ctx context.Context, res model.Resource, fs ...store.UpdateOptionsFunc) error {
 	meta := rest_v1alpha1.ResourceMeta{
+<<<<<<< HEAD
 		Type: string(res.Descriptor().Name),
 		Name: res.GetMeta().GetName(),
 		Mesh: res.GetMeta().GetMesh(),
+=======
+		Type:   string(res.Descriptor().Name),
+		Name:   res.GetMeta().GetName(),
+		Mesh:   res.GetMeta().GetMesh(),
+		Labels: maps.Clone(opts.Labels),
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 	}
 	if err := s.upsert(ctx, res, meta); err != nil {
 		return err

pkg/plugins/secrets/k8s/store.go

@@ -105,6 +105,23 @@ func (s *KubernetesStore) Update(ctx context.Context, r core_model.Resource, fs
 	return nil
 }
 
+<<<<<<< HEAD
+=======
+func setLabelsAnnotationsAndMesh(s *kube_core.Secret, mesh string, labels map[string]string) {
+	labels = maps.Clone(labels)
+	if labels == nil {
+		labels = map[string]string{}
+	}
+	if mesh != "" {
+		labels[metadata.KumaMeshLabel] = mesh
+	}
+
+	labels, annotations := k8s.SplitLabelsAndAnnotations(labels, s.GetAnnotations())
+	s.GetObjectMeta().SetLabels(labels)
+	s.GetObjectMeta().SetAnnotations(annotations)
+}
+
+>>>>>>> c3d7187c7 (fix(kuma-cp): avoid concurrent access on resource meta (#11997))
 func (s *KubernetesStore) Delete(ctx context.Context, r core_model.Resource, fs ...core_store.DeleteOptionsFunc) error {
 	opts := core_store.NewDeleteOptions(fs...)
 	if err := s.Get(ctx, r, core_store.GetByKey(opts.Name, opts.Mesh)); err != nil {