Remove Velero generated client from the pkg/podvolume directory.

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

pkg/client/retry.go

@@ -25,20 +25,18 @@ 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{})
-	})
+	return CreateRetryGenerateNameWithFunc(obj, client)
 }
 
-func CreateRetryGenerateNameWithFunc(obj kbclient.Object, createFn func() error) error {
+func CreateRetryGenerateNameWithFunc(obj kbclient.Object, crClient kbclient.Client) 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 crClient.Create(context.TODO(), obj)
 	}
 	if obj.GetGenerateName() != "" && obj.GetName() == "" {
 		return retry.OnError(retry.DefaultRetry, apierrors.IsAlreadyExists, retryCreateFn)
 	} else {
-		return createFn()
+		return crClient.Create(context.TODO(), obj)
 	}
 }

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"
@@ -246,30 +245,30 @@ func NewCommand(f client.Factory) *cobra.Command {
 }
 
 type server struct {
-	namespace             string
-	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
-	kubeWatchClient       ctrlclient.WithWatch
-	ctx                   context.Context
-	cancelFunc            context.CancelFunc
-	logger                logrus.FieldLogger
-	logLevel              logrus.Level
-	pluginRegistry        process.Registry
-	repoManager           repository.Manager
-	repoLocker            *repository.RepoLocker
-	repoEnsurer           *repository.Ensurer
-	metrics               *metrics.ServerMetrics
-	config                serverConfig
-	mgr                   manager.Manager
-	credentialFileStore   credentials.FileStore
-	credentialSecretStore credentials.SecretStore
+	namespace                    string
+	metricsAddress               string
+	kubeClientConfig             *rest.Config
+	kubeClient                   kubernetes.Interface
+	discoveryClient              discovery.DiscoveryInterface
+	discoveryHelper              velerodiscovery.Helper
+	dynamicClient                dynamic.Interface
+	csiSnapshotClient            *snapshotv1client.Clientset
+	csiSnapshotLister            snapshotv1listers.VolumeSnapshotLister
+	controllerRuntimeClient      ctrlclient.Client
+	controllerRuntimeWatchClient ctrlclient.WithWatch
+	ctx                          context.Context
+	cancelFunc                   context.CancelFunc
+	logger                       logrus.FieldLogger
+	logLevel                     logrus.Level
+	pluginRegistry               process.Registry
+	repoManager                  repository.Manager
+	repoLocker                   *repository.RepoLocker
+	repoEnsurer                  *repository.Ensurer
+	metrics                      *metrics.ServerMetrics
+	config                       serverConfig
+	mgr                          manager.Manager
+	credentialFileStore          credentials.FileStore
+	credentialSecretStore        credentials.SecretStore
 }
 
 func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*server, error) {
@@ -296,17 +295,17 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
 		return nil, err
 	}
 
-	veleroClient, err := f.Client()
+	dynamicClient, err := f.DynamicClient()
 	if err != nil {
 		return nil, err
 	}
 
-	dynamicClient, err := f.DynamicClient()
+	controllerRuntimeClient, err := f.KubebuilderClient()
 	if err != nil {
 		return nil, err
 	}
 
-	kubeWatchClient, err := f.KubebuilderWatchClient()
+	controllerRuntimeWatchClient, err := f.KubebuilderWatchClient()
 	if err != nil {
 		return nil, err
 	}
