feat(kds): add support for kuma.io/kds-sync label

api/mesh/v1alpha1/dataplane_helpers.go

@@ -16,6 +16,9 @@ const (
 	KubeNamespaceTag = "k8s.kuma.io/namespace"
 	KubeServiceTag   = "k8s.kuma.io/service-name"
 	KubePortTag      = "k8s.kuma.io/service-port"
+	// KDSSyncLabel a label that controls properties of the KDS sync.
+	// currently only disabled/enabled is supported
+	KDSSyncLabel = "kuma.io/kds-sync"
 )
 
 const (

pkg/core/resources/model/resource.go

@@ -67,6 +67,10 @@ const (
 	// there is a mechanism to avoid sending it back to the original Zone (.e.g producer policies, zone origin labels...).
 	// but this is outside of the resourceType sync mechanism.
 	SyncedAcrossZonesFlag = KDSFlagType(1 << 5)
+	// KDSDisabledByDefault is a flag that indicates that this resource type is not sent using KDS by default.
+	// only resources that are explicitly marked with `kuma.io/kds-sync: enabled` are sent.
+	// this is only used for Config resources atm.
+	KDSDisabledByDefault = KDSFlagType(1 << 6)
 	// ZoneToGlobalFlag gets sent from Zone to Global
 	ZoneToGlobalFlag = ConsumedByGlobalFlag | ProvidedByZoneFlag
 	// GlobalToZonesFlag gets sent from Global to Zone
@@ -289,6 +293,22 @@ func (d ResourceTypeDescriptor) IsReadOnly(isGlobal bool, isFederated bool) bool
 	return (isGlobal && !d.KDSFlags.Has(ProvidedByGlobalFlag)) || (isFederated && !d.KDSFlags.Has(ProvidedByZoneFlag))
 }
 
+// HasKDSDisabled returns whether this resource type is not sent using KDS.
+// Cases where KDS is disabled:
+// - KDSDisabledFlag is set
+// - KDSDisabledByDefault is set for the whole resource and there's no explicit allowance
+// zone is the name of the zone (empty when sending to global) it's potentially to be able to do more options on the flag in the future
+func (d ResourceTypeDescriptor) HasKDSDisabled(zone string, labels map[string]string) bool {
+	switch labels[mesh_proto.KDSSyncLabel] {
+	case "enabled":
+		return false
+	case "disabled":
+		return true
+	default:
+		return d.KDSFlags.Has(KDSDisabledByDefault)
+	}
+}
+
 type TypeFilter interface {
 	Apply(descriptor ResourceTypeDescriptor) bool
 }
@@ -311,12 +331,6 @@ func SentFromZoneToGlobal() TypeFilter {
 	})
 }
 
