data mover micro service restore

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

pkg/cmd/cli/datamover/restore.go

@@ -14,16 +14,37 @@ limitations under the License.
 package datamover
 
 import (
+	"context"
 	"fmt"
+	"os"
 	"strings"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
+	"github.com/vmware-tanzu/velero/internal/credentials"
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/buildinfo"
 	"github.com/vmware-tanzu/velero/pkg/client"
+	"github.com/vmware-tanzu/velero/pkg/cmd/util/signals"
+	"github.com/vmware-tanzu/velero/pkg/datamover"
+	"github.com/vmware-tanzu/velero/pkg/datapath"
+	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 	"github.com/vmware-tanzu/velero/pkg/util/logging"
+
+	ctlcache "sigs.k8s.io/controller-runtime/pkg/cache"
+	ctlclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type dataMoverRestoreConfig struct {
@@ -52,7 +73,10 @@ func NewRestoreCommand(f client.Factory) *cobra.Command {
 			logger.Infof("Starting Velero data-mover restore %s (%s)", buildinfo.Version, buildinfo.FormattedGitSHA())
 
 			f.SetBasename(fmt.Sprintf("%s-%s", c.Parent().Name(), c.Name()))
-			s := newdataMoverRestore(logger, config)
+			s, err := newdataMoverRestore(logger, f, config)
+			if err != nil {
+				exitWithMessage(logger, false, "Failed to create data mover restore, %v", err)
+			}
 
 			s.run()
 		},
@@ -74,19 +98,169 @@ func NewRestoreCommand(f client.Factory) *cobra.Command {
 }
 
 type dataMoverRestore struct {
-	logger logrus.FieldLogger
-	config dataMoverRestoreConfig
+	logger      logrus.FieldLogger
+	ctx         context.Context
+	cancelFunc  context.CancelFunc
+	client      ctlclient.Client
+	cache       ctlcache.Cache
+	namespace   string
+	nodeName    string
+	config      dataMoverRestoreConfig
+	kubeClient  kubernetes.Interface
+	dataPathMgr *datapath.Manager
 }
 
-func newdataMoverRestore(logger logrus.FieldLogger, config dataMoverRestoreConfig) *dataMoverRestore {
+func newdataMoverRestore(logger logrus.FieldLogger, factory client.Factory, config dataMoverRestoreConfig) (*dataMoverRestore, error) {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+
+	clientConfig, err := factory.ClientConfig()
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create client config")
+	}
+
+	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
+
+	scheme := runtime.NewScheme()
+	if err := velerov1api.AddToScheme(scheme); err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to add velero v1 scheme")
+	}
+
+	if err := velerov2alpha1api.AddToScheme(scheme); err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to add velero v2alpha1 scheme")
+	}
+
+	if err := v1.AddToScheme(scheme); err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to add core v1 scheme")
+	}
+
+	nodeName := os.Getenv("NODE_NAME")
+
+	// use a field selector to filter to only pods scheduled on this node.
+	cacheOption := ctlcache.Options{
+		Scheme: scheme,
+		ByObject: map[ctlclient.Object]ctlcache.ByObject{
+			&v1.Pod{}: {
+				Field: fields.Set{"spec.nodeName": nodeName}.AsSelector(),
+			},
+			&velerov2alpha1api.DataDownload{}: {
+				Field: fields.Set{"metadata.namespace": factory.Namespace()}.AsSelector(),
+			},
+		},
+	}
+
+	cli, err := ctlclient.New(clientConfig, ctlclient.Options{
+		Scheme: scheme,
+	})
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create client")
+	}
+
+	cache, err := ctlcache.New(clientConfig, cacheOption)
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create client cache")
+	}
+
 	s := &dataMoverRestore{
-		logger: logger,
-		config: config,
+		logger:     logger,
+		ctx:        ctx,
+		cancelFunc: cancelFunc,
+		client:     cli,
+		cache:      cache,
+		config:     config,
+		namespace:  factory.Namespace(),
+		nodeName:   nodeName,
+	}
+
+	s.kubeClient, err = factory.KubeClient()
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create kube client")
 	}
 
-	return s
+	s.dataPathMgr = datapath.NewManager(1)
+
+	return s, nil
 }
 
+var funcCreateDataPathRestore = (*dataMoverRestore).createDataPathService
+
 func (s *dataMoverRestore) run() {
+	signals.CancelOnShutdown(s.cancelFunc, s.logger)
+	go func() {
+		if err := s.cache.Start(s.ctx); err != nil {
+			s.logger.WithError(err).Warn("error starting cache")
+		}
+	}()
+
+	// TODOOO: call s.runDataPath()
 	time.Sleep(time.Duration(1<<63 - 1))
 }