@@ -382,23 +381,23 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
 	}
 
 	s := &server{
-		namespace:             f.Namespace(),
-		metricsAddress:        config.metricsAddress,
-		kubeClientConfig:      clientConfig,
-		kubeClient:            kubeClient,
-		veleroClient:          veleroClient,
-		discoveryClient:       veleroClient.Discovery(),
-		dynamicClient:         dynamicClient,
-		kubeWatchClient:       kubeWatchClient,
-		ctx:                   ctx,
-		cancelFunc:            cancelFunc,
-		logger:                logger,
-		logLevel:              logger.Level,
-		pluginRegistry:        pluginRegistry,
-		config:                config,
-		mgr:                   mgr,
-		credentialFileStore:   credentialFileStore,
-		credentialSecretStore: credentialSecretStore,
+		namespace:                    f.Namespace(),
+		metricsAddress:               config.metricsAddress,
+		kubeClientConfig:             clientConfig,
+		kubeClient:                   kubeClient,
+		discoveryClient:              kubeClient.Discovery(),
+		dynamicClient:                dynamicClient,
+		controllerRuntimeClient:      controllerRuntimeClient,
+		controllerRuntimeWatchClient: controllerRuntimeWatchClient,
+		ctx:                          ctx,
+		cancelFunc:                   cancelFunc,
+		logger:                       logger,
+		logLevel:                     logger.Level,
+		pluginRegistry:               pluginRegistry,
+		config:                       config,
+		mgr:                          mgr,
+		credentialFileStore:          credentialFileStore,
+		credentialSecretStore:        credentialSecretStore,
 	}
 
 	// Setup CSI snapshot client and lister
