feat(kds): sync mesh service status (#10337)

Signed-off-by: Jakub Dyszkiewicz <jakub.dyszkiewicz@gmail.com>

api/mesh/v1alpha1/kds.proto

@@ -22,6 +22,7 @@ message KumaResource {
   }
   Meta meta = 1;
   google.protobuf.Any spec = 2;
+  google.protobuf.Any status = 3;
 }
 
 message ZoneHealthCheckRequest {}

docs/generated/raw/protos/KumaResource.json

@@ -23,6 +23,22 @@
                     },
                     "additionalProperties": true,
                     "type": "object"
+                },
+                "status": {
+                    "properties": {
+                        "type_url": {
+                            "type": "string",
+                            "description": "A URL/resource name that uniquely identifies the type of the serialized protocol buffer message. This string must contain at least one \"/\" character. The last segment of the URL's path must represent the fully qualified name of the type (as in `path/google.protobuf.Duration`). The name should be in a canonical form (e.g., leading \".\" is not accepted). In practice, teams usually precompile into the binary all types that they expect it to use in the context of Any. However, for URLs which use the scheme `http`, `https`, or no scheme, one can optionally set up a type server that maps type URLs to message definitions as follows: * If no scheme is provided, `https` is assumed. * An HTTP GET on the URL must yield a [google.protobuf.Type][]   value in binary format, or produce an error. * Applications are allowed to cache lookup results based on the   URL, or have them precompiled into a binary to avoid any   lookup. Therefore, binary compatibility needs to be preserved   on changes to types. (Use versioned type names to manage   breaking changes.) Note: this functionality is not currently available in the official protobuf release, and it is not used for type URLs beginning with type.googleapis.com. Schemes other than `http`, `https` (or the empty scheme) might be used with implementation specific semantics."
+                        },
+                        "value": {
+                            "type": "string",
+                            "description": "Must be a valid serialized protocol buffer of the above specified type.",
+                            "format": "binary",
+                            "binaryEncoding": "base64"
+                        }
+                    },
+                    "additionalProperties": true,
+                    "type": "object"
                 }
             },
             "additionalProperties": true,

pkg/core/resources/model/rest/converter.go

