Merge pull request #8569 from Lyndon-Li/uploaders-windows-support

Uploaders windows support

changelogs/unreleased/8569-Lyndon-Li

@@ -0,0 +1 @@
+fs uploader and block uploader support Windows nodes

pkg/controller/data_upload_controller.go

@@ -808,6 +808,10 @@ func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload
 			return nil, errors.Wrapf(err, "failed to get attaching node OS for PVC %s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC)
 		}
 
+		if err := kube.HasNodeWithOS(context.Background(), nodeOS, r.kubeClient.CoreV1()); err != nil {
+			return nil, errors.Wrapf(err, "no appropriate node to run data upload for PVC %s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC)
+		}
+
 		accessMode := exposer.AccessModeFileSystem
 		if pvc.Spec.VolumeMode != nil && *pvc.Spec.VolumeMode == corev1.PersistentVolumeBlock {
 			accessMode = exposer.AccessModeBlock

pkg/controller/data_upload_controller_test.go

@@ -59,6 +59,7 @@ import (
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
 const dataUploadName = "dataupload-1"
@@ -187,6 +188,8 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci
 		},
 	}
 
+	node := builder.ForNode("fake-node").Labels(map[string]string{kube.NodeOSLabel: kube.NodeOSLinux}).Result()
+
 	dataPathMgr := datapath.NewManager(1)
 
 	now, err := time.Parse(time.RFC1123, time.RFC1123)
@@ -229,7 +232,7 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci
 	}
 
 	fakeSnapshotClient := snapshotFake.NewSimpleClientset(vsObject, vscObj)
-	fakeKubeClient := clientgofake.NewSimpleClientset(daemonSet)
+	fakeKubeClient := clientgofake.NewSimpleClientset(daemonSet, node)
 
 	return NewDataUploadReconciler(
 		fakeClient,

pkg/uploader/kopia/progress.go

@@ -138,7 +138,9 @@ func (p *Progress) HashingFile(fname string) {}
 func (p *Progress) ExcludedFile(fname string, numBytes int64) {}
 
 // ExcludedDir statistic the dir been excluded currently
-func (p *Progress) ExcludedDir(dirname string) {}
+func (p *Progress) ExcludedDir(dirname string) {
+	p.log.Infof("Excluded dir %s", dirname)
+}
 
 // FinishedHashingFile which will called when specific file finished hash
 func (p *Progress) FinishedHashingFile(fname string, numBytes int64) {

pkg/uploader/kopia/snapshot.go

@@ -127,6 +127,10 @@ func setupPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceInfo snap
 		curPolicy.UploadPolicy.ParallelUploadAboveSize = newOptionalInt64(2 << 30)
 	}
 
+	if runtime.GOOS == "windows" {
+		curPolicy.FilesPolicy.IgnoreRules = []string{"/System Volume Information/", "/$Recycle.Bin/"}
+	}
+
 	err := setPolicyFunc(ctx, rep, sourceInfo, curPolicy)
 	if err != nil {
 		return nil, errors.Wrap(err, "error to set policy")

pkg/util/kube/node.go

@@ -17,6 +17,7 @@ package kube
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -99,3 +100,20 @@ func GetNodeOS(ctx context.Context, nodeName string, nodeClient corev1client.Cor
 
 	return node.Labels[NodeOSLabel], nil
 }
+
+func HasNodeWithOS(ctx context.Context, os string, nodeClient corev1client.CoreV1Interface) error {
+	if os == "" {
+		return errors.New("invalid node OS")
+	}
+
+	nodes, err := nodeClient.Nodes().List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", NodeOSLabel, os)})
+	if err != nil {
+		return errors.Wrapf(err, "error listing nodes with OS %s", os)
+	}
+
+	if len(nodes.Items) == 0 {
+		return errors.Errorf("node with OS %s doesn't exist", os)
+	}
+
+	return nil
+}

pkg/util/kube/node_test.go

@@ -20,13 +20,15 @@ import (
 	"context"
 	"testing"
 
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 
 	"github.com/vmware-tanzu/velero/pkg/builder"
 
 	kubeClientFake "k8s.io/client-go/kubernetes/fake"
+	clientTesting "k8s.io/client-go/testing"
 	clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
@@ -181,3 +183,79 @@ func TestGetNodeOSType(t *testing.T) {
 		})
 	}
 }
