Remove dependency of generated client part 2

Remove dependecy of generate client from pkg/cmd/cli/snapshotLocation.
Remove the Velero generated informer from PVB and PVR. 
Remove dependency of generated client from pkg/podvolume directory.
Replace generated codec with runtime codec. 

Signed-off-by: Xun Jiang <[email protected]>

changelogs/unreleased/7041-blackpiglet

@@ -0,0 +1 @@
+Remove the Velero generated client.

internal/storage/storagelocation_test.go

@@ -26,8 +26,8 @@ import (
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
-	"github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/scheme"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util"
 )
 
 func TestIsReadyToValidate(t *testing.T) {
@@ -163,7 +163,7 @@ func TestListBackupStorageLocations(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			client := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithRuntimeObjects(tt.backupLocations).Build()
+			client := fake.NewClientBuilder().WithScheme(util.VeleroScheme).WithRuntimeObjects(tt.backupLocations).Build()
 			if tt.expectError {
 				_, err := ListBackupStorageLocations(context.Background(), client, "ns-1")
 				g.Expect(err).NotTo(BeNil())

pkg/client/retry.go

@@ -25,20 +25,14 @@ import (
 )
 
 func CreateRetryGenerateName(client kbclient.Client, ctx context.Context, obj kbclient.Object) error {
-	return CreateRetryGenerateNameWithFunc(obj, func() error {
-		return client.Create(ctx, obj, &kbclient.CreateOptions{})
-	})
-}
-
-func CreateRetryGenerateNameWithFunc(obj kbclient.Object, createFn func() error) error {
 	retryCreateFn := func() error {
 		// needed to ensure that the name from the failed create isn't left on the object between retries
 		obj.SetName("")
-		return createFn()
+		return client.Create(ctx, obj, &kbclient.CreateOptions{})
 	}
 	if obj.GetGenerateName() != "" && obj.GetName() == "" {
 		return retry.OnError(retry.DefaultRetry, apierrors.IsAlreadyExists, retryCreateFn)
 	} else {
-		return createFn()
+		return client.Create(ctx, obj, &kbclient.CreateOptions{})
 	}
 }

pkg/cmd/server/server.go

@@ -66,7 +66,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/controller"
 	velerodiscovery "github.com/vmware-tanzu/velero/pkg/discovery"
 	"github.com/vmware-tanzu/velero/pkg/features"
-	clientset "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned"
 	"github.com/vmware-tanzu/velero/pkg/itemoperationmap"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
@@ -250,12 +249,12 @@ type server struct {
 	metricsAddress        string
 	kubeClientConfig      *rest.Config
 	kubeClient            kubernetes.Interface
-	veleroClient          clientset.Interface
 	discoveryClient       discovery.DiscoveryInterface
 	discoveryHelper       velerodiscovery.Helper
 	dynamicClient         dynamic.Interface
 	csiSnapshotClient     *snapshotv1client.Clientset
 	csiSnapshotLister     snapshotv1listers.VolumeSnapshotLister
+	crClient              ctrlclient.Client
 	ctx                   context.Context
 	cancelFunc            context.CancelFunc
 	logger                logrus.FieldLogger
@@ -305,6 +304,11 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
 		return nil, err
 	}
 
+	crClient, err := f.KubebuilderClient()
+	if err != nil {
+		return nil, err
+	}
+
 	pluginRegistry := process.NewRegistry(config.pluginDir, logger, logger.Level)
 	if err := pluginRegistry.DiscoverPlugins(); err != nil {
 		return nil, err
@@ -380,9 +384,9 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
 		metricsAddress:        config.metricsAddress,
 		kubeClientConfig:      clientConfig,
 		kubeClient:            kubeClient,
-		veleroClient:          veleroClient,
 		discoveryClient:       veleroClient.Discovery(),
 		dynamicClient:         dynamicClient,
+		crClient:              crClient,
 		ctx:                   ctx,
 		cancelFunc:            cancelFunc,
 		logger:                logger,
@@ -727,6 +731,11 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		s.logger.Fatal(err, "unable to create controller", "controller", controller.BackupStorageLocation)
 	}
 
+	pvbInformer, err := s.mgr.GetCache().GetInformer(s.ctx, &velerov1api.PodVolumeBackup{})
+	if err != nil {
+		s.logger.Fatal(err, "fail to get controller-runtime informer from manager for PVB")
+	}
+
 	if _, ok := enabledRuntimeControllers[controller.Backup]; ok {
 		backupper, err := backup.NewKubernetesBackupper(
 			s.mgr.GetClient(),
@@ -736,10 +745,8 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			podvolume.NewBackupperFactory(
 				s.repoLocker,
 				s.repoEnsurer,
-				s.veleroClient,
-				s.kubeClient.CoreV1(),
-				s.kubeClient.CoreV1(),
-				s.kubeClient.CoreV1(),
+				s.crClient,
+				pvbInformer,
 				s.logger,
 			),
 			s.config.podVolumeOperationTimeout,
@@ -818,10 +825,8 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			podvolume.NewBackupperFactory(
 				s.repoLocker,
 				s.repoEnsurer,
-				s.veleroClient,
-				s.kubeClient.CoreV1(),
-				s.kubeClient.CoreV1(),
-				s.kubeClient.CoreV1(),
+				s.crClient,
+				pvbInformer,
 				s.logger,
 			),
 			s.config.podVolumeOperationTimeout,
@@ -909,6 +914,11 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		}
 	}
 
+	pvrInformer, err := s.mgr.GetCache().GetInformer(s.ctx, &velerov1api.PodVolumeRestore{})
+	if err != nil {
+		s.logger.Fatal(err, "fail to get controller-runtime informer from manager for PVR")
+	}
+
 	if _, ok := enabledRuntimeControllers[controller.Restore]; ok {
 		restorer, err := restore.NewKubernetesRestorer(
 			s.discoveryHelper,
@@ -918,10 +928,9 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			podvolume.NewRestorerFactory(
 				s.repoLocker,
 				s.repoEnsurer,
-				s.veleroClient,
-				s.kubeClient.CoreV1(),
-				s.kubeClient.CoreV1(),
 				s.kubeClient,
+				s.crClient,
+				pvrInformer,
 				s.logger,
 			),
 			s.config.podVolumeOperationTimeout,

pkg/nodeagent/node_agent.go

@@ -22,13 +22,13 @@ import (
 
 	"github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
-
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
 const (
@@ -52,12 +52,18 @@ func IsRunning(ctx context.Context, kubeClient kubernetes.Interface, namespace s
 }
 
 // IsRunningInNode checks if the node agent pod is running properly in a specified node. If not, return the error found
-func IsRunningInNode(ctx context.Context, namespace string, nodeName string, podClient corev1client.PodsGetter) error {
+func IsRunningInNode(ctx context.Context, namespace string, nodeName string, crClient ctrlclient.Client) error {
 	if nodeName == "" {
 		return errors.New("node name is empty")
 	}
 
-	pods, err := podClient.Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("name=%s", daemonSet)})
+	pods := new(v1.PodList)
+	parsedSelector, err := labels.Parse(fmt.Sprintf("name=%s", daemonSet))
+	if err != nil {
+		return errors.Wrap(err, "fail to parse selector")
+	}
+
+	err = crClient.List(ctx, pods, &ctrlclient.ListOptions{LabelSelector: parsedSelector})
 	if err != nil {
 		return errors.Wrap(err, "failed to list daemonset pods")
 	}

pkg/persistence/object_store.go

@@ -27,13 +27,14 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	"github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/scheme"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/volume"
 )
 
@@ -302,7 +303,9 @@ func (s *objectBackupStore) GetBackupMetadata(name string) (*velerov1api.Backup,
 		return nil, errors.WithStack(err)
 	}
 
-	decoder := scheme.Codecs.UniversalDecoder(velerov1api.SchemeGroupVersion)
+	codecFactory := serializer.NewCodecFactory(util.VeleroScheme)
+
+	decoder := codecFactory.UniversalDecoder(velerov1api.SchemeGroupVersion)
 	obj, _, err := decoder.Decode(data, nil, nil)
 	if err != nil {
 		return nil, errors.WithStack(err)

pkg/podvolume/backupper.go

@@ -26,13 +26,13 @@ import (
 	corev1api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
-	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/tools/cache"
+	ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	veleroclient "github.com/vmware-tanzu/velero/pkg/client"
-	clientset "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned"
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	"github.com/vmware-tanzu/velero/pkg/repository"
@@ -50,10 +50,7 @@ type backupper struct {
 	ctx          context.Context
 	repoLocker   *repository.RepoLocker
 	repoEnsurer  *repository.Ensurer
-	veleroClient clientset.Interface
-	pvcClient    corev1client.PersistentVolumeClaimsGetter
-	pvClient     corev1client.PersistentVolumesGetter
-	podClient    corev1client.PodsGetter
+	crClient     ctrlclient.Client
 	uploaderType string
 
 	results     map[string]chan *velerov1api.PodVolumeBackup
@@ -103,32 +100,31 @@ func newBackupper(
 	ctx context.Context,
 	repoLocker *repository.RepoLocker,
 	repoEnsurer *repository.Ensurer,
-	podVolumeBackupInformer cache.SharedIndexInformer,
-	veleroClient clientset.Interface,
-	pvcClient corev1client.PersistentVolumeClaimsGetter,
-	pvClient corev1client.PersistentVolumesGetter,
-	podClient corev1client.PodsGetter,
+	pvbInformer ctrlcache.Informer,
+	crClient ctrlclient.Client,
 	uploaderType string,
+	backup *velerov1api.Backup,
 	log logrus.FieldLogger,
 ) *backupper {
 	b := &backupper{
 		ctx:          ctx,
 		repoLocker:   repoLocker,
 		repoEnsurer:  repoEnsurer,
-		veleroClient: veleroClient,
-		pvcClient:    pvcClient,
-		pvClient:     pvClient,
-		podClient:    podClient,
+		crClient:     crClient,
 		uploaderType: uploaderType,
 
 		results: make(map[string]chan *velerov1api.PodVolumeBackup),
 	}
 
-	podVolumeBackupInformer.AddEventHandler(
+	pvbInformer.AddEventHandler(
 		cache.ResourceEventHandlerFuncs{
 			UpdateFunc: func(_, obj interface{}) {
 				pvb := obj.(*velerov1api.PodVolumeBackup)
 
+				if pvb.GetLabels()[velerov1api.BackupUIDLabel] != string(backup.UID) {
+					return
+				}
+
 				if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseCompleted || pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseFailed {
 					b.resultsLock.Lock()
 					defer b.resultsLock.Unlock()
@@ -153,7 +149,8 @@ func resultsKey(ns, name string) string {
 
 func (b *backupper) getMatchAction(resPolicies *resourcepolicies.Policies, pvc *corev1api.PersistentVolumeClaim, volume *corev1api.Volume) (*resourcepolicies.Action, error) {
 	if pvc != nil {
-		pv, err := b.pvClient.PersistentVolumes().Get(context.TODO(), pvc.Spec.VolumeName, metav1.GetOptions{})
+		pv := new(corev1api.PersistentVolume)
+		err := b.crClient.Get(context.TODO(), ctrlclient.ObjectKey{Name: pvc.Spec.VolumeName}, pv)
 		if err != nil {
 			return nil, errors.Wrapf(err, "error getting pv for pvc %s", pvc.Spec.VolumeName)
 		}
@@ -173,7 +170,7 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 	}
 	log.Infof("pod %s/%s has volumes to backup: %v", pod.Namespace, pod.Name, volumesToBackup)
 
-	err := nodeagent.IsRunningInNode(b.ctx, backup.Namespace, pod.Spec.NodeName, b.podClient)
+	err := nodeagent.IsRunningInNode(b.ctx, backup.Namespace, pod.Spec.NodeName, b.crClient)
 	if err != nil {
 		return nil, nil, []error{err}
 	}
@@ -213,7 +210,8 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 	for _, podVolume := range pod.Spec.Volumes {
 		podVolumes[podVolume.Name] = podVolume
 		if podVolume.PersistentVolumeClaim != nil {
-			pvc, err := b.pvcClient.PersistentVolumeClaims(pod.Namespace).Get(context.TODO(), podVolume.PersistentVolumeClaim.ClaimName, metav1.GetOptions{})
+			pvc := new(corev1api.PersistentVolumeClaim)
+			err := b.crClient.Get(context.TODO(), ctrlclient.ObjectKey{Namespace: pod.Namespace, Name: podVolume.PersistentVolumeClaim.ClaimName}, pvc)
 			if err != nil {
 				errs = append(errs, errors.Wrap(err, "error getting persistent volume claim for volume"))
 				continue
@@ -263,7 +261,7 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 
 		// hostPath volumes are not supported because they're not mounted into /var/lib/kubelet/pods, so our
 		// daemonset pod has no way to access their data.
-		isHostPath, err := isHostPathVolume(&volume, pvc, b.pvClient.PersistentVolumes())
+		isHostPath, err := isHostPathVolume(&volume, pvc, b.crClient)
 		if err != nil {
 			errs = append(errs, errors.Wrap(err, "error checking if volume is a hostPath volume"))
 			continue
@@ -303,11 +301,7 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 		}
 
 		volumeBackup := newPodVolumeBackup(backup, pod, volume, repoIdentifier, b.uploaderType, pvc)
-		// TODO: once backupper is refactored to use controller-runtime, just pass client instead of anonymous func
-		if err := veleroclient.CreateRetryGenerateNameWithFunc(volumeBackup, func() error {
-			_, err := b.veleroClient.VeleroV1().PodVolumeBackups(volumeBackup.Namespace).Create(context.TODO(), volumeBackup, metav1.CreateOptions{})
-			return err
-		}); err != nil {
+		if err := veleroclient.CreateRetryGenerateName(b.crClient, b.ctx, volumeBackup); err != nil {
 			errs = append(errs, err)
 			continue
 		}
@@ -339,13 +333,9 @@ ForEachVolume:
 	return podVolumeBackups, pvcSummary, errs
 }
 
-type pvGetter interface {
-	Get(ctx context.Context, name string, opts metav1.GetOptions) (*corev1api.PersistentVolume, error)
-}
-
 // isHostPathVolume returns true if the volume is either a hostPath pod volume or a persistent
 // volume claim on a hostPath persistent volume, or false otherwise.
-func isHostPathVolume(volume *corev1api.Volume, pvc *corev1api.PersistentVolumeClaim, pvGetter pvGetter) (bool, error) {
+func isHostPathVolume(volume *corev1api.Volume, pvc *corev1api.PersistentVolumeClaim, crClient ctrlclient.Client) (bool, error) {
 	if volume.HostPath != nil {
 		return true, nil
 	}
@@ -354,7 +344,8 @@ func isHostPathVolume(volume *corev1api.Volume, pvc *corev1api.PersistentVolumeC
 		return false, nil
 	}
 
-	pv, err := pvGetter.Get(context.TODO(), pvc.Spec.VolumeName, metav1.GetOptions{})
+	pv := new(corev1api.PersistentVolume)
+	err := crClient.Get(context.TODO(), ctrlclient.ObjectKey{Name: pvc.Spec.VolumeName}, pv)
 	if err != nil {
 		return false, errors.WithStack(err)
 	}
@@ -421,7 +412,3 @@ func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volume c
 
 	return pvb
 }
-
-func errorOnly(_ interface{}, err error) error {
-	return err
-}