@@ -41,11 +41,14 @@ func (f *from) Resource(r core_model.Resource) Resource {
 
 	meta := f.Meta(r)
 	if r.Descriptor().IsPluginOriginated {
-		return &v1alpha1.Resource{
+		res := &v1alpha1.Resource{
 			ResourceMeta: meta,
 			Spec:         r.GetSpec(),
-			Status:       r.GetStatus(),
 		}
+		if r.Descriptor().HasStatus {
+			res.Status = r.GetStatus()
+		}
+		return res
 	} else {
 		return &unversioned.Resource{
 			Meta: meta,

pkg/core/resources/model/rest/unmarshaller.go

@@ -101,6 +101,11 @@ func (u *unmarshaler) UnmarshalListToCore(b []byte, rs core_model.ResourceList)
 		if err := r.SetSpec(ri.GetSpec()); err != nil {
 			return err
 		}
+		if r.Descriptor().HasStatus {
+			if err := r.SetStatus(ri.GetStatus()); err != nil {
+				return err
+			}
+		}
 		r.SetMeta(ri.GetMeta())
 		_ = rs.AddItem(r)
 	}

pkg/kds/context/context.go

@@ -190,13 +190,9 @@ func MapZoneTokenSigningKeyGlobalToPublicKey(_ kds.Features, r core_model.Resour
 // from names of resources if resources are stored in kubernetes.
 func RemoveK8sSystemNamespaceSuffixMapper(k8sSystemNamespace string) reconcile.ResourceMapper {
 	return func(_ kds.Features, r core_model.Resource) (core_model.Resource, error) {
-		newObj := r.Descriptor().NewObject()
 		dotSuffix := fmt.Sprintf(".%s", k8sSystemNamespace)
 		newName := strings.TrimSuffix(r.GetMeta().GetName(), dotSuffix)
-		newMeta := util.CloneResourceMeta(r.GetMeta(), util.WithName(newName))
-		newObj.SetMeta(newMeta)
-		_ = newObj.SetSpec(r.GetSpec())
-		return newObj, nil
+		return util.CloneResource(r, util.WithName(newName)), nil
 	}
 }
 
@@ -213,12 +209,7 @@ func HashSuffixMapper(checkKDSFeature bool, labelsToUse ...string) reconcile.Res
 			values = append(values, r.GetMeta().GetLabels()[lbl])
 		}
 
-		newObj := r.Descriptor().NewObject()
-		newMeta := util.CloneResourceMeta(r.GetMeta(), util.WithName(hash.HashedName(r.GetMeta().GetMesh(), name, values...)))
-		newObj.SetMeta(newMeta)
-		_ = newObj.SetSpec(r.GetSpec())
-
-		return newObj, nil
+		return util.CloneResource(r, util.WithName(hash.HashedName(r.GetMeta().GetMesh(), name, values...))), nil
 	}
 }
 

pkg/kds/util/resource.go

@@ -6,10 +6,12 @@ import (
 
 	envoy_sd "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	envoy_types "github.com/envoyproxy/go-control-plane/pkg/cache/types"
+	"google.golang.org/protobuf/types/known/anypb"
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
 	system_proto "github.com/kumahq/kuma/api/system/v1alpha1"
 	"github.com/kumahq/kuma/pkg/core/resources/model"
+	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	"github.com/kumahq/kuma/pkg/core/resources/registry"
 	cache_v2 "github.com/kumahq/kuma/pkg/kds/v2/cache"
 	util_proto "github.com/kumahq/kuma/pkg/util/proto"
@@ -49,14 +51,22 @@ func ToEnvoyResources(rlist model.ResourceList) ([]envoy_types.Resource, error)
 		if err != nil {
 			return nil, err
 		}
+		var pbanyStatus *anypb.Any
+		if r.Descriptor().HasStatus {
+			pbanyStatus, err = model.ToAny(r.GetStatus())
+			if err != nil {
+				return nil, err
+			}
+		}
 		rv = append(rv, &mesh_proto.KumaResource{
 			Meta: &mesh_proto.KumaResource_Meta{
 				Name:    r.GetMeta().GetName(),
 				Mesh:    r.GetMeta().GetMesh(),
 				Labels:  r.GetMeta().GetLabels(),
 				Version: "",
 			},
-			Spec: pbany,
+			Spec:   pbany,
+			Status: pbanyStatus,
 		})
 	}
 	return rv, nil
@@ -133,6 +143,11 @@ func toResources(resourceType model.ResourceType, krs []*mesh_proto.KumaResource
 		if err = model.FromAny(kr.Spec, obj.GetSpec()); err != nil {
 			return nil, err
 		}
+		if obj.Descriptor().HasStatus && kr.Status != nil {
+			if err = model.FromAny(kr.Status, obj.GetStatus()); err != nil {
+				return nil, err
+			}
+		}
 		obj.SetMeta(kumaResourceMetaToResourceMeta(kr.Meta))
 		if err := list.AddItem(obj); err != nil {
 			return nil, err
@@ -152,3 +167,14 @@ func StatsOf(status *system_proto.KDSSubscriptionStatus, resourceType model.Reso
 	}
 	return stat
 }
+
+func CloneResource(res core_model.Resource, fs ...CloneResourceMetaOpt) core_model.Resource {
+	newObj := res.Descriptor().NewObject()
+	newMeta := CloneResourceMeta(res.GetMeta(), fs...)
+	newObj.SetMeta(newMeta)
+	_ = newObj.SetSpec(res.GetSpec())
+	if newObj.Descriptor().HasStatus {
+		_ = newObj.SetStatus(res.GetStatus())
+	}
+	return newObj
+}

pkg/kds/v2/store/sync.go

@@ -186,7 +186,9 @@ func (s *syncResourceStore) Sync(syncCtx context.Context, upstreamResponse clien
 			continue
 		}
 		newLabels := r.GetMeta().GetLabels()
-		if !core_model.Equal(existing.GetSpec(), r.GetSpec()) || !maps.Equal(existing.GetMeta().GetLabels(), newLabels) {
+		if !core_model.Equal(existing.GetSpec(), r.GetSpec()) ||
+			!maps.Equal(existing.GetMeta().GetLabels(), newLabels) ||
+			!core_model.Equal(existing.GetStatus(), r.GetStatus()) {
 			// we have to use meta of the current Store during update, because some Stores (Kubernetes, Memory)
 			// expect to receive ResourceMeta of own type.
 			r.SetMeta(existing.GetMeta())

pkg/plugins/resources/remote/store.go

@@ -164,6 +164,11 @@ func (s *remoteStore) Get(ctx context.Context, res model.Resource, fs ...store.G
 		return err
 	}
 	res.SetMeta(restRes.GetMeta())
+	if res.Descriptor().HasStatus {
+		if err := res.SetStatus(restRes.GetStatus()); err != nil {
+			return err
+		}
+	}
 	return res.SetSpec(restRes.GetSpec())
 }
 

test/e2e_env/multizone/meshservice/status.go

@@ -0,0 +1,67 @@
+package meshservice
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	. "github.com/kumahq/kuma/test/framework"
+	"github.com/kumahq/kuma/test/framework/envs/multizone"
+)
+
+func MeshService() {
+	meshName := "meshservice"
+
+	BeforeAll(func() {
+		Expect(NewClusterSetup().
+			Install(MTLSMeshUniversal(meshName)).
+			Install(MeshTrafficPermissionAllowAllUniversal(meshName)).
+			Setup(multizone.Global)).To(Succeed())
+		Expect(WaitForMesh(meshName, multizone.Zones())).To(Succeed())
+	})
+
+	AfterEachFailure(func() {
+		DebugUniversal(multizone.Global, meshName)
+		DebugUniversal(multizone.UniZone1, meshName)
+		DebugUniversal(multizone.UniZone2, meshName)
+		DebugKube(multizone.KubeZone1, meshName)
+		DebugKube(multizone.KubeZone2, meshName)
+	})
+
+	E2EAfterAll(func() {
+		Expect(multizone.UniZone1.DeleteMeshApps(meshName)).To(Succeed())
+		Expect(multizone.UniZone2.DeleteMeshApps(meshName)).To(Succeed())
+		Expect(multizone.Global.DeleteMesh(meshName)).To(Succeed())
+	})
+
+	It("should sync MeshService to global with VIP status", func() {
+		ms := `
+type: MeshService
+name: backend
+mesh: meshservice
+labels:
+  kuma.io/origin: zone
+spec:
+  selector:
+    dataplaneTags:
+      x: aaa
+`
+
+		// when
+		Expect(multizone.UniZone1.Install(YamlUniversal(ms))).To(Succeed())
+
+		// then VIP is assigned
+		Eventually(func(g Gomega) {
+			out, err := multizone.UniZone1.GetKumactlOptions().RunKumactlAndGetOutput("get", "meshservices", "-m", meshName, "-oyaml")
+			g.Expect(err).ToNot(HaveOccurred())
+			g.Expect(out).To(ContainSubstring("ip: 241.0.0."))
+		}, "30s", "1s").Should(Succeed())
+
+		// and MeshService is synced to global with status
+		Eventually(func(g Gomega) {
+			out, err := multizone.Global.GetKumactlOptions().RunKumactlAndGetOutput("get", "meshservices", "-m", meshName, "-oyaml")
+			g.Expect(err).ToNot(HaveOccurred())
+			g.Expect(out).To(ContainSubstring("kuma.io/display-name: backend"))
+			g.Expect(out).To(ContainSubstring("ip: 241.0.0."))
+		}, "30s", "1s").Should(Succeed())
+	})
+}

test/e2e_env/multizone/multizone_suite_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/kumahq/kuma/test/e2e_env/multizone/inspect"
 	"github.com/kumahq/kuma/test/e2e_env/multizone/localityawarelb"
 	"github.com/kumahq/kuma/test/e2e_env/multizone/meshhttproute"
+	"github.com/kumahq/kuma/test/e2e_env/multizone/meshservice"
 	"github.com/kumahq/kuma/test/e2e_env/multizone/meshtcproute"
 	"github.com/kumahq/kuma/test/e2e_env/multizone/meshtimeout"
 	"github.com/kumahq/kuma/test/e2e_env/multizone/meshtrafficpermission"
@@ -77,4 +78,5 @@ var (
 	_ = Describe("Advanced LocalityAwareness with MeshLoadBalancingStrategy with Gateway", localityawarelb.LocalityAwareLBGateway, Ordered)
 	_ = Describe("Advanced LocalityAwareness with MeshLoadBalancingStrategy and Enabled Egress", localityawarelb.LocalityAwareLBEgress, Ordered)
 	_ = Describe("Defaults", defaults.Defaults, Ordered)
+	_ = Describe("MeshService", meshservice.MeshService, Ordered)
 )