+
+func TestHasNodeWithOS(t *testing.T) {
+	nodeNoOSLabel := builder.ForNode("fake-node-1").Result()
+	nodeWindows := builder.ForNode("fake-node-2").Labels(map[string]string{"kubernetes.io/os": "windows"}).Result()
+	nodeLinux := builder.ForNode("fake-node-3").Labels(map[string]string{"kubernetes.io/os": "linux"}).Result()
+
+	scheme := runtime.NewScheme()
+	corev1.AddToScheme(scheme)
+
+	tests := []struct {
+		name          string
+		kubeClientObj []runtime.Object
+		kubeReactors  []reactor
+		os            string
+		err           string
+	}{
+		{
+			name: "os is empty",
+			err:  "invalid node OS",
+		},
+		{
+			name: "error to list node",
+			kubeReactors: []reactor{
+				{
+					verb:     "list",
+					resource: "nodes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("fake-list-error")
+					},
+				},
+			},
+			os:  "linux",
+			err: "error listing nodes with OS linux: fake-list-error",
+		},
+		{
+			name: "no expected node - no node",
+			os:   "linux",
+			err:  "node with OS linux doesn't exist",
+		},
+		{
+			name: "no expected node - no node with label",
+			kubeClientObj: []runtime.Object{
+				nodeNoOSLabel,
+				nodeWindows,
+			},
+			os:  "linux",
+			err: "node with OS linux doesn't exist",
+		},
+		{
+			name: "succeed",
+			kubeClientObj: []runtime.Object{
+				nodeNoOSLabel,
+				nodeWindows,
+				nodeLinux,
+			},
+			os: "windows",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeKubeClient := kubeClientFake.NewSimpleClientset(test.kubeClientObj...)
+
+			for _, reactor := range test.kubeReactors {
+				fakeKubeClient.Fake.PrependReactor(reactor.verb, reactor.resource, reactor.reactorFunc)
+			}
+
+			err := HasNodeWithOS(context.TODO(), test.os, fakeKubeClient.CoreV1())
+			if test.err != "" {
+				assert.EqualError(t, err, test.err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

pkg/util/kube/pvc_pv.go

@@ -434,6 +434,11 @@ func GetPVCAttachingNodeOS(pvc *corev1api.PersistentVolumeClaim, nodeClient core
 	var nodeOS string
 	var scFsType string
 
+	if pvc.Spec.VolumeMode != nil && *pvc.Spec.VolumeMode == corev1api.PersistentVolumeBlock {
+		log.Infof("Use linux node for block mode PVC %s/%s", pvc.Namespace, pvc.Name)
+		return NodeOSLinux, nil
+	}
+
 	if value := pvc.Annotations[KubeAnnSelectedNode]; value != "" {
 		os, err := GetNodeOS(context.Background(), value, nodeClient)
 		if err != nil {

pkg/util/kube/pvc_pv_test.go

@@ -1564,6 +1564,17 @@ func TestGetPVCAttachingNodeOS(t *testing.T) {
 		},
 	}
 
+	blockMode := corev1api.PersistentVolumeBlock
+	pvcObjBlockMode := &corev1api.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "fake-namespace",
+			Name:      "fake-pvc",
+		},
+		Spec: corev1api.PersistentVolumeClaimSpec{
+			VolumeMode: &blockMode,
+		},
+	}
+
 	pvcObjWithNode := &corev1api.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:   "fake-namespace",
@@ -1662,6 +1673,11 @@ func TestGetPVCAttachingNodeOS(t *testing.T) {
 			},
 			expectedNodeOS: NodeOSWindows,
 		},
+		{
+			name:           "block access",
+			pvc:            pvcObjBlockMode,
+			expectedNodeOS: NodeOSLinux,
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {