vault: Initial wan fed support (tls and gossip only)

acceptance/framework/consul/cli_cluster.go

@@ -74,8 +74,8 @@ func NewCLICluster(
 	require.NoError(t, err)
 
 	// Merge all helm values
-	MergeMaps(values, valuesFromConfig)
-	MergeMaps(values, helmValues)
+	helpers.MergeMaps(values, valuesFromConfig)
+	helpers.MergeMaps(values, helmValues)
 
 	logger := terratestLogger.New(logger.TestLogger{})
 
@@ -144,7 +144,7 @@ func (h *CLICluster) Create(t *testing.T) {
 	}
 	require.NoError(t, err)
 
-	helpers.WaitForAllPodsToBeReady(t, h.kubernetesClient, consulNS, fmt.Sprintf("release=%s", h.releaseName))
+	k8s.WaitForAllPodsToBeReady(t, h.kubernetesClient, consulNS, fmt.Sprintf("release=%s", h.releaseName))
 }
 
 func (h *CLICluster) Destroy(t *testing.T) {
@@ -179,9 +179,9 @@ func (h *CLICluster) Destroy(t *testing.T) {
 func (h *CLICluster) Upgrade(t *testing.T, helmValues map[string]string) {
 	t.Helper()
 
-	MergeMaps(h.helmOptions.SetValues, helmValues)
+	helpers.MergeMaps(h.helmOptions.SetValues, helmValues)
 	helm.Upgrade(t, h.helmOptions, config.HelmChartPath, h.releaseName)
-	helpers.WaitForAllPodsToBeReady(t, h.kubernetesClient, h.helmOptions.KubectlOptions.Namespace, fmt.Sprintf("release=%s", h.releaseName))
+	k8s.WaitForAllPodsToBeReady(t, h.kubernetesClient, h.helmOptions.KubectlOptions.Namespace, fmt.Sprintf("release=%s", h.releaseName))
 }
 
 func (h *CLICluster) SetupConsulClient(t *testing.T, secure bool) *api.Client {

acceptance/framework/consul/consul_cluster.go

@@ -75,8 +75,8 @@ func NewHelmCluster(
 	require.NoError(t, err)
 
 	// Merge all helm values
-	MergeMaps(values, valuesFromConfig)
-	MergeMaps(values, helmValues)
+	helpers.MergeMaps(values, valuesFromConfig)
+	helpers.MergeMaps(values, helmValues)
 
 	logger := terratestLogger.New(logger.TestLogger{})
 
@@ -118,7 +118,7 @@ func (h *HelmCluster) Create(t *testing.T) {
 
 	helm.Install(t, h.helmOptions, config.HelmChartPath, h.releaseName)
 
-	helpers.WaitForAllPodsToBeReady(t, h.kubernetesClient, h.helmOptions.KubectlOptions.Namespace, fmt.Sprintf("release=%s", h.releaseName))
+	k8s.WaitForAllPodsToBeReady(t, h.kubernetesClient, h.helmOptions.KubectlOptions.Namespace, fmt.Sprintf("release=%s", h.releaseName))
 }
 
 func (h *HelmCluster) Destroy(t *testing.T) {
@@ -213,9 +213,9 @@ func (h *HelmCluster) Destroy(t *testing.T) {
 func (h *HelmCluster) Upgrade(t *testing.T, helmValues map[string]string) {
 	t.Helper()
 
-	MergeMaps(h.helmOptions.SetValues, helmValues)
+	helpers.MergeMaps(h.helmOptions.SetValues, helmValues)
 	helm.Upgrade(t, h.helmOptions, config.HelmChartPath, h.releaseName)
-	helpers.WaitForAllPodsToBeReady(t, h.kubernetesClient, h.helmOptions.KubectlOptions.Namespace, fmt.Sprintf("release=%s", h.releaseName))
+	k8s.WaitForAllPodsToBeReady(t, h.kubernetesClient, h.helmOptions.KubectlOptions.Namespace, fmt.Sprintf("release=%s", h.releaseName))
 }
 
 func (h *HelmCluster) SetupConsulClient(t *testing.T, secure bool) *api.Client {
@@ -472,11 +472,3 @@ func defaultValues() map[string]string {
 	}
 	return values
 }
-
-// MergeMaps will merge the values in b with values in a and save in a.
-// If there are conflicts, the values in b will overwrite the values in a.
-func MergeMaps(a, b map[string]string) {
-	for k, v := range b {
-		a[k] = v
-	}
-}

acceptance/framework/environment/environment.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/gruntwork-io/terratest/modules/k8s"
 	"github.com/hashicorp/consul-k8s/acceptance/framework/config"
-	"github.com/hashicorp/consul-k8s/acceptance/framework/helpers"
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
@@ -87,6 +86,27 @@ type kubernetesContext struct {
 	options *k8s.KubectlOptions
 }
 
+// KubernetesContextFromOptions returns the Kubernetes context from options.
+// If context is explicitly set in options, it returns that context.
+// Otherwise, it returns the current context.
+func KubernetesContextFromOptions(t *testing.T, options *k8s.KubectlOptions) string {
+	t.Helper()
+
+	// First, check if context set in options and return that
+	if options.ContextName != "" {
+		return options.ContextName
+	}
+
+	// Otherwise, get current context from config
+	configPath, err := options.GetConfigPath(t)
+	require.NoError(t, err)
+
+	rawConfig, err := k8s.LoadConfigFromPath(configPath).RawConfig()
+	require.NoError(t, err)
+
+	return rawConfig.CurrentContext
+}
+
 func (k kubernetesContext) KubectlOptions(t *testing.T) *k8s.KubectlOptions {
 	if k.options != nil {
 		return k.options
@@ -107,7 +127,7 @@ func (k kubernetesContext) KubectlOptions(t *testing.T) *k8s.KubectlOptions {
 		rawConfig, err := k8s.LoadConfigFromPath(configPath).RawConfig()
 		require.NoError(t, err)
 
-		contextName := helpers.KubernetesContextFromOptions(t, k.options)
+		contextName := KubernetesContextFromOptions(t, k.options)
 		if rawConfig.Contexts[contextName].Namespace != "" {
 			k.options.Namespace = rawConfig.Contexts[contextName].Namespace
 		} else {
@@ -117,12 +137,26 @@ func (k kubernetesContext) KubectlOptions(t *testing.T) *k8s.KubectlOptions {
 	return k.options
 }
 
+// KubernetesClientFromOptions takes KubectlOptions and returns Kubernetes API client.
+func KubernetesClientFromOptions(t *testing.T, options *k8s.KubectlOptions) kubernetes.Interface {
+	configPath, err := options.GetConfigPath(t)
+	require.NoError(t, err)
+
+	config, err := k8s.LoadApiClientConfigE(configPath, options.ContextName)
+	require.NoError(t, err)
+
+	client, err := kubernetes.NewForConfig(config)
+	require.NoError(t, err)
+
+	return client
+}
+
 func (k kubernetesContext) KubernetesClient(t *testing.T) kubernetes.Interface {
 	if k.client != nil {
 		return k.client
 	}
 
-	k.client = helpers.KubernetesClientFromOptions(t, k.KubectlOptions(t))
+	k.client = KubernetesClientFromOptions(t, k.KubectlOptions(t))
 
 	return k.client
 }

acceptance/framework/helpers/helpers.go

@@ -12,13 +12,12 @@ import (
 	"time"
 
 	"github.com/gruntwork-io/terratest/modules/helm"
+	"github.com/hashicorp/consul/api"
 
-	terratestk8s "github.com/gruntwork-io/terratest/modules/k8s"
 	"github.com/gruntwork-io/terratest/modules/random"
 	"github.com/hashicorp/consul-k8s/acceptance/framework/logger"
 	"github.com/hashicorp/consul/sdk/testutil/retry"
 	"github.com/stretchr/testify/require"
-	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
@@ -70,37 +69,6 @@ func CheckForPriorInstallations(t *testing.T, client kubernetes.Interface, optio
 	})
 }
 
-// WaitForAllPodsToBeReady waits until all pods with the provided podLabelSelector
-// are in the ready status. It checks every 5 seconds for a total of 20 tries.
-// If there is at least one container in a pod that isn't ready after that,
-// it fails the test.
-func WaitForAllPodsToBeReady(t *testing.T, client kubernetes.Interface, namespace, podLabelSelector string) {
-	t.Helper()
-
-	logger.Logf(t, "Waiting for pods with label %q to be ready.", podLabelSelector)
-
-	// Wait up to 10m.
-	// On Azure, volume provisioning can sometimes take close to 5 min,
-	// so we need to give a bit more time for pods to become healthy.
-	counter := &retry.Counter{Count: 600, Wait: 1 * time.Second}
-	retry.RunWith(counter, t, func(r *retry.R) {
-		pods, err := client.CoreV1().Pods(namespace).List(context.Background(), metav1.ListOptions{LabelSelector: podLabelSelector})
-		require.NoError(r, err)
-		require.NotEmpty(r, pods.Items)
-
-		var notReadyPods []string
-		for _, pod := range pods.Items {
-			if !IsReady(pod) {
-				notReadyPods = append(notReadyPods, pod.Name)
-			}
-		}
-		if len(notReadyPods) > 0 {
-			r.Errorf("%d pods are not ready: %s", len(notReadyPods), strings.Join(notReadyPods, ","))
-		}
-	})
-	logger.Log(t, "Finished waiting for pods to be ready.")
-}
-
 // SetupInterruptHandler sets up a goroutine that will wait for interrupt signals
 // and call cleanup function when it catches it.
 func SetupInterruptHandler(cleanup func()) {
@@ -138,69 +106,45 @@ func Cleanup(t *testing.T, noCleanupOnFailure bool, cleanup func()) {
 	t.Cleanup(wrappedCleanupFunc)
 }
 
-// KubernetesClientFromOptions takes KubectlOptions and returns Kubernetes API client.
-func KubernetesClientFromOptions(t *testing.T, options *terratestk8s.KubectlOptions) kubernetes.Interface {
-	configPath, err := options.GetConfigPath(t)
-	require.NoError(t, err)
-
-	config, err := terratestk8s.LoadApiClientConfigE(configPath, options.ContextName)
-	require.NoError(t, err)
-
-	client, err := kubernetes.NewForConfig(config)
-	require.NoError(t, err)
-
-	return client
-}
-
-// KubernetesContextFromOptions returns the Kubernetes context from options.
-// If context is explicitly set in options, it returns that context.
-// Otherwise, it returns the current context.
-func KubernetesContextFromOptions(t *testing.T, options *terratestk8s.KubectlOptions) string {
-	t.Helper()
-
-	// First, check if context set in options and return that
-	if options.ContextName != "" {
-		return options.ContextName
-	}
-
-	// Otherwise, get current context from config
-	configPath, err := options.GetConfigPath(t)
-	require.NoError(t, err)
-
-	rawConfig, err := terratestk8s.LoadConfigFromPath(configPath).RawConfig()
-	require.NoError(t, err)
-
-	return rawConfig.CurrentContext
-}
-
-// KubernetesAPIServerHostFromOptions returns the Kubernetes API server host from options.
-func KubernetesAPIServerHostFromOptions(t *testing.T, options *terratestk8s.KubectlOptions) string {
-	t.Helper()
+// VerifyFederation checks that the WAN federation between servers is successful
+// by first checking members are alive from the perspective of both servers.
+// If secure is true, it will also check that the ACL replication is running on the secondary server.
+func VerifyFederation(t *testing.T, primaryClient, secondaryClient *api.Client, releaseName string, secure bool) {
+	retrier := &retry.Timer{Timeout: 5 * time.Minute, Wait: 1 * time.Second}
+	start := time.Now()
+
+	// Check that server in dc1 is healthy from the perspective of the server in dc2, and vice versa.
+	// We're calling the Consul health API, as opposed to checking serf membership status,
+	// because we need to make sure that the federated servers can make API calls and forward requests
+	// from one server to another. From running tests in CI for a while and using serf membership status before,
+	// we've noticed that the status could be "alive" as soon as the server in the secondary cluster joins the primary
+	// and then switch to "failed". This would require us to check that the status is "alive" is showing consistently for
+	// some amount of time, which could be quite flakey. Calling the API in another datacenter allows us to check that
+	// each server can forward calls to another, which is what we need for connect.
+	retry.RunWith(retrier, t, func(r *retry.R) {
+		secondaryServerHealth, _, err := primaryClient.Health().Node(fmt.Sprintf("%s-consul-server-0", releaseName), &api.QueryOptions{Datacenter: "dc2"})
+		require.NoError(r, err)
+		require.Equal(r, secondaryServerHealth.AggregatedStatus(), api.HealthPassing)
 
-	configPath, err := options.GetConfigPath(t)
-	require.NoError(t, err)
+		primaryServerHealth, _, err := secondaryClient.Health().Node(fmt.Sprintf("%s-consul-server-0", releaseName), &api.QueryOptions{Datacenter: "dc1"})
+		require.NoError(r, err)
+		require.Equal(r, primaryServerHealth.AggregatedStatus(), api.HealthPassing)
 
-	config, err := terratestk8s.LoadApiClientConfigE(configPath, options.ContextName)
-	require.NoError(t, err)
+		if secure {
+			replicationStatus, _, err := secondaryClient.ACL().Replication(nil)
+			require.NoError(r, err)
+			require.True(r, replicationStatus.Enabled)
+			require.True(r, replicationStatus.Running)
+		}
+	})
 
-	return config.Host
+	logger.Logf(t, "Took %s to verify federation", time.Since(start))
 }
 
-// IsReady returns true if pod is ready.
-func IsReady(pod corev1.Pod) bool {
-	if pod.Status.Phase == corev1.PodPending {
-		return false
-	}
-
-	for _, cond := range pod.Status.Conditions {
-		if cond.Type == corev1.PodReady {
-			if cond.Status == corev1.ConditionTrue {
-				return true
-			} else {
-				return false
-			}
-		}
+// MergeMaps will merge the values in b with values in a and save in a.
+// If there are conflicts, the values in b will overwrite the values in a.
+func MergeMaps(a, b map[string]string) {
+	for k, v := range b {
+		a[k] = v
 	}
-
-	return false
 }

acceptance/framework/k8s/debug.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/gruntwork-io/terratest/modules/k8s"
 	terratestLogger "github.com/gruntwork-io/terratest/modules/logger"
-	"github.com/hashicorp/consul-k8s/acceptance/framework/helpers"
+	"github.com/hashicorp/consul-k8s/acceptance/framework/environment"
 	"github.com/hashicorp/consul-k8s/acceptance/framework/logger"
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -23,9 +23,9 @@ func WritePodsDebugInfoIfFailed(t *testing.T, kubectlOptions *k8s.KubectlOptions
 
 	if t.Failed() {
 		// Create k8s client from kubectl options.
-		client := helpers.KubernetesClientFromOptions(t, kubectlOptions)
+		client := environment.KubernetesClientFromOptions(t, kubectlOptions)
 
-		contextName := helpers.KubernetesContextFromOptions(t, kubectlOptions)
+		contextName := environment.KubernetesContextFromOptions(t, kubectlOptions)
 
 		// Create a directory for the test.
 		testDebugDirectory := filepath.Join(debugDirectory, t.Name(), contextName)