@@ -596,7 +595,7 @@ var defaultRestorePriorities = restore.Priorities{
 
 func (s *server) checkNodeAgent() {
 	// warn if node agent does not exist
-	if err := nodeagent.IsRunning(s.ctx, s.kubeClient, s.namespace); err == nodeagent.ErrDaemonSetNotFound {
+	if err := nodeagent.IsRunning(s.ctx, s.controllerRuntimeClient, s.namespace); err == nodeagent.ErrDaemonSetNotFound {
 		s.logger.Warn("Velero node agent not found; pod volume backups/restores will not work until it's created")
 	} else if err != nil {
 		s.logger.WithError(errors.WithStack(err)).Warn("Error checking for existence of velero node agent")
@@ -743,11 +742,11 @@ 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.kubeWatchClient,
+				s.controllerRuntimeClient,
+				s.controllerRuntimeWatchClient,
 				s.logger,
 			),
 			s.config.podVolumeOperationTimeout,
@@ -826,11 +825,11 @@ 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.kubeWatchClient,
+				s.controllerRuntimeClient,
+				s.controllerRuntimeWatchClient,
 				s.logger,
 			),
 			s.config.podVolumeOperationTimeout,
@@ -927,11 +926,11 @@ 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.kubeWatchClient,
+				s.controllerRuntimeClient,
+				s.controllerRuntimeWatchClient,
 				s.logger,
 			),
 			s.config.podVolumeOperationTimeout,

pkg/nodeagent/node_agent.go

@@ -21,14 +21,15 @@ import (
 	"fmt"
 
 	"github.com/pkg/errors"
+	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/client-go/kubernetes"
-
-	"github.com/vmware-tanzu/velero/pkg/util/kube"
-
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
 const (
@@ -41,8 +42,9 @@ var (
 )
 
 // IsRunning checks if the node agent daemonset is running properly. If not, return the error found
-func IsRunning(ctx context.Context, kubeClient kubernetes.Interface, namespace string) error {
-	if _, err := kubeClient.AppsV1().DaemonSets(namespace).Get(ctx, daemonSet, metav1.GetOptions{}); apierrors.IsNotFound(err) {
+func IsRunning(ctx context.Context, kubeClient ctrlclient.Client, namespace string) error {
+	daemonSet := new(appsv1.DaemonSet)
+	if err := kubeClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonSet.String()}, daemonSet); apierrors.IsNotFound(err) {
 		return ErrDaemonSetNotFound
 	} else if err != nil {
 		return err

pkg/podvolume/backupper.go

@@ -28,11 +28,11 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/tools/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,10 @@ 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
@@ -104,10 +104,10 @@ func newBackupper(
 	repoLocker *repository.RepoLocker,
 	repoEnsurer *repository.Ensurer,
 	sharedInformer cache.SharedInformer,
-	veleroClient clientset.Interface,
 	pvcClient corev1client.PersistentVolumeClaimsGetter,
 	pvClient corev1client.PersistentVolumesGetter,
 	podClient corev1client.PodsGetter,
+	crClient ctrlclient.Client,
 	uploaderType string,
 	backup *velerov1api.Backup,
 	log logrus.FieldLogger,
@@ -116,10 +116,10 @@ func newBackupper(
 		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),
@@ -308,11 +308,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.CreateRetryGenerateNameWithFunc(volumeBackup, b.crClient); err != nil {
 			errs = append(errs, err)
 			continue
 		}

pkg/podvolume/backupper_factory.go

@@ -27,7 +27,6 @@ import (
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	clientset "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
@@ -41,20 +40,20 @@ type BackupperFactory interface {
 func NewBackupperFactory(
 	repoLocker *repository.RepoLocker,
 	repoEnsurer *repository.Ensurer,
-	veleroClient clientset.Interface,
 	pvcClient corev1client.PersistentVolumeClaimsGetter,
 	pvClient corev1client.PersistentVolumesGetter,
 	podClient corev1client.PodsGetter,
+	crClient ctrlclient.Client,
 	kubeWatchClient ctrlclient.WithWatch,
 	log logrus.FieldLogger,
 ) BackupperFactory {
 	return &backupperFactory{
 		repoLocker:      repoLocker,
 		repoEnsurer:     repoEnsurer,
-		veleroClient:    veleroClient,
 		pvcClient:       pvcClient,
 		pvClient:        pvClient,
 		podClient:       podClient,
+		crClient:        crClient,
 		kubeWatchClient: kubeWatchClient,
 		log:             log,
 	}
@@ -63,10 +62,10 @@ func NewBackupperFactory(
 type backupperFactory struct {
 	repoLocker      *repository.RepoLocker
 	repoEnsurer     *repository.Ensurer
-	veleroClient    clientset.Interface
 	pvcClient       corev1client.PersistentVolumeClaimsGetter
 	pvClient        corev1client.PersistentVolumesGetter
 	podClient       corev1client.PodsGetter
+	crClient        ctrlclient.Client
 	kubeWatchClient ctrlclient.WithWatch
 	log             logrus.FieldLogger
 }
@@ -80,7 +79,8 @@ func (bf *backupperFactory) NewBackupper(ctx context.Context, backup *velerov1ap
 
 	informer := cache.NewSharedInformer(&lw, &velerov1api.PodVolumeBackup{}, time.Second)
 
-	b := newBackupper(ctx, bf.repoLocker, bf.repoEnsurer, informer, bf.veleroClient, bf.pvcClient, bf.pvClient, bf.podClient, uploaderType, backup, bf.log)
+	b := newBackupper(ctx, bf.repoLocker, bf.repoEnsurer, informer, bf.pvcClient,
+		bf.pvClient, bf.podClient, bf.crClient, uploaderType, backup, bf.log)
 
 	go informer.Run(ctx.Done())
 	if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {

pkg/podvolume/backupper_test.go

@@ -39,7 +39,6 @@ import (
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	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"
 	velerofake "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/fake"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
@@ -639,7 +638,6 @@ func TestBackupPodVolumes(t *testing.T) {
 			for _, reactor := range test.veleroReactors {
 				fakeVeleroClient.Fake.PrependReactor(reactor.verb, reactor.resource, reactor.reactorFunc)
 			}
-			var veleroClient versioned.Interface = fakeVeleroClient
 
 			fakeKubeWatchClient := velerotest.NewFakeControllerRuntimeWatchClient(t, test.kubeClientObj...)
 
@@ -648,7 +646,8 @@ func TestBackupPodVolumes(t *testing.T) {
 			backupObj := builder.ForBackup(velerov1api.DefaultNamespace, "fake-backup").Result()
 			backupObj.Spec.StorageLocation = test.bsl
 
-			factory := NewBackupperFactory(repository.NewRepoLocker(), ensurer, veleroClient, kubeClient.CoreV1(), kubeClient.CoreV1(), kubeClient.CoreV1(), fakeKubeWatchClient, velerotest.NewLogger())
+			factory := NewBackupperFactory(repository.NewRepoLocker(), ensurer, kubeClient.CoreV1(),
+				kubeClient.CoreV1(), kubeClient.CoreV1(), fakeCtlClient, fakeKubeWatchClient, velerotest.NewLogger())
 			bp, err := factory.NewBackupper(ctx, backupObj, test.uploaderType)
 
 			require.NoError(t, err)