e2e: add tests for external-snapshot-metadata sidecar

The tests validate the sidecar's functionality, integration with the CSI driver and correctness of metadata retrieval for snapshot backups. This will help CSI vendors test their implementation of the snapshot-metadata feature. Issue: kubernetes-csi/external-snapshot-metadata#120 Signed-off-by: Praveen M <m.praveen@ibm.com>
2026-07-24 12:43:56 +00:00 · 2025-03-19 12:55:38 +05:30
parent 0cf6c382a0
commit a74bf84787
13 changed files with 975 additions and 1 deletions
--- a/test/e2e/feature/feature.go
+++ b/test/e2e/feature/feature.go
@@ -407,6 +407,15 @@ var (
 	// and the networking.k8s.io/v1alpha1 API.
 	ServiceCIDRs = framework.WithFeature(framework.ValidFeatures.Add("ServiceCIDRs"))

+	// Owner: sig-storage
+	// KEP: https://kep.k8s.io/3314
+	// Tests marked with this feature require:
+	// - A CSI driver that supports the snapshot metadata service (e.g., CSI hostpath driver with --enable-snapshot-metadata)
+	// - The external-snapshot-metadata sidecar deployed alongside the CSI driver
+	// - The SnapshotMetadataService CRD (cbt.storage.k8s.io/v1alpha1) installed
+	// - A storage driver that implements the CapSnapshotMetadata capability
+	SnapshotMetadata = framework.WithFeature(framework.ValidFeatures.Add("SnapshotMetadata"))
+
 	// TODO: document the feature (owning SIG, when to use this feature for a test)
 	StackdriverAcceleratorMonitoring = framework.WithFeature(framework.ValidFeatures.Add("StackdriverAcceleratorMonitoring"))

--- a/test/e2e/storage/drivers/csi.go
+++ b/test/e2e/storage/drivers/csi.go
@@ -150,6 +150,7 @@ func InitHostPathCSIDriver() storageframework.TestDriver {
 	capabilities := map[storageframework.Capability]bool{
 		storageframework.CapPersistence:                    true,
 		storageframework.CapSnapshotDataSource:             true,
+		storageframework.CapSnapshotMetadata:               true,
 		storageframework.CapMultiPODs:                      true,
 		storageframework.CapBlock:                          true,
 		storageframework.CapPVCDataSource:                  true,
@@ -182,6 +183,7 @@ func InitHostPathCSIDriver() storageframework.TestDriver {
 		},
 		"test/e2e/testing-manifests/storage-csi/external-attacher/rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/external-provisioner/rbac.yaml",
+		"test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/external-snapshotter/csi-snapshotter/rbac-csi-snapshotter.yaml",
 		"test/e2e/testing-manifests/storage-csi/external-health-monitor/external-health-monitor-controller/rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/external-resizer/rbac.yaml",
@@ -341,6 +343,12 @@ func (h *hostpathCSIDriver) PrepareTest(ctx context.Context, f *framework.Framew
 		framework.Failf("deploying %s driver: %v", h.driverInfo.Name, err)
 	}

+	// SnapshotMetadata E2E resources creation
+	err = utils.CreateSnapshotMetadataResources(ctx, f, h.driverInfo.Name, driverNamespace.Name)
+	if err != nil {
+		framework.Failf("creating snapshot-metadata resources for %s driver: %v", h.driverInfo.Name, err)
+	}
+
 	cleanupFunc := generateDriverCleanupFunc(
 		f,
 		h.driverInfo.Name,
--- a/test/e2e/storage/framework/testdriver.go
+++ b/test/e2e/storage/framework/testdriver.go
@@ -130,6 +130,10 @@ type SnapshottableTestDriver interface {
 	GetSnapshotClass(ctx context.Context, config *PerTestConfig, parameters map[string]string) *unstructured.Unstructured
 }

+type SnapshotMetadataTestDriver interface {
+	TestDriver
+}
+
 // VolumeGroupSnapshottableTestDriver represents an interface for a TestDriver that supports DynamicGroupSnapshot
 type VolumeGroupSnapshottableTestDriver interface {
 	TestDriver
@@ -171,6 +175,7 @@ const (
 	CapVolumeMountGroup    Capability = "volumeMountGroup"   // Driver has the VolumeMountGroup CSI node capability. Because this is a FSGroup feature, the fsGroup capability must also be set to true.
 	CapExec                Capability = "exec"               // exec a file in the volume
 	CapSnapshotDataSource  Capability = "snapshotDataSource" // support populate data from snapshot
+	CapSnapshotMetadata    Capability = "snapshotMetadata"   // support group snapshot
 	CapVolumeGroupSnapshot Capability = "groupSnapshot"      // support group snapshot
 	CapPVCDataSource       Capability = "pvcDataSource"      // support populate data from pvc

--- a/test/e2e/storage/framework/testpattern.go
+++ b/test/e2e/storage/framework/testpattern.go
@@ -312,6 +312,15 @@ var (
 		AllowExpansion: true,
 	}

+	// Deginition for snapshot metadata
+	SnapshotMetadata = TestPattern{
+		Name:                   "SnapshotMetadata",
+		VolType:                DynamicPV,
+		VolMode:                v1.PersistentVolumeBlock,
+		SnapshotType:           DynamicCreatedSnapshot,
+		SnapshotDeletionPolicy: DeleteSnapshot,
+	}
+
 	// Definitions for snapshot case

 	// DynamicSnapshotDelete is TestPattern for "Dynamic snapshot"
--- a/test/e2e/storage/testsuites/base.go
+++ b/test/e2e/storage/testsuites/base.go
@@ -82,6 +82,7 @@ var CSISuites = append(BaseSuites,
 		return InitCustomEphemeralTestSuite(CSIEphemeralTestPatterns())
 	},
 	InitSnapshottableTestSuite,
+	InitSnapshotMetadataTestSuite,
 	InitSnapshottableStressTestSuite,
 	InitVolumePerformanceTestSuite,
 	InitPvcDeletionPerformanceTestSuite,
--- a/test/e2e/storage/testsuites/snapshot-metadata.go
+++ b/test/e2e/storage/testsuites/snapshot-metadata.go
@@ -0,0 +1,422 @@
+/*
+Copyright The Kubernetes Authors.
+
+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 testsuites
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/test/e2e/feature"
+	"k8s.io/kubernetes/test/e2e/framework"
+
+	e2ekubectl "k8s.io/kubernetes/test/e2e/framework/kubectl"
+	e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
+	e2epv "k8s.io/kubernetes/test/e2e/framework/pv"
+	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
+	e2evolume "k8s.io/kubernetes/test/e2e/framework/volume"
+	storageframework "k8s.io/kubernetes/test/e2e/storage/framework"
+	storageutils "k8s.io/kubernetes/test/e2e/storage/utils"
+	admissionapi "k8s.io/pod-security-admission/api"
+)
+
+type SnapshotMetadataTest struct{}
+
+type snapshotMetadataTestSuite struct {
+	tsInfo storageframework.TestSuiteInfo
+}
+
+func InitCustomSnapshotMetadataTestSuite(patterns []storageframework.TestPattern) storageframework.TestSuite {
+	return &snapshotMetadataTestSuite{
+		tsInfo: storageframework.TestSuiteInfo{
+			Name:         "snapshotmetadata",
+			TestPatterns: patterns,
+			SupportedSizeRange: e2evolume.SizeRange{
+				// Min: "1Gi",
+				Max: "1Gi",
+			},
+			TestTags: []interface{}{feature.SnapshotMetadata},
+		},
+	}
+}
+
+func InitSnapshotMetadataTestSuite() storageframework.TestSuite {
+	patterns := []storageframework.TestPattern{
+		storageframework.SnapshotMetadata,
+	}
+
+	return InitCustomSnapshotMetadataTestSuite(patterns)
+}
+
+// GetTestSuiteInfo implements framework.TestSuite.
+func (s *snapshotMetadataTestSuite) GetTestSuiteInfo() storageframework.TestSuiteInfo {
+	return s.tsInfo
+}
+
+// SkipUnsupportedTests implements framework.TestSuite.
+func (s *snapshotMetadataTestSuite) SkipUnsupportedTests(driver storageframework.TestDriver, pattern storageframework.TestPattern) {
+	_, ok := driver.(storageframework.SnapshotMetadataTestDriver)
+	driverInfo := driver.GetDriverInfo()
+	if !driverInfo.Capabilities[storageframework.CapSnapshotMetadata] || !ok {
+		e2eskipper.Skipf("Driver %q does not support snapshot metadata", driverInfo.Name)
+	}
+}
+
+func constructVerifierCommand(
+	namespace, snapshotName, previousSnapshotName,
+	sourceDevicePath, targetDevicePath string,
+) string {
+	command := []string{
+		"/tools/snapshot-metadata-verifier",
+		"-max-results 10",
+		"-starting-offset 0",
+		"-kubeconfig=\"\"",
+		fmt.Sprintf("-namespace=%s", namespace),
+	}
+	if snapshotName != "" {
+		command = append(command, fmt.Sprintf("-snapshot=%s", snapshotName))
+	}
+	if previousSnapshotName != "" {
+		command = append(command, fmt.Sprintf("-previous-snapshot=%s", previousSnapshotName))
+	}
+	if sourceDevicePath != "" {
+		command = append(command, fmt.Sprintf("-source-device-path=%s", sourceDevicePath))
+	}
+	if targetDevicePath != "" {
+		command = append(command, fmt.Sprintf("-target-device-path=%s", targetDevicePath))
+	}
+
+	// return command
+	return strings.Join(command, " ")
+}
+
+func runSnapshotMetadataVerifier(pod *v1.Pod, toolCommand string) {
+	ginkgo.By("run snapshot-metadata-verifier")
+	stdout, stderr, err := e2ekubectl.RunKubectlWithFullOutput(
+		pod.Namespace,
+		strings.Split(toolCommand, " ")...,
+	)
+	framework.ExpectNoError(err, "failed to run snapshot-metadata-verifier tool")
+	if stderr != "" {
+		framework.Failf("failed to run snapshot-metadata-verifier tool:\nstdout:%s\nstderr:%s\n", stdout, stderr)
+	}
+}
+
+func createBackupClientResources(ctx context.Context, f *framework.Framework) {
+	ginkgo.By("creating backup client resources")
+	// create from manifest
+	backupClientManifests := []string{
+		"test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/backup-client-rbac.yaml",
+	}
+	err := storageutils.CreateFromManifests(ctx, f, f.Namespace,
+		func(item interface{}) error { return nil },
+		backupClientManifests...)
+	framework.ExpectNoError(err)
+}
+
+func runSyncInPod(pod *v1.Pod) {
+	ginkgo.By(fmt.Sprintf("running sync in pod %s", pod.Name))
+	writeCmd := fmt.Sprintf("exec %s -c write-pod -- sync", pod.Name)
+	stdout, stderr, err := e2ekubectl.RunKubectlWithFullOutput(
+		pod.Namespace,
+		strings.Split(writeCmd, " ")...,
+	)
+	framework.ExpectNoError(err, "failed running sync command")
+	if stderr != "" {
+		framework.Failf("failed running sync command:\nstdout:%s\nstderr:%s\n", stdout, stderr)
+	}
+}
+
+func writeToDeviceInPod(pod *v1.Pod, writeCmd string) {
+	ginkgo.By(fmt.Sprintf("writing content into pod %s", pod.Name))
+	stdout, stderr, err := e2ekubectl.RunKubectlWithFullOutput(
+		pod.Namespace,
+		strings.Split(writeCmd, " ")...,
+	)
+	framework.ExpectNoError(err, "failed writing the contents")
+	if stderr != "" {
+		framework.Failf("failed writing the contents:\nstdout:%s\nstderr:%s\n", stdout, stderr)
+	}
+	runSyncInPod(pod)
+}
+
+const (
+	backupClientServiceAccountName = "backup-app-service-account"
+	sourceDevicePvcName            = "source-device"
+	targetDevicePvcName            = "target-device"
+	installToolContainerName       = "install-tool"
+	installToolImage               = "golang:1.25.0"
+	installToolCommand             = "/bin/sh -c 'go install github.com/kubernetes-csi/external-snapshot-metadata/tools/snapshot-metadata-verifier@main && cp $(go env GOPATH)/bin/snapshot-metadata-verifier /output'"
+	sharedVolumeName               = "shared-volume"
+	sharedVolumeMountPath          = "/tools"
+)
+
+func createBackupClientPod(ctx context.Context, f *framework.Framework, clientSet clientset.Interface, source, target *v1.PersistentVolumeClaim) (*v1.Pod, error) {
+	ginkgo.By("creating backup client pod")
+	backupClientPodConfig := e2epod.Config{
+		NS:   f.Namespace.Name,
+		PVCs: []*v1.PersistentVolumeClaim{source, target},
+	}
+
+	backupClientPodObject, err := e2epod.MakeSecPod(&backupClientPodConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get pod config: %w", err)
+	}
+
+	backupClientPodObject.Spec.InitContainers = append(backupClientPodObject.Spec.InitContainers, v1.Container{
+		Name:    installToolContainerName,
+		Image:   installToolImage,
+		Command: []string{"/bin/sh", "-c", installToolCommand},
+		VolumeMounts: []v1.VolumeMount{
+			{
+				Name:      sharedVolumeName,
+				MountPath: "/output",
+			},
+		},
+	})
+
+	backupClientPodObject.Spec.Containers[0].VolumeMounts = append(
+		backupClientPodObject.Spec.Containers[0].VolumeMounts,
+		v1.VolumeMount{
+			Name:      sharedVolumeName,
+			MountPath: sharedVolumeMountPath,
+		},
+	)
+	backupClientPodObject.Spec.Volumes = append(
+		backupClientPodObject.Spec.Volumes,
+		v1.Volume{
+			Name: sharedVolumeName,
+			VolumeSource: v1.VolumeSource{
+				EmptyDir: &v1.EmptyDirVolumeSource{},
+			},
+		},
+	)
+
+	backupClientPodObject.Spec.ServiceAccountName = backupClientServiceAccountName
+	backupClientPod, err := clientSet.CoreV1().Pods(f.Namespace.Name).Create(ctx, backupClientPodObject, metav1.CreateOptions{})
+	framework.ExpectNoError(err, "Failed to create backup client pod")
+	err = e2epod.WaitForPodNameRunningInNamespace(ctx, clientSet, backupClientPod.Name, f.Namespace.Name)
+	framework.ExpectNoError(err, "Failed to wait for pod %s to turn into running status", backupClientPod.Name)
+	backupClientPod, err = clientSet.CoreV1().Pods(f.Namespace.Name).Get(ctx, backupClientPod.Name, metav1.GetOptions{})
+	framework.ExpectNoError(err, "Failed to get pod %s", backupClientPod.Name)
+
+	return backupClientPod, nil
+}
+
+func createPVCFromSnapshot(ctx context.Context, clientSet clientset.Interface, f *framework.Framework, pvcName, snapshotName string, testPVC *v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error) {
+	ginkgo.By(fmt.Sprintf("creating pvc %s from the snapshot %s", pvcName, snapshotName))
+	sourceDevicePvcObject := e2epv.MakePersistentVolumeClaim(e2epv.PersistentVolumeClaimConfig{
+		Name:             pvcName,
+		ClaimSize:        testPVC.Spec.Resources.Requests.Storage().String(),
+		StorageClassName: testPVC.Spec.StorageClassName,
+		VolumeMode:       testPVC.Spec.VolumeMode,
+	}, f.Namespace.Name)
+
+	group := "snapshot.storage.k8s.io"
+	sourceDevicePvcObject.Spec.DataSource = &v1.TypedLocalObjectReference{
+		APIGroup: &group,
+		Kind:     "VolumeSnapshot",
+		Name:     snapshotName,
+	}
+	restorePvc, err := clientSet.CoreV1().PersistentVolumeClaims(f.Namespace.Name).Create(ctx, sourceDevicePvcObject, metav1.CreateOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create pvc: %w", err)
+	}
+
+	return restorePvc, nil
+}
+
+// DefineTests implements framework.TestSuite.
+func (s *snapshotMetadataTestSuite) DefineTests(driver storageframework.TestDriver, pattern storageframework.TestPattern) {
+	var (
+		smDriver  storageframework.SnapshotMetadataTestDriver
+		config    *storageframework.PerTestConfig
+		clientSet clientset.Interface
+
+		//
+		volume         *storageframework.VolumeResource
+		testPVC        *v1.PersistentVolumeClaim
+		testPod        *v1.Pod
+		testDevicePath string
+		err            error
+
+		// backup client
+		sourceDeviceName string
+		sourveDevicePath string
+		targetDevicePath string
+		targetDeviceName string
+	)
+
+	f := framework.NewDefaultFramework("snapshotmetadata")
+	f.NamespacePodSecurityLevel = admissionapi.LevelPrivileged
+
+	createBackupClientPod := func(ctx context.Context, source, target *v1.PersistentVolumeClaim) *v1.Pod {
+		createBackupClientResources(ctx, f)
+		backupClientPod, err := createBackupClientPod(ctx, f, clientSet, source, target)
+		framework.ExpectNoError(err, "Failed to create backup client pod")
+
+		// Update source and target device path
+		for _, volume := range backupClientPod.Spec.Volumes {
+			if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName == sourceDevicePvcName {
+				sourceDeviceName = volume.Name
+			}
+			if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName == targetDevicePvcName {
+				targetDeviceName = volume.Name
+			}
+		}
+		for _, device := range backupClientPod.Spec.Containers[0].VolumeDevices {
+			if device.Name == sourceDeviceName {
+				sourveDevicePath = device.DevicePath
+			}
+			if device.Name == targetDeviceName {
+				targetDevicePath = device.DevicePath
+			}
+		}
+
+		err = framework.Gomega().Expect(sourveDevicePath).NotTo(gomega.BeEmpty())
+		framework.ExpectNoError(err, "Failed to get source device path")
+		err = framework.Gomega().Expect(targetDevicePath).NotTo(gomega.BeEmpty())
+		framework.ExpectNoError(err, "Failed to get target device path")
+
+		return backupClientPod
+	}
+
+	init := func(ctx context.Context) {
+		framework.Logf("Initializing test")
+		smDriver = driver.(storageframework.SnapshotMetadataTestDriver)
+		clientSet = f.ClientSet
+
+		config = smDriver.PrepareTest(ctx, f)
+
+		pattern.VolMode = v1.PersistentVolumeBlock
+		volume = storageframework.CreateVolumeResource(ctx, smDriver, config, pattern, s.GetTestSuiteInfo().SupportedSizeRange)
+		testPVC = volume.Pvc
+		podConfig := e2epod.Config{
+			NS:           f.Namespace.Name,
+			PVCs:         []*v1.PersistentVolumeClaim{testPVC},
+			SeLinuxLabel: e2epv.SELinuxLabel,
+		}
+
+		testPod, err = e2epod.MakeSecPod(&podConfig)
+		framework.ExpectNoError(err, "Failed to get pod config for %s", testPod.Name)
+		testPod, err = clientSet.CoreV1().Pods(f.Namespace.Name).Create(ctx, testPod, metav1.CreateOptions{})
+		framework.ExpectNoError(err, "Failed to create pod %s", testPod.Name)
+		err = e2epod.WaitForPodRunningInNamespace(ctx, clientSet, testPod)
+		framework.ExpectNoError(err, "Failed to wait for pod %s to turn into running status. Error", testPod.Name)
+		testPod, err = clientSet.CoreV1().Pods(testPod.Namespace).Get(ctx, testPod.Name, metav1.GetOptions{})
+		framework.ExpectNoError(err, "Failed to get pod %s", testPod.Name)
+		testDevicePath = testPod.Spec.Containers[0].VolumeDevices[0].DevicePath // Other way?
+	}
+
+	createPVCFromSnapshot := func(ctx context.Context, pvcName, snaphotName string) *v1.PersistentVolumeClaim {
+		restorePvc, err := createPVCFromSnapshot(ctx, clientSet, f, pvcName, snaphotName, testPVC)
+		framework.ExpectNoError(err, "Failed to create pvc from snapshot")
+		return restorePvc
+	}
+
+	ginkgo.AfterEach(func(ctx context.Context) {
+		// framework.Logf("Don't cleanup")
+		// ginkgo.DeferCleanup(nil)
+	})
+
+	ginkgo.It("should verify GetMetadataDelta", func(ctx context.Context) {
+		init(ctx)
+
+		parameters := map[string]string{}
+		sDriver := smDriver.(storageframework.SnapshottableTestDriver)
+
+		// Write content
+		writeCmd := fmt.Sprintf("exec %s -c write-pod -- dd if=/dev/urandom of=%s bs=4K count=6 oflag=direct status=none", testPod.Name, testDevicePath)
+		writeToDeviceInPod(testPod, writeCmd)
+
+		// Create snap-1
+		ginkgo.By("taking snapshot snap-1")
+		snapResource1 := storageframework.CreateSnapshotResource(ctx, sDriver, config, pattern, testPVC.Name, testPVC.Namespace, f.Timeouts, parameters)
+		ginkgo.DeferCleanup(snapResource1.CleanupResource, f.Timeouts)
+
+		// Write more content
+		writeCmd = fmt.Sprintf("exec %s -c write-pod -- dd if=/dev/urandom of=%s bs=4K count=3 oflag=direct status=none", testPod.Name, testDevicePath)
+		writeToDeviceInPod(testPod, writeCmd)
+
+		// Create snap-2
+		ginkgo.By("taking snapshot snap-2")
+		snapResource2 := storageframework.CreateSnapshotResource(ctx, sDriver, config, pattern, testPVC.Name, testPVC.Namespace, f.Timeouts, parameters)
+		ginkgo.DeferCleanup(snapResource2.CleanupResource, f.Timeouts)
+
+		// Restore PVC (source device) from snapshot snap-2
+		sourceDevicePvc := createPVCFromSnapshot(ctx, sourceDevicePvcName, snapResource2.Vs.GetName())
+
+		// Restore PVC (target device) from snapshot snap-1
+		targetDevicePvc := createPVCFromSnapshot(ctx, targetDevicePvcName, snapResource1.Vs.GetName())
+
+		// create backup client
+		backupClientPod := createBackupClientPod(ctx, sourceDevicePvc, targetDevicePvc)
+
+		// Run snapshot-metadata-verifier
+		ginkgo.By("run snapshot-metadata-verifier")
+		toolCommand := fmt.Sprintf("exec %s -c write-pod -- %s",
+			backupClientPod.Name,
+			constructVerifierCommand(f.Namespace.Name, snapResource2.Vs.GetName(), snapResource1.Vs.GetName(), sourveDevicePath, targetDevicePath))
+		runSnapshotMetadataVerifier(backupClientPod, toolCommand)
+	})
+
+	ginkgo.It("should verify GetAllocatedMetadata", func(ctx context.Context) {
+		init(ctx)
+
+		parameters := map[string]string{}
+		sDriver := smDriver.(storageframework.SnapshottableTestDriver)
+
+		// Write content
+		writeCmd := fmt.Sprintf("exec %s -c write-pod -- dd if=/dev/urandom of=%s bs=4K count=6 oflag=direct status=none", testPod.Name, testDevicePath)
+		writeToDeviceInPod(testPod, writeCmd)
+
+		// Create snap-1
+		ginkgo.By("taking snapshot snap-1")
+		snapResource1 := storageframework.CreateSnapshotResource(ctx, sDriver, config, pattern, testPVC.Name, testPVC.Namespace, f.Timeouts, parameters)
+		ginkgo.DeferCleanup(snapResource1.CleanupResource, f.Timeouts)
+
+		// Restore PVC (source device) from snapshot snap-1
+		sourceDevicePvc := createPVCFromSnapshot(ctx, sourceDevicePvcName, snapResource1.Vs.GetName())
+
+		// create new PVC (target device)
+		ginkgo.By("creating pvc target-device")
+		targetPvcClaim := e2epv.MakePersistentVolumeClaim(e2epv.PersistentVolumeClaimConfig{
+			Name:             "target-device",
+			ClaimSize:        testPVC.Spec.Resources.Requests.Storage().String(),
+			StorageClassName: testPVC.Spec.StorageClassName,
+			VolumeMode:       testPVC.Spec.VolumeMode,
+		}, f.Namespace.Name)
+
+		targetDevicePvc, err := e2epv.CreatePVC(ctx, f.ClientSet, f.Namespace.Name, targetPvcClaim)
+		framework.ExpectNoError(err)
+
+		// create backup client
+		backupClientPod := createBackupClientPod(ctx, sourceDevicePvc, targetDevicePvc)
+
+		// Run snapshot-metadata-verifier
+		ginkgo.By("run snapshot-metadata-verifier")
+		toolCommand := fmt.Sprintf("exec %s -c write-pod -- %s",
+			backupClientPod.Name,
+			constructVerifierCommand(f.Namespace.Name, snapResource1.Vs.GetName(), "", sourveDevicePath, targetDevicePath))
+		runSnapshotMetadataVerifier(backupClientPod, toolCommand)
+	})
+}
--- a/test/e2e/storage/utils/snapshot-metadata.go
+++ b/test/e2e/storage/utils/snapshot-metadata.go
@@ -0,0 +1,329 @@
+/*
+Copyright The Kubernetes Authors.
+
+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 utils
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"fmt"
+	"math/big"
+	"time"
+
+	"github.com/onsi/ginkgo/v2"
+	v1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/kubernetes/test/e2e/framework"
+)
+
+// Constants for common values
+const (
+	CommonName                         = "csi-snapshot-metadata"
+	TLSSecretName                      = "csi-snapshot-metadata-server-certs"
+	ServicePort                        = 6443
+	TargetPort                         = 50051
+	CertificateValidityDays            = 365
+	ServerCertificateValidity          = 60
+	CertificateKeySize                 = 4096
+	AppSelectorKey                     = "app.kubernetes.io/name"
+	AppSelectorValue                   = "csi-hostpathplugin"
+	SnapshotMetadataCRAudience         = "csi-snapshot-metadata-test"
+	SnapshotMetadataServiceCRDManifest = "test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/cbt.storage.k8s.io_snapshotmetadataservices.yaml"
+
+	// SnapshotMetadataService API constants
+	SnapshotMetadataServiceGroup      = "cbt.storage.k8s.io"
+	SnapshotMetadataServiceVersion    = "v1alpha1"
+	SnapshotMetadataServiceAPIVersion = SnapshotMetadataServiceGroup + "/" + SnapshotMetadataServiceVersion
+	SnapshotMetadataServiceKind       = "SnapshotMetadataService"
+	SnapshotMetadataServiceResource   = "snapshotmetadataservices"
+)
+
+// generateCA creates a self-signed CA certificate and private key using RSA.
+func generateCA() (*x509.Certificate, *rsa.PrivateKey, error) {
+	ginkgo.By("Generating CA certificate and private key")
+
+	caPrivateKey, err := rsa.GenerateKey(rand.Reader, CertificateKeySize)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to generate CA private key: %w", err)
+	}
+
+	caTemplate := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject: pkix.Name{
+			CommonName: CommonName,
+		},
+		NotBefore:             time.Now(),
+		NotAfter:              time.Now().Add(CertificateValidityDays * 24 * time.Hour),
+		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature,
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+	}
+
+	caCertBytes, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caPrivateKey.PublicKey, caPrivateKey)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create CA certificate: %w", err)
+	}
+
+	caCert, err := x509.ParseCertificate(caCertBytes)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to parse CA certificate: %w", err)
+	}
+	return caCert, caPrivateKey, nil
+}
+
+// generateServerCert creates a server certificate signed by the given CA using RSA.
+func generateServerCert(driverNamespace string, caCert *x509.Certificate, caPrivateKey *rsa.PrivateKey) ([]byte, []byte, error) {
+	ginkgo.By("Generating server certificate and private key")
+
+	serverPrivateKey, err := rsa.GenerateKey(rand.Reader, CertificateKeySize)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to generate server private key: %w", err)
+	}
+
+	serverTemplate := &x509.Certificate{
+		SerialNumber: big.NewInt(2),
+		Subject: pkix.Name{
+			CommonName: CommonName,
+		},
+		DNSNames: []string{
+			fmt.Sprintf(".%s", driverNamespace),
+			fmt.Sprintf("%s.%s", CommonName, driverNamespace),
+		},
+		NotBefore:   time.Now(),
+		NotAfter:    time.Now().Add(ServerCertificateValidity * 24 * time.Hour),
+		KeyUsage:    x509.KeyUsageDigitalSignature,
+		ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}
+
+	serverCertBytes, err := x509.CreateCertificate(rand.Reader, serverTemplate, caCert, &serverPrivateKey.PublicKey, caPrivateKey)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create server certificate: %w", err)
+	}
+
+	serverKeyBytes := x509.MarshalPKCS1PrivateKey(serverPrivateKey)
+
+	return serverCertBytes, serverKeyBytes, nil
+}
+
+// createTLSSecret creates or updates a TLS secret in the specified namespace.
+func createTLSSecret(ctx context.Context, f *framework.Framework, driverNamespace string, certBytes, keyBytes []byte) error {
+	ginkgo.By("Creating TLS secret")
+
+	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certBytes})
+	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: keyBytes})
+
+	tlsSecret := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      TLSSecretName,
+			Namespace: driverNamespace,
+		},
+		Type: v1.SecretTypeTLS,
+		Data: map[string][]byte{
+			"tls.crt": certPEM,
+			"tls.key": keyPEM,
+		},
+	}
+
+	client := f.ClientSet.CoreV1().Secrets(driverNamespace)
+	if _, err := client.Create(ctx, tlsSecret, metav1.CreateOptions{}); err != nil {
+		return fmt.Errorf("failed to create or update TLS secret: %w", err)
+	}
+
+	framework.Logf("TLS secret created successfully: %s", tlsSecret.Name)
+	return nil
+}
+
+// createSnapshotMetadataSVC creates a Kubernetes service for the snapshot metadata.
+func createSnapshotMetadataSVC(ctx context.Context, f *framework.Framework, driverName, driverNamespace string) error {
+	ginkgo.By("Creating Kubernetes service for snapshot metadata")
+	svc := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      CommonName,
+			Namespace: driverNamespace,
+		},
+		Spec: v1.ServiceSpec{
+			Ports: []v1.ServicePort{
+				{
+					Name:       "snapshot-metadata-port",
+					Port:       ServicePort,
+					Protocol:   "TCP",
+					TargetPort: intstr.FromInt(TargetPort),
+				},
+			},
+			Selector: map[string]string{
+				AppSelectorKey: AppSelectorValue,
+			},
+		},
+	}
+
+	client := f.ClientSet.CoreV1().Services(driverNamespace)
+	if _, err := client.Create(ctx, svc, metav1.CreateOptions{}); err != nil {
+		return fmt.Errorf("failed to create service: %w", err)
+	}
+	framework.Logf("Service created successfully: %s", svc.Name)
+
+	return nil
+}
+
+// crdExistsInDiscovery checks to see if the given CRD exists in discovery at all served versions.
+func crdExistsInDiscovery(client apiextensionsclientset.Interface, crd *apiextensionsv1.CustomResourceDefinition) bool {
+	var versions []string
+	for _, v := range crd.Spec.Versions {
+		if v.Served {
+			versions = append(versions, v.Name)
+		}
+	}
+	for _, v := range versions {
+		if !crdVersionExistsInDiscovery(client, crd, v) {
+			return false
+		}
+	}
+	return true
+}
+
+func crdVersionExistsInDiscovery(client apiextensionsclientset.Interface, crd *apiextensionsv1.CustomResourceDefinition, version string) bool {
+	resourceList, err := client.Discovery().ServerResourcesForGroupVersion(crd.Spec.Group + "/" + version)
+	if err != nil {
+		return false
+	}
+	for _, resource := range resourceList.APIResources {
+		if resource.Name == crd.Spec.Names.Plural {
+			return true
+		}
+	}
+	return false
+}
+
+// createSnapshotMetadataServiceCRD creates the SnapshotMetadataService CRD from the manifest if it doesn't already exist.
+func createSnapshotMetadataServiceCRD(ctx context.Context, f *framework.Framework) error {
+	ginkgo.By("Creating SnapshotMetadataService CRD if not present")
+
+	// Load CRD from manifest
+	items, err := LoadFromManifests(SnapshotMetadataServiceCRDManifest)
+	if err != nil {
+		return fmt.Errorf("failed to load SnapshotMetadataService CRD manifest: %w", err)
+	}
+
+	if len(items) == 0 {
+		return fmt.Errorf("no items found in SnapshotMetadataService CRD manifest")
+	}
+
+	crd, ok := items[0].(*apiextensionsv1.CustomResourceDefinition)
+	if !ok {
+		return fmt.Errorf("expected CustomResourceDefinition, got %T", items[0])
+	}
+
+	// Create apiextensions client
+	apiextClient, err := apiextensionsclientset.NewForConfig(f.ClientConfig())
+	if err != nil {
+		return fmt.Errorf("failed to create apiextensions client: %w", err)
+	}
+
+	// Check if CRD already exists in discovery
+	if crdExistsInDiscovery(apiextClient, crd) {
+		framework.Logf("SnapshotMetadataService CRD already exists, skipping creation")
+		return nil
+	}
+
+	// Create the CRD
+	if _, err := apiextClient.ApiextensionsV1().CustomResourceDefinitions().Create(ctx, crd, metav1.CreateOptions{}); err != nil {
+		return fmt.Errorf("failed to create SnapshotMetadataService CRD: %w", err)
+	}
+
+	// Wait for CRD to appear in discovery
+	if err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) {
+		return crdExistsInDiscovery(apiextClient, crd), nil
+	}); err != nil {
+		return fmt.Errorf("failed to see SnapshotMetadataService CRD in discovery: %w", err)
+	}
+
+	framework.Logf("SnapshotMetadataService CRD created successfully: %s", crd.Name)
+	return nil
+}
+
+// createSnapshotMetdataServiceCR creates a SnapshotMetadataService custom resource.
+func createSnapshotMetdataServiceCR(ctx context.Context, f *framework.Framework, driverName, driverNamespace string, caCert *x509.Certificate) error {
+	ginkgo.By("Creating snapshot metadata service CR")
+
+	sms := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"kind":       SnapshotMetadataServiceKind,
+			"apiVersion": SnapshotMetadataServiceAPIVersion,
+			"metadata": map[string]interface{}{
+				"name": fmt.Sprintf("%s-%s", driverName, f.UniqueName),
+			},
+			"spec": map[string]interface{}{
+				"caCert":   pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caCert.Raw}),
+				"audience": SnapshotMetadataCRAudience,
+				"address":  fmt.Sprintf("%s.%s:%d", CommonName, driverNamespace, ServicePort),
+			},
+		},
+	}
+
+	gvr := schema.GroupVersionResource{
+		Group:    SnapshotMetadataServiceGroup,
+		Version:  SnapshotMetadataServiceVersion,
+		Resource: SnapshotMetadataServiceResource,
+	}
+
+	if _, err := f.DynamicClient.Resource(gvr).Create(ctx, sms, metav1.CreateOptions{}); err != nil {
+		return fmt.Errorf("failed to create SnapshotMetadataService: %w", err)
+	}
+
+	return nil
+}
+
+// CreateSnapshotMetadataResources sets up the snapshot metadata resources.
+func CreateSnapshotMetadataResources(ctx context.Context, f *framework.Framework, driverName, driverNamespace string) error {
+	caCert, caPrivateKey, err := generateCA()
+	if err != nil {
+		return fmt.Errorf("failed to generate CA certificate: %w", err)
+	}
+
+	serverCertBytes, serverKeyBytes, err := generateServerCert(driverNamespace, caCert, caPrivateKey)
+	if err != nil {
+		return fmt.Errorf("failed to generate server certificate: %w", err)
+	}
+
+	if err := createTLSSecret(ctx, f, driverNamespace, serverCertBytes, serverKeyBytes); err != nil {
+		return fmt.Errorf("failed to create TLS secret: %w", err)
+	}
+
+	if err := createSnapshotMetadataSVC(ctx, f, driverName, driverNamespace); err != nil {
+		return fmt.Errorf("failed to create snapshot metadata svc: %w", err)
+	}
+
+	// Create SnapshotMetadataService CRD, if not already present
+	if err := createSnapshotMetadataServiceCRD(ctx, f); err != nil {
+		return fmt.Errorf("failed to create SnapshotMetadataService CRD: %w", err)
+	}
+
+	if err := createSnapshotMetdataServiceCR(ctx, f, driverName, driverNamespace, caCert); err != nil {
+		return fmt.Errorf("failed to create SnapshotMetadataService CR: %w", err)
+	}
+
+	return nil
+}
--- a/test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/backup-client-rbac.yaml
+++ b/test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/backup-client-rbac.yaml
@@ -0,0 +1,47 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: backup-app-service-account
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: external-snapshot-metadata-client-runner
+rules:
+- apiGroups:
+  - snapshot.storage.k8s.io
+  resources:
+  - volumesnapshots
+  - volumesnapshotcontents
+  verbs:
+  - get
+  - list
+  - watch
+# Access to discover SnapshotMetadataService resources
+- apiGroups:
+  - cbt.storage.k8s.io
+  resources:
+  - snapshotmetadataservices
+  verbs:
+  - get
+  - list
+## Access to create sa tokens with tokenrequests API if client needs to generate SA token
+- apiGroups:
+  - ""
+  resources:
+  - serviceaccounts/token
+  verbs:
+  - create
+  - get
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: backup-app-cluster-role-binding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: external-snapshot-metadata-client-runner
+subjects:
+- kind: ServiceAccount
+  name: backup-app-service-account
--- a/test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/cbt.storage.k8s.io_snapshotmetadataservices.yaml
+++ b/test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/cbt.storage.k8s.io_snapshotmetadataservices.yaml
@@ -0,0 +1,76 @@
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.15.0
+    api-approved.kubernetes.io: "https://github.com/kubernetes-csi/external-snapshot-metadata/pull/2"
+  name: snapshotmetadataservices.cbt.storage.k8s.io
+spec:
+  group: cbt.storage.k8s.io
+  names:
+    kind: SnapshotMetadataService
+    listKind: SnapshotMetadataServiceList
+    plural: snapshotmetadataservices
+    shortNames:
+    - sms
+    singular: snapshotmetadataservice
+  scope: Cluster
+  versions:
+  - name: v1alpha1
+    schema:
+      openAPIV3Schema:
+        description: |-
+          SnapshotMetadataService is the Schema for the snapshotmetadataservices API
+          The presence of a SnapshotMetadataService CR advertises the existence of a CSI
+          driver's Kubernetes SnapshotMetadata gRPC service.
+          An audience scoped Kubernetes authentication bearer token must be passed in the
+          "security_token" field of each gRPC call made by a Kubernetes backup client.
+        properties:
+          apiVersion:
+            description: |-
+              APIVersion defines the versioned schema of this representation of an object.
+              Servers should convert recognized schemas to the latest internal value, and
+              may reject unrecognized values.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
+            type: string
+          kind:
+            description: |-
+              Kind is a string value representing the REST resource this object represents.
+              Servers may infer this from the endpoint the client submits requests to.
+              Cannot be updated.
+              In CamelCase.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
+            type: string
+          metadata:
+            type: object
+          spec:
+            description: Required.
+            properties:
+              address:
+                description: |-
+                  The TCP endpoint address of the gRPC service.
+                  Required.
+                type: string
+              audience:
+                description: |-
+                  The audience string value expected in a client's authentication token passed
+                  in the "security_token" field of each gRPC call.
+                  Required.
+                type: string
+              caCert:
+                description: |-
+                  Certificate authority bundle needed by the client to validate the service.
+                  Required.
+                format: byte
+                type: string
+            required:
+            - address
+            - audience
+            - caCert
+            type: object
+        required:
+        - spec
+        type: object
+    served: true
+    storage: true
--- a/test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/rbac.yaml
+++ b/test/e2e/testing-manifests/storage-csi/external-snapshot-metadata/rbac.yaml
@@ -0,0 +1,42 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  labels:
+    app.kubernetes.io/instance: hostpath.csi.k8s.io
+    app.kubernetes.io/part-of: csi-driver-host-path
+  name: external-snapshot-metadata-runner
+rules:
+- apiGroups:
+  - cbt.storage.k8s.io
+  resources:
+  - snapshotmetadataservices
+  verbs:
+  - get
+  - list
+  - watch
+  - create
+  - update
+  - patch
+  - delete
+- apiGroups:
+  - authentication.k8s.io
+  resources:
+  - tokenreviews
+  verbs:
+  - create
+  - get
+- apiGroups:
+  - authorization.k8s.io
+  resources:
+  - subjectaccessreviews
+  verbs:
+  - create
+  - get
+- apiGroups:
+  - snapshot.storage.k8s.io
+  resources:
+  - volumesnapshots
+  - volumesnapshotcontents
+  verbs:
+  - get
+  - list
--- a/test/e2e/testing-manifests/storage-csi/hostpath/hostpath/csi-hostpath-plugin.yaml
+++ b/test/e2e/testing-manifests/storage-csi/hostpath/hostpath/csi-hostpath-plugin.yaml
@@ -243,6 +243,7 @@ spec:
            - "--v=5"
            - "--endpoint=$(CSI_ENDPOINT)"
            - "--nodeid=$(KUBE_NODE_NAME)"
+            - "--enable-snapshot-metadata"
            # end hostpath args
          env:
            - name: CSI_ENDPOINT
@@ -386,6 +387,23 @@ spec:
            - mountPath: /csi
              name: socket-dir

+        - name: csi-snapshot-metadata
+          image: registry.k8s.io/sig-storage/csi-snapshot-metadata:v0.2.0
+          imagePullPolicy: IfNotPresent
+          args:
+          - --csi-address=/csi/csi.sock
+          - --tls-cert=/tmp/certificates/tls.crt
+          - --tls-key=/tmp/certificates/tls.key
+          resources: {}
+          terminationMessagePath: /dev/termination-log
+          terminationMessagePolicy: File
+          volumeMounts:
+          - mountPath: /csi
+            name: socket-dir
+          - mountPath: /tmp/certificates
+            name: csi-snapshot-metadata-server-certs
+            readOnly: true
+
        # end csi containers

      volumes:
@@ -415,4 +433,8 @@ spec:
            path: /dev
            type: Directory
          name: dev-dir
+        - name: csi-snapshot-metadata-server-certs
+          secret:
+            defaultMode: 420
+            secretName: csi-snapshot-metadata-server-certs
        # end csi volumes
--- a/test/images/.permitted-images
+++ b/test/images/.permitted-images
@@ -37,9 +37,10 @@ registry.k8s.io/sig-storage/csi-external-health-monitor-controller
 registry.k8s.io/sig-storage/csi-node-driver-registrar
 registry.k8s.io/sig-storage/csi-provisioner
 registry.k8s.io/sig-storage/csi-resizer
+registry.k8s.io/sig-storage/csi-snapshot-metadata
 registry.k8s.io/sig-storage/csi-snapshotter
 registry.k8s.io/sig-storage/hello-populator
 registry.k8s.io/sig-storage/hostpathplugin
 registry.k8s.io/sig-storage/livenessprobe
 registry.k8s.io/sig-storage/nfs-provisioner
-registry.k8s.io/sig-storage/volume-data-source-validator
+registry.k8s.io/sig-storage/volume-data-source-validator
--- a/test/utils/image/csi_manifests_test.go
+++ b/test/utils/image/csi_manifests_test.go
@@ -53,6 +53,9 @@ func TestCSIImageConfigs(t *testing.T) {
 		// For AnyVolumeDataSource feature tests.
 		"volume-data-source-validator",
 		"hello-populator",
+
+		// For SnapshotMetadata feature tests.
+		"csi-snapshot-metadata",
 	}
 	actualImages := sets.NewString()
 	for _, config := range configs {