Backport of Proxy Lifecycle helm, connect-inject and acceptance tests into release/1.0.x (#2233)

.changelog/2233.txt

@@ -0,0 +1,3 @@
+```release-note:feature
+Add support for configuring graceful shutdown proxy lifecycle management settings.
+```

acceptance/framework/config/config.go

@@ -43,11 +43,13 @@ type TestConfig struct {
 
 	DisablePeering bool
 
-	HelmChartVersion string
-	ConsulImage      string
-	ConsulK8SImage   string
-	ConsulVersion    *version.Version
-	EnvoyImage       string
+	HelmChartVersion       string
+	ConsulImage            string
+	ConsulK8SImage         string
+	ConsulDataplaneImage   string
+	ConsulVersion          *version.Version
+	ConsulDataplaneVersion *version.Version
+	EnvoyImage             string
 
 	NoCleanupOnFailure bool
 	DebugDirectory     string
@@ -101,6 +103,7 @@ func (t *TestConfig) HelmValuesFromConfig() (map[string]string, error) {
 	setIfNotEmpty(helmValues, "global.image", t.ConsulImage)
 	setIfNotEmpty(helmValues, "global.imageK8S", t.ConsulK8SImage)
 	setIfNotEmpty(helmValues, "global.imageEnvoy", t.EnvoyImage)
+	setIfNotEmpty(helmValues, "global.imageConsulDataplane", t.ConsulDataplaneImage)
 
 	return helmValues, nil
 }

acceptance/framework/connhelper/connect_helper.go

@@ -47,8 +47,8 @@ type ConnectHelper struct {
 	// consulCluster is the cluster to use for the test.
 	consulCluster consul.Cluster
 
-	// consulClient is the client used to test service mesh connectivity.
-	consulClient *api.Client
+	// ConsulClient is the client used to test service mesh connectivity.
+	ConsulClient *api.Client
 }
 
 // Setup creates a new cluster using the New*Cluster function and assigns it
@@ -66,14 +66,14 @@ func (c *ConnectHelper) Setup(t *testing.T) {
 func (c *ConnectHelper) Install(t *testing.T) {
 	logger.Log(t, "Installing Consul cluster")
 	c.consulCluster.Create(t)
-	c.consulClient, _ = c.consulCluster.SetupConsulClient(t, c.Secure)
+	c.ConsulClient, _ = c.consulCluster.SetupConsulClient(t, c.Secure)
 }
 
 // Upgrade uses the existing Consul cluster and upgrades it using Helm values
 // set by the Secure, AutoEncrypt, and HelmValues fields.
 func (c *ConnectHelper) Upgrade(t *testing.T) {
 	require.NotNil(t, c.consulCluster, "consulCluster must be set before calling Upgrade (Call Install first).")
-	require.NotNil(t, c.consulClient, "consulClient must be set before calling Upgrade (Call Install first).")
+	require.NotNil(t, c.ConsulClient, "ConsulClient must be set before calling Upgrade (Call Install first).")
 
 	logger.Log(t, "upgrading Consul cluster")
 	c.consulCluster.Upgrade(t, c.helmValues())
@@ -93,7 +93,7 @@ func (c *ConnectHelper) DeployClientAndServer(t *testing.T) {
 		t.Cleanup(func() {
 			retrier := &retry.Timer{Timeout: 30 * time.Second, Wait: 100 * time.Millisecond}
 			retry.RunWith(retrier, t, func(r *retry.R) {
-				tokens, _, err := c.consulClient.ACL().TokenList(nil)
+				tokens, _, err := c.ConsulClient.ACL().TokenList(nil)
 				require.NoError(r, err)
 				for _, token := range tokens {
 					require.NotContains(r, token.Description, StaticServerName)
@@ -139,7 +139,7 @@ func (c *ConnectHelper) TestConnectionFailureWithoutIntention(t *testing.T) {
 // the static-client pod.
 func (c *ConnectHelper) CreateIntention(t *testing.T) {
 	logger.Log(t, "creating intention")
-	_, _, err := c.consulClient.ConfigEntries().Set(&api.ServiceIntentionsConfigEntry{
+	_, _, err := c.ConsulClient.ConfigEntries().Set(&api.ServiceIntentionsConfigEntry{
 		Kind: api.ServiceIntentions,
 		Name: StaticServerName,
 		Sources: []*api.SourceIntention{

acceptance/framework/flags/flags.go

@@ -31,11 +31,13 @@ type TestFlags struct {
 
 	flagEnableTransparentProxy bool
 
-	flagHelmChartVersion string
-	flagConsulImage      string
-	flagConsulK8sImage   string
-	flagConsulVersion    string
-	flagEnvoyImage       string
+	flagHelmChartVersion       string
+	flagConsulImage            string
+	flagConsulK8sImage         string
+	flagConsulDataplaneImage   string
+	flagConsulVersion          string
+	flagConsulDataplaneVersion string
+	flagEnvoyImage             string
 
 	flagNoCleanupOnFailure bool
 
@@ -67,7 +69,9 @@ func (t *TestFlags) init() {
 
 	flag.StringVar(&t.flagConsulImage, "consul-image", "", "The Consul image to use for all tests.")
 	flag.StringVar(&t.flagConsulK8sImage, "consul-k8s-image", "", "The consul-k8s image to use for all tests.")
+	flag.StringVar(&t.flagConsulDataplaneImage, "consul-dataplane-image", "", "The consul-dataplane image to use for all tests.")
 	flag.StringVar(&t.flagConsulVersion, "consul-version", "", "The consul version used for all tests.")
+	flag.StringVar(&t.flagConsulDataplaneVersion, "consul-dataplane-version", "", "The consul-dataplane version used for all tests.")
 	flag.StringVar(&t.flagHelmChartVersion, "helm-chart-version", config.HelmChartPath, "The helm chart used for all tests.")
 	flag.StringVar(&t.flagEnvoyImage, "envoy-image", "", "The Envoy image to use for all tests.")
 
@@ -142,6 +146,8 @@ func (t *TestFlags) TestConfigFromFlags() *config.TestConfig {
 
 	// if the Version is empty consulVersion will be nil
 	consulVersion, _ := version.NewVersion(t.flagConsulVersion)
+	consulDataplaneVersion, _ := version.NewVersion(t.flagConsulDataplaneVersion)
+	//vaultserverVersion, _ := version.NewVersion(t.flagVaultServerVersion)
 
 	return &config.TestConfig{
 		Kubeconfig:    t.flagKubeconfig,
@@ -166,11 +172,13 @@ func (t *TestFlags) TestConfigFromFlags() *config.TestConfig {
 
 		DisablePeering: t.flagDisablePeering,
 
-		HelmChartVersion: t.flagHelmChartVersion,
-		ConsulImage:      t.flagConsulImage,
-		ConsulK8SImage:   t.flagConsulK8sImage,
-		ConsulVersion:    consulVersion,
-		EnvoyImage:       t.flagEnvoyImage,
+		HelmChartVersion:       t.flagHelmChartVersion,
+		ConsulImage:            t.flagConsulImage,
+		ConsulK8SImage:         t.flagConsulK8sImage,
+		ConsulDataplaneImage:   t.flagConsulDataplaneImage,
+		ConsulVersion:          consulVersion,
+		ConsulDataplaneVersion: consulDataplaneVersion,
+		EnvoyImage:             t.flagEnvoyImage,
 
 		NoCleanupOnFailure: t.flagNoCleanupOnFailure,
 		DebugDirectory:     tempDir,

acceptance/tests/connect/connect_proxy_lifecycle_test.go

@@ -0,0 +1,214 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package connect
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/consul-k8s/acceptance/framework/connhelper"
+	"github.com/hashicorp/consul-k8s/acceptance/framework/consul"
+	"github.com/hashicorp/consul-k8s/acceptance/framework/helpers"
+	"github.com/hashicorp/consul-k8s/acceptance/framework/k8s"
+	"github.com/hashicorp/consul-k8s/acceptance/framework/logger"
+	"github.com/hashicorp/consul/sdk/testutil/retry"
+	"github.com/hashicorp/go-version"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type LifecycleShutdownConfig struct {
+	secure     bool
+	helmValues map[string]string
+}
+
+const (
+	helmDrainListenersKey     = "connectInject.sidecarProxy.lifecycle.defaultEnableShutdownDrainListeners"
+	helmGracePeriodSecondsKey = "connectInject.sidecarProxy.lifecycle.defaultShutdownGracePeriodSeconds"
+)
+
+// Test the endpoints controller cleans up force-killed pods.
+func TestConnectInject_ProxyLifecycleShutdown(t *testing.T) {
+	cfg := suite.Config()
+
+	ver, err := version.NewVersion("1.2.0")
+	require.NoError(t, err)
+	if cfg.ConsulDataplaneVersion != nil && cfg.ConsulDataplaneVersion.LessThan(ver) {
+		t.Skipf("skipping this test because proxy lifecycle management is not supported in consul-dataplane version %v", cfg.ConsulDataplaneVersion.String())
+	}
+
+	for _, testCfg := range []LifecycleShutdownConfig{
+		{secure: false, helmValues: map[string]string{}},
+		{secure: true, helmValues: map[string]string{}},
+		{secure: false, helmValues: map[string]string{
+			helmDrainListenersKey:     "true",
+			helmGracePeriodSecondsKey: "15",
+		}},
+		{secure: true, helmValues: map[string]string{
+			helmDrainListenersKey:     "true",
+			helmGracePeriodSecondsKey: "15",
+		}},
+		{secure: false, helmValues: map[string]string{
+			helmDrainListenersKey:     "false",
+			helmGracePeriodSecondsKey: "15",
+		}},
+		{secure: true, helmValues: map[string]string{
+			helmDrainListenersKey:     "false",
+			helmGracePeriodSecondsKey: "15",
+		}},
+		{secure: false, helmValues: map[string]string{
+			helmDrainListenersKey:     "false",
+			helmGracePeriodSecondsKey: "0",
+		}},
+		{secure: true, helmValues: map[string]string{
+			helmDrainListenersKey:     "false",
+			helmGracePeriodSecondsKey: "0",
+		}},
+	} {
+		// Determine if listeners should be expected to drain inbound connections
+		var drainListenersEnabled bool
+		val, ok := testCfg.helmValues[helmDrainListenersKey]
+		if ok {
+			drainListenersEnabled, err = strconv.ParseBool(val)
+			require.NoError(t, err)
+		}
+
+		// Determine expected shutdown grace period
+		var gracePeriodSeconds int64
+		val, ok = testCfg.helmValues[helmGracePeriodSecondsKey]
+		if ok {
+			gracePeriodSeconds, err = strconv.ParseInt(val, 10, 64)
+			require.NoError(t, err)
+		} else {
+			// Half of the helm default to speed tests up
+			gracePeriodSeconds = 15
+		}
+
+		name := fmt.Sprintf("secure: %t, drainListeners: %t, gracePeriodSeconds: %d", testCfg.secure, drainListenersEnabled, gracePeriodSeconds)
+		t.Run(name, func(t *testing.T) {
+			ctx := suite.Environment().DefaultContext(t)
+			releaseName := helpers.RandomName()
+
+			connHelper := connhelper.ConnectHelper{
+				ClusterKind: consul.Helm,
+				Secure:      testCfg.secure,
+				ReleaseName: releaseName,
+				Ctx:         ctx,
+				Cfg:         cfg,
+				HelmValues:  testCfg.helmValues,
+			}
+
+			connHelper.Setup(t)
+			connHelper.Install(t)
+			connHelper.DeployClientAndServer(t)
+
+			// TODO: should this move into connhelper.DeployClientAndServer?
+			logger.Log(t, "waiting for static-client and static-server to be registered with Consul")
+			retry.Run(t, func(r *retry.R) {
+				for _, name := range []string{
+					"static-client",
+					"static-client-sidecar-proxy",
+					"static-server",
+					"static-server-sidecar-proxy",
+				} {
+					logger.Logf(t, "checking for %s service in Consul catalog", name)
+					instances, _, err := connHelper.ConsulClient.Catalog().Service(name, "", nil)
+					r.Check(err)
+
+					if len(instances) != 1 {
+						r.Errorf("expected 1 instance of %s", name)
+					}
+				}
+			})
+
+			if testCfg.secure {
+				connHelper.TestConnectionFailureWithoutIntention(t)
+				connHelper.CreateIntention(t)
+			}
+
+			connHelper.TestConnectionSuccess(t)
+
+			// Get static-client pod name
+			ns := ctx.KubectlOptions(t).Namespace
+			pods, err := ctx.KubernetesClient(t).CoreV1().Pods(ns).List(
+				context.Background(),
+				metav1.ListOptions{
+					LabelSelector: "app=static-client",
+				},
+			)
+			require.NoError(t, err)
+			require.Len(t, pods.Items, 1)
+			clientPodName := pods.Items[0].Name
+
+			var terminationGracePeriod int64 = 60
+			logger.Logf(t, "killing the %q pod with %dseconds termination grace period", clientPodName, terminationGracePeriod)
+			err = ctx.KubernetesClient(t).CoreV1().Pods(ns).Delete(context.Background(), clientPodName, metav1.DeleteOptions{GracePeriodSeconds: &terminationGracePeriod})
+			require.NoError(t, err)
+
+			// Exec into terminating pod, not just any static-client pod
+			args := []string{"exec", clientPodName, "-c", connhelper.StaticClientName, "--", "curl", "-vvvsSf"}
+
+			if cfg.EnableTransparentProxy {
+				args = append(args, "http://static-server")
+			} else {
+				args = append(args, "http://localhost:1234")
+			}
+
+			if gracePeriodSeconds > 0 {
+				// Ensure outbound requests are still successful during grace
+				// period.
+				retry.RunWith(&retry.Timer{Timeout: time.Duration(gracePeriodSeconds) * time.Second, Wait: 2 * time.Second}, t, func(r *retry.R) {
+					output, err := k8s.RunKubectlAndGetOutputE(t, ctx.KubectlOptions(t), args...)
+					require.NoError(r, err)
+					require.Condition(r, func() bool {
+						exists := false
+						if strings.Contains(output, "curl: (7) Failed to connect") {
+							exists = true
+						}
+						return !exists
+					})
+				})
+
+				// If listener draining is enabled, ensure inbound
+				// requests are rejected during grace period.
+				// connHelper.TestConnectionSuccess(t)
+			} else {
+				// Ensure outbound requests fail because proxy has terminated
+				retry.RunWith(&retry.Timer{Timeout: time.Duration(terminationGracePeriod) * time.Second, Wait: 2 * time.Second}, t, func(r *retry.R) {
+					output, err := k8s.RunKubectlAndGetOutputE(t, ctx.KubectlOptions(t), args...)
+					require.Error(r, err)
+					require.Condition(r, func() bool {
+						exists := false
+						if strings.Contains(output, "curl: (7) Failed to connect") {
+							exists = true
+						}
+						return exists
+					})
+				})
+			}
+
+			logger.Log(t, "ensuring pod is deregistered after termination")
+			retry.Run(t, func(r *retry.R) {
+				for _, name := range []string{
+					"static-client",
+					"static-client-sidecar-proxy",
+				} {
+					logger.Logf(t, "checking for %s service in Consul catalog", name)
+					instances, _, err := connHelper.ConsulClient.Catalog().Service(name, "", nil)
+					r.Check(err)
+
+					for _, instance := range instances {
+						if strings.Contains(instance.ServiceID, clientPodName) {
+							r.Errorf("%s is still registered", instance.ServiceID)
+						}
+					}
+				}
+			})
+		})
+	}
+}

charts/consul/templates/connect-inject-deployment.yaml

@@ -234,6 +234,19 @@ spec:
                 -default-sidecar-proxy-cpu-request={{ $resources.requests.cpu }} \
                 {{- end }}
                 -default-envoy-proxy-concurrency={{ .Values.connectInject.sidecarProxy.concurrency }} \
+                {{- if .Values.connectInject.sidecarProxy.lifecycle.defaultEnabled }}
+                -default-enable-sidecar-proxy-lifecycle=true \
+                {{- else }}
+                -default-enable-sidecar-proxy-lifecycle=false \
+                {{- end }}
+                {{- if .Values.connectInject.sidecarProxy.lifecycle.defaultEnableShutdownDrainListeners }}
+                -default-enable-sidecar-proxy-lifecycle-shutdown-drain-listeners=true \
+                {{- else }}
+                -default-enable-sidecar-proxy-lifecycle-shutdown-drain-listeners=false \
+                {{- end }}
+                -default-sidecar-proxy-lifecycle-shutdown-grace-period-seconds={{ .Values.connectInject.sidecarProxy.lifecycle.defaultShutdownGracePeriodSeconds }} \
+                -default-sidecar-proxy-lifecycle-graceful-port={{ .Values.connectInject.sidecarProxy.lifecycle.defaultGracefulPort }} \
+                -default-sidecar-proxy-lifecycle-graceful-shutdown-path="{{ .Values.connectInject.sidecarProxy.lifecycle.defaultGracefulShutdownPath }}" \
 
                 {{- if .Values.connectInject.initContainer }}
                 {{- $initResources := .Values.connectInject.initContainer.resources }}