fix(dpp): save and update labels from Dataplane resource on Universal (#12975)

## Motivation

We need to handle dpp labels properly in order to be able to select them
in policies

<!--
> Changelog: skip
-->
<!--
Uncomment the above section to explicitly set a [`> Changelog:` entry
here](https://github.com/kumahq/kuma/blob/master/CONTRIBUTING.md#submitting-a-patch)?
-->

---------

Signed-off-by: Marcin Skalski <[email protected]>

Author: Automaat

pkg/core/managers/apis/dataplane/dataplane_manager.go

@@ -94,10 +94,12 @@ func (m *dataplaneManager) Update(ctx context.Context, resource core_model.Resou
 
 	m.setInboundsClusterTag(dp)
 	m.setGatewayClusterTag(dp)
+
+	opts := core_store.NewUpdateOptions(fs...)
 	labels, err := core_model.ComputeLabels(
 		resource.Descriptor(),
 		resource.GetSpec(),
-		resource.GetMeta().GetLabels(),
+		opts.Labels,
 		core_model.GetNamespace(resource.GetMeta(), m.systemNamespace),
 		resource.GetMeta().GetMesh(),
 		m.mode,

pkg/core/resources/manager/manager.go

@@ -127,6 +127,7 @@ type ConflictRetry struct {
 type UpsertOpts struct {
 	ConflictRetry ConflictRetry
 	Transactions  store.Transactions
+	Labels        map[string]string
 }
 
 type UpsertFunc func(opts *UpsertOpts)
@@ -155,6 +156,12 @@ func NewUpsertOpts(fs ...UpsertFunc) UpsertOpts {
 	return opts
 }
 
+func UpsertWithLabels(labels map[string]string) UpsertFunc {
+	return func(opts *UpsertOpts) {
+		opts.Labels = labels
+	}
+}
+
 var ErrSkipUpsert = errors.New("don't do upsert")
 
 func Upsert(ctx context.Context, manager ResourceManager, key model.ResourceKey, resource model.Resource, fn func(resource model.Resource) error, fs ...UpsertFunc) error {
@@ -177,9 +184,9 @@ func Upsert(ctx context.Context, manager ResourceManager, key model.ResourceKey,
 				return err
 			}
 			if create {
-				return manager.Create(ctx, resource, store.CreateBy(key))
+				return manager.Create(ctx, resource, store.CreateBy(key), store.CreateWithLabels(opts.Labels))
 			} else {
-				return manager.Update(ctx, resource)
+				return manager.Update(ctx, resource, store.UpdateWithLabels(opts.Labels))
 			}
 		})
 	}

pkg/test/resources/builders/dataplane_builder.go

@@ -67,6 +67,11 @@ func (d *DataplaneBuilder) WithMesh(mesh string) *DataplaneBuilder {
 	return d
 }
 
+func (d *DataplaneBuilder) WithLabels(labels map[string]string) *DataplaneBuilder {
+	d.res.Meta.(*test_model.ResourceMeta).Labels = labels
+	return d
+}
+
 func (d *DataplaneBuilder) WithVersion(version string) *DataplaneBuilder {
 	d.res.Meta.(*test_model.ResourceMeta).Version = version
 	return d

pkg/xds/server/callbacks/dataplane_lifecycle.go

@@ -125,7 +125,7 @@ func (d *DataplaneLifecycle) register(
 			return errors.Wrap(err, "you are trying to override existing proxy to which you don't have an access.")
 		}
 		return existing.SetSpec(md.Resource.GetSpec())
-	})
+	}, manager.UpsertWithLabels(md.Resource.GetMeta().GetLabels()))
 	if err != nil {
 		log.Info("cannot register proxy", "reason", err.Error())
 		if !loaded {

pkg/xds/server/callbacks/dataplane_lifecycle_test.go

@@ -73,6 +73,9 @@ var _ = Describe("Dataplane Lifecycle", func() {
                                   "type": "Dataplane",
                                   "mesh": "default",
                                   "name": "backend-01",
+                                  "labels": {
+                                    "app": "backend",
+                                  },
                                   "networking": {
                                     "address": "127.0.0.1",
                                     "inbound": [
@@ -106,8 +109,10 @@ var _ = Describe("Dataplane Lifecycle", func() {
 
 		// then dp is created
 		Expect(err).ToNot(HaveOccurred())
-		err = resManager.Get(context.Background(), core_mesh.NewDataplaneResource(), core_store.GetByKey("backend-01", "default"))
+		dp := core_mesh.NewDataplaneResource()
+		err = resManager.Get(context.Background(), dp, core_store.GetByKey("backend-01", "default"))
 		Expect(err).ToNot(HaveOccurred())
+		Expect(dp.GetMeta().GetLabels()).To(HaveKeyWithValue("app", "backend"))
 
 		// when
 		callbacks.OnStreamClosed(streamId, node)

test/e2e_env/universal/auth/dp_auth.go

@@ -49,20 +49,27 @@ func DpAuth() {
 	It("should be able to override old Dataplane of same service", func() {
 		// given
 		dp := builders.Dataplane().
-			WithName("dp-01").
+			WithName("dp-02").
 			WithMesh(meshName).
+			WithLabels(map[string]string{"app": "not-test-server"}).
 			WithAddress("192.168.0.2").
 			WithServices("test-server").
 			Build()
 		Expect(universal.Cluster.Install(ResourceUniversal(dp))).To(Succeed())
 
 		// when
-		err := TestServerUniversal("dp-02", meshName, WithServiceName("test-server"))(universal.Cluster)
+		err := TestServerUniversal("dp-02", meshName, WithServiceName("test-server"), WithAppLabel("test-server"))(universal.Cluster)
 		Expect(err).ToNot(HaveOccurred())
 
 		// then
 		Eventually(func() (string, error) {
-			return universal.Cluster.GetKumactlOptions().RunKumactlAndGetOutput("get", "dataplanes", "-oyaml")
-		}, "30s", "1s").ShouldNot(ContainSubstring("192.168.0.2"))
+			return universal.Cluster.GetKumactlOptions().RunKumactlAndGetOutput("get", "dataplanes", "--mesh", meshName, "-oyaml")
+		}, "30s", "1s").Should(
+			And(
+				Not(ContainSubstring("192.168.0.2")),
+				Not(ContainSubstring("not-test-server")),
+				ContainSubstring("test-server"),
+			),
+		)
 	})
 }

test/framework/interface.go

@@ -95,6 +95,7 @@ type appDeploymentOptions struct {
 	name                  string
 	appYaml               string
 	appArgs               []string
+	appLabel              string
 	token                 string
 	transparent           *bool
 	builtindns            *bool // true by default
@@ -424,6 +425,12 @@ func WithServiceName(name string) AppDeploymentOption {
 	})
 }
 
+func WithAppLabel(app string) AppDeploymentOption {
+	return AppOptionFunc(func(o *appDeploymentOptions) {
+		o.appLabel = app
+	})
+}
+
 func WithAppendDataplaneYaml(config string) AppDeploymentOption {
 	return AppOptionFunc(func(o *appDeploymentOptions) {
 		o.appendDataplaneConfig = config

test/framework/setup.go

@@ -926,6 +926,8 @@ func TestServerUniversal(name string, mesh string, opt ...AppDeploymentOption) I
 type: Dataplane
 mesh: %s
 name: {{ name }}
+labels:
+  app: %s
 networking:
   address:  {{ address }}
   inbound:
@@ -942,7 +944,7 @@ networking:
 %s
 %s
 %s
-`, mesh, "80", "8080", serviceAddress, opts.serviceName, opts.protocol, opts.serviceVersion, opts.serviceInstance, additionalTags, serviceProbe, transparentProxy, opts.appendDataplaneConfig)
+`, mesh, opts.appLabel, "80", "8080", serviceAddress, opts.serviceName, opts.protocol, opts.serviceVersion, opts.serviceInstance, additionalTags, serviceProbe, transparentProxy, opts.appendDataplaneConfig)
 
 		opt = append(opt,
 			WithName(name),