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

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

@@ -191,6 +191,14 @@ func (b *remoteBootstrap) requestForBootstrap(ctx context.Context, client *http.
 		},
 		SystemCaPath: params.SystemCaPath,
 	}
+	switch cfg.DataplaneRuntime.XdsConfigMode {
+	case "delta":
+		request.XdsConfigMode = types.DELTA
+	case "sotw":
+		request.XdsConfigMode = types.SOTW
+	default:
+		request.XdsConfigMode = types.NOT_DEFINED
+	}
 	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/remote_bootstrap_test.go

@@ -149,6 +149,7 @@ var _ = Describe("Remote Bootstrap", func() {
 				cfg.Dataplane.Mesh = "demo"
 				cfg.Dataplane.Name = "sample"
 				cfg.DataplaneRuntime.Token = "token"
+				cfg.DataplaneRuntime.XdsConfigMode = "delta"
 
 				return testCase{
 					config:                       cfg,

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

@@ -29,5 +29,6 @@
   "certPath": "",
   "keyPath": ""
  },
- "systemCaPath": ""
+ "systemCaPath": "",
+ "xdsConfigMode": 2
 }

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,10 @@ 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"`
+	// XdsConfigMode defines xDS communication type between Envoy and control-plane.
+	// Available values are: sotw, delta
+	// By default it takes configuration provided by the control-plane.
+	XdsConfigMode string `json:"xdsConfigMode,omitempty" envconfig:"kuma_dataplane_runtime_dynamic_xds_config_mode"`
 }
 
 type Metrics struct {
@@ -335,6 +339,13 @@ func (d *DataplaneRuntime) Validate() error {
 	if d.BinaryPath == "" {
 		errs = multierr.Append(errs, errors.Errorf(".BinaryPath must be non-empty"))
 	}
+	if d.XdsConfigMode != "" {
+		switch d.XdsConfigMode {
+		case "delta", "sotw":
+		default:
+			errs = multierr.Append(errs, errors.Errorf(".XdsConfigMode 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; .XdsConfigMode 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:
+  xdsConfigMode: 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
+  xdsConfigMode: delta

pkg/config/loader_test.go

@@ -372,6 +372,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)))
@@ -759,6 +760,7 @@ experimental:
   sidecarContainers: true
   generateMeshServices: true
   skipPersistedVIPs: true
+  useDeltaXds: true
 proxy:
   gateway:
     globalDownstreamMaxConnections: 1
@@ -1065,6 +1067,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

@@ -41,7 +41,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(
@@ -57,10 +58,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 {

pkg/xds/auth/callbacks.go

@@ -10,6 +10,7 @@ import (
 	"google.golang.org/grpc/metadata"
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
+	"github.com/kumahq/kuma/pkg/core"
 	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	core_manager "github.com/kumahq/kuma/pkg/core/resources/manager"
 	"github.com/kumahq/kuma/pkg/core/resources/model"
@@ -30,14 +31,15 @@ type DPNotFoundRetry struct {
 	MaxTimes uint
 }
 
-func NewCallbacks(resManager core_manager.ReadOnlyResourceManager, authenticator Authenticator, dpNotFoundRetry DPNotFoundRetry) util_xds.Callbacks {
+func NewCallbacks(resManager core_manager.ReadOnlyResourceManager, authenticator Authenticator, dpNotFoundRetry DPNotFoundRetry) util_xds.MultiXDSCallbacks {
 	if dpNotFoundRetry.Backoff == 0 { // backoff cannot be 0
 		dpNotFoundRetry.Backoff = 1 * time.Millisecond
 	}
 	return &authCallbacks{
 		resManager:      resManager,
 		authenticator:   authenticator,
 		streams:         map[core_xds.StreamID]stream{},
+		deltaStreams:    map[core_xds.StreamID]stream{},
 		dpNotFoundRetry: dpNotFoundRetry,
 	}
 }
@@ -51,6 +53,7 @@ type authCallbacks struct {
 
 	sync.RWMutex // protects streams
 	streams      map[core_xds.StreamID]stream
+	deltaStreams map[core_xds.StreamID]stream
 }
 
 type stream struct {
@@ -62,7 +65,7 @@ type stream struct {
 	nodeID string
 }
 
-var _ util_xds.Callbacks = &authCallbacks{}
+var _ util_xds.MultiXDSCallbacks = &authCallbacks{}
 
 func (a *authCallbacks) OnStreamOpen(ctx context.Context, streamID core_xds.StreamID, _ string) error {
 	a.Lock()
@@ -82,10 +85,38 @@ func (a *authCallbacks) OnStreamClosed(streamID core_xds.StreamID) {
 }
 
 func (a *authCallbacks) OnStreamRequest(streamID core_xds.StreamID, req util_xds.DiscoveryRequest) error {
-	s, err := a.stream(streamID, req)
+	return a.onStreamRequest(streamID, req, false)
+}
+
+func (a *authCallbacks) OnDeltaStreamOpen(ctx context.Context, streamID core_xds.StreamID, _ string) error {
+	a.Lock()
+	defer a.Unlock()
+
+	a.deltaStreams[streamID] = stream{
+		ctx:      ctx,
+		resource: nil,
+	}
+
+	core.Log.V(1).Info("OnDeltaStreamOpen", "streamID", streamID)
+	return nil
+}
+
+func (a *authCallbacks) OnDeltaStreamClosed(streamID int64) {
+	a.Lock()
+	delete(a.deltaStreams, streamID)
+	a.Unlock()
+}
+
+func (a *authCallbacks) OnStreamDeltaRequest(streamID core_xds.StreamID, req util_xds.DeltaDiscoveryRequest) error {
+	return a.onStreamRequest(streamID, req, true)
+}
+
+func (a *authCallbacks) onStreamRequest(streamID core_xds.StreamID, req util_xds.Request, isDelta bool) error {
+	s, err := a.stream(streamID, req, isDelta)
 	if err != nil {
 		return err
 	}
+	core.Log.V(1).Info("OnStreamDeltaRequest auth", "req", req)
 
 	credential, err := ExtractCredential(s.ctx)
 	if err != nil {
@@ -95,14 +126,24 @@ func (a *authCallbacks) OnStreamRequest(streamID core_xds.StreamID, req util_xds
 		return errors.Wrap(err, "authentication failed")
 	}
 	a.Lock()
-	a.streams[streamID] = s
+	if isDelta {
+		a.deltaStreams[streamID] = s
+	} else {
+		a.streams[streamID] = s
+	}
 	a.Unlock()
 	return nil
 }
 
-func (a *authCallbacks) stream(streamID core_xds.StreamID, req util_xds.DiscoveryRequest) (stream, error) {
+func (a *authCallbacks) stream(streamID core_xds.StreamID, req util_xds.Request, isDelta bool) (stream, error) {
 	a.RLock()
-	s, ok := a.streams[streamID]
+	var s stream
+	var ok bool
+	if isDelta {
+		s, ok = a.deltaStreams[streamID]
+	} else {
+		s, ok = a.streams[streamID]
+	}
 	a.RUnlock()
 	if !ok {
 		return stream{}, errors.New("stream is not present")

pkg/xds/bootstrap/components.go

@@ -21,6 +21,7 @@ func RegisterBootstrap(rt core_runtime.Runtime) error {
 		rt.Config().DpServer.Authn.EnableReloadableTokens || rt.Config().Store.Type == store.KubernetesStore,
 		rt.Config().DpServer.Hds.Enabled,
 		rt.Config().GetEnvoyAdminPort(),
+		rt.Config().Experimental.UseDeltaXds,
 	)
 	if err != nil {
 		return err