Repository: kumahq/kuma
Pull request #1862 (2021-04-21 12:46:46)

fix(tests) e2e framework env functions fix (backport #1860)
This is an automatic backport of pull request #1860 done by [Mergify](https://mergify.io).

---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.io/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.io/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.io/
</details>



Patch:
diff --git a/pkg/xds/sync/components.go b/pkg/xds/sync/components.go
index 17896fcc0..32cd4419b 100644
--- a/pkg/xds/sync/components.go
+++ b/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,
diff --git a/pkg/xds/sync/dataplane_proxy_builder.go b/pkg/xds/sync/dataplane_proxy_builder.go
index feca51162..e176704b2 100644
--- a/pkg/xds/sync/dataplane_proxy_builder.go
+++ b/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
 	}
diff --git a/pkg/xds/topology/outbound.go b/pkg/xds/topology/outbound.go
index 691dc8d11..72004e262 100644
--- a/pkg/xds/topology/outbound.go
+++ b/pkg/xds/topology/outbound.go
@@ -179,18 +179,25 @@ func convertToEnvoy(ds *v1alpha1.DataSource, mesh string, loader datasource.Load
 }
 
 func localityFromTags(mesh *mesh_core.MeshResource, priority uint32, tags map[string]string) *core_xds.Locality {
-	if !mesh.Spec.GetRouting().GetLocalityAwareLoadBalancing() {
-		return nil
-	}
-
 	region, regionPresent := tags[mesh_proto.RegionTag]
 	zone, zonePresent := tags[mesh_proto.ZoneTag]
 	subZone, subZonePresent := tags[mesh_proto.SubZoneTag]
 
 	if !regionPresent && !zonePresent && !subZonePresent {
+		// this means that we are running in standalone since in multi-zone Kuma always adds Zone tag automatically
 		return nil
 	}
 
+	if !mesh.Spec.GetRouting().GetLocalityAwareLoadBalancing() {
+		// we want to set the Locality even when localityAwareLoadbalancing is enabled
+		// If we set the locality we have an extra visibility about this in /clusters etc.
+		// Kuma's LocalityAwareLoadBalancing feature is based only on Priority therefore when it's disabled we need to set Priority to local
+		//
+		// Setting this regardless of LocalityAwareLoadBalancing on the mesh also solves the problem that endpoints have problems when moving from one locality to another
+		// https://github.com/envoyproxy/envoy/issues/12392
+		priority = priorityLocal
+	}
+
 	return &core_xds.Locality{
 		Region:   region,
 		Zone:     zone,


Test patch:
diff --git a/mk/e2e.new.mk b/mk/e2e.new.mk
index 9a400c972..05e603007 100644
--- a/mk/e2e.new.mk
+++ b/mk/e2e.new.mk
@@ -15,6 +15,52 @@ KUMA_IN_EKS ?=
 KUMA_DEFAULT_RETRIES ?=
 KUMA_DEFAULT_TIMEOUT ?=
 
+ENV_VARS ?= API_VERSION="$(API_VERSION)"
+
+ifdef KUMA_UNIVERSAL_IMAGE
+	ENV_VARS += KUMA_UNIVERSAL_IMAGE=$(KUMA_UNIVERSAL_IMAGE)
+endif
+
+ifdef HELM_CHART_PATH
+	ENV_VARS += HELM_CHART_PATH=$(HELM_CHART_PATH)
+endif
+
+ifdef KUMA_GLOBAL_IMAGE_TAG
+	ENV_VARS += KUMA_GLOBAL_IMAGE_TAG=$(KUMA_GLOBAL_IMAGE_TAG)
+endif
+
+ifdef KUMA_GLOBAL_IMAGE_REGISTRY
+	ENV_VARS += KUMA_GLOBAL_IMAGE_REGISTRY=$(KUMA_GLOBAL_IMAGE_REGISTRY)
+endif
+
+ifdef KUMA_CP_IMAGE_REPOSITORY
+	ENV_VARS += KUMA_CP_IMAGE_REPOSITORY=$(KUMA_CP_IMAGE_REPOSITORY)
+endif
+
+ifdef KUMA_DP_IMAGE_REPOSITORY
+	ENV_VARS += KUMA_DP_IMAGE_REPOSITORY=$(KUMA_DP_IMAGE_REPOSITORY)
+endif
+
+ifdef KUMA_DP_INIT_IMAGE_REPOSITORY
+	ENV_VARS += KUMA_DP_INIT_IMAGE_REPOSITORY=$(KUMA_DP_INIT_IMAGE_REPOSITORY)
+endif
+
+ifdef KUMA_USE_LOAD_BALANCER
+	ENV_VARS += KUMA_USE_LOAD_BALANCER=$(KUMA_USE_LOAD_BALANCER)
+endif
+
+ifdef KUMA_IN_EKS
+	ENV_VARS += KUMA_IN_EKS=$(KUMA_IN_EKS)
+endif
+
+ifdef KUMA_DEFAULT_RETRIES
+	ENV_VARS += KUMA_DEFAULT_RETRIES=$(KUMA_DEFAULT_RETRIES)
+endif
+
+ifdef KUMA_DEFAULT_TIMEOUT
+	ENV_VARS += KUMA_DEFAULT_TIMEOUT=$(KUMA_DEFAULT_TIMEOUT)
+endif
+
 TEST_NAMES = $(shell ls -1 ./test/e2e)
 ALL_TESTS = $(addprefix ./test/e2e/, $(addsuffix /..., $(TEST_NAMES)))
 E2E_PKG_LIST ?= $(ALL_TESTS)
@@ -56,19 +102,8 @@ test/e2e/test:
 	for t in $(E2E_PKG_LIST); do \
 		K8SCLUSTERS="$(K8SCLUSTERS)" \
 		KUMACTLBIN=${BUILD_ARTIFACTS_DIR}/kumactl/kumactl \
-		API_VERSION="$(API_VERSION)" \
-		HELM_CHART_PATH="$(HELM_CHART_PATH)" \
-		KUMA_GLOBAL_IMAGE_TAG="$(KUMA_GLOBAL_IMAGE_TAG)" \
-		KUMA_GLOBAL_IMAGE_REGISTRY="$(KUMA_GLOBAL_IMAGE_REGISTRY)" \
-		KUMA_CP_IMAGE_REPOSITORY="$(KUMA_CP_IMAGE_REPOSITORY)" \
-		KUMA_DP_IMAGE_REPOSITORY="$(KUMA_DP_IMAGE_REPOSITORY)" \
-		KUMA_DP_INIT_IMAGE_REPOSITORY="$(KUMA_DP_INIT_IMAGE_REPOSITORY)" \
-		KUMA_USE_LOAD_BALANCER="$(KUMA_USE_LOAD_BALANCER)" \
-		KUMA_IN_EKS="$(KUMA_IN_EKS)" \
-		KUMA_UNIVERSAL_IMAGE="$(KUMA_UNIVERSAL_IMAGE)" \
-		KUMA_DEFAULT_RETRIES="$(KUMA_DEFAULT_RETRIES)" \
-        KUMA_DEFAULT_TIMEOUT="$(KUMA_DEFAULT_TIMEOUT)" \
-			$(GO_TEST) -v -timeout=45m $$t || exit; \
+		$(ENV_VARS) \
+		$(GO_TEST) -v -timeout=45m $$t || exit; \
 	done
 
 # test/e2e/debug is used for quicker feedback of E2E tests (ex. debugging flaky tests)
diff --git a/pkg/xds/topology/outbound_test.go b/pkg/xds/topology/outbound_test.go
index 33603ab22..4c806db59 100644
--- a/pkg/xds/topology/outbound_test.go
+++ b/pkg/xds/topology/outbound_test.go
@@ -59,7 +59,6 @@ var _ = Describe("TrafficRoute", func() {
 		dataSourceLoader = datasource.NewDataSourceLoader(secretManager)
 	})
 	Describe("GetOutboundTargets()", func() {
-
 		It("should pick proper dataplanes for each outbound destination", func() {
 			// given
 			backend := &mesh_core.DataplaneResource{ // dataplane that is a source of traffic
@@ -197,12 +196,18 @@ var _ = Describe("TrafficRoute", func() {
 						mesh_proto.ServiceTag: "elastic",
 						mesh_proto.RegionTag:  "eu",
 					},
+					Locality: &core_xds.Locality{
+						Region: "eu",
+					},
 					Weight: 1,
 				},
 				{
 					Target: "192.168.0.6",
 					Port:   9200,
 					Tags:   map[string]string{mesh_proto.ServiceTag: "elastic", mesh_proto.RegionTag: "us"},
+					Locality: &core_xds.Locality{
+						Region: "us",
+					},
 					Weight: 1,
 				},
 			}))
@@ -214,6 +219,9 @@ var _ = Describe("TrafficRoute", func() {
 						mesh_proto.ServiceTag: "frontend",
 						mesh_proto.RegionTag:  "eu",
 					},
+					Locality: &core_xds.Locality{
+						Region: "eu",
+					},
 					Weight: 1,
 				},
 			}))
@@ -225,6 +233,9 @@ var _ = Describe("TrafficRoute", func() {
 						mesh_proto.ServiceTag: "backend",
 						mesh_proto.RegionTag:  "eu",
 					},
+					Locality: &core_xds.Locality{
+						Region: "eu",
+					},
 					Weight: 1,
 				},
 			}))
