Merge pull request #1672 from souleb/main

Adds a watch flag to the get command

cmd/flux/create_tenant.go

@@ -148,7 +148,7 @@ func createTenantCmdRun(cmd *cobra.Command, args []string) error {
 	}
 
 	if createArgs.export {
-		for i, _ := range tenantArgs.namespaces {
+		for i := range tenantArgs.namespaces {
 			if err := exportTenant(namespaces[i], accounts[i], roleBindings[i]); err != nil {
 				return err
 			}
@@ -164,7 +164,7 @@ func createTenantCmdRun(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
-	for i, _ := range tenantArgs.namespaces {
+	for i := range tenantArgs.namespaces {
 		logger.Actionf("applying namespace %s", namespaces[i].Name)
 		if err := upsertNamespace(ctx, kubeClient, namespaces[i]); err != nil {
 			return err

cmd/flux/get.go

@@ -25,6 +25,9 @@ import (
 	"github.com/spf13/cobra"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	watchtools "k8s.io/client-go/tools/watch"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/fluxcd/pkg/apis/meta"
@@ -32,6 +35,26 @@ import (
 	"github.com/fluxcd/flux2/internal/utils"
 )
 
+type deriveType func(runtime.Object) (summarisable, error)
+
+type typeMap map[string]deriveType
+
+func (m typeMap) registerCommand(t string, f deriveType) error {
+	if _, ok := m[t]; ok {
+		return fmt.Errorf("duplicate type function %s", t)
+	}
+	m[t] = f
+	return nil
+}
+
+func (m typeMap) execute(t string, obj runtime.Object) (summarisable, error) {
+	f, ok := m[t]
+	if !ok {
+		return nil, fmt.Errorf("unsupported type %s", t)
+	}
+	return f(obj)
+}
+
 var getCmd = &cobra.Command{
 	Use:   "get",
 	Short: "Get the resources and their status",
@@ -42,6 +65,7 @@ type GetFlags struct {
 	allNamespaces  bool
 	noHeader       bool
 	statusSelector string
+	watch          bool
 }
 
 var getArgs GetFlags
@@ -50,6 +74,7 @@ func init() {
 	getCmd.PersistentFlags().BoolVarP(&getArgs.allNamespaces, "all-namespaces", "A", false,
 		"list the requested object(s) across all namespaces")
 	getCmd.PersistentFlags().BoolVarP(&getArgs.noHeader, "no-header", "", false, "skip the header when printing the results")
+	getCmd.PersistentFlags().BoolVarP(&getArgs.watch, "watch", "w", false, "After listing/getting the requested object, watch for changes.")
 	getCmd.PersistentFlags().StringVar(&getArgs.statusSelector, "status-selector", "",
 		"specify the status condition name and the desired state to filter the get result, e.g. ready=false")
 	rootCmd.AddCommand(getCmd)
@@ -103,6 +128,7 @@ var namespaceHeader = []string{"Namespace"}
 type getCommand struct {
 	apiType
 	list    summarisable
+	funcMap typeMap
 }
 
 func (get getCommand) run(cmd *cobra.Command, args []string) error {
@@ -123,13 +149,17 @@ func (get getCommand) run(cmd *cobra.Command, args []string) error {
 		listOpts = append(listOpts, client.MatchingFields{"metadata.name": args[0]})
 	}
 
+	getAll := cmd.Use == "all"
+
+	if getArgs.watch {
+		return get.watch(ctx, kubeClient, cmd, args, listOpts)
+	}
+
 	err = kubeClient.List(ctx, get.list.asClientList(), listOpts...)
 	if err != nil {
 		return err
 	}
 
-	getAll := cmd.Use == "all"
-
 	if get.list.len() == 0 {
 		if !getAll {
 			logger.Failuref("no %s objects found in %s namespace", get.kind, rootArgs.namespace)
@@ -141,28 +171,93 @@ func (get getCommand) run(cmd *cobra.Command, args []string) error {
 	if !getArgs.noHeader {
 		header = get.list.headers(getArgs.allNamespaces)
 	}
+
+	rows, err := getRowsToPrint(getAll, get.list)
+	if err != nil {
+		return err
+	}
+
+	utils.PrintTable(os.Stdout, header, rows)
+
+	if getAll {
+		fmt.Println()
+	}
+
+	return nil
+}
+
+func getRowsToPrint(getAll bool, list summarisable) ([][]string, error) {
 	noFilter := true
 	var conditionType, conditionStatus string
 	if getArgs.statusSelector != "" {
 		parts := strings.SplitN(getArgs.statusSelector, "=", 2)
 		if len(parts) != 2 {
-			return fmt.Errorf("expected status selector in type=status format, but found: %s", getArgs.statusSelector)
+			return nil, fmt.Errorf("expected status selector in type=status format, but found: %s", getArgs.statusSelector)
 		}
 		conditionType = parts[0]
 		conditionStatus = parts[1]
 		noFilter = false
 	}
 	var rows [][]string
-	for i := 0; i < get.list.len(); i++ {
-		if noFilter || get.list.statusSelectorMatches(i, conditionType, conditionStatus) {
-			row := get.list.summariseItem(i, getArgs.allNamespaces, getAll)
+	for i := 0; i < list.len(); i++ {
+		if noFilter || list.statusSelectorMatches(i, conditionType, conditionStatus) {
+			row := list.summariseItem(i, getArgs.allNamespaces, getAll)
 			rows = append(rows, row)
 		}
 	}
+	return rows, nil
+}
+
+//
+// watch starts a client-side watch of one or more resources.
+func (get *getCommand) watch(ctx context.Context, kubeClient client.WithWatch, cmd *cobra.Command, args []string, listOpts []client.ListOption) error {
+	w, err := kubeClient.Watch(ctx, get.list.asClientList(), listOpts...)
+	if err != nil {
+		return err
+	}
+
+	_, err = watchUntil(ctx, w, get)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func watchUntil(ctx context.Context, w watch.Interface, get *getCommand) (bool, error) {
+	firstIteration := true
+	_, error := watchtools.UntilWithoutRetry(ctx, w, func(e watch.Event) (bool, error) {
+		objToPrint := e.Object
+		sink, err := get.funcMap.execute(get.apiType.kind, objToPrint)
+		if err != nil {
+			return false, err
+		}
+
+		var header []string
+		if !getArgs.noHeader {
+			header = sink.headers(getArgs.allNamespaces)
+		}
+		rows, err := getRowsToPrint(false, sink)
+		if err != nil {
+			return false, err
+		}
+		if firstIteration {
 			utils.PrintTable(os.Stdout, header, rows)
+			firstIteration = false
+		} else {
+			utils.PrintTable(os.Stdout, []string{}, rows)
+		}
 
-	if getAll {
-		fmt.Println()
+		return false, nil
+	})
+
+	return false, error
+}
+
+func validateWatchOption(cmd *cobra.Command, toMatch string) error {
+	w, _ := cmd.Flags().GetBool("watch")
+	if cmd.Use == toMatch && w {
+		return fmt.Errorf("expected a single resource type, but found %s", cmd.Use)
 	}
 	return nil
 }

cmd/flux/get_alert.go

@@ -17,10 +17,12 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	notificationv1 "github.com/fluxcd/notification-controller/api/v1beta1"
 )
@@ -32,10 +34,39 @@ var getAlertCmd = &cobra.Command{
 	Long:    "The get alert command prints the statuses of the resources.",
 	Example: `  # List all Alerts and their status
   flux get alerts`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: alertType,
 			list:    &alertListAdapter{&notificationv1.AlertList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*notificationv1.Alert)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v alert", obj)
+			}
+
+			sink := alertListAdapter{
+				&notificationv1.AlertList{
+					Items: []notificationv1.Alert{
+						*o,
+					},
+				},
+			}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_alertprovider.go

@@ -17,7 +17,10 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
+
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	notificationv1 "github.com/fluxcd/notification-controller/api/v1beta1"
 )
@@ -29,10 +32,39 @@ var getAlertProviderCmd = &cobra.Command{
 	Long:    "The get alert-provider command prints the statuses of the resources.",
 	Example: `  # List all Providers and their status
   flux get alert-providers`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: alertProviderType,
 			list:    alertProviderListAdapter{&notificationv1.ProviderList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*notificationv1.Provider)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v alert-provider", obj)
+			}
+
+			sink := alertProviderListAdapter{
+				&notificationv1.ProviderList{
+					Items: []notificationv1.Provider{
+						*o,
+					},
+				},
+			}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_all.go

@@ -36,7 +36,12 @@ var getAllCmd = &cobra.Command{
   # List all resources in all namespaces
   flux get all --all-namespaces`,
 	RunE: func(cmd *cobra.Command, args []string) error {
-		err := getSourceAllCmd.RunE(cmd, args)
+		err := validateWatchOption(cmd, "all")
+		if err != nil {
+			return err
+		}
+
+		err = getSourceAllCmd.RunE(cmd, args)
 		if err != nil {
 			logError(err)
 		}

cmd/flux/get_helmrelease.go

@@ -17,11 +17,13 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	helmv2 "github.com/fluxcd/helm-controller/api/v2beta1"
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 )
 
 var getHelmReleaseCmd = &cobra.Command{
@@ -31,10 +33,36 @@ var getHelmReleaseCmd = &cobra.Command{
 	Long:    "The get helmreleases command prints the statuses of the resources.",
 	Example: `  # List all Helm releases and their status
   flux get helmreleases`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: helmReleaseType,
 			list:    &helmReleaseListAdapter{&helmv2.HelmReleaseList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*helmv2.HelmRelease)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v helmrelease", obj)
+			}
+
+			sink := helmReleaseListAdapter{&helmv2.HelmReleaseList{
+				Items: []helmv2.HelmRelease{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_image.go

@@ -25,6 +25,9 @@ var getImageCmd = &cobra.Command{
 	Aliases: []string{"image"},
 	Short:   "Get image automation object status",
 	Long:    "The get image sub-commands print the status of image automation objects.",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return validateWatchOption(cmd, "images")
+	},
 }
 
 func init() {

cmd/flux/get_image_all.go

@@ -35,6 +35,11 @@ var getImageAllCmd = &cobra.Command{
   # List all image objects in all namespaces
   flux get images all --all-namespaces`,
 	RunE: func(cmd *cobra.Command, args []string) error {
+		err := validateWatchOption(cmd, "all")
+		if err != nil {
+			return err
+		}
+
 		var allImageCmd = []getCommand{
 			{
 				apiType: imageRepositoryType,

cmd/flux/get_image_policy.go

@@ -17,7 +17,10 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
+
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	imagev1 "github.com/fluxcd/image-reflector-controller/api/v1beta1"
 )
@@ -31,10 +34,36 @@ var getImagePolicyCmd = &cobra.Command{
 
  # List image policies from all namespaces
   flux get image policy --all-namespaces`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: imagePolicyType,
 			list:    &imagePolicyListAdapter{&imagev1.ImagePolicyList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*imagev1.ImagePolicy)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v policy", obj)
+			}
+
+			sink := imagePolicyListAdapter{&imagev1.ImagePolicyList{
+				Items: []imagev1.ImagePolicy{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_image_repository.go

@@ -17,11 +17,13 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	imagev1 "github.com/fluxcd/image-reflector-controller/api/v1beta1"
 )
@@ -35,10 +37,36 @@ var getImageRepositoryCmd = &cobra.Command{
 
  # List image repositories from all namespaces
   flux get image repository --all-namespaces`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: imageRepositoryType,
 			list:    imageRepositoryListAdapter{&imagev1.ImageRepositoryList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*imagev1.ImageRepository)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v repository", obj)
+			}
+
+			sink := imageRepositoryListAdapter{&imagev1.ImageRepositoryList{
+				Items: []imagev1.ImageRepository{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_image_update.go

@@ -17,11 +17,13 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	autov1 "github.com/fluxcd/image-automation-controller/api/v1beta1"
 )
@@ -35,10 +37,36 @@ var getImageUpdateCmd = &cobra.Command{
 
  # List image update automations from all namespaces
   flux get image update --all-namespaces`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: imageUpdateAutomationType,
 			list:    &imageUpdateAutomationListAdapter{&autov1.ImageUpdateAutomationList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*autov1.ImageUpdateAutomation)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v update", obj)
+			}
+
+			sink := imageUpdateAutomationListAdapter{&autov1.ImageUpdateAutomationList{
+				Items: []autov1.ImageUpdateAutomation{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_kustomization.go

@@ -17,10 +17,12 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	kustomizev1 "github.com/fluxcd/kustomize-controller/api/v1beta1"
 )
@@ -32,10 +34,39 @@ var getKsCmd = &cobra.Command{
 	Long:    "The get kustomizations command prints the statuses of the resources.",
 	Example: `  # List all kustomizations and their status
   flux get kustomizations`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: kustomizationType,
 			list:    &kustomizationListAdapter{&kustomizev1.KustomizationList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*kustomizev1.Kustomization)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v kustomization", obj)
+			}
+
+			sink := kustomizationListAdapter{
+				&kustomizev1.KustomizationList{
+					Items: []kustomizev1.Kustomization{
+						*o,
+					},
+				},
+			}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_receiver.go

@@ -17,10 +17,12 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	notificationv1 "github.com/fluxcd/notification-controller/api/v1beta1"
 )
@@ -32,10 +34,36 @@ var getReceiverCmd = &cobra.Command{
 	Long:    "The get receiver command prints the statuses of the resources.",
 	Example: `  # List all Receiver and their status
   flux get receivers`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: receiverType,
 			list:    receiverListAdapter{&notificationv1.ReceiverList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*notificationv1.Receiver)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v receiver", obj)
+			}
+
+			sink := receiverListAdapter{&notificationv1.ReceiverList{
+				Items: []notificationv1.Receiver{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_source.go

@@ -25,6 +25,10 @@ var getSourceCmd = &cobra.Command{
 	Aliases: []string{"source"},
 	Short:   "Get source statuses",
 	Long:    "The get source sub-commands print the statuses of the sources.",
+	RunE: func(cmd *cobra.Command, args []string) error {
+
+		return validateWatchOption(cmd, "sources")
+	},
 }
 
 func init() {

cmd/flux/get_source_all.go

@@ -34,6 +34,11 @@ var getSourceAllCmd = &cobra.Command{
   # List all sources in all namespaces
   flux get sources all --all-namespaces`,
 	RunE: func(cmd *cobra.Command, args []string) error {
+		err := validateWatchOption(cmd, "all")
+		if err != nil {
+			return err
+		}
+
 		var allSourceCmd = []getCommand{
 			{
 				apiType: bucketType,

cmd/flux/get_source_bucket.go

@@ -17,10 +17,12 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
 )
@@ -34,10 +36,36 @@ var getSourceBucketCmd = &cobra.Command{
 
  # List buckets from all namespaces
   flux get sources helm --all-namespaces`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: bucketType,
 			list:    &bucketListAdapter{&sourcev1.BucketList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*sourcev1.Bucket)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v bucket", obj)
+			}
+
+			sink := &bucketListAdapter{&sourcev1.BucketList{
+				Items: []sourcev1.Bucket{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_source_chart.go

@@ -17,10 +17,12 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
 )
@@ -34,10 +36,36 @@ var getSourceHelmChartCmd = &cobra.Command{
 
  # List Helm charts from all namespaces
   flux get sources chart --all-namespaces`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: helmChartType,
 			list:    &helmChartListAdapter{&sourcev1.HelmChartList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*sourcev1.HelmChart)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v chart", obj)
+			}
+
+			sink := &helmChartListAdapter{&sourcev1.HelmChartList{
+				Items: []sourcev1.HelmChart{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_source_git.go

@@ -17,10 +17,12 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
 )
@@ -34,10 +36,36 @@ var getSourceGitCmd = &cobra.Command{
 
  # List Git repositories from all namespaces
   flux get sources git --all-namespaces`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: gitRepositoryType,
 			list:    &gitRepositoryListAdapter{&sourcev1.GitRepositoryList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*sourcev1.GitRepository)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v git", obj)
+			}
+
+			sink := &gitRepositoryListAdapter{&sourcev1.GitRepositoryList{
+				Items: []sourcev1.GitRepository{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/get_source_helm.go

@@ -17,10 +17,12 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
 )
@@ -34,10 +36,36 @@ var getSourceHelmCmd = &cobra.Command{
 
  # List Helm repositories from all namespaces
   flux get sources helm --all-namespaces`,
-	RunE: getCommand{
+	RunE: func(cmd *cobra.Command, args []string) error {
+		get := getCommand{
 			apiType: helmRepositoryType,
 			list:    &helmRepositoryListAdapter{&sourcev1.HelmRepositoryList{}},
-	}.run,
+			funcMap: make(typeMap),
+		}
+
+		err := get.funcMap.registerCommand(get.apiType.kind, func(obj runtime.Object) (summarisable, error) {
+			o, ok := obj.(*sourcev1.HelmRepository)
+			if !ok {
+				return nil, fmt.Errorf("Impossible to cast type %#v helm", obj)
+			}
+
+			sink := &helmRepositoryListAdapter{&sourcev1.HelmRepositoryList{
+				Items: []sourcev1.HelmRepository{
+					*o,
+				}}}
+			return sink, nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		if err := get.run(cmd, args); err != nil {
+			return err
+		}
+
+		return nil
+	},
 }
 
 func init() {

cmd/flux/main_test.go

@@ -50,10 +50,10 @@ func readYamlObjects(objectFile string) ([]client.Object, error) {
 
 // A KubeManager that can create objects that are subject to a test.
 type fakeKubeManager struct {
-	fakeClient client.Client
+	fakeClient client.WithWatch
 }
 
-func (m *fakeKubeManager) NewClient(kubeconfig string, kubecontext string) (client.Client, error) {
+func (m *fakeKubeManager) NewClient(kubeconfig string, kubecontext string) (client.WithWatch, error) {
 	return m.fakeClient, nil
 }
 

go.sum

@@ -336,6 +336,7 @@ github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4er
 github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
+github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e h1:1r7pUrabqp18hOBcwBwiTsbnFeTZHV9eER/QT5JVZxY=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
@@ -398,6 +399,7 @@ github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
@@ -449,6 +451,7 @@ github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.3/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=

internal/utils/utils.go

@@ -134,7 +134,7 @@ func KubeConfig(kubeConfigPath string, kubeContext string) (*rest.Config, error)
 // KubeManger creates a Kubernetes client.Client. This interface exists to
 // facilitate unit testing and provide a fake client.
 type KubeManager interface {
-	NewClient(string, string) (client.Client, error)
+	NewClient(string, string) (client.WithWatch, error)
 }
 
 type defaultKubeManager struct{}
@@ -144,14 +144,14 @@ func DefaultKubeManager() KubeManager {
 	return manager
 }
 
-func (m defaultKubeManager) NewClient(kubeConfigPath string, kubeContext string) (client.Client, error) {
+func (m defaultKubeManager) NewClient(kubeConfigPath string, kubeContext string) (client.WithWatch, error) {
 	cfg, err := KubeConfig(kubeConfigPath, kubeContext)
 	if err != nil {
 		return nil, fmt.Errorf("kubernetes client initialization failed: %w", err)
 	}
 
 	scheme := NewScheme()
-	kubeClient, err := client.New(cfg, client.Options{
+	kubeClient, err := client.NewWithWatch(cfg, client.Options{
 		Scheme: scheme,
 	})
 	if err != nil {
@@ -179,7 +179,7 @@ func NewScheme() *apiruntime.Scheme {
 	return scheme
 }
 
-func KubeClient(kubeConfigPath string, kubeContext string) (client.Client, error) {
+func KubeClient(kubeConfigPath string, kubeContext string) (client.WithWatch, error) {
 	m := DefaultKubeManager()
 	kubeClient, err := m.NewClient(kubeConfigPath, kubeContext)
 	return kubeClient, err