fix(ha): make more resilient with a job

Author: emosbaugh

cmd/installer/cli/enable_ha.go

@@ -0,0 +1,92 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/replicatedhq/embedded-cluster/pkg/addons"
+	"github.com/replicatedhq/embedded-cluster/pkg/helm"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
+	"github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig"
+	rcutil "github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig/util"
+	"github.com/replicatedhq/embedded-cluster/pkg/versions"
+	"github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+)
+
+// EnableHACmd is the command for enabling HA mode.
+func EnableHACmd(ctx context.Context, name string) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:    "enable-ha",
+		Short:  fmt.Sprintf("Enable high availability for the %s cluster", name),
+		Hidden: true,
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			if os.Getuid() != 0 {
+				return fmt.Errorf("enable-ha command must be run as root")
+			}
+
+			rcutil.InitBestRuntimeConfig(cmd.Context())
+
+			os.Setenv("KUBECONFIG", runtimeconfig.PathToKubeConfig())
+			os.Setenv("TMPDIR", runtimeconfig.EmbeddedClusterTmpSubDir())
+
+			return nil
+		},
+		PostRun: func(cmd *cobra.Command, args []string) {
+			runtimeconfig.Cleanup()
+		},
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if err := runEnableHA(cmd.Context()); err != nil {
+				return err
+			}
+
+			return nil
+		},
+	}
+
+	return cmd
+}
+
+func runEnableHA(ctx context.Context) error {
+	kcli, err := kubeutils.KubeClient()
+	if err != nil {
+		return fmt.Errorf("unable to get kube client: %w", err)
+	}
+
+	canEnableHA, reason, err := addons.CanEnableHA(ctx, kcli)
+	if err != nil {
+		return fmt.Errorf("unable to check if HA can be enabled: %w", err)
+	}
+	if !canEnableHA {
+		logrus.Warnf("High availability cannot be enabled: %s", reason)
+		return NewErrorNothingElseToAdd(fmt.Errorf("high availability cannot be enabled: %s", reason))
+	}
+
+	kclient, err := kubeutils.GetClientset()
+	if err != nil {
+		return fmt.Errorf("unable to create kubernetes client: %w", err)
+	}
+
+	in, err := kubeutils.GetLatestInstallation(ctx, kcli)
+	if err != nil {
+		return fmt.Errorf("unable to get latest installation: %w", err)
+	}
+
+	airgapChartsPath := ""
+	if in.Spec.AirGap {
+		airgapChartsPath = runtimeconfig.EmbeddedClusterChartsSubDir()
+	}
+
+	hcli, err := helm.NewClient(helm.HelmOptions{
+		KubeConfig: runtimeconfig.PathToKubeConfig(),
+		K0sVersion: versions.K0sVersion,
+		AirgapPath: airgapChartsPath,
+	})
+	if err != nil {
+		return fmt.Errorf("unable to create helm client: %w", err)
+	}
+	defer hcli.Close()
+
+	return addons.EnableHA(ctx, kcli, kclient, hcli, in.Spec.AirGap, in.Spec.Network.ServiceCIDR, in.Spec.Proxy, in.Spec.Config)
+}

cmd/installer/cli/install.go

@@ -78,9 +78,6 @@ type InstallCmdFlags struct {
 }
 
 // InstallCmd returns a cobra command for installing the embedded cluster.
-// This is the upcoming version of install without the operator and where
-// install does all of the work. This is a hidden command until it's tested
-// and ready.
 func InstallCmd(ctx context.Context, name string) *cobra.Command {
 	var flags InstallCmdFlags
 

cmd/installer/cli/join.go

@@ -27,6 +27,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"gopkg.in/yaml.v2"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	k8syaml "sigs.k8s.io/yaml"
 )
@@ -41,9 +42,7 @@ type JoinCmdFlags struct {
 	ignoreHostPreflights   bool
 }
 
