Factor out upsert and upsertAndWait

It's a common pattern in the create commands to construct a value,
then (if not exporting it) upsert it and wait for it to
reconcile. This commit factors `upsert`, which does the update/insert
bit, and `upsertAndWait`, which does the whole thing.

Since these output messages, they are methods of `apiType` (previously
`names`), so that they have access to the name of the kind they are
operating on.

Signed-off-by: Michael Bridgen <[email protected]>

cmd/flux/create.go

@@ -17,13 +17,19 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
 
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/validation"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
-	"github.com/spf13/cobra"
+	"github.com/fluxcd/flux2/internal/utils"
 )
 
 var createCmd = &cobra.Command{
@@ -46,6 +52,78 @@ func init() {
 	rootCmd.AddCommand(createCmd)
 }
 
+// upsertable is an interface for values that can be used in `upsert`.
+type upsertable interface {
+	adapter
+	named
+}
+
+// upsert updates or inserts an object. Instead of providing the
+// object itself, you provide a named (as in Name and Namespace)
+// template value, and a mutate function which sets the values you
+// want to update. The mutate function is nullary -- you mutate a
+// value in the closure, e.g., by doing this:
+//
+//     var existing Value
+//     existing.Name = name
+//     existing.Namespace = ns
+//     upsert(ctx, client, valueAdapter{&value}, func() error {
+//       value.Spec = onePreparedEarlier
+//     })
+func (names apiType) upsert(ctx context.Context, kubeClient client.Client, object upsertable, mutate func() error) (types.NamespacedName, error) {
+	nsname := types.NamespacedName{
+		Namespace: object.GetNamespace(),
+		Name:      object.GetName(),
+	}
+
+	op, err := controllerutil.CreateOrUpdate(ctx, kubeClient, object.asRuntimeObject(), mutate)
+	if err != nil {
+		return nsname, err
+	}
+
+	switch op {
+	case controllerutil.OperationResultCreated:
+		logger.Successf("%s created", names.kind)
+	case controllerutil.OperationResultUpdated:
+		logger.Successf("%s updated", names.kind)
+	}
+	return nsname, nil
+}
+
+type upsertWaitable interface {
+	upsertable
+	statusable
+}
+
+// upsertAndWait encodes the pattern of creating or updating a
+// resource, then waiting for it to reconcile. See the note on
+// `upsert` for how to work with the `mutate` argument.
+func (names apiType) upsertAndWait(object upsertWaitable, mutate func() error) error {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	kubeClient, err := utils.KubeClient(kubeconfig, kubecontext) // NB globals
+	if err != nil {
+		return err
+	}
+
+	logger.Generatef("generating %s", names.kind)
+	logger.Actionf("applying %s", names.kind)
+
+	namespacedName, err := imageRepositoryType.upsert(ctx, kubeClient, object, mutate)
+	if err != nil {
+		return err
+	}
+
+	logger.Waitingf("waiting for %s reconciliation", names.kind)
+	if err := wait.PollImmediate(pollInterval, timeout,
+		isReady(ctx, kubeClient, namespacedName, object)); err != nil {
+		return err
+	}
+	logger.Successf("%s reconciliation completed", names.kind)
+	return nil
+}
+
 func parseLabels() (map[string]string, error) {
 	result := make(map[string]string)
 	for _, label := range labels {

cmd/flux/create_image_policy.go

@@ -17,18 +17,12 @@ limitations under the License.
 package main
 
 import (
-	"context"
 	"fmt"
 
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
-	"github.com/fluxcd/flux2/internal/utils"
 	imagev1 "github.com/fluxcd/image-reflector-controller/api/v1alpha1"
 )
 
@@ -105,57 +99,12 @@ func createImagePolicyRun(cmd *cobra.Command, args []string) error {
 		return printExport(exportImagePolicy(&policy))
 	}
 
-	// I don't need these until attempting to upsert the object, but
-	// for consistency with other create commands, the following are
-	// given a chance to error out before reporting any progress.
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
-	kubeClient, err := utils.KubeClient(kubeconfig, kubecontext)
-	if err != nil {
-		return err
-	}
-
-	logger.Generatef("generating ImagePolicy")
-	logger.Actionf("applying ImagePolicy")
-	namespacedName, err := upsertImagePolicy(ctx, kubeClient, &policy)
-	if err != nil {
-		return err
-	}
-
-	logger.Waitingf("waiting for ImagePolicy reconciliation")
-	if err := wait.PollImmediate(pollInterval, timeout,
-		isReady(ctx, kubeClient, namespacedName, imagePolicyAdapter{&policy})); err != nil {
-		return err
-	}
-	logger.Successf("ImagePolicy reconciliation completed")
-
-	return nil
-}
-
-func upsertImagePolicy(ctx context.Context, kubeClient client.Client, policy *imagev1.ImagePolicy) (types.NamespacedName, error) {
-	nsname := types.NamespacedName{
-		Namespace: policy.GetNamespace(),
-		Name:      policy.GetName(),
-	}
-
 	var existing imagev1.ImagePolicy
-	existing.SetName(nsname.Name)
-	existing.SetNamespace(nsname.Namespace)
-	op, err := controllerutil.CreateOrUpdate(ctx, kubeClient, &existing, func() error {
+	copyName(&existing, &policy)
+	err = imagePolicyType.upsertAndWait(imagePolicyAdapter{&existing}, func() error {
 		existing.Spec = policy.Spec
 		existing.SetLabels(policy.Labels)
 		return nil
 	})
-	if err != nil {
-		return nsname, err
-	}
-
-	switch op {
-	case controllerutil.OperationResultCreated:
-		logger.Successf("ImagePolicy created")
-	case controllerutil.OperationResultUpdated:
-		logger.Successf("ImagePolicy updated")
-	}
-	return nsname, nil
+	return err
 }

cmd/flux/create_image_repository.go

@@ -17,20 +17,14 @@ limitations under the License.
 package main
 
 import (
-	"context"
 	"fmt"
 	"time"
 
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
-	"github.com/fluxcd/flux2/internal/utils"
 	imagev1 "github.com/fluxcd/image-reflector-controller/api/v1alpha1"
 )
 
@@ -104,57 +98,13 @@ func createImageRepositoryRun(cmd *cobra.Command, args []string) error {
 		return printExport(exportImageRepository(&repo))
 	}
 
-	// I don't need these until attempting to upsert the object, but
-	// for consistency with other create commands, the following are
-	// given a chance to error out before reporting any progress.
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
-	kubeClient, err := utils.KubeClient(kubeconfig, kubecontext)
-	if err != nil {
-		return err
-	}
-
-	logger.Generatef("generating ImageRepository")
-	logger.Actionf("applying ImageRepository")
-	namespacedName, err := upsertImageRepository(ctx, kubeClient, &repo)
-	if err != nil {
-		return err
-	}
-
-	logger.Waitingf("waiting for ImageRepository reconciliation")
-	if err := wait.PollImmediate(pollInterval, timeout,
-		isReady(ctx, kubeClient, namespacedName, imageRepositoryAdapter{&repo})); err != nil {
-		return err
-	}
-	logger.Successf("ImageRepository reconciliation completed")
-
-	return nil
-}
-
-func upsertImageRepository(ctx context.Context, kubeClient client.Client, repo *imagev1.ImageRepository) (types.NamespacedName, error) {
-	nsname := types.NamespacedName{
-		Namespace: repo.GetNamespace(),
-		Name:      repo.GetName(),
-	}
-
+	// a temp value for use with the rest
 	var existing imagev1.ImageRepository
-	existing.SetName(nsname.Name)
-	existing.SetNamespace(nsname.Namespace)
-	op, err := controllerutil.CreateOrUpdate(ctx, kubeClient, &existing, func() error {
+	copyName(&existing, &repo)
+	err = imageRepositoryType.upsertAndWait(imageRepositoryAdapter{&existing}, func() error {
 		existing.Spec = repo.Spec
 		existing.Labels = repo.Labels
 		return nil
 	})
-	if err != nil {
-		return nsname, err
-	}
-
-	switch op {
-	case controllerutil.OperationResultCreated:
-		logger.Successf("ImageRepository created")
-	case controllerutil.OperationResultUpdated:
-		logger.Successf("ImageRepository updated")
-	}
-	return nsname, nil
+	return err
 }

cmd/flux/create_image_updateauto.go

@@ -17,18 +17,12 @@ limitations under the License.
 package main
 
 import (
-	"context"
 	"fmt"
 
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
-	"github.com/fluxcd/flux2/internal/utils"
 	autov1 "github.com/fluxcd/image-automation-controller/api/v1alpha1"
 )
 
@@ -108,57 +102,12 @@ func createImageUpdateRun(cmd *cobra.Command, args []string) error {
 		return printExport(exportImageUpdate(&update))
 	}
 
-	// I don't need these until attempting to upsert the object, but
-	// for consistency with other create commands, the following are
-	// given a chance to error out before reporting any progress.
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
-	kubeClient, err := utils.KubeClient(kubeconfig, kubecontext)
-	if err != nil {
-		return err
-	}
-
-	logger.Generatef("generating ImageUpdateAutomation")
-	logger.Actionf("applying ImageUpdateAutomation")
-	namespacedName, err := upsertImageUpdateAutomation(ctx, kubeClient, &update)
-	if err != nil {
-		return err
-	}
-
-	logger.Waitingf("waiting for ImageUpdateAutomation reconciliation")
-	if err := wait.PollImmediate(pollInterval, timeout,
-		isReady(ctx, kubeClient, namespacedName, imageUpdateAutomationAdapter{&update})); err != nil {
-		return err
-	}
-	logger.Successf("ImageUpdateAutomation reconciliation completed")
-
-	return nil
-}
-
-func upsertImageUpdateAutomation(ctx context.Context, kubeClient client.Client, update *autov1.ImageUpdateAutomation) (types.NamespacedName, error) {
-	nsname := types.NamespacedName{
-		Namespace: update.GetNamespace(),
-		Name:      update.GetName(),
-	}
-
 	var existing autov1.ImageUpdateAutomation
-	existing.SetName(nsname.Name)
-	existing.SetNamespace(nsname.Namespace)
-	op, err := controllerutil.CreateOrUpdate(ctx, kubeClient, &existing, func() error {
+	copyName(&existing, &update)
+	err = imageUpdateAutomationType.upsertAndWait(imageUpdateAutomationAdapter{&existing}, func() error {
 		existing.Spec = update.Spec
 		existing.Labels = update.Labels
 		return nil
 	})
-	if err != nil {
-		return nsname, err
-	}
-
-	switch op {
-	case controllerutil.OperationResultCreated:
-		logger.Successf("ImageUpdateAutomation created")
-	case controllerutil.OperationResultUpdated:
-		logger.Successf("ImageUpdateAutomation updated")
-	}
-	return nsname, nil
+	return err
 }

cmd/flux/delete.go

@@ -45,7 +45,7 @@ func init() {
 }
 
 type deleteCommand struct {
-	names
+	apiType
 	object adapter // for getting the value, and later deleting it
 }
 

cmd/flux/delete_image_policy.go

@@ -30,8 +30,8 @@ var deleteImagePolicyCmd = &cobra.Command{
   flux delete auto image-policy alpine3.x
 `,
 	RunE: deleteCommand{
-		names:  imagePolicyNames,
-		object: universalAdapter{&imagev1.ImagePolicy{}},
+		apiType: imagePolicyType,
+		object:  universalAdapter{&imagev1.ImagePolicy{}},
 	}.run,
 }
 

cmd/flux/delete_image_repository.go

@@ -30,8 +30,8 @@ var deleteImageRepositoryCmd = &cobra.Command{
   flux delete auto image-repository alpine
 `,
 	RunE: deleteCommand{
-		names:  imageRepositoryNames,
-		object: universalAdapter{&imagev1.ImageRepository{}},
+		apiType: imageRepositoryType,
+		object:  universalAdapter{&imagev1.ImageRepository{}},
 	}.run,
 }
 

cmd/flux/delete_image_updateauto.go

@@ -30,8 +30,8 @@ var deleteImageUpdateCmd = &cobra.Command{
   flux delete auto image-update latest-images
 `,
 	RunE: deleteCommand{
-		names:  imageUpdateAutomationNames,
-		object: universalAdapter{&autov1.ImageUpdateAutomation{}},
+		apiType: imageUpdateAutomationType,
+		object:  universalAdapter{&autov1.ImageUpdateAutomation{}},
 	}.run,
 }