feat(kuma-cp): implement delta xDS for envoy config exchange

app/kuma-dp/pkg/dataplane/envoy/remote_bootstrap.go

@@ -191,6 +191,11 @@ func (b *remoteBootstrap) requestForBootstrap(ctx context.Context, client *http.
 		},
 		SystemCaPath: params.SystemCaPath,
 	}
+	if cfg.DataplaneRuntime.XDSConfigType == "" {
+		request.XDSConfigType = "sotw"
+	} else {
+		request.XDSConfigType = "delta"
+	}
 	jsonBytes, err := json.MarshalIndent(request, "", " ")
 	if err != nil {
 		return nil, errors.Wrap(err, "could not marshal request to json")

app/kuma-dp/pkg/dataplane/envoy/testdata/bootstrap-request-0.golden.json

@@ -31,5 +31,6 @@
   "certPath": "/tmp/cert.pem",
   "keyPath": "/tmp/key.pem"
  },
- "systemCaPath": ""
+ "systemCaPath": "",
+ "xdsConfigType": "sotw"
 }

app/kuma-dp/pkg/dataplane/envoy/testdata/bootstrap-request-1.golden.json

@@ -30,5 +30,6 @@
   "certPath": "",
   "keyPath": ""
  },
- "systemCaPath": ""
+ "systemCaPath": "",
+ "xdsConfigType": "sotw"
 }

app/kuma-dp/pkg/dataplane/envoy/testdata/bootstrap-request-2.golden.json

@@ -29,5 +29,6 @@
   "certPath": "",
   "keyPath": ""
  },
- "systemCaPath": ""
+ "systemCaPath": "",
+ "xdsConfigType": "sotw"
 }

app/kuma-dp/pkg/dataplane/envoy/testdata/bootstrap-request-3.golden.json

@@ -29,5 +29,6 @@
   "certPath": "",
   "keyPath": ""
  },
- "systemCaPath": ""
+ "systemCaPath": "",
+ "xdsConfigType": "sotw"
 }

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/containernetworking/cni v1.2.3
 	github.com/containernetworking/plugins v1.5.1
 	github.com/emicklei/go-restful/v3 v3.12.1
-	github.com/envoyproxy/go-control-plane v0.12.0
+	github.com/envoyproxy/go-control-plane v0.13.0
 	github.com/envoyproxy/protoc-gen-validate v1.1.0
 	github.com/evanphx/json-patch/v5 v5.9.0
 	github.com/exaring/otelpgx v0.6.2