-// This is the upcoming version of join without the operator and where
-// join does all of the work. This is a hidden command until it's tested
-// and ready.
+// JoinCmd returns a cobra command for joining a node to the cluster.
 func JoinCmd(ctx context.Context, name string) *cobra.Command {
 	var flags JoinCmdFlags
 
@@ -202,23 +201,28 @@ func runJoin(ctx context.Context, name string, flags JoinCmdFlags, jcmd *kotsadm
 		return nil
 	}
 
-	airgapChartsPath := ""
-	if flags.isAirgap {
-		airgapChartsPath = runtimeconfig.EmbeddedClusterChartsSubDir()
-	}
+	if flags.enableHighAvailability {
+		kclient, err := kubeutils.GetClientset()
+		if err != nil {
+			return fmt.Errorf("unable to create kubernetes client: %w", err)
+		}
 
-	hcli, err := helm.NewClient(helm.HelmOptions{
-		KubeConfig: runtimeconfig.PathToKubeConfig(),
-		K0sVersion: versions.K0sVersion,
-		AirgapPath: airgapChartsPath,
-	})
-	if err != nil {
-		return fmt.Errorf("unable to create helm client: %w", err)
-	}
-	defer hcli.Close()
+		airgapChartsPath := ""
+		if flags.isAirgap {
+			airgapChartsPath = runtimeconfig.EmbeddedClusterChartsSubDir()
+		}
 
-	if flags.enableHighAvailability {
-		if err := maybeEnableHA(ctx, kcli, hcli, flags.isAirgap, cidrCfg.ServiceCIDR, jcmd.InstallationSpec.Proxy, jcmd.InstallationSpec.Config); err != nil {
+		hcli, err := helm.NewClient(helm.HelmOptions{
+			KubeConfig: runtimeconfig.PathToKubeConfig(),
+			K0sVersion: versions.K0sVersion,
+			AirgapPath: airgapChartsPath,
+		})
+		if err != nil {
+			return fmt.Errorf("unable to create helm client: %w", err)
+		}
+		defer hcli.Close()
+
+		if err := maybeEnableHA(ctx, kcli, kclient, hcli, flags.isAirgap, cidrCfg.ServiceCIDR, jcmd.InstallationSpec.Proxy, jcmd.InstallationSpec.Config); err != nil {
 			return fmt.Errorf("unable to enable high availability: %w", err)
 		}
 	}
@@ -493,8 +497,8 @@ func waitForNodeToJoin(ctx context.Context, kcli client.Client, hostname string,
 	return nil
 }
 
-func maybeEnableHA(ctx context.Context, kcli client.Client, hcli helm.Client, isAirgap bool, serviceCIDR string, proxy *ecv1beta1.ProxySpec, cfgspec *ecv1beta1.ConfigSpec) error {
-	canEnableHA, err := addons.CanEnableHA(ctx, kcli)
+func maybeEnableHA(ctx context.Context, kcli client.Client, kclient kubernetes.Interface, hcli helm.Client, isAirgap bool, serviceCIDR string, proxy *ecv1beta1.ProxySpec, cfgspec *ecv1beta1.ConfigSpec) error {
+	canEnableHA, _, err := addons.CanEnableHA(ctx, kcli)
 	if err != nil {
 		return fmt.Errorf("unable to check if HA can be enabled: %w", err)
 	}
@@ -509,5 +513,5 @@ func maybeEnableHA(ctx context.Context, kcli client.Client, hcli helm.Client, is
 		return nil
 	}
 	logrus.Info("")
-	return addons.EnableHA(ctx, kcli, hcli, isAirgap, serviceCIDR, proxy, cfgspec)
+	return addons.EnableHA(ctx, kcli, kclient, hcli, isAirgap, serviceCIDR, proxy, cfgspec)
 }

cmd/installer/cli/root.go

@@ -100,6 +100,7 @@ func RootCmd(ctx context.Context, name string) *cobra.Command {
 	cmd.AddCommand(JoinCmd(ctx, name))
 	cmd.AddCommand(ShellCmd(ctx, name))
 	cmd.AddCommand(NodeCmd(ctx, name))
+	cmd.AddCommand(EnableHACmd(ctx, name))
 	cmd.AddCommand(VersionCmd(ctx, name))
 	cmd.AddCommand(ResetCmd(ctx, name))
 	cmd.AddCommand(MaterializeCmd(ctx, name))

go.mod

@@ -41,7 +41,6 @@ require (
 	github.com/vmware-tanzu/velero v1.15.2
 	go.uber.org/multierr v1.11.0
 	golang.org/x/crypto v0.33.0
-	golang.org/x/sync v0.11.0
 	golang.org/x/term v0.29.0
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
@@ -272,6 +271,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.34.0 // indirect
 	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 // indirect
 	golang.org/x/mod v0.23.0 // indirect
+	golang.org/x/sync v0.11.0 // indirect
 	golang.org/x/tools v0.28.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/api v0.197.0 // indirect

operator/pkg/cli/migrate.go

@@ -0,0 +1,46 @@
+package cli
+
+import (
+	"fmt"
+
+	"github.com/replicatedhq/embedded-cluster/pkg/addons/registry/migrate"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
+	"github.com/spf13/cobra"
+)
+
+func MigrateCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "migrate",
+		Short: "Run the specified migration",
+	}
+
+	cmd.AddCommand(
+		MigrateRegistryDataCmd(),
+	)
+
+	return cmd
+}
+
+func MigrateRegistryDataCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:          "registry-data",
+		Short:        "Run the registry-data migration",
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := cmd.Context()
+
+			cli, err := kubeutils.KubeClient()
+			if err != nil {
+				return fmt.Errorf("failed to create kubernetes client: %w", err)
+			}
+
+			err = migrate.RegistryData(ctx, cli)
+			if err != nil {
+				return fmt.Errorf("failed to migrate registry data: %w", err)
+			}
+			return nil
+		},
+	}
+
+	return cmd
+}

operator/pkg/cli/root.go

@@ -117,6 +117,7 @@ func addSubcommands(cmd *cobra.Command) {
 	cmd.AddCommand(
 		UpgradeCmd(),
 		UpgradeJobCmd(),
+		MigrateCmd(),
 		MigrateV2Cmd(),
 		VersionCmd(),
 	)

pkg/addons/embeddedclusteroperator/upgrade.go

@@ -2,10 +2,10 @@ package embeddedclusteroperator
 
 import (
 	"context"
-	"log/slog"
 
 	"github.com/pkg/errors"
 	"github.com/replicatedhq/embedded-cluster/pkg/helm"
+	"github.com/sirupsen/logrus"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -15,7 +15,7 @@ func (e *EmbeddedClusterOperator) Upgrade(ctx context.Context, kcli client.Clien
 		return errors.Wrap(err, "check if release exists")
 	}
 	if !exists {
-		slog.Info("Release not found, installing", "release", releaseName, "namespace", namespace)
+		logrus.Debugf("Release not found, installing release %s in namespace %s", releaseName, namespace)
 		if err := e.Install(ctx, kcli, hcli, overrides, nil); err != nil {
 			return errors.Wrap(err, "install")
 		}