@@ -362,6 +373,9 @@ var _ = Describe("TrafficRoute", func() {
 							Target: "192.168.0.100",
 							Port:   12345,
 							Tags:   map[string]string{mesh_proto.ServiceTag: "redis", "version": "v2", mesh_proto.RegionTag: "eu"},
+							Locality: &core_xds.Locality{
+								Region: "eu",
+							},
 							Weight: 2,
 						},
 						{
@@ -374,6 +388,9 @@ var _ = Describe("TrafficRoute", func() {
 							Target: "192.168.0.101",
 							Port:   12345,
 							Tags:   map[string]string{mesh_proto.ServiceTag: "redis", "version": "v2", mesh_proto.RegionTag: "eu"},
+							Locality: &core_xds.Locality{
+								Region: "eu",
+							},
 							Weight: 2,
 						},
 						{
@@ -445,6 +462,9 @@ var _ = Describe("TrafficRoute", func() {
 							Target: "192.168.0.100",
 							Port:   12345,
 							Tags:   map[string]string{mesh_proto.ServiceTag: "redis", "version": "v2", mesh_proto.RegionTag: "eu"},
+							Locality: &core_xds.Locality{
+								Region: "eu",
+							},
 							Weight: 2,
 						},
 						{
diff --git a/test/e2e/healthcheck/healthcheck_hybrid_test.go b/test/e2e/healthcheck/healthcheck_hybrid_test.go
index 7364822bb..ffb76a024 100644
--- a/test/e2e/healthcheck/healthcheck_hybrid_test.go
+++ b/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
diff --git a/test/framework/env.go b/test/framework/env.go
index 8bf9732d9..7409cd207 100644
--- a/test/framework/env.go
+++ b/test/framework/env.go
@@ -36,7 +36,7 @@ func GetGlobalImageRegistry() string {
 }
 
 func HasGlobalImageRegistry() bool {
-	return envBool("KUMA_GLOBAL_IMAGE_REGISTRY")
+	return envIsPresent("KUMA_GLOBAL_IMAGE_REGISTRY")
 }
 
 func GetGlobalImageTag() string {
@@ -44,7 +44,7 @@ func GetGlobalImageTag() string {
 }
 
 func HasGlobalImageTag() bool {
-	return envBool("KUMA_GLOBAL_IMAGE_TAG")
+	return envIsPresent("KUMA_GLOBAL_IMAGE_TAG")
 }
 
 func GetCpImageRegistry() string {
@@ -52,7 +52,7 @@ func GetCpImageRegistry() string {
 }
 
 func HasCpImageRegistry() bool {
-	return envBool("KUMA_CP_IMAGE_REPOSITORY")
+	return envIsPresent("KUMA_CP_IMAGE_REPOSITORY")
 }
 
 func GetDpImageRegistry() string {
@@ -60,7 +60,7 @@ func GetDpImageRegistry() string {
 }
 
 func HasDpImageRegistry() bool {
-	return envBool("KUMA_DP_IMAGE_REPOSITORY")
+	return envIsPresent("KUMA_DP_IMAGE_REPOSITORY")
 }
 
 func GetDpInitImageRegistry() string {
@@ -68,7 +68,7 @@ func GetDpInitImageRegistry() string {
 }
 
 func HasDpInitImageRegistry() bool {
-	return envBool("KUMA_DP_INIT_IMAGE_REPOSITORY")
+	return envIsPresent("KUMA_DP_INIT_IMAGE_REPOSITORY")
 }
 
 func GetUniversalImage() string {
@@ -84,7 +84,7 @@ func GetApiVersion() string {
 }
 
 func HasApiVersion() bool {
-	return envBool(envAPIVersion)
+	return envIsPresent(envAPIVersion)
 }
 
 func GetHelmChartPath() string {
@@ -92,7 +92,7 @@ func GetHelmChartPath() string {
 }
 
 func HasHelmChartPath() bool {
-	return envBool("HELM_CHART_PATH")
+	return envIsPresent("HELM_CHART_PATH")
 }
 
 func GetCniConfName() string {
@@ -100,7 +100,7 @@ func GetCniConfName() string {
 }
 
 func HasCniConfName() bool {
-	return envBool("KUMA_CNI_CONF_NAME")
+	return envIsPresent("KUMA_CNI_CONF_NAME")
 }
 
 func UseLoadBalancer() bool {
@@ -120,8 +120,13 @@ func GetKumactlBin() string {
 }
 
 func IsK8sClustersStarted() bool {
-	_, found := os.LookupEnv(envK8SCLUSTERS)
-	return found
+	return envIsPresent(envK8SCLUSTERS)
+}
+
+func envIsPresent(env string) bool {
+	_, ok := os.LookupEnv(env)
+
+	return ok
 }
 
 func envBool(env string) bool {