@@ -186,6 +186,7 @@ require (
 	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.0 // indirect
+	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/pquerna/otp v1.2.0 // indirect
@@ -227,6 +228,3 @@ require (
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 )
-
-// remove once https://github.com/envoyproxy/go-control-plane/issues/875 is resolved
-replace github.com/envoyproxy/go-control-plane v0.12.0 => github.com/envoyproxy/go-control-plane v0.11.2-0.20231010133108-1dfbe83bcebc

go.sum

@@ -83,8 +83,8 @@ github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4
 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/emicklei/go-restful/v3 v3.12.1 h1:PJMDIM/ak7btuL8Ex0iYET9hxM3CI2sjZtzpL63nKAU=
 github.com/emicklei/go-restful/v3 v3.12.1/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
-github.com/envoyproxy/go-control-plane v0.11.2-0.20231010133108-1dfbe83bcebc h1:k6n7EmQYjNHEKr8XI3rdtFIhb0UdJSyCWnt9gvgLx5g=
-github.com/envoyproxy/go-control-plane v0.11.2-0.20231010133108-1dfbe83bcebc/go.mod h1:9ODlpdyEVNyci9DZ6cdQYkwYSW1YMrnSz3xnku3cjL0=
+github.com/envoyproxy/go-control-plane v0.13.0 h1:HzkeUz1Knt+3bK+8LG1bxOO/jzWZmdxpwC51i202les=
+github.com/envoyproxy/go-control-plane v0.13.0/go.mod h1:GRaKG3dwvFoTg4nj7aXdZnvMg4d7nvT/wl9WgVXn3Q8=
 github.com/envoyproxy/protoc-gen-validate v1.1.0 h1:tntQDh69XqOCOZsDz0lVJQez/2L6Uu2PdjCQwWCJ3bM=
 github.com/envoyproxy/protoc-gen-validate v1.1.0/go.mod h1:sXRDRVmzEbkM7CVcM06s9shE/m23dg3wzjl0UWqJ2q4=
 github.com/evanphx/json-patch v5.7.0+incompatible h1:vgGkfT/9f8zE6tvSCe74nfpAVDQ2tG6yudJd8LBksgI=
@@ -328,6 +328,8 @@ github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaR
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
+github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=

pkg/config/app/kuma-cp/config.go

@@ -464,6 +464,8 @@ type ExperimentalConfig struct {
 	// If true skips persisted VIPs. Change to true only if generateMeshServices is enabled.
 	// Do not enable on production.
 	SkipPersistedVIPs bool `json:"skipPersistedVIPs" envconfig:"KUMA_EXPERIMENTAL_SKIP_PERSISTED_VIPS"`
+	// If true uses Delta xDS to deliver changes to sidecars.
+	UseDeltaXDS bool `json:"useDeltaXDS" envconfig:"KUMA_EXPERIMENTAL_USE_DELTA_XDS"`
 }
 
 type ExperimentalKDSEventBasedWatchdog struct {

pkg/config/app/kuma-dp/config.go

@@ -217,6 +217,8 @@ type DataplaneRuntime struct {
 	DynamicConfiguration DynamicConfiguration `json:"dynamicConfiguration" envconfig:"kuma_dataplane_runtime_dynamic_configuration"`
 	// SystemCaPath defines path of system provided Ca
 	SystemCaPath string `json:"systemCaPath,omitempty" envconfig:"kuma_dataplane_runtime_dynamic_system_ca_path"`
+	// XDSConfigType defines xDS communication type between Envoy and control-plane
+	XDSConfigType string `json:"xdsConfigType,omitempty" envconfig:"kuma_dataplane_runtime_dynamic_xds_config_type"`
 }
 
 type Metrics struct {
@@ -335,6 +337,13 @@ func (d *DataplaneRuntime) Validate() error {
 	if d.BinaryPath == "" {
 		errs = multierr.Append(errs, errors.Errorf(".BinaryPath must be non-empty"))
 	}
+	if d.XDSConfigType != "" {
+		switch d.XDSConfigType {
+		case "delta", "sotw":
+		default:
+			errs = multierr.Append(errs, errors.Errorf(".XDSConfigType can be one of: delta, sotw"))
+		}
+	}
 	return errs
 }
 

pkg/config/app/kuma-dp/testdata/invalid-config.golden.txt

@@ -1 +1 @@
-parsing configuration from file 'testdata/invalid-config.input.yaml' failed: configuration validation failed: .ControlPlane is not valid: .Retry is not valid: .Backoff must be a positive duration; .Dataplane is not valid: .ProxyType is not valid: not-a-proxy is not a valid proxy type; .Mesh must be non-empty; .Name must be non-empty; .DrainTime must be positive; .DataplaneRuntime is not valid: .BinaryPath must be non-empty
+parsing configuration from file 'testdata/invalid-config.input.yaml' failed: configuration validation failed: .ControlPlane is not valid: .Retry is not valid: .Backoff must be a positive duration; .Dataplane is not valid: .ProxyType is not valid: not-a-proxy is not a valid proxy type; .Mesh must be non-empty; .Name must be non-empty; .DrainTime must be positive; .DataplaneRuntime is not valid: .BinaryPath must be non-empty; .XDSConfigType can be one of: delta, sotw

pkg/config/app/kuma-dp/testdata/invalid-config.input.yaml

@@ -12,3 +12,4 @@ dataplane:
   proxyType: not-a-proxy
 dataplaneRuntime:
   binaryPath:
+  xdsConfigType: a

pkg/config/app/kuma-dp/testdata/valid-config.input.yaml

@@ -12,3 +12,4 @@ dataplaneRuntime:
   binaryPath: envoy.sh
   configDir: /var/run/envoy
   envoyLogLevel: trace
+  xdsConfigType: delta

pkg/config/loader_test.go

@@ -370,6 +370,7 @@ var _ = Describe("Config loader", func() {
 			Expect(cfg.Experimental.SidecarContainers).To(BeTrue())
 			Expect(cfg.Experimental.SkipPersistedVIPs).To(BeTrue())
 			Expect(cfg.Experimental.GenerateMeshServices).To(BeTrue())
+			Expect(cfg.Experimental.UseDeltaXDS).To(BeTrue())
 
 			Expect(cfg.Proxy.Gateway.GlobalDownstreamMaxConnections).To(BeNumerically("==", 1))
 			Expect(cfg.EventBus.BufferSize).To(Equal(uint(30)))
@@ -754,6 +755,7 @@ experimental:
   sidecarContainers: true
   generateMeshServices: true
   skipPersistedVIPs: true
+  useDeltaXDS: true
 proxy:
   gateway:
     globalDownstreamMaxConnections: 1
@@ -1058,6 +1060,7 @@ meshService:
 				"KUMA_EXPERIMENTAL_SIDECAR_CONTAINERS":                                                     "true",
 				"KUMA_EXPERIMENTAL_GENERATE_MESH_SERVICES":                                                 "true",
 				"KUMA_EXPERIMENTAL_SKIP_PERSISTED_VIPS":                                                    "true",
+				"KUMA_EXPERIMENTAL_USE_DELTA_XDS":                                                          "true",
 				"KUMA_PROXY_GATEWAY_GLOBAL_DOWNSTREAM_MAX_CONNECTIONS":                                     "1",
 				"KUMA_TRACING_OPENTELEMETRY_ENDPOINT":                                                      "otel-collector:4317",
 				"KUMA_TRACING_OPENTELEMETRY_ENABLED":                                                       "true",

pkg/hds/tracker/callbacks.go

@@ -40,7 +40,8 @@ type tracker struct {
 
 	sync.RWMutex       // protects access to the fields below
 	streamsAssociation map[xds.StreamID]core_model.ResourceKey
-	dpStreams          map[core_model.ResourceKey]streams
+	// deltaStreamsAssociation map[xds.StreamID]core_model.ResourceKey
+	dpStreams map[core_model.ResourceKey]streams
 }
 
 func NewCallbacks(
@@ -56,10 +57,11 @@ func NewCallbacks(
 	return &tracker{
 		resourceManager:    resourceManager,
 		streamsAssociation: map[xds.StreamID]core_model.ResourceKey{},
-		dpStreams:          map[core_model.ResourceKey]streams{},
-		config:             config,
-		log:                log,
-		metrics:            metrics,
+		// deltaStreamsAssociation: map[xds.StreamID]core_model.ResourceKey{},
+		dpStreams: map[core_model.ResourceKey]streams{},
+		config:    config,
+		log:       log,
+		metrics:   metrics,
 		reconciler: &reconciler{
 			cache:     cache,
 			hasher:    hasher,
@@ -69,6 +71,14 @@ func NewCallbacks(
 	}
 }
 
+func (t *tracker) OnDeltaStreamOpen(ctx context.Context, streamID int64) error {
+	return t.OnStreamOpen(ctx, streamID)
+}
+
+func (t *tracker) OnDeltaStreamClosed(streamID xds.StreamID) {
+	t.OnStreamClosed(streamID)
+}
+
 func (t *tracker) OnStreamOpen(ctx context.Context, streamID int64) error {
 	t.metrics.StreamsActiveInc()
 	return nil

pkg/util/xds/callbacks.go

@@ -3,52 +3,50 @@ package xds
 import (
 	"context"
 
-	discoveryv3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/structpb"
 )
 
-// DiscoveryRequest defines interface over real Envoy's DiscoveryRequest.
-type DiscoveryRequest interface {
+type Request interface {
 	NodeId() string
-	// Node returns either a v2 or v3 Node
 	Node() interface{}
 	Metadata() *structpb.Struct
-	VersionInfo() string
-	GetTypeUrl() string
 	GetResponseNonce() string
-	GetResourceNames() []string
+	GetTypeUrl() string
 	HasErrors() bool
 	ErrorMsg() string
+	VersionInfo() string
+	GetResourceNames() []string
 }
 
-// DiscoveryResponse defines interface over real Envoy's DiscoveryResponse.
-type DiscoveryResponse interface {
+type Response interface {
 	GetTypeUrl() string
-	VersionInfo() string
 	GetResources() []*anypb.Any
 	GetNonce() string
+	VersionInfo() string
+	GetNumberOfResources() int
+}
+
+// DiscoveryRequest defines interface over real Envoy's DiscoveryRequest.
+type DiscoveryRequest interface {
+	Request
+}
+
+// DiscoveryResponse defines interface over real Envoy's DiscoveryResponse.
+type DiscoveryResponse interface {
+	Response
 }
 
 type DeltaDiscoveryRequest interface {
-	NodeId() string
-	// Node returns either a v2 or v3 Node
-	Node() interface{}
-	Metadata() *structpb.Struct
-	GetTypeUrl() string
-	GetResponseNonce() string
+	Request
 	GetResourceNamesSubscribe() []string
 	GetInitialResourceVersions() map[string]string
-	HasErrors() bool
-	ErrorMsg() string
 }
 
 // DeltaDiscoveryResponse defines interface over real Envoy's DeltaDiscoveryResponse.
 type DeltaDiscoveryResponse interface {
-	GetTypeUrl() string
-	GetResources() []*discoveryv3.Resource
+	Response
 	GetRemovedResources() []string
-	GetNonce() string
 }
 
 // Callbacks defines Callbacks for xDS streaming requests. The difference over real go-control-plane Callbacks is that it takes an DiscoveryRequest / DiscoveryResponse interface.
@@ -94,4 +92,10 @@ type RestCallbacks interface {
 type MultiCallbacks interface {
 	Callbacks
 	RestCallbacks
+	DeltaCallbacks
+}
+
+type MultiXDSCallbacks interface {
+	Callbacks
+	DeltaCallbacks
 }

pkg/util/xds/v3/callbacks.go

@@ -6,6 +6,7 @@ import (
 	envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	envoy_sd "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	envoy_xds "github.com/envoyproxy/go-control-plane/pkg/server/v3"
+	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/kumahq/kuma/pkg/util/xds"
@@ -175,6 +176,10 @@ type discoveryResponse struct {
 	*envoy_sd.DiscoveryResponse
 }
 
+func (d *discoveryResponse) GetNumberOfResources() int {
+	return len(d.Resources)
+}
+
 func (d *discoveryResponse) VersionInfo() string {
 	return d.GetVersionInfo()
 }
@@ -211,12 +216,36 @@ func (d *deltaDiscoveryRequest) GetInitialResourceVersions() map[string]string {
 	return d.InitialResourceVersions
 }
 
+func (d *deltaDiscoveryRequest) GetResourceNames() []string {
+	return d.GetResourceNamesSubscribe()
+}
+
+func (d *deltaDiscoveryRequest) VersionInfo() string {
+	return ""
+}
+
 var _ xds.DeltaDiscoveryRequest = &deltaDiscoveryRequest{}
 
 type deltaDiscoveryResponse struct {
 	*envoy_sd.DeltaDiscoveryResponse
 }
 
+func (d *deltaDiscoveryResponse) VersionInfo() string {
+	return d.SystemVersionInfo
+}
+
+func (d *deltaDiscoveryResponse) GetResources() []*anypb.Any {
+	resources := []*anypb.Any{}
+	for _, res := range d.Resources {
+		resources = append(resources, res.Resource)
+	}
+	return resources
+}
+
+func (d *deltaDiscoveryResponse) GetNumberOfResources() int {
+	return len(d.Resources)
+}
+
 var _ xds.DeltaDiscoveryResponse = &deltaDiscoveryResponse{}
 
 func (d *deltaDiscoveryResponse) GetTypeUrl() string {