+
+func (s *dataMoverRestore) runDataPath() {
+	s.logger.Infof("Starting micro service in node %s for dd %s", s.nodeName, s.config.ddName)
+
+	dpService, err := funcCreateDataPathRestore(s)
+	if err != nil {
+		s.cancelFunc()
+		funcExitWithMessage(s.logger, false, "Failed to create data path service for DataDownload %s: %v", s.config.ddName, err)
+		return
+	}
+
+	s.logger.Infof("Starting data path service %s", s.config.ddName)
+
+	dpService.Init()
+
+	result, err := dpService.RunCancelableDataPath(s.ctx)
+	if err != nil {
+		s.cancelFunc()
+		funcExitWithMessage(s.logger, false, "Failed to run data path service for DataDownload %s: %v", s.config.ddName, err)
+		return
+	}
+
+	s.logger.WithField("dd", s.config.ddName).Info("Data path service completed")
+
+	dpService.Shutdown()
+
+	s.logger.WithField("dd", s.config.ddName).Info("Data path service is shut down")
+
+	s.cancelFunc()
+
+	funcExitWithMessage(s.logger, true, result)
+}
+
+func (s *dataMoverRestore) createDataPathService() (dataPathService, error) {
+	credentialFileStore, err := funcNewCredentialFileStore(
+		s.client,
+		s.namespace,
+		defaultCredentialsDirectory,
+		filesystem.NewFileSystem(),
+	)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error to create credential file store")
+	}
+
+	credSecretStore, err := funcNewCredentialSecretStore(s.client, s.namespace)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error to create credential secret store")
+	}
+
+	credGetter := &credentials.CredentialGetter{FromFile: credentialFileStore, FromSecret: credSecretStore}
+
+	duInformer, err := s.cache.GetInformer(s.ctx, &velerov2alpha1api.DataDownload{})
+	if err != nil {
+		return nil, errors.Wrap(err, "error to get controller-runtime informer from manager")
+	}
+
+	repoEnsurer := repository.NewEnsurer(s.client, s.logger, s.config.resourceTimeout)
+
+	return datamover.NewRestoreMicroService(s.ctx, s.client, s.kubeClient, s.config.ddName, s.namespace, s.nodeName, datapath.AccessPoint{
+		ByPath:  s.config.volumePath,
+		VolMode: uploader.PersistentVolumeMode(s.config.volumeMode),
+	}, s.dataPathMgr, repoEnsurer, credGetter, duInformer, s.logger), nil
+}

pkg/cmd/cli/datamover/restore_test.go

@@ -0,0 +1,158 @@
+/*
+Copyright The Velero Contributors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package datamover
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	cacheMock "github.com/vmware-tanzu/velero/pkg/cmd/cli/datamover/mocks"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func fakeCreateDataPathRestoreWithErr(_ *dataMoverRestore) (dataPathService, error) {
+	return nil, errors.New("fake-create-data-path-error")
+}
+
+func fakeCreateDataPathRestore(_ *dataMoverRestore) (dataPathService, error) {
+	return frHelper, nil
+}
+
+func TestRunDataPathRestore(t *testing.T) {
+	tests := []struct {
+		name                        string
+		ddName                      string
+		createDataPathFail          bool
+		runCancelableDataPathErr    error
+		runCancelableDataPathResult string
+		expectedMessage             string
+		expectedSucceed             bool
+	}{
+		{
+			name:               "create data path failed",
+			ddName:             "fake-name",
+			createDataPathFail: true,
+			expectedMessage:    "Failed to create data path service for DataDownload fake-name: fake-create-data-path-error",
+		},
+		{
+			name:                     "run data path failed",
+			ddName:                   "fake-name",
+			runCancelableDataPathErr: errors.New("fake-run-data-path-error"),
+			expectedMessage:          "Failed to run data path service for DataDownload fake-name: fake-run-data-path-error",
+		},
+		{
+			name:                        "succeed",
+			ddName:                      "fake-name",
+			runCancelableDataPathResult: "fake-run-data-path-result",
+			expectedMessage:             "fake-run-data-path-result",
+			expectedSucceed:             true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			frHelper = &fakeRunHelper{
+				runCancelableDataPathErr:    test.runCancelableDataPathErr,
+				runCancelableDataPathResult: test.runCancelableDataPathResult,
+			}
+
+			if test.createDataPathFail {
+				funcCreateDataPathRestore = fakeCreateDataPathRestoreWithErr
+			} else {
+				funcCreateDataPathRestore = fakeCreateDataPathRestore
+			}
+
+			funcExitWithMessage = frHelper.ExitWithMessage
+
+			s := &dataMoverRestore{
+				logger:     velerotest.NewLogger(),
+				cancelFunc: func() {},
+				config: dataMoverRestoreConfig{
+					ddName: test.ddName,
+				},
+			}
+
+			s.runDataPath()
+
+			assert.Equal(t, test.expectedMessage, frHelper.exitMessage)
+			assert.Equal(t, test.expectedSucceed, frHelper.succeed)
+		})
+	}
+}
+
+func TestCreateDataPathRestore(t *testing.T) {
+	tests := []struct {
+		name            string
+		fileStoreErr    error
+		secretStoreErr  error
+		mockGetInformer bool
+		getInformerErr  error
+		expectedError   string
+	}{
+		{
+			name:          "create credential file store error",
+			fileStoreErr:  errors.New("fake-file-store-error"),
+			expectedError: "error to create credential file store: fake-file-store-error",
+		},
+		{
+			name:           "create credential secret store",
+			secretStoreErr: errors.New("fake-secret-store-error"),
+			expectedError:  "error to create credential secret store: fake-secret-store-error",
+		},
+		{
+			name:            "get informer error",
+			mockGetInformer: true,
+			getInformerErr:  errors.New("fake-get-informer-error"),
+			expectedError:   "error to get controller-runtime informer from manager: fake-get-informer-error",
+		},
+		{
+			name:            "succeed",
+			mockGetInformer: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fcHelper := &fakeCreateDataPathServiceHelper{
+				fileStoreErr:   test.fileStoreErr,
+				secretStoreErr: test.secretStoreErr,
+			}
+
+			funcNewCredentialFileStore = fcHelper.NewNamespacedFileStore
+			funcNewCredentialSecretStore = fcHelper.NewNamespacedSecretStore
+
+			cache := cacheMock.NewCache(t)
+			if test.mockGetInformer {
+				cache.On("GetInformer", mock.Anything, mock.Anything).Return(nil, test.getInformerErr)
+			}
+
+			funcExitWithMessage = frHelper.ExitWithMessage
+
+			s := &dataMoverRestore{
+				cache: cache,
+			}
+
+			_, err := s.createDataPathService()
+
+			if test.expectedError != "" {
+				assert.EqualError(t, err, test.expectedError)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}