From 7e236479c060839630cda24d6d1b93b33cf95450 Mon Sep 17 00:00:00 2001 From: cskh Date: Tue, 14 Feb 2023 21:10:26 -0500 Subject: [PATCH] upgrade test: fix flaky peering through mesh gateway --- .../consul-container/libs/assert/envoy.go | 2 +- .../consul-container/libs/service/connect.go | 7 ++++ .../consul-container/libs/service/examples.go | 7 ++++ .../consul-container/libs/service/gateway.go | 7 ++++ .../consul-container/libs/service/service.go | 1 + .../libs/topology/peering_topology.go | 42 +++++++++++++++---- .../rotate_server_and_ca_then_fail_test.go | 2 +- .../upgrade/peering_control_plane_mgw_test.go | 26 ++++-------- .../test/upgrade/peering_http_test.go | 2 +- 9 files changed, 67 insertions(+), 29 deletions(-) diff --git a/test/integration/consul-container/libs/assert/envoy.go b/test/integration/consul-container/libs/assert/envoy.go index e62118c4f1d8..6713c4fb6490 100644 --- a/test/integration/consul-container/libs/assert/envoy.go +++ b/test/integration/consul-container/libs/assert/envoy.go @@ -127,7 +127,7 @@ func AssertEnvoyMetricAtLeast(t *testing.T, adminPort int, prefix, metric string err error ) failer := func() *retry.Timer { - return &retry.Timer{Timeout: 30 * time.Second, Wait: 500 * time.Millisecond} + return &retry.Timer{Timeout: 60 * time.Second, Wait: 500 * time.Millisecond} } retry.RunWith(failer(), t, func(r *retry.R) { diff --git a/test/integration/consul-container/libs/service/connect.go b/test/integration/consul-container/libs/service/connect.go index ac4907d4e583..b5a8087d2d67 100644 --- a/test/integration/consul-container/libs/service/connect.go +++ b/test/integration/consul-container/libs/service/connect.go @@ -109,6 +109,13 @@ func (g ConnectContainer) Start() error { return g.container.Start(g.ctx) } +func (g ConnectContainer) Stop() error { + if g.container == nil { + return fmt.Errorf("container has not been initialized") + } + return g.container.Stop(context.Background(), nil) +} + func (g ConnectContainer) Terminate() error { return cluster.TerminateContainer(g.ctx, g.container, true) } diff --git a/test/integration/consul-container/libs/service/examples.go b/test/integration/consul-container/libs/service/examples.go index 9d95f6e9099f..da075f5aec9c 100644 --- a/test/integration/consul-container/libs/service/examples.go +++ b/test/integration/consul-container/libs/service/examples.go @@ -101,6 +101,13 @@ func (g exampleContainer) Start() error { return g.container.Start(context.Background()) } +func (g exampleContainer) Stop() error { + if g.container == nil { + return fmt.Errorf("container has not been initialized") + } + return g.container.Stop(context.Background(), nil) +} + func (c exampleContainer) Terminate() error { return cluster.TerminateContainer(c.ctx, c.container, true) } diff --git a/test/integration/consul-container/libs/service/gateway.go b/test/integration/consul-container/libs/service/gateway.go index 5fb3a36184b4..70897fc7b099 100644 --- a/test/integration/consul-container/libs/service/gateway.go +++ b/test/integration/consul-container/libs/service/gateway.go @@ -86,6 +86,13 @@ func (g gatewayContainer) Start() error { return g.container.Start(context.Background()) } +func (g gatewayContainer) Stop() error { + if g.container == nil { + return fmt.Errorf("container has not been initialized") + } + return g.container.Stop(context.Background(), nil) +} + func (c gatewayContainer) Terminate() error { return cluster.TerminateContainer(c.ctx, c.container, true) } diff --git a/test/integration/consul-container/libs/service/service.go b/test/integration/consul-container/libs/service/service.go index 57a3539a6412..99da55822690 100644 --- a/test/integration/consul-container/libs/service/service.go +++ b/test/integration/consul-container/libs/service/service.go @@ -18,6 +18,7 @@ type Service interface { GetName() string GetServiceName() string Start() (err error) + Stop() (err error) Terminate() error Restart() error GetStatus() (string, error) diff --git a/test/integration/consul-container/libs/topology/peering_topology.go b/test/integration/consul-container/libs/topology/peering_topology.go index 1c764c45c53c..ba36978c72f4 100644 --- a/test/integration/consul-container/libs/topology/peering_topology.go +++ b/test/integration/consul-container/libs/topology/peering_topology.go @@ -41,6 +41,7 @@ type BuiltCluster struct { func BasicPeeringTwoClustersSetup( t *testing.T, consulVersion string, + peeringThroughMeshgateway bool, ) (*BuiltCluster, *BuiltCluster) { // acceptingCluster, acceptingCtx, acceptingClient := NewPeeringCluster(t, "dc1", 3, consulVersion, true) acceptingCluster, acceptingCtx, acceptingClient := NewPeeringCluster(t, 3, &libcluster.BuildOptions{ @@ -53,6 +54,38 @@ func BasicPeeringTwoClustersSetup( ConsulVersion: consulVersion, InjectAutoEncryption: true, }) + + // Create the mesh gateway for dataplane traffic and peering control plane traffic (if enabled) + acceptingClusterGateway, err := libservice.NewGatewayService(context.Background(), "mesh", "mesh", acceptingCluster.Clients()[0]) + require.NoError(t, err) + dialingClusterGateway, err := libservice.NewGatewayService(context.Background(), "mesh", "mesh", dialingCluster.Clients()[0]) + require.NoError(t, err) + + // Enable peering control plane traffic through mesh gateway + if peeringThroughMeshgateway { + req := &api.MeshConfigEntry{ + Peering: &api.PeeringMeshConfig{ + PeerThroughMeshGateways: true, + }, + } + configCluster := func(cli *api.Client) error { + libassert.CatalogServiceExists(t, cli, "mesh") + ok, _, err := cli.ConfigEntries().Set(req, &api.WriteOptions{}) + if !ok { + return fmt.Errorf("config entry is not set") + } + + if err != nil { + return fmt.Errorf("error writing config entry: %s", err) + } + return nil + } + err = configCluster(dialingClient) + require.NoError(t, err) + err = configCluster(acceptingClient) + require.NoError(t, err) + } + require.NoError(t, dialingCluster.PeerWithCluster(acceptingClient, AcceptingPeerName, DialingPeerName)) libassert.PeeringStatus(t, acceptingClient, AcceptingPeerName, api.PeeringStateActive) @@ -60,7 +93,6 @@ func BasicPeeringTwoClustersSetup( // Register an static-server service in acceptingCluster and export to dialing cluster var serverService, serverSidecarService libservice.Service - var acceptingClusterGateway libservice.Service { clientNode := acceptingCluster.Clients()[0] @@ -81,15 +113,10 @@ func BasicPeeringTwoClustersSetup( libassert.CatalogServiceExists(t, acceptingClient, "static-server-sidecar-proxy") require.NoError(t, serverService.Export("default", AcceptingPeerName, acceptingClient)) - - // Create the mesh gateway for dataplane traffic - acceptingClusterGateway, err = libservice.NewGatewayService(context.Background(), "mesh", "mesh", clientNode) - require.NoError(t, err) } // Register an static-client service in dialing cluster and set upstream to static-server service var clientSidecarService *libservice.ConnectContainer - var dialingClusterGateway libservice.Service { clientNode := dialingCluster.Clients()[0] @@ -100,9 +127,6 @@ func BasicPeeringTwoClustersSetup( libassert.CatalogServiceExists(t, dialingClient, "static-client-sidecar-proxy") - // Create the mesh gateway for dataplane traffic - dialingClusterGateway, err = libservice.NewGatewayService(context.Background(), "mesh", "mesh", clientNode) - require.NoError(t, err) } _, adminPort := clientSidecarService.GetAdminAddr() diff --git a/test/integration/consul-container/test/peering/rotate_server_and_ca_then_fail_test.go b/test/integration/consul-container/test/peering/rotate_server_and_ca_then_fail_test.go index 223effa449b2..bbac9cc03401 100644 --- a/test/integration/consul-container/test/peering/rotate_server_and_ca_then_fail_test.go +++ b/test/integration/consul-container/test/peering/rotate_server_and_ca_then_fail_test.go @@ -50,7 +50,7 @@ import ( func TestPeering_RotateServerAndCAThenFail_(t *testing.T) { t.Parallel() - accepting, dialing := libtopology.BasicPeeringTwoClustersSetup(t, utils.TargetVersion) + accepting, dialing := libtopology.BasicPeeringTwoClustersSetup(t, utils.TargetVersion, false) var ( acceptingCluster = accepting.Cluster dialingCluster = dialing.Cluster diff --git a/test/integration/consul-container/test/upgrade/peering_control_plane_mgw_test.go b/test/integration/consul-container/test/upgrade/peering_control_plane_mgw_test.go index f4112b6f6b83..5ccba9567739 100644 --- a/test/integration/consul-container/test/upgrade/peering_control_plane_mgw_test.go +++ b/test/integration/consul-container/test/upgrade/peering_control_plane_mgw_test.go @@ -42,7 +42,7 @@ func TestPeering_Upgrade_ControlPlane_MGW(t *testing.T) { } run := func(t *testing.T, tc testcase) { - accepting, dialing := libtopology.BasicPeeringTwoClustersSetup(t, tc.oldversion) + accepting, dialing := libtopology.BasicPeeringTwoClustersSetup(t, tc.oldversion, true) var ( acceptingCluster = accepting.Cluster dialingCluster = dialing.Cluster @@ -54,19 +54,6 @@ func TestPeering_Upgrade_ControlPlane_MGW(t *testing.T) { acceptingClient, err := acceptingCluster.GetClient(nil, false) require.NoError(t, err) - // Enable peering control plane traffic through mesh gateway - req := &api.MeshConfigEntry{ - Peering: &api.PeeringMeshConfig{ - PeerThroughMeshGateways: true, - }, - } - ok, _, err := dialingClient.ConfigEntries().Set(req, &api.WriteOptions{}) - require.True(t, ok) - require.NoError(t, err) - ok, _, err = acceptingClient.ConfigEntries().Set(req, &api.WriteOptions{}) - require.True(t, ok) - require.NoError(t, err) - // Verify control plane endpoints and traffic in gateway _, gatewayAdminPort := dialing.Gateway.GetAdminAddr() libassert.AssertUpstreamEndpointStatus(t, gatewayAdminPort, "server.dc1.peering", "HEALTHY", 1) @@ -74,6 +61,9 @@ func TestPeering_Upgrade_ControlPlane_MGW(t *testing.T) { libassert.AssertEnvoyMetricAtLeast(t, gatewayAdminPort, "cluster.static-server.default.default.accepting-to-dialer.external", "upstream_cx_total", 1) + libassert.AssertEnvoyMetricAtLeast(t, gatewayAdminPort, + "cluster.server.dc1.peering", + "upstream_cx_total", 1) // Upgrade the accepting cluster and assert peering is still ACTIVE require.NoError(t, acceptingCluster.StandardUpgrade(t, context.Background(), tc.targetVersion)) @@ -90,11 +80,12 @@ func TestPeering_Upgrade_ControlPlane_MGW(t *testing.T) { // - Register a new static-client service in dialing cluster and // - set upstream to static-server service in peered cluster - // Restart the gateway & proxy sidecar + // Stop the accepting gateway and restart dialing gateway + // to force peering control plane traffic through dialing mesh gateway + require.NoError(t, accepting.Gateway.Stop()) require.NoError(t, dialing.Gateway.Restart()) - require.NoError(t, dialing.Container.Restart()) - // Restarted gateway should not have any measurement on data plane traffic + // Restarted dialing gateway should not have any measurement on data plane traffic libassert.AssertEnvoyMetricAtMost(t, gatewayAdminPort, "cluster.static-server.default.default.accepting-to-dialer.external", "upstream_cx_total", 0) @@ -102,6 +93,7 @@ func TestPeering_Upgrade_ControlPlane_MGW(t *testing.T) { libassert.AssertEnvoyMetricAtLeast(t, gatewayAdminPort, "cluster.server.dc1.peering", "upstream_cx_total", 1) + require.NoError(t, accepting.Gateway.Start()) clientSidecarService, err := libservice.CreateAndRegisterStaticClientSidecar(dialingCluster.Servers()[0], libtopology.DialingPeerName, true) require.NoError(t, err) diff --git a/test/integration/consul-container/test/upgrade/peering_http_test.go b/test/integration/consul-container/test/upgrade/peering_http_test.go index aec03a3edb41..fe91f7653098 100644 --- a/test/integration/consul-container/test/upgrade/peering_http_test.go +++ b/test/integration/consul-container/test/upgrade/peering_http_test.go @@ -99,7 +99,7 @@ func TestPeering_UpgradeToTarget_fromLatest(t *testing.T) { } run := func(t *testing.T, tc testcase) { - accepting, dialing := libtopology.BasicPeeringTwoClustersSetup(t, tc.oldversion) + accepting, dialing := libtopology.BasicPeeringTwoClustersSetup(t, tc.oldversion, false) var ( acceptingCluster = accepting.Cluster dialingCluster = dialing.Cluster