fix(xds): don't read metadata in ProxyBuilders (backport #5414) (#5416)

We were using a `DataplaneMetadataTracker` in all proxy builders.
This was causing issues as the proxy may disconnect while reconciliation is in progress.

We now set the proxy metadata in the DataplaneWatchdog so that it's read once only and nil checked.
We therefore protect ourselves against this race

Signed-off-by: Charly Molter <charly.molter@konghq.com>
(cherry picked from commit 39ba902)

Signed-off-by: Charly Molter <charly.molter@konghq.com>

pkg/api-server/inspect_endpoints.go

@@ -23,7 +23,6 @@ import (
 	"github.com/kumahq/kuma/pkg/plugins/runtime/gateway/route"
 	xds_context "github.com/kumahq/kuma/pkg/xds/context"
 	"github.com/kumahq/kuma/pkg/xds/envoy"
-	"github.com/kumahq/kuma/pkg/xds/server/callbacks"
 	"github.com/kumahq/kuma/pkg/xds/sync"
 )
 
@@ -34,10 +33,7 @@ func getMatchedPolicies(
 ) (
 	*core_xds.MatchedPolicies, []gateway.GatewayListenerInfo, core_xds.Proxy, error,
 ) {
-	proxyBuilder := sync.DefaultDataplaneProxyBuilder(
-		*cfg,
-		callbacks.NewDataplaneMetadataTracker(),
-		envoy.APIV3)
+	proxyBuilder := sync.DefaultDataplaneProxyBuilder(*cfg, envoy.APIV3)
 	if proxy, err := proxyBuilder.Build(ctx, dataplaneKey, meshContext); err != nil {
 		return nil, nil, core_xds.Proxy{}, err
 	} else {

pkg/plugins/runtime/gateway/suite_test.go

@@ -96,17 +96,10 @@ func MakeProtoSnapshot(snap cache_v3.ResourceSnapshot) ProtoSnapshot {
 	}
 }
 
-type mockMetadataTracker struct{}
-
-func (m mockMetadataTracker) Metadata(dpKey core_model.ResourceKey) *core_xds.DataplaneMetadata {
-	return nil
-}
-
 func MakeGeneratorContext(rt runtime.Runtime, key core_model.ResourceKey) (*xds_context.Context, *core_xds.Proxy) {
 	b := sync.DataplaneProxyBuilder{
-		MetadataTracker: mockMetadataTracker{},
-		Zone:            rt.Config().Multizone.Zone.Name,
-		APIVersion:      envoy.APIV3,
+		Zone:       rt.Config().Multizone.Zone.Name,
+		APIVersion: envoy.APIV3,
 	}
 
 	cache, err := cla.NewCache(rt.Config().Store.Cache.ExpirationTime, rt.Metrics())

pkg/xds/server/v3/snapshot_generator_test.go

@@ -37,7 +37,6 @@ import (
 	"github.com/kumahq/kuma/pkg/xds/generator"
 	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
 	"github.com/kumahq/kuma/pkg/xds/server"
-	"github.com/kumahq/kuma/pkg/xds/server/callbacks"
 	v3 "github.com/kumahq/kuma/pkg/xds/server/v3"
 	"github.com/kumahq/kuma/pkg/xds/sync"
 	"github.com/kumahq/kuma/pkg/xds/template"
@@ -118,7 +117,7 @@ var _ = Describe("GenerateSnapshot", func() {
 			cfg.DNSServer.ServiceVipPort,
 		)
 
-		proxyBuilder = sync.DefaultDataplaneProxyBuilder(cfg, callbacks.NewDataplaneMetadataTracker(), envoy_common.APIV3)
+		proxyBuilder = sync.DefaultDataplaneProxyBuilder(cfg, envoy_common.APIV3)
 	})
 
 	create := func(r core_model.Resource) {

pkg/xds/sync/components.go

@@ -18,26 +18,22 @@ var (
 
 func DefaultDataplaneProxyBuilder(
 	config kuma_cp.Config,
-	metadataTracker DataplaneMetadataTracker,
 	apiVersion envoy.APIVersion,
 ) *DataplaneProxyBuilder {
 	return &DataplaneProxyBuilder{
-		MetadataTracker: metadataTracker,
-		Zone:            config.Multizone.Zone.Name,
-		APIVersion:      apiVersion,
+		Zone:       config.Multizone.Zone.Name,
+		APIVersion: apiVersion,
 	}
 }
 
 func DefaultIngressProxyBuilder(
 	rt core_runtime.Runtime,
-	metadataTracker DataplaneMetadataTracker,
 	apiVersion envoy.APIVersion,
 ) *IngressProxyBuilder {
 	return &IngressProxyBuilder{
 		ResManager:         rt.ResourceManager(),
 		ReadOnlyResManager: rt.ReadOnlyResourceManager(),
 		LookupIP:           rt.LookupIP(),
-		MetadataTracker:    metadataTracker,
 		apiVersion:         apiVersion,
 		meshCache:          rt.MeshCache(),
 		zone:               rt.Config().Multizone.Zone.Name,
@@ -47,15 +43,13 @@ func DefaultIngressProxyBuilder(
 func DefaultEgressProxyBuilder(
 	ctx context.Context,
 	rt core_runtime.Runtime,
-	metadataTracker DataplaneMetadataTracker,
 	apiVersion envoy.APIVersion,
 ) *EgressProxyBuilder {
 	return &EgressProxyBuilder{
 		ctx:                ctx,
 		ResManager:         rt.ResourceManager(),
 		ReadOnlyResManager: rt.ReadOnlyResourceManager(),
 		LookupIP:           rt.LookupIP(),
-		MetadataTracker:    metadataTracker,
 		meshCache:          rt.MeshCache(),
 		apiVersion:         apiVersion,
 		zone:               rt.Config().Multizone.Zone.Name,
@@ -77,20 +71,17 @@ func DefaultDataplaneWatchdogFactory(
 
 	dataplaneProxyBuilder := DefaultDataplaneProxyBuilder(
 		config,
-		metadataTracker,
 		apiVersion,
 	)
 
 	ingressProxyBuilder := DefaultIngressProxyBuilder(
 		rt,
-		metadataTracker,
 		apiVersion,
 	)
 
 	egressProxyBuilder := DefaultEgressProxyBuilder(
 		ctx,
 		rt,
-		metadataTracker,
 		apiVersion,
 	)
 

pkg/xds/sync/dataplane_proxy_builder.go

@@ -27,8 +27,6 @@ import (
 var syncLog = core.Log.WithName("sync")
 
 type DataplaneProxyBuilder struct {
-	MetadataTracker DataplaneMetadataTracker
-
 	Zone       string
 	APIVersion envoy.APIVersion
 }
@@ -64,7 +62,6 @@ func (p *DataplaneProxyBuilder) Build(ctx context.Context, key core_model.Resour
 		Id:             core_xds.FromResourceKey(key),
 		APIVersion:     p.APIVersion,
 		Dataplane:      dp,
-		Metadata:       p.MetadataTracker.Metadata(key),
 		Routing:        *routing,
 		Policies:       *matchedPolicies,
 		SecretsTracker: secretsTracker,

pkg/xds/sync/dataplane_watchdog.go

@@ -63,11 +63,11 @@ func (d *DataplaneWatchdog) Sync(ctx context.Context) error {
 	}
 	switch d.dpType {
 	case mesh_proto.DataplaneProxyType:
-		return d.syncDataplane(ctx)
+		return d.syncDataplane(ctx, metadata)
 	case mesh_proto.IngressProxyType:
-		return d.syncIngress(ctx)
+		return d.syncIngress(ctx, metadata)
 	case mesh_proto.EgressProxyType:
-		return d.syncEgress(ctx)
+		return d.syncEgress(ctx, metadata)
 	default:
 		// It might be a case that dp type is not yet inferred because there is no Dataplane definition yet.
 		return nil
@@ -91,7 +91,7 @@ func (d *DataplaneWatchdog) Cleanup() error {
 
 // syncDataplane syncs state of the Dataplane.
 // It uses Mesh Hash to decide if we need to regenerate configuration or not.
-func (d *DataplaneWatchdog) syncDataplane(ctx context.Context) error {
+func (d *DataplaneWatchdog) syncDataplane(ctx context.Context, metadata *core_xds.DataplaneMetadata) error {
 	meshCtx, err := d.MeshCache.GetMeshContext(ctx, syncLog, d.key.Mesh)
 	if err != nil {
 		return err
@@ -126,6 +126,7 @@ func (d *DataplaneWatchdog) syncDataplane(ctx context.Context) error {
 	if !envoyCtx.Mesh.Resource.MTLSEnabled() {
 		d.EnvoyCpCtx.Secrets.Cleanup(d.key) // we need to cleanup secrets if mtls is disabled
 	}
+	proxy.Metadata = metadata
 	if err := d.DataplaneReconciler.Reconcile(*envoyCtx, proxy); err != nil {
 		return err
 	}
@@ -134,7 +135,7 @@ func (d *DataplaneWatchdog) syncDataplane(ctx context.Context) error {
 }
 
 // syncIngress synces state of Ingress Dataplane. Notice that it does not use Mesh Hash yet because Ingress supports many Meshes.
-func (d *DataplaneWatchdog) syncIngress(ctx context.Context) error {
+func (d *DataplaneWatchdog) syncIngress(ctx context.Context, metadata *core_xds.DataplaneMetadata) error {
 	envoyCtx := &xds_context.Context{
 		ControlPlane: d.EnvoyCpCtx,
 		Mesh:         xds_context.MeshContext{}, // ZoneIngress does not have a mesh!
@@ -148,12 +149,13 @@ func (d *DataplaneWatchdog) syncIngress(ctx context.Context) error {
 		return errors.Wrap(err, "could not get Envoy Admin mTLS certs")
 	}
 	proxy.EnvoyAdminMTLSCerts = envoyAdminMTLS
+	proxy.Metadata = metadata
 	return d.IngressReconciler.Reconcile(*envoyCtx, proxy)
 }
 
 // syncEgress syncs state of Egress Dataplane. Notice that it does not use
 // Mesh Hash yet because Egress supports many Meshes.
-func (d *DataplaneWatchdog) syncEgress(ctx context.Context) error {
+func (d *DataplaneWatchdog) syncEgress(ctx context.Context, metadata *core_xds.DataplaneMetadata) error {
 	envoyCtx := &xds_context.Context{
 		ControlPlane: d.EnvoyCpCtx,
 		Mesh:         xds_context.MeshContext{}, // ZoneEgress does not have a mesh!
@@ -168,7 +170,7 @@ func (d *DataplaneWatchdog) syncEgress(ctx context.Context) error {
 		return errors.Wrap(err, "could not get Envoy Admin mTLS certs")
 	}
 	proxy.EnvoyAdminMTLSCerts = envoyAdminMTLS
-
+	proxy.Metadata = metadata
 	return d.EgressReconciler.Reconcile(*envoyCtx, proxy)
 }
 

pkg/xds/sync/dataplane_watchdog_test.go

@@ -87,9 +87,8 @@ var _ = Describe("Dataplane Watchdog", func() {
 
 		deps = sync.DataplaneWatchdogDependencies{
 			DataplaneProxyBuilder: &sync.DataplaneProxyBuilder{
-				MetadataTracker: metadataTracker,
-				APIVersion:      envoy.APIV3,
-				Zone:            zone,
+				APIVersion: envoy.APIV3,
+				Zone:       zone,
 			},
 			DataplaneReconciler: snapshotReconciler,
 			EnvoyCpCtx: &xds_context.ControlPlaneContext{

pkg/xds/sync/egress_proxy_builder.go

@@ -12,7 +12,7 @@ import (
 	"github.com/kumahq/kuma/pkg/core/resources/manager"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	core_store "github.com/kumahq/kuma/pkg/core/resources/store"
-	"github.com/kumahq/kuma/pkg/core/xds"
+	core_xds "github.com/kumahq/kuma/pkg/core/xds"
 	xds_cache "github.com/kumahq/kuma/pkg/xds/cache/mesh"
 	envoy_common "github.com/kumahq/kuma/pkg/xds/envoy"
 	xds_topology "github.com/kumahq/kuma/pkg/xds/topology"
@@ -24,7 +24,6 @@ type EgressProxyBuilder struct {
 	ResManager         manager.ResourceManager
 	ReadOnlyResManager manager.ReadOnlyResourceManager
 	LookupIP           lookup.LookupIPFunc
-	MetadataTracker    DataplaneMetadataTracker
 	meshCache          *xds_cache.Cache
 
 	zone       string
@@ -34,7 +33,7 @@ type EgressProxyBuilder struct {
 func (p *EgressProxyBuilder) Build(
 	ctx context.Context,
 	key core_model.ResourceKey,
-) (*xds.Proxy, error) {
+) (*core_xds.Proxy, error) {
 	zoneEgress := core_mesh.NewZoneEgressResource()
 
 	if err := p.ReadOnlyResManager.Get(
@@ -80,7 +79,7 @@ func (p *EgressProxyBuilder) Build(
 		return zoneIngresses[a].GetMeta().GetName() < zoneIngresses[b].GetMeta().GetName()
 	})
 
-	var meshResourcesList []*xds.MeshResources
+	var meshResourcesList []*core_xds.MeshResources
 
 	for _, mesh := range meshes {
 		meshName := mesh.GetMeta().GetName()
@@ -96,7 +95,7 @@ func (p *EgressProxyBuilder) Build(
 		faultInjections := meshCtx.Resources.FaultInjections().Items
 		rateLimits := meshCtx.Resources.RateLimits().Items
 
-		meshResources := &xds.MeshResources{
+		meshResources := &core_xds.MeshResources{
 			Mesh:             mesh,
 			TrafficRoutes:    trafficRoutes,
 			ExternalServices: externalServices,
@@ -125,15 +124,14 @@ func (p *EgressProxyBuilder) Build(
 		meshResourcesList = append(meshResourcesList, meshResources)
 	}
 
-	proxy := &xds.Proxy{
-		Id:         xds.FromResourceKey(key),
+	proxy := &core_xds.Proxy{
+		Id:         core_xds.FromResourceKey(key),
 		APIVersion: p.apiVersion,
-		ZoneEgressProxy: &xds.ZoneEgressProxy{
+		ZoneEgressProxy: &core_xds.ZoneEgressProxy{
 			ZoneEgressResource: zoneEgress,
 			ZoneIngresses:      zoneIngresses,
 			MeshResourcesList:  meshResourcesList,
 		},
-		Metadata: p.MetadataTracker.Metadata(key),
 	}
 
 	return proxy, nil

pkg/xds/sync/ingress_proxy_builder.go

@@ -10,7 +10,7 @@ import (
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	"github.com/kumahq/kuma/pkg/core/resources/registry"
 	core_store "github.com/kumahq/kuma/pkg/core/resources/store"
-	"github.com/kumahq/kuma/pkg/core/xds"
+	core_xds "github.com/kumahq/kuma/pkg/core/xds"
 	xds_cache "github.com/kumahq/kuma/pkg/xds/cache/mesh"
 	"github.com/kumahq/kuma/pkg/xds/envoy"
 	"github.com/kumahq/kuma/pkg/xds/ingress"
@@ -21,14 +21,13 @@ type IngressProxyBuilder struct {
 	ResManager         manager.ResourceManager
 	ReadOnlyResManager manager.ReadOnlyResourceManager
 	LookupIP           lookup.LookupIPFunc
-	MetadataTracker    DataplaneMetadataTracker
 	meshCache          *xds_cache.Cache
 
 	apiVersion envoy.APIVersion
 	zone       string
 }
 
-func (p *IngressProxyBuilder) Build(ctx context.Context, key core_model.ResourceKey) (*xds.Proxy, error) {
+func (p *IngressProxyBuilder) Build(ctx context.Context, key core_model.ResourceKey) (*core_xds.Proxy, error) {
 	zoneIngress, err := p.getZoneIngress(ctx, key)
 	if err != nil {
 		return nil, err
@@ -62,18 +61,17 @@ func (p *IngressProxyBuilder) Build(ctx context.Context, key core_model.Resource
 
 	routing := p.resolveRouting(zoneIngress, zoneEgressesList, allMeshDataplanes, availableExternalServices, zoneIngressProxy.MeshGateways)
 
-	proxy := &xds.Proxy{
-		Id:               xds.FromResourceKey(key),
+	proxy := &core_xds.Proxy{
+		Id:               core_xds.FromResourceKey(key),
 		APIVersion:       p.apiVersion,
 		ZoneIngress:      zoneIngress,
-		Metadata:         p.MetadataTracker.Metadata(key),
 		Routing:          *routing,
 		ZoneIngressProxy: zoneIngressProxy,
 	}
 	return proxy, nil
 }
 
-func (p *IngressProxyBuilder) buildZoneIngressProxy(ctx context.Context) (*xds.ZoneIngressProxy, error) {
+func (p *IngressProxyBuilder) buildZoneIngressProxy(ctx context.Context) (*core_xds.ZoneIngressProxy, error) {
 	routes := &core_mesh.TrafficRouteResourceList{}
 	if err := p.ReadOnlyResManager.List(ctx, routes); err != nil {
 		return nil, err
@@ -91,7 +89,7 @@ func (p *IngressProxyBuilder) buildZoneIngressProxy(ctx context.Context) (*xds.Z
 		return nil, err
 	}
 
-	return &xds.ZoneIngressProxy{
+	return &core_xds.ZoneIngressProxy{
 		TrafficRouteList: routes,
 		GatewayRoutes:    gatewayRoutes,
 		MeshGateways:     gateways,
@@ -118,13 +116,13 @@ func (p *IngressProxyBuilder) resolveRouting(
 	dataplanes *core_mesh.DataplaneResourceList,
 	externalServices *core_mesh.ExternalServiceResourceList,
 	meshGateways *core_mesh.MeshGatewayResourceList,
-) *xds.Routing {
+) *core_xds.Routing {
 	destinations := ingress.BuildDestinationMap(zoneIngress)
 	endpoints := ingress.BuildEndpointMap(
 		destinations, dataplanes.Items, externalServices.Items, zoneEgresses.Items, meshGateways.Items,
 	)
 
-	routing := &xds.Routing{
+	routing := &core_xds.Routing{
 		OutboundTargets: endpoints,
 	}
 	return routing

pkg/xds/sync/proxy_builder_test.go

@@ -30,12 +30,6 @@ import (
 	"github.com/kumahq/kuma/pkg/xds/sync"
 )
 
-type mockMetadataTracker struct{}
-
-func (m mockMetadataTracker) Metadata(dpKey core_model.ResourceKey) *core_xds.DataplaneMetadata {
-	return nil
-}
-
 func initializeStore(ctx context.Context, resourceManager core_manager.ResourceManager, fileWithResourcesName string) {
 	resourcePath := filepath.Join(
 		"testdata", "input", fileWithResourcesName,
@@ -75,7 +69,6 @@ func initializeStore(ctx context.Context, resourceManager core_manager.ResourceM
 }
 
 var _ = Describe("Proxy Builder", func() {
-	tracker := mockMetadataTracker{}
 	localZone := "zone-1"
 
 	ctx := context.Background()
@@ -112,7 +105,6 @@ var _ = Describe("Proxy Builder", func() {
 		egressProxyBuilder := sync.DefaultEgressProxyBuilder(
 			ctx,
 			rt,
-			tracker,
 			envoy_common.APIV3,
 		)
 
@@ -195,7 +187,6 @@ var _ = Describe("Proxy Builder", func() {
 	Describe("Build() zone ingress", func() {
 		ingressProxyBuilder := sync.DefaultIngressProxyBuilder(
 			rt,
-			tracker,
 			envoy_common.APIV3,
 		)