-func HasKDSEnabled() TypeFilter {
-	return TypeFilterFn(func(descriptor ResourceTypeDescriptor) bool {
-		return descriptor.KDSFlags != KDSDisabledFlag
-	})
-}
-
 func HasKumactlEnabled() TypeFilter {
 	return TypeFilterFn(func(descriptor ResourceTypeDescriptor) bool {
 		return descriptor.KumactlArg != ""

pkg/kds/context/context.go

@@ -257,8 +257,13 @@ func UpdateResourceMeta(fs ...util.CloneResourceMetaOpt) reconcile_v2.ResourceMa
 
 func GlobalProvidedFilter(rm manager.ResourceManager) reconcile_v2.ResourceFilter {
 	return func(ctx context.Context, zoneName string, features kds.Features, r core_model.Resource) bool {
+		// There's explicit flag to disable KDS for a resource
+		if r.Descriptor().HasKDSDisabled(zoneName, r.GetMeta().GetLabels()) {
+			return false
+		}
 		// for Config, Secret and GlobalSecret there are resources that are internal to each CP
 		// we filter them out to not send them to other CPs
+		// We could instead leverage the KDSDisabled label, but this would be complex for backward compatibility
 		switch r.Descriptor().Name {
 		case system.ConfigType:
 			// We only sync specific entries for configs.
@@ -356,5 +361,8 @@ func SkipUnsupportedHostnameGenerator(ctx context.Context, clusterID string, fea
 // ZoneProvidedFilter filters resources that should be sent from Zone to Global
 // a resource is sent from zone to global only if it was created in this zone.
 func ZoneProvidedFilter(_ context.Context, localZone string, _ kds.Features, r core_model.Resource) bool {
+	if r.Descriptor().HasKDSDisabled("", r.GetMeta().GetLabels()) {
+		return false
+	}
 	return core_model.IsLocallyOriginated(config_core.Zone, r.GetMeta().GetLabels())
 }

pkg/kds/context/full_sync_test.go

@@ -0,0 +1,100 @@
+package context_test
+
+import (
+	"fmt"
+	"os"
+	"path"
+	"strings"
+	"sync"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	kuma_cp "github.com/kumahq/kuma/pkg/config/app/kuma-cp"
+	config_core "github.com/kumahq/kuma/pkg/config/core"
+	config_store "github.com/kumahq/kuma/pkg/config/core/resources/store"
+	config_types "github.com/kumahq/kuma/pkg/config/types"
+	"github.com/kumahq/kuma/pkg/core/resources/store"
+	"github.com/kumahq/kuma/pkg/kds/global"
+	"github.com/kumahq/kuma/pkg/kds/zone"
+	"github.com/kumahq/kuma/pkg/plugins/resources/memory"
+	"github.com/kumahq/kuma/pkg/test"
+	"github.com/kumahq/kuma/pkg/test/kds/setup"
+	"github.com/kumahq/kuma/pkg/test/matchers"
+	test_store "github.com/kumahq/kuma/pkg/test/store"
+)
+
+var _ = Describe("Full sync tests", func() {
+	DescribeTable("Full sync tests", func(ctx SpecContext, folder string) {
+		files, err := os.ReadDir(folder)
+		Expect(err).ToNot(HaveOccurred())
+		zones := make(map[string]store.ResourceStore)
+		wg := sync.WaitGroup{}
+		done := make(chan struct{})
+
+		for _, file := range files {
+			if strings.HasSuffix(file.Name(), ".input.yaml") {
+				zoneName := strings.TrimSuffix(file.Name(), ".input.yaml")
+				resourceStore := memory.NewStore()
+				fullPath := path.Join(folder, file.Name())
+				Expect(test_store.LoadResourcesFromFile(ctx, resourceStore, fullPath)).To(Succeed())
+				zones[zoneName] = resourceStore
+			}
+		}
+
+		// Starts all the things
+
+		globalStore := zones["global"]
+		Expect(globalStore).ToNot(BeNil(), "global must be present")
+		// start global
+		cfg := kuma_cp.DefaultConfig()
+		cfg.Store.Type = config_store.MemoryStore
+		globalPort, err := test.GetFreePort()
+		Expect(err).ToNot(HaveOccurred())
+		cfg.Multizone.Global.KDS.GrpcPort = uint32(globalPort)
+		cfg.Multizone.Global.KDS.TlsEnabled = false
+		cfg.Multizone.Global.KDS.ZoneInsightFlushInterval = config_types.Duration{Duration: 100 * time.Millisecond}
+		cfg.Mode = config_core.Global
+		rt := setup.NewTestRuntime(ctx, cfg, globalStore)
+		Expect(global.Setup(rt)).To(Succeed())
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			defer GinkgoRecover()
+			Expect(rt.Start(done)).To(Succeed())
+		}()
+		// start zones
+		for zoneName, zoneStore := range zones {
+			if zoneName == "global" {
+				continue
+			}
+			cfg := kuma_cp.DefaultConfig()
+			cfg.Store.Type = config_store.MemoryStore
+			cfg.Mode = config_core.Zone
+			cfg.Multizone.Zone.Name = zoneName
+			cfg.Multizone.Zone.GlobalAddress = fmt.Sprintf("grpc://localhost:%d", globalPort)
+			cfg.Multizone.Global.KDS.ZoneInsightFlushInterval = config_types.Duration{Duration: 100 * time.Millisecond}
+			rt := setup.NewTestRuntime(ctx, cfg, zoneStore)
+			Expect(zone.Setup(rt)).To(Succeed())
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				defer GinkgoRecover()
+				Expect(rt.Start(done)).To(Succeed())
+			}()
+		}
+
+		// Wait for some time to ensure sync was complete
+		time.Sleep(time.Second * 2)
+		close(done)
+		wg.Wait()
+
+		// Compare golden files
+		for zoneName, zoneStore := range zones {
+			out, err := test_store.ExtractResources(ctx, zoneStore)
+			Expect(err).To(Succeed())
+			Expect(out).To(matchers.MatchGoldenYAML(folder, zoneName+".golden.yaml"), "zone %s", zoneName)
+		}
+	}, test.EntriesAsFolder("full_sync"))
+})

pkg/kds/context/testdata/full_sync/mesh-only/global.golden.yaml

@@ -0,0 +1,39 @@
+# Mesh
+---
+creationTime: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: default
+type: Mesh
+
+# Zone
+---
+creationTime: "0001-01-01T00:00:00Z"
+enabled: true
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-1
+type: Zone
+
+---
+creationTime: "0001-01-01T00:00:00Z"
+enabled: true
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-2
+type: Zone
+
+# ZoneInsight
+---
+creationTime: "0001-01-01T00:00:00Z"
+healthCheck:
+  time: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-2
+type: ZoneInsight
+
+---
+creationTime: "0001-01-01T00:00:00Z"
+healthCheck:
+  time: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-1
+type: ZoneInsight
+

pkg/kds/context/testdata/full_sync/mesh-only/global.input.yaml

@@ -0,0 +1,2 @@
+type: Mesh
+name: default

pkg/kds/context/testdata/full_sync/mesh-only/zone-1.golden.yaml

@@ -0,0 +1,16 @@
+# Mesh
+---
+creationTime: "0001-01-01T00:00:00Z"
+labels:
+  kuma.io/origin: global
+modificationTime: "0001-01-01T00:00:00Z"
+name: default
+type: Mesh
+
+# ZoneInsight
+---
+creationTime: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-1
+type: ZoneInsight
+

pkg/kds/context/testdata/full_sync/mesh-only/zone-2.golden.yaml

@@ -0,0 +1,16 @@
+# Mesh
+---
+creationTime: "0001-01-01T00:00:00Z"
+labels:
+  kuma.io/origin: global
+modificationTime: "0001-01-01T00:00:00Z"
+name: default
+type: Mesh
+
+# ZoneInsight
+---
+creationTime: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-2
+type: ZoneInsight
+

pkg/kds/context/testdata/full_sync/policy-simple/global.golden.yaml

@@ -0,0 +1,76 @@
+# Mesh
+---
+creationTime: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: default
+type: Mesh
+
+# MeshTimeout
+---
+creationTime: "0001-01-01T00:00:00Z"
+mesh: default
+modificationTime: "0001-01-01T00:00:00Z"
+name: default-timeout-global
+spec:
+  to:
+  - default:
+      connectionTimeout: 2s
+      http:
+        requestTimeout: 2s
+      idleTimeout: 20s
+    targetRef:
+      kind: Mesh
+type: MeshTimeout
+
+---
+creationTime: "0001-01-01T00:00:00Z"
+labels:
+  kuma.io/display-name: default-timeout-zone-1
+  kuma.io/origin: zone
+  kuma.io/zone: zone-1
+mesh: default
+modificationTime: "0001-01-01T00:00:00Z"
+name: default-timeout-zone-1-c58w24w4vw27f454
+spec:
+  to:
+  - default:
+      connectionTimeout: 2s
+      http:
+        requestTimeout: 2s
+      idleTimeout: 20s
+    targetRef:
+      kind: Mesh
+type: MeshTimeout
+
+# Zone
+---
+creationTime: "0001-01-01T00:00:00Z"
+enabled: true
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-1
+type: Zone
+
+---
+creationTime: "0001-01-01T00:00:00Z"
+enabled: true
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-2
+type: Zone
+
+# ZoneInsight
+---
+creationTime: "0001-01-01T00:00:00Z"
+healthCheck:
+  time: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-1
+type: ZoneInsight
+
+---
+creationTime: "0001-01-01T00:00:00Z"
+healthCheck:
+  time: "0001-01-01T00:00:00Z"
+modificationTime: "0001-01-01T00:00:00Z"
+name: zone-2
+type: ZoneInsight
+

pkg/kds/context/testdata/full_sync/policy-simple/global.input.yaml

@@ -0,0 +1,15 @@
+type: Mesh
+name: default
+---
+type: MeshTimeout
+name: default-timeout-global
+mesh: default
+spec:
+  to:
+    - targetRef:
+        kind: Mesh
+      default:
+        idleTimeout: 20s
+        connectionTimeout: 2s
+        http:
+          requestTimeout: 2s