wait for rqlite to be synced when enabling HA (#2067)

* wait for rqlite to be synced when enabling HA

* maybe fix integration test

* add timestamps to HA kind test logs, add logs to HA migration

* log ss replicas

* f

* remove rqlite sync check from go code

* use helm chart that waits for rqlite sync state

* use staging chart

* use real chart version

* t.Logf

Author: laverya

cmd/buildtools/adminconsole.go

@@ -44,7 +44,7 @@ var updateAdminConsoleAddonCommand = &cli.Command{
 		defer hcli.Close()
 
 		logrus.Infof("getting admin console latest tag")
-		latest, err := GetLatestGitHubTag(c.Context, "replicatedhq", "kots-helm")
+		latest, err := GetLatestKotsHelmTag(c.Context)
 		if err != nil {
 			return fmt.Errorf("failed to get admin console latest tag: %w", err)
 		}

cmd/buildtools/utils.go

@@ -211,6 +211,37 @@ func GetLatestGitHubTag(ctx context.Context, owner, repo string) (string, error)
 	return tags[0].GetName(), nil
 }
 
+// GetLatestKotsHelmTag returns the correct tag from the kots-helm repository.
+// this is not quite the same as the latest tag from the kots-helm repository, as github
+// will list "v1.124.12" as being newer than "v1.124.12-build.0" and it is not in our usage.
+func GetLatestKotsHelmTag(ctx context.Context) (string, error) {
+	client := github.NewClient(nil)
+	tags, _, err := client.Repositories.ListTags(ctx, "replicatedhq", "kots-helm", &github.ListOptions{PerPage: 100})
+	if err != nil {
+		return "", fmt.Errorf("list tags: %w", err)
+	}
+	if len(tags) == 0 {
+		return "", fmt.Errorf("no tags found")
+	}
+	latestTag := tags[0].GetName()
+	logrus.Infof("latest tag: %s", latestTag)
+
+	// check to see if there is a 'build.x' tag - if so, return that
+	for _, tag := range tags {
+		logrus.Infof("checkingtag: %s", tag.GetName())
+		if !strings.HasPrefix(tag.GetName(), latestTag) {
+			// tags are sorted, so once we find a tag that doesn't have the same prefix, we can break
+			logrus.Infof("tag does not have same prefix: %s", tag.GetName())
+			break
+		}
+		if strings.Contains(tag.GetName(), "-build.") {
+			logrus.Infof("tag is a build tag, returning: %s", tag.GetName())
+			return tag.GetName(), nil
+		}
+	}
+	return latestTag, nil
+}
+
 // GetGreatestGitHubTag returns the greatest non-prerelease semver tag from a GitHub repository
 // that matches the provided constraints.
 func GetGreatestGitHubTag(ctx context.Context, owner, repo string, constrants *semver.Constraints) (string, error) {

pkg/addons/adminconsole/static/metadata.yaml

@@ -5,7 +5,7 @@
 # $ make buildtools
 # $ output/bin/buildtools update addon <addon name>
 #
-version: 1.124.12
+version: 1.124.12-build.0
 location: oci://proxy.replicated.com/anonymous/registry.replicated.com/library/admin-console
 images:
     kotsadm:

pkg/addons/highavailability.go

@@ -3,8 +3,6 @@ package addons
 import (
 	"context"
 	"fmt"
-	"strings"
-
 	"github.com/pkg/errors"
 	ecv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
 	"github.com/replicatedhq/embedded-cluster/pkg/addons/adminconsole"
@@ -23,6 +21,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"strings"
 )
 
 // CanEnableHA checks if high availability can be enabled in the cluster.
@@ -254,6 +253,10 @@ func EnableAdminConsoleHA(ctx context.Context, kcli client.Client, hcli helm.Cli
 		return errors.Wrap(err, "upgrade admin console")
 	}
 
+	if err := kubeutils.WaitForStatefulset(ctx, kcli, runtimeconfig.KotsadmNamespace, "kotsadm-rqlite", nil); err != nil {
+		return errors.Wrap(err, "wait for rqlite to be ready")
+	}
+
 	return nil
 }
 

pkg/dryrun/kubeutils.go

@@ -22,6 +22,10 @@ func (k *KubeUtils) WaitForDeployment(ctx context.Context, cli client.Client, ns
 	return nil
 }
 
+func (k *KubeUtils) WaitForStatefulset(ctx context.Context, cli client.Client, ns, name string, opts *kubeutils.WaitOptions) error {
+	return nil
+}
+
 func (k *KubeUtils) WaitForDaemonset(ctx context.Context, cli client.Client, ns, name string, opts *kubeutils.WaitOptions) error {
 	return nil
 }

pkg/kubeutils/interface.go

@@ -23,6 +23,7 @@ func Set(_kb KubeUtilsInterface) {
 type KubeUtilsInterface interface {
 	WaitForNamespace(ctx context.Context, cli client.Client, ns string, opts *WaitOptions) error
 	WaitForDeployment(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error
+	WaitForStatefulset(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error
 	WaitForDaemonset(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error
 	WaitForService(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error
 	WaitForJob(ctx context.Context, cli client.Client, ns, name string, completions int32, opts *WaitOptions) error
@@ -62,6 +63,10 @@ func WaitForDeployment(ctx context.Context, cli client.Client, ns, name string,
 	return kb.WaitForDeployment(ctx, cli, ns, name, opts)
 }
 
+func WaitForStatefulset(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error {
+	return kb.WaitForStatefulset(ctx, cli, ns, name, opts)
+}
+
 func WaitForDaemonset(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error {
 	return kb.WaitForDaemonset(ctx, cli, ns, name, opts)
 }

pkg/kubeutils/kubeutils.go

@@ -75,6 +75,34 @@ func (k *KubeUtils) WaitForDeployment(ctx context.Context, cli client.Client, ns
 	return nil
 }
 
+// WaitForStatefulset waits for the provided statefulset to be ready.
+func (k *KubeUtils) WaitForStatefulset(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error {
+	backoff := opts.GetBackoff()
+	var lasterr error
+	if err := wait.ExponentialBackoffWithContext(
+		ctx, backoff, func(ctx context.Context) (bool, error) {
+			ready, err := k.IsStatefulSetReady(ctx, cli, ns, name)
+			if err != nil {
+				lasterr = fmt.Errorf("unable to get statefulset %s status: %v", name, err)
+				return false, nil
+			}
+			return ready, nil
+		},
+	); err != nil {
+		if errors.Is(err, context.Canceled) {
+			if lasterr != nil {
+				err = errors.Join(err, lasterr)
+			}
+			return err
+		} else if lasterr != nil {
+			return fmt.Errorf("timed out waiting for %s to statefulset: %w", name, lasterr)
+		} else {
+			return fmt.Errorf("timed out waiting for %s to statefulset", name)
+		}
+	}
+	return nil
+}
+
 // WaitForDaemonset waits for the provided daemonset to be ready.
 func (k *KubeUtils) WaitForDaemonset(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error {
 	backoff := opts.GetBackoff()

tests/integration/kind/registry/ha_test.go

@@ -55,29 +55,29 @@ func TestRegistry_EnableHAAirgap(t *testing.T) {
 	kclient := util.KubeClient(t, kubeconfig)
 	hcli := util.HelmClient(t, kubeconfig)
 
-	t.Log("installing openebs")
+	t.Logf("%s installing openebs", formattedTime())
 	addon := &openebs.OpenEBS{
 		ProxyRegistryDomain: "proxy.replicated.com",
 	}
 	if err := addon.Install(ctx, kcli, hcli, nil, nil); err != nil {
 		t.Fatalf("failed to install openebs: %v", err)
 	}
 
-	t.Log("waiting for storageclass")
+	t.Logf("%s waiting for storageclass", formattedTime())
 	util.WaitForStorageClass(t, kubeconfig, "openebs-hostpath", 30*time.Second)
 
-	t.Log("installing registry")
+	t.Logf("%s installing registry", formattedTime())
 	registryAddon := &registry.Registry{
 		ServiceCIDR:         "10.96.0.0/12",
 		ProxyRegistryDomain: "proxy.replicated.com",
 		IsHA:                false,
 	}
 	require.NoError(t, registryAddon.Install(ctx, kcli, hcli, nil, nil))
 
-	t.Log("creating hostport service")
+	t.Logf("%s creating hostport service", formattedTime())
 	registryAddr := createHostPortService(t, clusterName, kubeconfig)
 
-	t.Log("installing admin console")
+	t.Logf("%s installing admin console", formattedTime())
 	adminConsoleAddon := &adminconsole.AdminConsole{
 		IsAirgap:            true,
 		ServiceCIDR:         "10.96.0.0/12",
@@ -86,13 +86,13 @@ func TestRegistry_EnableHAAirgap(t *testing.T) {
 	}
 	require.NoError(t, adminConsoleAddon.Install(ctx, kcli, hcli, nil, nil))
 
-	t.Log("pushing image to registry")
+	t.Logf("%s pushing image to registry", formattedTime())
 	copyImageToRegistry(t, registryAddr, "docker.io/library/busybox:1.36.1")
 
-	t.Log("running pod to validate image pull")
+	t.Logf("%s running pod to validate image pull", formattedTime())
 	runPodAndValidateImagePull(t, kubeconfig, "pod-1", "pod1.yaml")
 
-	t.Log("creating installation with HA disabled")
+	t.Logf("%s creating installation with HA disabled", formattedTime())
 	util.EnsureInstallation(t, kcli, ecv1beta1.InstallationSpec{
 		HighAvailability: false,
 	})
@@ -119,21 +119,21 @@ func TestRegistry_EnableHAAirgap(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, canEnable, "should be able to enable HA: %s", reason)
 
-	t.Log("enabling HA")
+	t.Logf("%s enabling HA", formattedTime())
 	loading := newTestingSpinner(t)
 	func() {
 		defer loading.Close()
 		err = addons.EnableHA(ctx, kcli, kclient, hcli, true, "10.96.0.0/12", nil, cfgSpec, loading)
 		require.NoError(t, err)
 	}()
 
-	t.Log("pushing a second image to registry")
+	t.Logf("%s pushing a second image to registry", formattedTime())
 	copyImageToRegistry(t, registryAddr, "docker.io/library/busybox:1.37.0")
 
-	t.Log("running pod to validate image pull")
+	t.Logf("%s running pod to validate image pull", formattedTime())
 	runPodAndValidateImagePull(t, kubeconfig, "pod-1", "pod1.yaml")
 
-	t.Log("running second pod to validate image pull")
+	t.Logf("%s running second pod to validate image pull", formattedTime())
 	runPodAndValidateImagePull(t, kubeconfig, "pod-2", "pod2.yaml")
 }
 
@@ -175,7 +175,7 @@ func enableHAAndCancelContextOnMessage(
 		defer pr.Close()
 		got := waitForMatchingMessage(t, pr, re)
 		if got {
-			t.Log("cancelling context")
+			t.Logf("%s cancelling context", formattedTime())
 			cancel()
 		}
 		io.Copy(io.Discard, pr) // discard the rest of the output
@@ -184,19 +184,18 @@ func enableHAAndCancelContextOnMessage(
 	loading := newTestingSpinner(t)
 	defer loading.Close()
 
-	t.Log("enabling HA and cancelling context on message")
+	t.Logf("%s enabling HA and cancelling context on message", formattedTime())
 	err = addons.EnableHA(ctx, kcli, kclient, hcli, true, "10.96.0.0/12", nil, cfgSpec, loading)
 	require.ErrorIs(t, err, context.Canceled, "expected context to be cancelled")
-	t.Logf("cancelled context and got error: %v", err)
+	t.Logf("%s cancelled context and got error: %v", formattedTime(), err)
 }
 
 func waitForMatchingMessage(t *testing.T, r io.Reader, re *regexp.Regexp) bool {
 	scanner := bufio.NewScanner(r)
 	for scanner.Scan() {
 		b := scanner.Bytes()
-		// t.Logf("got message: %s", string(b))
 		if re.Match(b) {
-			t.Logf("got matching message: %s", string(b))
+			t.Logf("%s got matching message: %s", formattedTime(), string(b))
 			return true
 		}
 	}
@@ -212,7 +211,7 @@ func buildOperatorImage(t *testing.T) string {
 	operatorDir := filepath.Join(workspaceRoot, "operator")
 
 	if os.Getenv("SKIP_OPERATOR_IMAGE_BUILD") == "" {
-		t.Log("building operator image")
+		t.Logf("%s building operator image", formattedTime())
 
 		cmd := exec.CommandContext(
 			t.Context(), "make", "-C", operatorDir, "build-ttl.sh", "USE_CHAINGUARD=0",
@@ -253,7 +252,7 @@ func newTestingSpinner(t *testing.T) *spinner.MessageWriter {
 		spinner.WithWriter(func(format string, args ...any) (int, error) {
 			// discard the output
 			out := fmt.Sprintf(format, args...)
-			t.Log("[spinner]", strings.TrimSpace(out))
+			t.Logf("%s [spinner] %s", formattedTime(), strings.TrimSpace(out))
 			return len(out), nil
 		}),
 		spinner.WithTTY(false),
@@ -272,3 +271,7 @@ func (h *logrusHook) Fire(entry *logrus.Entry) error {
 	h.writer.Write([]byte(entry.Message + "\n"))
 	return nil
 }
+
+func formattedTime() string {
+	return time.Now().Format("2006-01-02 15:04:05")
+}