mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-09-20 15:04:17 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/qiuming-best/velero into kopia-parallelism

Browse Source
This commit is contained in:
Ming Qiu
2023-11-22 01:42:39 +00:00
parent 481cb60493 e58a7808e0
commit fea22bbbc9
192 changed files with 6052 additions and 1305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cmd/cli/debug/debug.go
+16 -11
View File
@@ -30,9 +30,12 @@ import (
"github.com/spf13/cobra"
"github.com/spf13/pflag"
"github.com/vmware-tanzu/crash-diagnostics/exec"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
appsv1 "k8s.io/api/apps/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/client-go/tools/clientcmd"
ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/client"
"github.com/vmware-tanzu/velero/pkg/cmd"
)
@@ -110,30 +113,32 @@ func (o *option) complete(f client.Factory, fs *pflag.FlagSet) error {
}
func (o *option) validate(f client.Factory) error {
kubeClient, err := f.KubeClient()
crClient, err := f.KubebuilderClient()
if err != nil {
return err
}
l, err := kubeClient.AppsV1().Deployments(o.namespace).List(context.TODO(), metav1.ListOptions{
LabelSelector: "component=velero",
deploymentList := new(appsv1.DeploymentList)
selector, err := labels.Parse("component=velero")
cmd.CheckError(err)
err = crClient.List(context.TODO(), deploymentList, &ctrlclient.ListOptions{
Namespace: o.namespace,
LabelSelector: selector,
})
if err != nil {
return errors.Wrap(err, "failed to check velero deployment")
}
if len(l.Items) == 0 {
if len(deploymentList.Items) == 0 {
return fmt.Errorf("velero deployment does not exist in namespace: %s", o.namespace)
}
veleroClient, err := f.Client()
if err != nil {
return err
}
if len(o.backup) > 0 {
if _, err := veleroClient.VeleroV1().Backups(o.namespace).Get(context.TODO(), o.backup, metav1.GetOptions{}); err != nil {
backup := new(velerov1api.Backup)
if err := crClient.Get(context.TODO(), ctrlclient.ObjectKey{Namespace: o.namespace, Name: o.backup}, backup); err != nil {
return err
}
}
if len(o.restore) > 0 {
if _, err := veleroClient.VeleroV1().Restores(o.namespace).Get(context.TODO(), o.restore, metav1.GetOptions{}); err != nil {
restore := new(velerov1api.Restore)
if err := crClient.Get(context.TODO(), ctrlclient.ObjectKey{Namespace: o.namespace, Name: o.restore}, restore); err != nil {
return err
}
}
pkg/cmd/cli/nodeagent/server.go
+76 -4
View File
@@ -19,6 +19,7 @@ package nodeagent
import (
"context"
"fmt"
"math"
"net/http"
"os"
"strings"
@@ -32,6 +33,7 @@ import (
storagev1api "k8s.io/api/storage/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
@@ -53,7 +55,9 @@ import (
"github.com/vmware-tanzu/velero/pkg/cmd"
"github.com/vmware-tanzu/velero/pkg/cmd/util/signals"
"github.com/vmware-tanzu/velero/pkg/controller"
"github.com/vmware-tanzu/velero/pkg/datapath"
"github.com/vmware-tanzu/velero/pkg/metrics"
"github.com/vmware-tanzu/velero/pkg/nodeagent"
"github.com/vmware-tanzu/velero/pkg/repository"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
"github.com/vmware-tanzu/velero/pkg/util/logging"
@@ -73,6 +77,7 @@ const (
defaultResourceTimeout = 10 * time.Minute
defaultDataMoverPrepareTimeout = 30 * time.Minute
defaultDataPathConcurrentNum = 1
)
type nodeAgentServerConfig struct {
@@ -132,6 +137,7 @@ type nodeAgentServer struct {
config nodeAgentServerConfig
kubeClient kubernetes.Interface
csiSnapshotClient *snapshotv1client.Clientset
dataPathMgr *datapath.Manager
}
func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, config nodeAgentServerConfig) (*nodeAgentServer, error) {
@@ -219,6 +225,10 @@ func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, confi
if err != nil {
return nil, err
}
dataPathConcurrentNum := s.getDataPathConcurrentNum(defaultDataPathConcurrentNum)
s.dataPathMgr = datapath.NewManager(dataPathConcurrentNum)
return s, nil
}
@@ -263,24 +273,24 @@ func (s *nodeAgentServer) run() {
credentialGetter := &credentials.CredentialGetter{FromFile: credentialFileStore, FromSecret: credSecretStore}
repoEnsurer := repository.NewEnsurer(s.mgr.GetClient(), s.logger, s.config.resourceTimeout)
pvbReconciler := controller.NewPodVolumeBackupReconciler(s.mgr.GetClient(), repoEnsurer,
pvbReconciler := controller.NewPodVolumeBackupReconciler(s.mgr.GetClient(), s.dataPathMgr, repoEnsurer,
credentialGetter, s.nodeName, s.mgr.GetScheme(), s.metrics, s.logger)
if err := pvbReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.Fatal(err, "unable to create controller", "controller", controller.PodVolumeBackup)
}
if err = controller.NewPodVolumeRestoreReconciler(s.mgr.GetClient(), repoEnsurer, credentialGetter, s.logger).SetupWithManager(s.mgr); err != nil {
if err = controller.NewPodVolumeRestoreReconciler(s.mgr.GetClient(), s.dataPathMgr, repoEnsurer, credentialGetter, s.logger).SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the pod volume restore controller")
}
dataUploadReconciler := controller.NewDataUploadReconciler(s.mgr.GetClient(), s.kubeClient, s.csiSnapshotClient.SnapshotV1(), repoEnsurer, clock.RealClock{}, credentialGetter, s.nodeName, s.fileSystem, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
dataUploadReconciler := controller.NewDataUploadReconciler(s.mgr.GetClient(), s.kubeClient, s.csiSnapshotClient.SnapshotV1(), s.dataPathMgr, repoEnsurer, clock.RealClock{}, credentialGetter, s.nodeName, s.fileSystem, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
s.markDataUploadsCancel(dataUploadReconciler)
if err = dataUploadReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the data upload controller")
}
dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.kubeClient, repoEnsurer, credentialGetter, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.kubeClient, s.dataPathMgr, repoEnsurer, credentialGetter, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
s.markDataDownloadsCancel(dataDownloadReconciler)
if err = dataDownloadReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the data download controller")
@@ -478,3 +488,65 @@ func (s *nodeAgentServer) markInProgressPVRsFailed(client ctrlclient.Client) {
s.logger.WithField("podvolumerestore", pvr.GetName()).Warn(pvr.Status.Message)
}
}
var getConfigsFunc = nodeagent.GetConfigs
func (s *nodeAgentServer) getDataPathConcurrentNum(defaultNum int) int {
configs, err := getConfigsFunc(s.ctx, s.namespace, s.kubeClient)
if err != nil {
s.logger.WithError(err).Warn("Failed to get node agent configs")
return defaultNum
}
if configs == nil || configs.DataPathConcurrency == nil {
s.logger.Infof("Concurrency configs are not found, use the default number %v", defaultNum)
return defaultNum
}
globalNum := configs.DataPathConcurrency.GlobalConfig
if globalNum <= 0 {
s.logger.Warnf("Global number %v is invalid, use the default value %v", globalNum, defaultNum)
globalNum = defaultNum
}
if len(configs.DataPathConcurrency.PerNodeConfig) == 0 {
return globalNum
}
curNode, err := s.kubeClient.CoreV1().Nodes().Get(s.ctx, s.nodeName, metav1.GetOptions{})
if err != nil {
s.logger.WithError(err).Warnf("Failed to get node info for %s, use the global number %v", s.nodeName, globalNum)
return globalNum
}
concurrentNum := math.MaxInt32
for _, rule := range configs.DataPathConcurrency.PerNodeConfig {
selector, err := metav1.LabelSelectorAsSelector(&rule.NodeSelector)
if err != nil {
s.logger.WithError(err).Warnf("Failed to parse rule with label selector %s, skip it", rule.NodeSelector.String())
continue
}
if rule.Number <= 0 {
s.logger.Warnf("Rule with label selector %s is with an invalid number %v, skip it", rule.NodeSelector.String(), rule.Number)
continue
}
if selector.Matches(labels.Set(curNode.GetLabels())) {
if concurrentNum > rule.Number {
concurrentNum = rule.Number
}
}
}
if concurrentNum == math.MaxInt32 {
s.logger.Infof("Per node number for node %s is not found, use the global number %v", s.nodeName, globalNum)
concurrentNum = globalNum
} else {
s.logger.Infof("Use the per node number %v over global number %v for node %s", concurrentNum, globalNum, s.nodeName)
}
return concurrentNum
}
pkg/cmd/cli/nodeagent/server_test.go
+262
View File
@@ -17,16 +17,22 @@ package nodeagent
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/fake"
"github.com/vmware-tanzu/velero/pkg/builder"
"github.com/vmware-tanzu/velero/pkg/nodeagent"
testutil "github.com/vmware-tanzu/velero/pkg/test"
)
@@ -107,3 +113,259 @@ func Test_validatePodVolumesHostPath(t *testing.T) {
})
}
}
func Test_getDataPathConcurrentNum(t *testing.T) {
defaultNum := 100001
globalNum := 6
nodeName := "node-agent-node"
node1 := builder.ForNode("node-agent-node").Result()
node2 := builder.ForNode("node-agent-node").Labels(map[string]string{
"host-name": "node-1",
"xxxx": "yyyyy",
}).Result()
invalidLabelSelector := metav1.LabelSelector{
MatchLabels: map[string]string{
"inva/lid": "inva/lid",
},
}
validLabelSelector1 := metav1.LabelSelector{
MatchLabels: map[string]string{
"host-name": "node-1",
},
}
validLabelSelector2 := metav1.LabelSelector{
MatchLabels: map[string]string{
"xxxx": "yyyyy",
},
}
tests := []struct {
name string
getFunc func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error)
setKubeClient bool
kubeClientObj []runtime.Object
expectNum int
expectLog string
}{
{
name: "failed to get configs",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return nil, errors.New("fake-get-error")
},
expectLog: "Failed to get node agent configs",
expectNum: defaultNum,
},
{
name: "configs cm not found",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return nil, nil
},
expectLog: fmt.Sprintf("Concurrency configs are not found, use the default number %v", defaultNum),
expectNum: defaultNum,
},
{
name: "configs cm's data path concurrency is nil",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{}, nil
},
expectLog: fmt.Sprintf("Concurrency configs are not found, use the default number %v", defaultNum),
expectNum: defaultNum,
},
{
name: "global number is invalid",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: -1,
},
}, nil
},
expectLog: fmt.Sprintf("Global number %v is invalid, use the default value %v", -1, defaultNum),
expectNum: defaultNum,
},
{
name: "global number is valid",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
},
}, nil
},
expectNum: globalNum,
},
{
name: "node is not found",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
PerNodeConfig: []nodeagent.RuledConfigs{
{
Number: 100,
},
},
},
}, nil
},
setKubeClient: true,
expectLog: fmt.Sprintf("Failed to get node info for %s, use the global number %v", nodeName, globalNum),
expectNum: globalNum,
},
{
name: "failed to get selector",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
PerNodeConfig: []nodeagent.RuledConfigs{
{
NodeSelector: invalidLabelSelector,
Number: 100,
},
},
},
}, nil
},
setKubeClient: true,
kubeClientObj: []runtime.Object{node1},
expectLog: fmt.Sprintf("Failed to parse rule with label selector %s, skip it", invalidLabelSelector.String()),
expectNum: globalNum,
},
{
name: "rule number is invalid",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
PerNodeConfig: []nodeagent.RuledConfigs{
{
NodeSelector: validLabelSelector1,
Number: -1,
},
},
},
}, nil
},
setKubeClient: true,
kubeClientObj: []runtime.Object{node1},
expectLog: fmt.Sprintf("Rule with label selector %s is with an invalid number %v, skip it", validLabelSelector1.String(), -1),
expectNum: globalNum,
},
{
name: "label doesn't match",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
PerNodeConfig: []nodeagent.RuledConfigs{
{
NodeSelector: validLabelSelector1,
Number: -1,
},
},
},
}, nil
},
setKubeClient: true,
kubeClientObj: []runtime.Object{node1},
expectLog: fmt.Sprintf("Per node number for node %s is not found, use the global number %v", nodeName, globalNum),
expectNum: globalNum,
},
{
name: "match one rule",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
PerNodeConfig: []nodeagent.RuledConfigs{
{
NodeSelector: validLabelSelector1,
Number: 66,
},
},
},
}, nil
},
setKubeClient: true,
kubeClientObj: []runtime.Object{node2},
expectLog: fmt.Sprintf("Use the per node number %v over global number %v for node %s", 66, globalNum, nodeName),
expectNum: 66,
},
{
name: "match multiple rules",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
PerNodeConfig: []nodeagent.RuledConfigs{
{
NodeSelector: validLabelSelector1,
Number: 66,
},
{
NodeSelector: validLabelSelector2,
Number: 36,
},
},
},
}, nil
},
setKubeClient: true,
kubeClientObj: []runtime.Object{node2},
expectLog: fmt.Sprintf("Use the per node number %v over global number %v for node %s", 36, globalNum, nodeName),
expectNum: 36,
},
{
name: "match multiple rules 2",
getFunc: func(context.Context, string, kubernetes.Interface) (*nodeagent.Configs, error) {
return &nodeagent.Configs{
DataPathConcurrency: &nodeagent.DataPathConcurrency{
GlobalConfig: globalNum,
PerNodeConfig: []nodeagent.RuledConfigs{
{
NodeSelector: validLabelSelector1,
Number: 36,
},
{
NodeSelector: validLabelSelector2,
Number: 66,
},
},
},
}, nil
},
setKubeClient: true,
kubeClientObj: []runtime.Object{node2},
expectLog: fmt.Sprintf("Use the per node number %v over global number %v for node %s", 36, globalNum, nodeName),
expectNum: 36,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
logBuffer := ""
s := &nodeAgentServer{
nodeName: nodeName,
logger: testutil.NewSingleLogger(&logBuffer),
}
if test.setKubeClient {
s.kubeClient = fakeKubeClient
}
getConfigsFunc = test.getFunc
num := s.getDataPathConcurrentNum(defaultNum)
assert.Equal(t, test.expectNum, num)
if test.expectLog == "" {
assert.Equal(t, "", logBuffer)
} else {
assert.True(t, strings.Contains(logBuffer, test.expectLog))
}
})
}
}
pkg/cmd/cli/repo/get.go
+11 -3
View File
@@ -21,6 +21,8 @@ import (
"github.com/spf13/cobra"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/client"
@@ -38,19 +40,25 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
err := output.ValidateFlags(c)
cmd.CheckError(err)
veleroClient, err := f.Client()
crClient, err := f.KubebuilderClient()
cmd.CheckError(err)
var repos *api.BackupRepositoryList
if len(args) > 0 {
repos = new(api.BackupRepositoryList)
for _, name := range args {
repo, err := veleroClient.VeleroV1().BackupRepositories(f.Namespace()).Get(context.TODO(), name, metav1.GetOptions{})
repo := new(api.BackupRepository)
err := crClient.Get(context.TODO(), ctrlclient.ObjectKey{Namespace: f.Namespace(), Name: name}, repo)
cmd.CheckError(err)
repos.Items = append(repos.Items, *repo)
}
} else {
repos, err = veleroClient.VeleroV1().BackupRepositories(f.Namespace()).List(context.TODO(), listOptions)
selector := labels.NewSelector()
if listOptions.LabelSelector != "" {
selector, err = labels.Parse(listOptions.LabelSelector)
cmd.CheckError(err)
}
err = crClient.List(context.TODO(), repos, &ctrlclient.ListOptions{LabelSelector: selector})
cmd.CheckError(err)
}
pkg/cmd/cli/restore/logs.go
+3 -5
View File
@@ -24,7 +24,7 @@ import (
"github.com/spf13/cobra"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/client"
@@ -49,13 +49,11 @@ func NewLogsCommand(f client.Factory) *cobra.Command {
Run: func(c *cobra.Command, args []string) {
restoreName := args[0]
veleroClient, err := f.Client()
cmd.CheckError(err)
kbClient, err := f.KubebuilderClient()
cmd.CheckError(err)
restore, err := veleroClient.VeleroV1().Restores(f.Namespace()).Get(context.TODO(), restoreName, metav1.GetOptions{})
restore := new(velerov1api.Restore)
err = kbClient.Get(context.TODO(), ctrlclient.ObjectKey{Namespace: f.Namespace(), Name: restoreName}, restore)
if apierrors.IsNotFound(err) {
cmd.Exit("Restore %q does not exist.", restoreName)
} else if err != nil {
pkg/cmd/cli/schedule/create.go
+2 -2
View File
@@ -115,7 +115,7 @@ func (o *CreateOptions) Complete(args []string, f client.Factory) error {
func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
var orders map[string]string
veleroClient, err := f.Client()
crClient, err := f.KubebuilderClient()
if err != nil {
return err
}
@@ -171,7 +171,7 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
return err
}
_, err = veleroClient.VeleroV1().Schedules(schedule.Namespace).Create(context.TODO(), schedule, metav1.CreateOptions{})
err = crClient.Create(context.TODO(), schedule)
if err != nil {
return err
}
pkg/cmd/cli/schedule/describe.go
+11 -3
View File
@@ -22,6 +22,8 @@ import (
"github.com/spf13/cobra"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/client"
@@ -36,19 +38,25 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
Use: use + " [NAME1] [NAME2] [NAME...]",
Short: "Describe schedules",
Run: func(c *cobra.Command, args []string) {
veleroClient, err := f.Client()
crClient, err := f.KubebuilderClient()
cmd.CheckError(err)
var schedules *v1.ScheduleList
if len(args) > 0 {
schedules = new(v1.ScheduleList)
for _, name := range args {
schedule, err := veleroClient.VeleroV1().Schedules(f.Namespace()).Get(context.TODO(), name, metav1.GetOptions{})
schedule := new(v1.Schedule)
err := crClient.Get(context.TODO(), ctrlclient.ObjectKey{Namespace: f.Namespace(), Name: name}, schedule)
cmd.CheckError(err)
schedules.Items = append(schedules.Items, *schedule)
}
} else {
schedules, err = veleroClient.VeleroV1().Schedules(f.Namespace()).List(context.TODO(), listOptions)
selector := labels.NewSelector()
if listOptions.LabelSelector != "" {
selector, err = labels.Parse(listOptions.LabelSelector)
cmd.CheckError(err)
}
err = crClient.List(context.TODO(), schedules, &ctrlclient.ListOptions{LabelSelector: selector})
cmd.CheckError(err)
}
pkg/cmd/cli/schedule/get.go
+11 -3
View File
@@ -21,6 +21,8 @@ import (
"github.com/spf13/cobra"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/client"
@@ -38,19 +40,25 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
err := output.ValidateFlags(c)
cmd.CheckError(err)
veleroClient, err := f.Client()
crClient, err := f.KubebuilderClient()
cmd.CheckError(err)
var schedules *api.ScheduleList
if len(args) > 0 {
schedules = new(api.ScheduleList)
for _, name := range args {
schedule, err := veleroClient.VeleroV1().Schedules(f.Namespace()).Get(context.TODO(), name, metav1.GetOptions{})
schedule := new(api.Schedule)
err := crClient.Get(context.TODO(), ctrlclient.ObjectKey{Name: name, Namespace: f.Namespace()}, schedule)
cmd.CheckError(err)
schedules.Items = append(schedules.Items, *schedule)
}
} else {
schedules, err = veleroClient.VeleroV1().Schedules(f.Namespace()).List(context.TODO(), listOptions)
selector := labels.NewSelector()
if listOptions.LabelSelector != "" {
selector, err = labels.Parse(listOptions.LabelSelector)
cmd.CheckError(err)
}
err := crClient.List(context.TODO(), schedules, &ctrlclient.ListOptions{LabelSelector: selector})
cmd.CheckError(err)
}
pkg/cmd/cli/schedule/pause.go
+13 -6
View File
@@ -25,6 +25,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
kubeerrs "k8s.io/apimachinery/pkg/util/errors"
ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/client"
@@ -63,7 +64,7 @@ func NewPauseCommand(f client.Factory, use string) *cobra.Command {
}
func runPause(f client.Factory, o *cli.SelectOptions, paused bool) error {
client, err := f.Client()
crClient, err := f.KubebuilderClient()
if err != nil {
return err
}
@@ -75,7 +76,8 @@ func runPause(f client.Factory, o *cli.SelectOptions, paused bool) error {
switch {
case len(o.Names) > 0:
for _, name := range o.Names {
schedule, err := client.VeleroV1().Schedules(f.Namespace()).Get(context.TODO(), name, metav1.GetOptions{})
schedule := new(velerov1api.Schedule)
err := crClient.Get(context.TODO(), ctrlclient.ObjectKey{Name: name, Namespace: f.Namespace()}, schedule)
if err != nil {
errs = append(errs, errors.WithStack(err))
continue
@@ -83,11 +85,16 @@ func runPause(f client.Factory, o *cli.SelectOptions, paused bool) error {
schedules = append(schedules, schedule)
}
default:
selector := labels.Everything().String()
selector := labels.Everything()
if o.Selector.LabelSelector != nil {
selector = o.Selector.String()
convertedSelector, err := metav1.LabelSelectorAsSelector(o.Selector.LabelSelector)
if err != nil {
return errors.WithStack(err)
}
selector = convertedSelector
}
res, err := client.VeleroV1().Schedules(f.Namespace()).List(context.TODO(), metav1.ListOptions{
res := new(velerov1api.ScheduleList)
err := crClient.List(context.TODO(), res, &ctrlclient.ListOptions{
LabelSelector: selector,
})
if err != nil {
@@ -113,7 +120,7 @@ func runPause(f client.Factory, o *cli.SelectOptions, paused bool) error {
continue
}
schedule.Spec.Paused = paused
if _, err := client.VeleroV1().Schedules(schedule.Namespace).Update(context.TODO(), schedule, metav1.UpdateOptions{}); err != nil {
if err := crClient.Update(context.TODO(), schedule); err != nil {
return errors.Wrapf(err, "failed to update schedule %s", schedule.Name)
}
fmt.Printf("Schedule %s %s successfully\n", schedule.Name, msg)
pkg/cmd/cli/snapshotlocation/create.go
+2 -2
View File
@@ -124,12 +124,12 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
return err
}
client, err := f.Client()
client, err := f.KubebuilderClient()
if err != nil {
return err
}
if _, err := client.VeleroV1().VolumeSnapshotLocations(volumeSnapshotLocation.Namespace).Create(context.TODO(), volumeSnapshotLocation, metav1.CreateOptions{}); err != nil {
if err := client.Create(context.TODO(), volumeSnapshotLocation); err != nil {
return errors.WithStack(err)
}
pkg/cmd/cli/snapshotlocation/get.go
+7 -5
View File
@@ -21,6 +21,7 @@ import (
"github.com/spf13/cobra"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
kbclient "sigs.k8s.io/controller-runtime/pkg/client"
api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/client"
@@ -36,18 +37,19 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
Run: func(c *cobra.Command, args []string) {
err := output.ValidateFlags(c)
cmd.CheckError(err)
veleroClient, err := f.Client()
client, err := f.KubebuilderClient()
cmd.CheckError(err)
var locations *api.VolumeSnapshotLocationList
locations := new(api.VolumeSnapshotLocationList)
if len(args) > 0 {
locations = new(api.VolumeSnapshotLocationList)
for _, name := range args {
location, err := veleroClient.VeleroV1().VolumeSnapshotLocations(f.Namespace()).Get(context.TODO(), name, metav1.GetOptions{})
location := new(api.VolumeSnapshotLocation)
err := client.Get(context.TODO(), kbclient.ObjectKey{Namespace: f.Namespace(), Name: name}, location)
cmd.CheckError(err)
locations.Items = append(locations.Items, *location)
}
} else {
locations, err = veleroClient.VeleroV1().VolumeSnapshotLocations(f.Namespace()).List(context.TODO(), listOptions)
err = client.List(context.TODO(), locations, &kbclient.ListOptions{Namespace: f.Namespace()})
cmd.CheckError(err)
}
_, err = output.PrintWithFormat(c, locations)
pkg/cmd/server/plugin/plugin.go
+2 -2
View File
@@ -152,12 +152,12 @@ func newPodVolumeRestoreItemAction(f client.Factory) plugincommon.HandlerInitial
return nil, err
}
veleroClient, err := f.Client()
crClient, err := f.KubebuilderClient()
if err != nil {
return nil, err
}
return restore.NewPodVolumeRestoreAction(logger, client.CoreV1().ConfigMaps(f.Namespace()), veleroClient.VeleroV1().PodVolumeBackups(f.Namespace())), nil
return restore.NewPodVolumeRestoreAction(logger, client.CoreV1().ConfigMaps(f.Namespace()), crClient), nil
}
}
pkg/cmd/server/server.go
+50 -27
View File
@@ -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"
@@ -239,7 +238,7 @@ func NewCommand(f client.Factory) *cobra.Command {
command.Flags().DurationVar(&config.resourceTimeout, "resource-timeout", config.resourceTimeout, "How long to wait for resource processes which are not covered by other specific timeout parameters. Default is 10 minutes.")
command.Flags().IntVar(&config.maxConcurrentK8SConnections, "max-concurrent-k8s-connections", config.maxConcurrentK8SConnections, "Max concurrent connections number that Velero can create with kube-apiserver. Default is 30.")
command.Flags().BoolVar(&config.defaultSnapshotMoveData, "default-snapshot-move-data", config.defaultSnapshotMoveData, "Move data by default for all snapshots supporting data movement.")
command.Flags().BoolVar(&config.disableInformerCache, "disable-informer-cache", config.disableInformerCache, "Disable informer cache for Get calls on restore. WIth this enabled, it will speed up restore in cases where there are backup resources which already exist in the cluster, but for very large clusters this will increase velero memory usage. Default is false (don't disable).")
command.Flags().BoolVar(&config.disableInformerCache, "disable-informer-cache", config.disableInformerCache, "Disable informer cache for Get calls on restore. With this enabled, it will speed up restore in cases where there are backup resources which already exist in the cluster, but for very large clusters this will increase velero memory usage. Default is false (don't disable).")
return command
}
@@ -249,12 +248,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
@@ -268,6 +267,7 @@ type server struct {
mgr manager.Manager
credentialFileStore credentials.FileStore
credentialSecretStore credentials.SecretStore
featureVerifier features.Verifier
}
func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*server, error) {
@@ -294,12 +294,12 @@ 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()
crClient, err := f.KubebuilderClient()
if err != nil {
return nil, err
}
@@ -309,6 +309,12 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
return nil, err
}
featureVerifier := features.NewVerifier(pluginRegistry)
if _, err := featureVerifier.Verify(velerov1api.CSIFeatureFlag); err != nil {
logger.WithError(err).Warn("CSI feature verification failed, the feature may not be ready.")
}
// cancelFunc is not deferred here because if it was, then ctx would immediately
// be canceled once this function exited, making it useless to any informers using later.
// That, in turn, causes the velero server to halt when the first informer tries to use it.
@@ -367,14 +373,20 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
return nil, err
}
var discoveryClient *discovery.DiscoveryClient
if discoveryClient, err = discovery.NewDiscoveryClientForConfig(clientConfig); err != nil {
cancelFunc()
return nil, err
}
s := &server{
namespace: f.Namespace(),
metricsAddress: config.metricsAddress,
kubeClientConfig: clientConfig,
kubeClient: kubeClient,
veleroClient: veleroClient,
discoveryClient: veleroClient.Discovery(),
discoveryClient: discoveryClient,
dynamicClient: dynamicClient,
crClient: crClient,
ctx: ctx,
cancelFunc: cancelFunc,
logger: logger,
@@ -384,6 +396,7 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
mgr: mgr,
credentialFileStore: credentialFileStore,
credentialSecretStore: credentialSecretStore,
featureVerifier: featureVerifier,
}
// Setup CSI snapshot client and lister
@@ -528,7 +541,7 @@ High priorities:
- PVs go before PVCs because PVCs depend on them.
- PVCs go before pods or controllers so they can be mounted as volumes.
- Service accounts go before secrets so service account token secrets can be filled automatically.
- Secrets and config maps go before pods or controllers so they can be mounted
- Secrets and ConfigMaps go before pods or controllers so they can be mounted
as volumes.
- Limit ranges go before pods or controllers so pods can use them.
- Pods go before controllers so they can be explicitly restored and potentially
@@ -719,6 +732,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(),
@@ -728,10 +746,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,
@@ -760,7 +776,6 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
backupStoreGetter,
s.config.formatFlag.Parse(),
s.csiSnapshotLister,
s.csiSnapshotClient,
s.credentialFileStore,
s.config.maxConcurrentK8SConnections,
s.config.defaultSnapshotMoveData,
@@ -810,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,
@@ -824,6 +837,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
cmd.CheckError(err)
r := controller.NewBackupFinalizerReconciler(
s.mgr.GetClient(),
s.csiSnapshotLister,
clock.RealClock{},
backupper,
newPluginManager,
@@ -901,6 +915,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,
@@ -910,10 +929,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,
@@ -924,6 +942,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
s.kubeClient.CoreV1().RESTClient(),
s.credentialFileStore,
s.mgr.GetClient(),
s.featureVerifier,
)
cmd.CheckError(err)
@@ -1062,13 +1081,13 @@ func markInProgressBackupsFailed(ctx context.Context, client ctrlclient.Client,
}
for i, backup := range backups.Items {
if backup.Status.Phase != velerov1api.BackupPhaseInProgress && backup.Status.Phase != velerov1api.BackupPhaseWaitingForPluginOperations {
if backup.Status.Phase != velerov1api.BackupPhaseInProgress {
log.Debugf("the status of backup %q is %q, skip", backup.GetName(), backup.Status.Phase)
continue
}
updated := backup.DeepCopy()
updated.Status.Phase = velerov1api.BackupPhaseFailed
updated.Status.FailureReason = fmt.Sprintf("found a backup with status %q during the server starting, mark it as %q", velerov1api.BackupPhaseInProgress, updated.Status.Phase)
updated.Status.FailureReason = fmt.Sprintf("found a backup with status %q during the server starting, mark it as %q", backup.Status.Phase, updated.Status.Phase)
updated.Status.CompletionTimestamp = &metav1.Time{Time: time.Now()}
if err := client.Patch(ctx, updated, ctrlclient.MergeFrom(&backups.Items[i])); err != nil {
log.WithError(errors.WithStack(err)).Errorf("failed to patch backup %q", backup.GetName())
@@ -1086,13 +1105,13 @@ func markInProgressRestoresFailed(ctx context.Context, client ctrlclient.Client,
return
}
for i, restore := range restores.Items {
if restore.Status.Phase != velerov1api.RestorePhaseInProgress && restore.Status.Phase != velerov1api.RestorePhaseWaitingForPluginOperations {
if restore.Status.Phase != velerov1api.RestorePhaseInProgress {
log.Debugf("the status of restore %q is %q, skip", restore.GetName(), restore.Status.Phase)
continue
}
updated := restore.DeepCopy()
updated.Status.Phase = velerov1api.RestorePhaseFailed
updated.Status.FailureReason = fmt.Sprintf("found a restore with status %q during the server starting, mark it as %q", velerov1api.RestorePhaseInProgress, updated.Status.Phase)
updated.Status.FailureReason = fmt.Sprintf("found a restore with status %q during the server starting, mark it as %q", restore.Status.Phase, updated.Status.Phase)
updated.Status.CompletionTimestamp = &metav1.Time{Time: time.Now()}
if err := client.Patch(ctx, updated, ctrlclient.MergeFrom(&restores.Items[i])); err != nil {
log.WithError(errors.WithStack(err)).Errorf("failed to patch restore %q", restore.GetName())
@@ -1115,7 +1134,9 @@ func markDataUploadsCancel(ctx context.Context, client ctrlclient.Client, backup
du := dataUploads.Items[i]
if du.Status.Phase == velerov2alpha1api.DataUploadPhaseAccepted ||
du.Status.Phase == velerov2alpha1api.DataUploadPhasePrepared ||
du.Status.Phase == velerov2alpha1api.DataUploadPhaseInProgress {
du.Status.Phase == velerov2alpha1api.DataUploadPhaseInProgress ||
du.Status.Phase == velerov2alpha1api.DataUploadPhaseNew ||
du.Status.Phase == "" {
err := controller.UpdateDataUploadWithRetry(ctx, client, types.NamespacedName{Namespace: du.Namespace, Name: du.Name}, log.WithField("dataupload", du.Name),
func(dataUpload *velerov2alpha1api.DataUpload) {
dataUpload.Spec.Cancel = true
@@ -1143,7 +1164,9 @@ func markDataDownloadsCancel(ctx context.Context, client ctrlclient.Client, rest
dd := dataDownloads.Items[i]
if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted ||
dd.Status.Phase == velerov2alpha1api.DataDownloadPhasePrepared ||
dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress ||
dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseNew ||
dd.Status.Phase == "" {
err := controller.UpdateDataDownloadWithRetry(ctx, client, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name}, log.WithField("datadownload", dd.Name),
func(dataDownload *velerov2alpha1api.DataDownload) {
dataDownload.Spec.Cancel = true
@@ -1151,7 +1174,7 @@ func markDataDownloadsCancel(ctx context.Context, client ctrlclient.Client, rest
})
if err != nil {
log.WithError(errors.WithStack(err)).Errorf("failed to mark dataupload %q cancel", dd.GetName())
log.WithError(errors.WithStack(err)).Errorf("failed to mark datadownload %q cancel", dd.GetName())
continue
}
log.WithField("datadownload", dd.GetName()).Warn(dd.Status.Message)