mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-09-18 22:14:29 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #7988 from Lyndon-Li/data-mover-ms-new-exposer

Browse Source 
New exposer for data mover ms
This commit is contained in:
lyndon-li
2024-07-24 10:11:54 +08:00
committed by GitHub
parent 2bf3bc9cc7 a1d6d1d698
commit 01aa657f0e
18 changed files with 905 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/exposer/csi_snapshot.go
+54 -17
View File
@@ -18,6 +18,7 @@ package exposer
import (
"context"
"fmt"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
@@ -174,7 +175,7 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1.Obje
}
}()
backupPod, err := e.createBackupPod(ctx, ownerObject, backupPVC, csiExposeParam.HostingPodLabels, csiExposeParam.Affinity)
backupPod, err := e.createBackupPod(ctx, ownerObject, backupPVC, csiExposeParam.OperationTimeout, csiExposeParam.HostingPodLabels, csiExposeParam.Affinity)
if err != nil {
return errors.Wrap(err, "error to create backup pod")
}
@@ -195,6 +196,8 @@ func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1.
backupPodName := ownerObject.Name
backupPVCName := ownerObject.Name
containerName := string(ownerObject.UID)
volumeName := string(ownerObject.UID)
curLog := e.log.WithFields(logrus.Fields{
@@ -237,7 +240,11 @@ func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1.
curLog.WithField("pod", pod.Name).Infof("Backup volume is found in pod at index %v", i)
return &ExposeResult{ByPod: ExposeByPod{HostingPod: pod, VolumeName: volumeName}}, nil
return &ExposeResult{ByPod: ExposeByPod{
HostingPod: pod,
HostingContainer: containerName,
VolumeName: volumeName,
}}, nil
}
func (e *csiSnapshotExposer) PeekExposed(ctx context.Context, ownerObject corev1.ObjectReference) error {
@@ -393,12 +400,12 @@ func (e *csiSnapshotExposer) createBackupPVC(ctx context.Context, ownerObject co
return created, err
}
func (e *csiSnapshotExposer) createBackupPod(ctx context.Context, ownerObject corev1.ObjectReference, backupPVC *corev1.PersistentVolumeClaim,
func (e *csiSnapshotExposer) createBackupPod(ctx context.Context, ownerObject corev1.ObjectReference, backupPVC *corev1.PersistentVolumeClaim, operationTimeout time.Duration,
label map[string]string, affinity *nodeagent.LoadAffinity) (*corev1.Pod, error) {
podName := ownerObject.Name
volumeName := string(ownerObject.UID)
containerName := string(ownerObject.UID)
volumeName := string(ownerObject.UID)
podInfo, err := getInheritedPodInfo(ctx, e.kubeClient, ownerObject.Namespace)
if err != nil {
@@ -406,14 +413,41 @@ func (e *csiSnapshotExposer) createBackupPod(ctx context.Context, ownerObject co
}
var gracePeriod int64 = 0
volumeMounts, volumeDevices := kube.MakePodPVCAttachment(volumeName, backupPVC.Spec.VolumeMode)
volumeMounts, volumeDevices, volumePath := kube.MakePodPVCAttachment(volumeName, backupPVC.Spec.VolumeMode)
volumeMounts = append(volumeMounts, podInfo.volumeMounts...)
volumes := []corev1.Volume{{
Name: volumeName,
VolumeSource: corev1.VolumeSource{
PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
ClaimName: backupPVC.Name,
},
},
}}
volumes = append(volumes, podInfo.volumes...)
if label == nil {
label = make(map[string]string)
}
label[podGroupLabel] = podGroupSnapshot
volumeMode := corev1.PersistentVolumeFilesystem
if backupPVC.Spec.VolumeMode != nil {
volumeMode = *backupPVC.Spec.VolumeMode
}
args := []string{
fmt.Sprintf("--volume-path=%s", volumePath),
fmt.Sprintf("--volume-mode=%s", volumeMode),
fmt.Sprintf("--data-upload=%s", ownerObject.Name),
fmt.Sprintf("--resource-timeout=%s", operationTimeout.String()),
}
args = append(args, podInfo.logFormatArgs...)
args = append(args, podInfo.logLevelArgs...)
userID := int64(0)
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: podName,
@@ -448,21 +482,24 @@ func (e *csiSnapshotExposer) createBackupPod(ctx context.Context, ownerObject co
Name: containerName,
Image: podInfo.image,
ImagePullPolicy: corev1.PullNever,
Command: []string{"/velero-helper", "pause"},
VolumeMounts: volumeMounts,
VolumeDevices: volumeDevices,
Command: []string{
"/velero",
"data-mover",
"backup",
},
Args: args,
VolumeMounts: volumeMounts,
VolumeDevices: volumeDevices,
Env: podInfo.env,
},
},
ServiceAccountName: podInfo.serviceAccount,
TerminationGracePeriodSeconds: &gracePeriod,
Volumes: []corev1.Volume{{
Name: volumeName,
VolumeSource: corev1.VolumeSource{
PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
ClaimName: backupPVC.Name,
},
},
}},
Volumes: volumes,
RestartPolicy: corev1.RestartPolicyNever,
SecurityContext: &corev1.PodSecurityContext{
RunAsUser: &userID,
},
},
}
pkg/exposer/csi_snapshot_test.go
+11 -1
View File
@@ -138,7 +138,17 @@ func TestExpose(t *testing.T) {
Kind: "DaemonSet",
APIVersion: appsv1.SchemeGroupVersion.String(),
},
Spec: appsv1.DaemonSetSpec{},
Spec: appsv1.DaemonSetSpec{
Template: corev1.PodTemplateSpec{
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Name: "node-agent",
},
},
},
},
},
}
tests := []struct {
pkg/exposer/generic_restore.go
+54 -17
View File
@@ -87,7 +87,7 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1.O
return errors.Errorf("Target PVC %s/%s has already been bound, abort", sourceNamespace, targetPVCName)
}
restorePod, err := e.createRestorePod(ctx, ownerObject, targetPVC, hostingPodLabels, selectedNode)
restorePod, err := e.createRestorePod(ctx, ownerObject, targetPVC, timeout, hostingPodLabels, selectedNode)
if err != nil {
return errors.Wrapf(err, "error to create restore pod")
}
@@ -119,6 +119,8 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1.O
func (e *genericRestoreExposer) GetExposed(ctx context.Context, ownerObject corev1.ObjectReference, nodeClient client.Client, nodeName string, timeout time.Duration) (*ExposeResult, error) {
restorePodName := ownerObject.Name
restorePVCName := ownerObject.Name
containerName := string(ownerObject.UID)
volumeName := string(ownerObject.UID)
curLog := e.log.WithFields(logrus.Fields{
@@ -162,7 +164,11 @@ func (e *genericRestoreExposer) GetExposed(ctx context.Context, ownerObject core
curLog.WithField("pod", pod.Name).Infof("Restore volume is found in pod at index %v", i)
return &ExposeResult{ByPod: ExposeByPod{HostingPod: pod, VolumeName: volumeName}}, nil
return &ExposeResult{ByPod: ExposeByPod{
HostingPod: pod,
HostingContainer: containerName,
VolumeName: volumeName,
}}, nil
}
func (e *genericRestoreExposer) PeekExposed(ctx context.Context, ownerObject corev1.ObjectReference) error {
@@ -291,12 +297,12 @@ func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject co
}
func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObject corev1.ObjectReference, targetPVC *corev1.PersistentVolumeClaim,
label map[string]string, selectedNode string) (*corev1.Pod, error) {
operationTimeout time.Duration, label map[string]string, selectedNode string) (*corev1.Pod, error) {
restorePodName := ownerObject.Name
restorePVCName := ownerObject.Name
volumeName := string(ownerObject.UID)
containerName := string(ownerObject.UID)
volumeName := string(ownerObject.UID)
podInfo, err := getInheritedPodInfo(ctx, e.kubeClient, ownerObject.Namespace)
if err != nil {
@@ -304,7 +310,35 @@ func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObjec
}
var gracePeriod int64 = 0
volumeMounts, volumeDevices := kube.MakePodPVCAttachment(volumeName, targetPVC.Spec.VolumeMode)
volumeMounts, volumeDevices, volumePath := kube.MakePodPVCAttachment(volumeName, targetPVC.Spec.VolumeMode)
volumeMounts = append(volumeMounts, podInfo.volumeMounts...)
volumes := []corev1.Volume{{
Name: volumeName,
VolumeSource: corev1.VolumeSource{
PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
ClaimName: restorePVCName,
},
},
}}
volumes = append(volumes, podInfo.volumes...)
volumeMode := corev1.PersistentVolumeFilesystem
if targetPVC.Spec.VolumeMode != nil {
volumeMode = *targetPVC.Spec.VolumeMode
}
args := []string{
fmt.Sprintf("--volume-path=%s", volumePath),
fmt.Sprintf("--volume-mode=%s", volumeMode),
fmt.Sprintf("--data-download=%s", ownerObject.Name),
fmt.Sprintf("--resource-timeout=%s", operationTimeout.String()),
}
args = append(args, podInfo.logFormatArgs...)
args = append(args, podInfo.logLevelArgs...)
userID := int64(0)
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
@@ -327,22 +361,25 @@ func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObjec
Name: containerName,
Image: podInfo.image,
ImagePullPolicy: corev1.PullNever,
Command: []string{"/velero-helper", "pause"},
VolumeMounts: volumeMounts,
VolumeDevices: volumeDevices,
Command: []string{
"/velero",
"data-mover",
"restore",
},
Args: args,
VolumeMounts: volumeMounts,
VolumeDevices: volumeDevices,
Env: podInfo.env,
},
},
ServiceAccountName: podInfo.serviceAccount,
TerminationGracePeriodSeconds: &gracePeriod,
Volumes: []corev1.Volume{{
Name: volumeName,
VolumeSource: corev1.VolumeSource{
PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
ClaimName: restorePVCName,
},
},
}},
NodeName: selectedNode,
Volumes: volumes,
NodeName: selectedNode,
RestartPolicy: corev1.RestartPolicyNever,
SecurityContext: &corev1.PodSecurityContext{
RunAsUser: &userID,
},
},
}
pkg/exposer/generic_restore_test.go
+12 -1
View File
@@ -31,6 +31,7 @@ import (
velerotest "github.com/vmware-tanzu/velero/pkg/test"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
corev1api "k8s.io/api/core/v1"
clientTesting "k8s.io/client-go/testing"
)
@@ -74,7 +75,17 @@ func TestRestoreExpose(t *testing.T) {
Kind: "DaemonSet",
APIVersion: appsv1.SchemeGroupVersion.String(),
},
Spec: appsv1.DaemonSetSpec{},
Spec: appsv1.DaemonSetSpec{
Template: corev1.PodTemplateSpec{
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Image: "fake-image",
},
},
},
},
},
}
tests := []struct {
pkg/exposer/image.go
+25 -1
View File
@@ -18,8 +18,10 @@ package exposer
import (
"context"
"strings"
"github.com/pkg/errors"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes"
"github.com/vmware-tanzu/velero/pkg/nodeagent"
@@ -28,6 +30,11 @@ import (
type inheritedPodInfo struct {
image string
serviceAccount string
env []v1.EnvVar
volumeMounts []v1.VolumeMount
volumes []v1.Volume
logLevelArgs []string
logFormatArgs []string
}
func getInheritedPodInfo(ctx context.Context, client kubernetes.Interface, veleroNamespace string) (inheritedPodInfo, error) {
@@ -39,11 +46,28 @@ func getInheritedPodInfo(ctx context.Context, client kubernetes.Interface, veler
}
if len(podSpec.Containers) != 1 {
return podInfo, errors.Wrap(err, "unexpected pod template from node-agent")
return podInfo, errors.New("unexpected pod template from node-agent")
}
podInfo.image = podSpec.Containers[0].Image
podInfo.serviceAccount = podSpec.ServiceAccountName
podInfo.env = podSpec.Containers[0].Env
podInfo.volumeMounts = podSpec.Containers[0].VolumeMounts
podInfo.volumes = podSpec.Volumes
args := podSpec.Containers[0].Args
for i, arg := range args {
if arg == "--log-format" {
podInfo.logFormatArgs = append(podInfo.logFormatArgs, args[i:i+2]...)
} else if strings.HasPrefix(arg, "--log-format") {
podInfo.logFormatArgs = append(podInfo.logFormatArgs, arg)
} else if arg == "--log-level" {
podInfo.logLevelArgs = append(podInfo.logLevelArgs, args[i:i+2]...)
} else if strings.HasPrefix(arg, "--log-level") {
podInfo.logLevelArgs = append(podInfo.logLevelArgs, arg)
}
}
return podInfo, nil
}
pkg/exposer/image_test.go
+271
View File
@@ -0,0 +1,271 @@
/*
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 exposer
import (
"context"
"reflect"
"testing"
"github.com/stretchr/testify/assert"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/kubernetes"
appsv1 "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes/fake"
)
func TestGetInheritedPodInfo(t *testing.T) {
daemonSet := &appsv1.DaemonSet{
ObjectMeta: metav1.ObjectMeta{
Namespace: "fake-ns",
Name: "node-agent",
},
TypeMeta: metav1.TypeMeta{
Kind: "DaemonSet",
},
}
daemonSetWithNoLog := &appsv1.DaemonSet{
ObjectMeta: metav1.ObjectMeta{
Namespace: "fake-ns",
Name: "node-agent",
},
TypeMeta: metav1.TypeMeta{
Kind: "DaemonSet",
},
Spec: appsv1.DaemonSetSpec{
Template: v1.PodTemplateSpec{
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "container-1",
Image: "image-1",
Env: []v1.EnvVar{
{
Name: "env-1",
Value: "value-1",
},
{
Name: "env-2",
Value: "value-2",
},
},
VolumeMounts: []v1.VolumeMount{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
},
},
Volumes: []v1.Volume{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
ServiceAccountName: "sa-1",
},
},
},
}
daemonSetWithLog := &appsv1.DaemonSet{
ObjectMeta: metav1.ObjectMeta{
Namespace: "fake-ns",
Name: "node-agent",
},
TypeMeta: metav1.TypeMeta{
Kind: "DaemonSet",
},
Spec: appsv1.DaemonSetSpec{
Template: v1.PodTemplateSpec{
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "container-1",
Image: "image-1",
Env: []v1.EnvVar{
{
Name: "env-1",
Value: "value-1",
},
{
Name: "env-2",
Value: "value-2",
},
},
VolumeMounts: []v1.VolumeMount{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
Args: []string{
"--log-format=json",
"--log-level",
"debug",
},
Command: []string{
"command-1",
},
},
},
Volumes: []v1.Volume{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
ServiceAccountName: "sa-1",
},
},
},
}
scheme := runtime.NewScheme()
appsv1.AddToScheme(scheme)
tests := []struct {
name string
namespace string
client kubernetes.Interface
kubeClientObj []runtime.Object
result inheritedPodInfo
expectErr string
}{
{
name: "ds is not found",
namespace: "fake-ns",
expectErr: "error to get node-agent pod template: error to get node-agent daemonset: daemonsets.apps \"node-agent\" not found",
},
{
name: "ds pod container number is invalidate",
namespace: "fake-ns",
kubeClientObj: []runtime.Object{
daemonSet,
},
expectErr: "unexpected pod template from node-agent",
},
{
name: "no log info",
namespace: "fake-ns",
kubeClientObj: []runtime.Object{
daemonSetWithNoLog,
},
result: inheritedPodInfo{
image: "image-1",
serviceAccount: "sa-1",
env: []v1.EnvVar{
{
Name: "env-1",
Value: "value-1",
},
{
Name: "env-2",
Value: "value-2",
},
},
volumeMounts: []v1.VolumeMount{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
volumes: []v1.Volume{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
},
},
{
name: "with log info",
namespace: "fake-ns",
kubeClientObj: []runtime.Object{
daemonSetWithLog,
},
result: inheritedPodInfo{
image: "image-1",
serviceAccount: "sa-1",
env: []v1.EnvVar{
{
Name: "env-1",
Value: "value-1",
},
{
Name: "env-2",
Value: "value-2",
},
},
volumeMounts: []v1.VolumeMount{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
volumes: []v1.Volume{
{
Name: "volume-1",
},
{
Name: "volume-2",
},
},
logFormatArgs: []string{
"--log-format=json",
},
logLevelArgs: []string{
"--log-level",
"debug",
},
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
info, err := getInheritedPodInfo(context.Background(), fakeKubeClient, test.namespace)
if test.expectErr == "" {
assert.NoError(t, err)
assert.True(t, reflect.DeepEqual(info, test.result))
} else {
assert.EqualError(t, err, test.expectErr)
}
})
}
}
pkg/exposer/types.go
+3 -2
View File
@@ -35,6 +35,7 @@ type ExposeResult struct {
// ExposeByPod defines the result for the expose method that a hosting pod is created
type ExposeByPod struct {
HostingPod *corev1.Pod
VolumeName string
HostingPod *corev1.Pod
HostingContainer string
VolumeName string
}