chore: unify kubeutils (#1831)

* chore: unify kubeutils

* f

pkg/logging/logging.go → cmd/installer/cli/logging.go

@@ -1,7 +1,4 @@
-// Package logging manages setup of common logging interfaces and settings. We set the log
-// level to all levels but we only show on stdout the info, error, and fatal levels. All
-// other error levels are written only to a log file.
-package logging
+package cli
 
 import (
 	"fmt"
@@ -13,6 +10,9 @@ import (
 	"github.com/fatih/color"
 	"github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig"
 	"github.com/sirupsen/logrus"
+	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	ctrlzap "sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
 
 // MaxLogFiles is the maximum number of log files we keep.
@@ -111,11 +111,22 @@ func SetupLogging() {
 	logpath := runtimeconfig.PathToLog(fname)
 	logfile, err := os.OpenFile(logpath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0400)
 	if err != nil {
-		logrus.Warnf("unable to setup logging: %v", err)
+		logrus.Warnf("Unable to setup logging: %v", err)
 		return
 	}
 	logrus.SetOutput(logfile)
 	logrus.AddHook(&StdoutLogger{})
 	logrus.Debugf("command line: %v", os.Args)
+
+	setupCtrlLogging(logfile)
+
 	trimLogDir()
 }
+
+// setupCtrlLogging sets up the logging for the controller-runtime package to the writer specified.
+func setupCtrlLogging(w io.Writer) {
+	k8slogger := ctrlzap.New(func(o *zap.Options) {
+		o.DestWriter = w
+	})
+	ctrllog.SetLogger(k8slogger)
+}

cmd/installer/cli/restore_test.go

@@ -11,12 +11,12 @@ import (
 	k0sv1beta1 "github.com/k0sproject/k0s/pkg/apis/k0s/v1beta1"
 	clitesting "github.com/replicatedhq/embedded-cluster/cmd/installer/cli/testing"
 	"github.com/replicatedhq/embedded-cluster/pkg/disasterrecovery"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/replicatedhq/embedded-cluster/pkg/release"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/kubectl/pkg/scheme"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -193,8 +193,7 @@ func Test_isReplicatedBackupRestorable(t *testing.T) {
 }
 
 func Test_waitForBackups(t *testing.T) {
-	scheme := scheme.Scheme
-	velerov1.AddToScheme(scheme)
+	scheme := kubeutils.Scheme
 
 	appendCommonAnnotations := func(annotations map[string]string) map[string]string {
 		annotations["kots.io/embedded-cluster-version"] = "v0.0.0"

cmd/installer/main.go

@@ -6,16 +6,14 @@ import (
 	"path"
 
 	"github.com/mattn/go-isatty"
-
 	"github.com/replicatedhq/embedded-cluster/cmd/installer/cli"
-	"github.com/replicatedhq/embedded-cluster/pkg/logging"
 	"github.com/replicatedhq/embedded-cluster/pkg/prompts"
 )
 
 func main() {
 	ctx := context.Background()
 
-	logging.SetupLogging()
+	cli.SetupLogging()
 
 	prompts.SetTerminal(isatty.IsTerminal(os.Stdout.Fd()))
 

operator/pkg/artifacts/upgrade.go

@@ -10,9 +10,9 @@ import (
 
 	autopilotv1beta2 "github.com/k0sproject/k0s/pkg/apis/autopilot/v1beta2"
 	clusterv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/release"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/util"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -181,7 +181,7 @@ func ensureArtifactsJobForNode(ctx context.Context, cli client.Client, in *clust
 		return nil, fmt.Errorf("get job for node: %w", err)
 	}
 
-	err = k8sutil.EnsureObject(ctx, cli, job, func(opts *k8sutil.EnsureObjectOptions) {
+	err = kubeutils.EnsureObject(ctx, cli, job, func(opts *kubeutils.EnsureObjectOptions) {
 		opts.DeleteOptions = append(opts.DeleteOptions, client.PropagationPolicy(metav1.DeletePropagationForeground))
 		opts.ShouldDelete = func(obj client.Object) bool {
 			// we need to check if the job is for the given installation otherwise we delete

operator/pkg/artifacts/upgrade_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/go-logr/logr/testr"
 	clusterv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	batchv1 "k8s.io/api/batch/v1"
@@ -233,7 +233,7 @@ func TestEnsureArtifactsJobForNodes(t *testing.T) {
 			require.NoError(t, err)
 			t.Cleanup(func() { _ = testEnv.Stop() })
 
-			cli, err := client.New(cfg, client.Options{Scheme: k8sutil.Scheme()})
+			cli, err := client.New(cfg, client.Options{Scheme: kubeutils.Scheme})
 			require.NoError(t, err)
 
 			for _, obj := range tt.initRuntimeObjs {
@@ -397,7 +397,7 @@ func TestListArtifactsJobForNodes(t *testing.T) {
 			require.NoError(t, err)
 			t.Cleanup(func() { _ = testEnv.Stop() })
 
-			cli, err := client.New(cfg, client.Options{Scheme: k8sutil.Scheme()})
+			cli, err := client.New(cfg, client.Options{Scheme: kubeutils.Scheme})
 			require.NoError(t, err)
 
 			for _, obj := range tt.initRuntimeObjs {

operator/pkg/cli/migrate_v2.go

@@ -5,7 +5,7 @@ import (
 
 	ecv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/cli/migratev2"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig"
 	"github.com/spf13/cobra"
 )
@@ -36,7 +36,7 @@ func MigrateV2Cmd() *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			ctx := cmd.Context()
 
-			cli, err := k8sutil.KubeClient()
+			cli, err := kubeutils.KubeClient()
 			if err != nil {
 				return fmt.Errorf("failed to create kubernetes client: %w", err)
 			}

operator/pkg/cli/root.go

@@ -5,7 +5,7 @@ import (
 	"os"
 
 	"github.com/replicatedhq/embedded-cluster/operator/controllers"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/replicatedhq/embedded-cluster/pkg/versions"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -44,7 +44,7 @@ func RootCmd() *cobra.Command {
 		},
 		Run: func(cmd *cobra.Command, args []string) {
 			mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
-				Scheme: k8sutil.Scheme(),
+				Scheme: kubeutils.Scheme,
 				Metrics: metricsserver.Options{
 					BindAddress: metricsAddr,
 				},

operator/pkg/cli/upgrade.go

@@ -7,7 +7,6 @@ import (
 	"os"
 
 	ecv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/upgrade"
 	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig"
@@ -42,7 +41,7 @@ func UpgradeCmd() *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			slog.Info("Upgrade job creation started")
 
-			cli, err := k8sutil.KubeClient()
+			cli, err := kubeutils.KubeClient()
 			if err != nil {
 				return fmt.Errorf("failed to create kubernetes client: %w", err)
 			}

operator/pkg/cli/upgrade_job.go

@@ -10,10 +10,10 @@ import (
 	"github.com/google/uuid"
 	ecv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/cli/migratev2"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/upgrade"
 	"github.com/replicatedhq/embedded-cluster/pkg/helm"
 	"github.com/replicatedhq/embedded-cluster/pkg/helpers"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/replicatedhq/embedded-cluster/pkg/metrics"
 	"github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig"
 	"github.com/replicatedhq/embedded-cluster/pkg/versions"
@@ -56,7 +56,7 @@ func UpgradeJobCmd() *cobra.Command {
 			slog.Info("Upgrade job started", "version", versions.Version)
 			slog.Info("Upgrading to installation", "name", in.Name, "version", in.Spec.Config.Version)
 
-			kcli, err := k8sutil.KubeClient()
+			kcli, err := kubeutils.KubeClient()
 			if err != nil {
 				return fmt.Errorf("failed to create kubernetes client: %w", err)
 			}
@@ -128,7 +128,7 @@ func maybeMarkAsFailed(ctx context.Context, kcli client.Client, in *ecv1beta1.In
 	if !lastAttempt {
 		return nil
 	}
-	if err := k8sutil.SetInstallationState(ctx, kcli, in.Name, ecv1beta1.InstallationStateFailed, helpers.CleanErrorMessage(upgradeErr)); err != nil {
+	if err := kubeutils.SetInstallationState(ctx, kcli, in, ecv1beta1.InstallationStateFailed, helpers.CleanErrorMessage(upgradeErr)); err != nil {
 		return fmt.Errorf("set installation state: %w", err)
 	}
 	return nil

operator/pkg/upgrade/autopilot.go

@@ -17,8 +17,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-// DetermineUpgradeTargets makes sure that we are listing all the nodes in the autopilot plan.
-func DetermineUpgradeTargets(ctx context.Context, cli client.Client) (apv1b2.PlanCommandTargets, error) {
+// determineUpgradeTargets makes sure that we are listing all the nodes in the autopilot plan.
+func determineUpgradeTargets(ctx context.Context, cli client.Client) (apv1b2.PlanCommandTargets, error) {
 	var nodes corev1.NodeList
 	if err := cli.List(ctx, &nodes); err != nil {
 		return apv1b2.PlanCommandTargets{}, fmt.Errorf("failed to list nodes: %w", err)
@@ -46,9 +46,9 @@ func DetermineUpgradeTargets(ctx context.Context, cli client.Client) (apv1b2.Pla
 	}, nil
 }
 
-// StartAutopilotUpgrade creates an autopilot plan to upgrade to version specified in spec.config.version.
-func StartAutopilotUpgrade(ctx context.Context, cli client.Client, in *v1beta1.Installation, meta *ectypes.ReleaseMetadata) error {
-	targets, err := DetermineUpgradeTargets(ctx, cli)
+// startAutopilotUpgrade creates an autopilot plan to upgrade to version specified in spec.config.version.
+func startAutopilotUpgrade(ctx context.Context, cli client.Client, in *v1beta1.Installation, meta *ectypes.ReleaseMetadata) error {
+	targets, err := determineUpgradeTargets(ctx, cli)
 	if err != nil {
 		return fmt.Errorf("failed to determine upgrade targets: %w", err)
 	}

operator/pkg/upgrade/installation.go

@@ -6,7 +6,6 @@ import (
 	"log/slog"
 
 	ecv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
 	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -27,7 +26,7 @@ func CreateInstallation(ctx context.Context, cli client.Client, original *ecv1be
 		return fmt.Errorf("create installation: %w", err)
 	}
 
-	err = k8sutil.SetInstallationState(ctx, cli, in.Name, ecv1beta1.InstallationStateInstalling, "Upgrading Kubernetes", "")
+	err = kubeutils.SetInstallationState(ctx, cli, in, ecv1beta1.InstallationStateInstalling, "Upgrading Kubernetes", "")
 	if err != nil {
 		return fmt.Errorf("update installation status: %w", err)
 	}

operator/pkg/upgrade/job.go

@@ -12,9 +12,9 @@ import (
 	ecv1beta1 "github.com/replicatedhq/embedded-cluster/kinds/apis/v1beta1"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/artifacts"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/autopilot"
-	"github.com/replicatedhq/embedded-cluster/operator/pkg/k8sutil"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/metadata"
 	"github.com/replicatedhq/embedded-cluster/operator/pkg/release"
+	"github.com/replicatedhq/embedded-cluster/pkg/kubeutils"
 	"github.com/replicatedhq/embedded-cluster/pkg/runtimeconfig"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -373,7 +373,7 @@ func autopilotEnsureAirgapArtifactsPlan(ctx context.Context, cli client.Client,
 		return fmt.Errorf("get autopilot airgap artifacts plan: %w", err)
 	}
 
-	err = k8sutil.EnsureObject(ctx, cli, plan, func(opts *k8sutil.EnsureObjectOptions) {
+	err = kubeutils.EnsureObject(ctx, cli, plan, func(opts *kubeutils.EnsureObjectOptions) {
 		opts.ShouldDelete = func(obj client.Object) bool {
 			return obj.GetAnnotations()[artifacts.InstallationNameAnnotation] != in.Name
 		}