fix(kuma-cp) Envoy config is created based on old Dataplane (#1848)

(cherry picked from commit 847a42d)

pkg/xds/sync/components.go

@@ -18,7 +18,8 @@ var (
 
 func defaultDataplaneProxyBuilder(rt core_runtime.Runtime, metadataTracker DataplaneMetadataTracker, apiVersion envoy.APIVersion) *DataplaneProxyBuilder {
 	return &DataplaneProxyBuilder{
-		ResManager:            rt.ReadOnlyResourceManager(),
+		CachingResManager:     rt.ReadOnlyResourceManager(),
+		NonCachingResManager:  rt.ResourceManager(),
 		LookupIP:              rt.LookupIP(),
 		DataSourceLoader:      rt.DataSourceLoader(),
 		MetadataTracker:       metadataTracker,

pkg/xds/sync/dataplane_proxy_builder.go

@@ -23,7 +23,8 @@ import (
 var syncLog = core.Log.WithName("sync")
 
 type DataplaneProxyBuilder struct {
-	ResManager            manager.ReadOnlyResourceManager
+	CachingResManager     manager.ReadOnlyResourceManager
+	NonCachingResManager  manager.ReadOnlyResourceManager
 	LookupIP              lookup.LookupIPFunc
 	DataSourceLoader      datasource.Loader
 	MetadataTracker       DataplaneMetadataTracker
@@ -69,7 +70,7 @@ func (p *DataplaneProxyBuilder) resolveDataplane(ctx context.Context, key core_m
 
 	// we use non-cached ResourceManager to always fetch fresh version of the Dataplane.
 	// Otherwise, technically MeshCache can use newer version because it uses List operation instead of Get
-	if err := p.ResManager.Get(ctx, dataplane, core_store.GetBy(key)); err != nil {
+	if err := p.NonCachingResManager.Get(ctx, dataplane, core_store.GetBy(key)); err != nil {
 		return nil, err
 	}
 
@@ -87,12 +88,12 @@ func (p *DataplaneProxyBuilder) resolveRouting(
 	dataplane *core_mesh.DataplaneResource,
 ) (*xds.Routing, xds.DestinationMap, error) {
 	externalServices := &core_mesh.ExternalServiceResourceList{}
-	if err := p.ResManager.List(ctx, externalServices, core_store.ListByMesh(dataplane.Meta.GetMesh())); err != nil {
+	if err := p.CachingResManager.List(ctx, externalServices, core_store.ListByMesh(dataplane.Meta.GetMesh())); err != nil {
 		return nil, nil, err
 	}
 
 	// pick a single the most specific route for each outbound interface
-	routes, err := xds_topology.GetRoutes(ctx, dataplane, p.ResManager)
+	routes, err := xds_topology.GetRoutes(ctx, dataplane, p.CachingResManager)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -111,17 +112,17 @@ func (p *DataplaneProxyBuilder) resolveRouting(
 }
 
 func (p *DataplaneProxyBuilder) matchPolicies(ctx context.Context, meshContext *xds_context.MeshContext, dataplane *core_mesh.DataplaneResource, outboundSelectors xds.DestinationMap) (*xds.MatchedPolicies, error) {
-	healthChecks, err := xds_topology.GetHealthChecks(ctx, dataplane, outboundSelectors, p.ResManager)
+	healthChecks, err := xds_topology.GetHealthChecks(ctx, dataplane, outboundSelectors, p.CachingResManager)
 	if err != nil {
 		return nil, err
 	}
 
-	circuitBreakers, err := xds_topology.GetCircuitBreakers(ctx, dataplane, outboundSelectors, p.ResManager)
+	circuitBreakers, err := xds_topology.GetCircuitBreakers(ctx, dataplane, outboundSelectors, p.CachingResManager)
 	if err != nil {
 		return nil, err
 	}
 
-	trafficTrace, err := xds_topology.GetTrafficTrace(ctx, dataplane, p.ResManager)
+	trafficTrace, err := xds_topology.GetTrafficTrace(ctx, dataplane, p.CachingResManager)
 	if err != nil {
 		return nil, err
 	}
@@ -130,7 +131,7 @@ func (p *DataplaneProxyBuilder) matchPolicies(ctx context.Context, meshContext *
 		tracingBackend = meshContext.Resource.GetTracingBackend(trafficTrace.Spec.GetConf().GetBackend())
 	}
 
-	retries, err := xds_topology.GetRetries(ctx, dataplane, outboundSelectors, p.ResManager)
+	retries, err := xds_topology.GetRetries(ctx, dataplane, outboundSelectors, p.CachingResManager)
 	if err != nil {
 		return nil, err
 	}
@@ -150,7 +151,7 @@ func (p *DataplaneProxyBuilder) matchPolicies(ctx context.Context, meshContext *
 		return nil, err
 	}
 
-	timeouts, err := xds_topology.GetTimeouts(ctx, dataplane, p.ResManager)
+	timeouts, err := xds_topology.GetTimeouts(ctx, dataplane, p.CachingResManager)
 	if err != nil {
 		return nil, err
 	}

test/e2e/healthcheck/healthcheck_hybrid_test.go

@@ -3,6 +3,7 @@ package healthcheck_test
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/gruntwork-io/terratest/modules/k8s"
 	"github.com/gruntwork-io/terratest/modules/retry"
@@ -125,23 +126,27 @@ metadata:
 
 		cmd := []string{"curl", "-v", "-m", "3", "--fail", "echo-server_kuma-test_svc_8080.mesh"}
 
-		checkInstance := func(instance string) {
-			_, err = retry.DoWithRetryE(remoteK8s.GetTesting(), fmt.Sprintf("kubectl exec %s -- %s", pods[0].GetName(), strings.Join(cmd, " ")),
-				DefaultRetries, DefaultTimeout, func() (string, error) {
-					stdout, _, err := remoteK8s.Exec(TestNamespace, pods[0].GetName(), "demo-client", cmd...)
-					if err != nil {
-						return "", err
+		instances := []string{"echo-universal-1", "echo-universal-3"}
+		instanceSet := map[string]bool{}
+
+		_, err = retry.DoWithRetryE(remoteK8s.GetTesting(), fmt.Sprintf("kubectl exec %s -- %s", pods[0].GetName(), strings.Join(cmd, " ")),
+			100, 500*time.Millisecond, func() (string, error) {
+				stdout, _, err := remoteK8s.Exec(TestNamespace, pods[0].GetName(), "demo-client", cmd...)
+				if err != nil {
+					return "", err
+				}
+				for _, instance := range instances {
+					if strings.Contains(stdout, instance) {
+						instanceSet[instance] = true
 					}
-					if !strings.Contains(stdout, instance) {
-						return "", errors.New("wrong instance")
-					}
-					return "", nil
-				},
-			)
-		}
-
-		checkInstance("echo-universal-1")
-		checkInstance("echo-universal-2")
+				}
+				if len(instanceSet) != len(instances) {
+					return "", errors.Errorf("checked %d/%d instances", len(instanceSet), len(instances))
+				}
+				return "", nil
+			},
+		)
+		Expect(err).ToNot(HaveOccurred())
 
 		var counter1, counter2, counter3 int
 		const numOfRequest = 100