Merge branch 'main' into kopia-repo-snapshot-operations

README.md

@@ -1,7 +1,7 @@
 ![100]
 
 [![Build Status][1]][2] [![CII Best Practices](https://bestpractices.coreinfrastructure.org/projects/3811/badge)](https://bestpractices.coreinfrastructure.org/projects/3811)
-![GitHub release (latest SemVer)](https://img.shields.io/github/v/release/vmware-tanzu/velero)
+![GitHub release (latest SemVer)](https://img.shields.io/github/v/release/velero-io/velero)
 
 ## Overview
 
@@ -52,14 +52,29 @@ Velero supports IPv4, IPv6, and dual stack environments. Support for this was te
 
 The Velero maintainers are continuously working to expand testing coverage, but are not able to test every combination of Velero and supported Kubernetes versions for each Velero release. The table above is meant to track the current testing coverage and the expected supported Kubernetes versions for each Velero version.
 
-If you are interested in using a different version of Kubernetes with a given Velero version, we'd recommend that you perform testing before installing or upgrading your environment. For full information around capabilities within a release, also see the Velero [release notes](https://github.com/vmware-tanzu/velero/releases) or Kubernetes [release notes](https://github.com/kubernetes/kubernetes/tree/master/CHANGELOG). See the Velero [support page](https://velero.io/docs/latest/support-process/) for information about supported versions of Velero.
+If you are interested in using a different version of Kubernetes with a given Velero version, we'd recommend that you perform testing before installing or upgrading your environment. For full information around capabilities within a release, also see the Velero [release notes](https://github.com/velero-io/velero/releases) or Kubernetes [release notes](https://github.com/kubernetes/kubernetes/tree/master/CHANGELOG). See the Velero [support page](https://velero.io/docs/latest/support-process/) for information about supported versions of Velero.
 
 For each release, Velero maintainers run the test to ensure the upgrade path from n-2 minor release.  For example, before the release of v1.10.x, the test will verify that the backup created by v1.9.x and v1.8.x can be restored using the build to be tagged as v1.10.x.
 
-[1]: https://github.com/vmware-tanzu/velero/workflows/Main%20CI/badge.svg
-[2]: https://github.com/vmware-tanzu/velero/actions?query=workflow%3A"Main+CI"
-[4]: https://github.com/vmware-tanzu/velero/issues
-[6]: https://github.com/vmware-tanzu/velero/releases
+## Cloud Native Computing Foundation
+<!-- remove sandbox once promoted -->
+Velero is a [Cloud Native Computing Foundation](https://www.cncf.io/) sandbox project.
+
+<p align="center">
+  <a href="https://www.cncf.io/">
+    <img src="https://raw.githubusercontent.com/cncf/artwork/main/other/cncf/horizontal/color/cncf-color.svg"
+         alt="Cloud Native Computing Foundation logo" width="300"/>
+  </a>
+</p>
+
+Copyright Contributors to Velero, established as Velero a Series of LF Projects, LLC.
+For website terms of use, trademark policy and other project policies please see
+<https://lfprojects.org/policies/>.
+
+[1]: https://github.com/velero-io/velero/workflows/Main%20CI/badge.svg
+[2]: https://github.com/velero-io/velero/actions?query=workflow%3A"Main+CI"
+[4]: https://github.com/velero-io/velero/issues
+[6]: https://github.com/velero-io/velero/releases
 [9]: https://kubernetes.io/docs/setup/
 [10]: https://kubernetes.io/docs/tasks/tools/install-kubectl/#install-with-homebrew-on-macos
 [11]: https://kubernetes.io/docs/tasks/tools/install-kubectl/#tabset-1

changelogs/unreleased/9791-christian-schlichtherle

@@ -0,0 +1 @@
+Fix DataUploadDeleteAction creating snapshot-info ConfigMaps labeled with the wrong backup name when a DataUpload CR from another backup is incidentally captured in the backup tarball, which caused Kopia snapshots to be leaked in object storage on expiry of the real owning backup.

changelogs/unreleased/9807-Lyndon-Li

@@ -0,0 +1 @@
+Uploader interface for block data mover

changelogs/unreleased/9817-Lyndon-Li

@@ -0,0 +1 @@
+Add metadata operation to Kopia repo for block data mover

changelogs/unreleased/9821-adam-jian-zhang

@@ -0,0 +1 @@
+Fix issue #9811, add interface to support ClusterScopedFilterPolicy and NamespacedFilterPolicy

internal/resourcepolicies/resource_policies.go

@@ -54,6 +54,31 @@ type Action struct {
 	Parameters map[string]any `yaml:"parameters,omitempty"`
 }
 
+// ResourceFilter defines a filter for specific resource kinds.
+type ResourceFilter struct {
+	Kinds            []string            `yaml:"kinds"`
+	LabelSelector    map[string]string   `yaml:"labelSelector,omitempty"`
+	OrLabelSelectors []map[string]string `yaml:"orLabelSelectors,omitempty"`
+	Names            []string            `yaml:"names,omitempty"`
+	ExcludedNames    []string            `yaml:"excludedNames,omitempty"`
+}
+
+// IsCatchAll returns true if the filter is a catch-all entry (empty kinds or ["*"])
+func (rf *ResourceFilter) IsCatchAll() bool {
+	return len(rf.Kinds) == 0 || (len(rf.Kinds) == 1 && rf.Kinds[0] == "*")
+}
+
+// ClusterScopedFilterPolicy defines backup filters scoped globally to cluster-scoped resources.
+type ClusterScopedFilterPolicy struct {
+	ResourceFilters []ResourceFilter `yaml:"resourceFilters"`
+}
+
+// NamespacedFilterPolicy defines backup filters scoped to specific namespaces.
+type NamespacedFilterPolicy struct {
+	Namespaces      []string         `yaml:"namespaces"`
+	ResourceFilters []ResourceFilter `yaml:"resourceFilters"`
+}
+
 // IncludeExcludePolicy defined policy to include or exclude resources based on the names
 type IncludeExcludePolicy struct {
 	// The following fields have the same semantics as those from the spec of backup.
@@ -95,17 +120,21 @@ type VolumePolicy struct {
 
 // ResourcePolicies currently defined slice of volume policies to handle backup
 type ResourcePolicies struct {
-	Version              string                `yaml:"version"`
-	VolumePolicies       []VolumePolicy        `yaml:"volumePolicies"`
-	IncludeExcludePolicy *IncludeExcludePolicy `yaml:"includeExcludePolicy"`
+	Version                   string                     `yaml:"version"`
+	VolumePolicies            []VolumePolicy             `yaml:"volumePolicies"`
+	IncludeExcludePolicy      *IncludeExcludePolicy      `yaml:"includeExcludePolicy"`
+	ClusterScopedFilterPolicy *ClusterScopedFilterPolicy `yaml:"clusterScopedFilterPolicy,omitempty"`
+	NamespacedFilterPolicies  []NamespacedFilterPolicy   `yaml:"namespacedFilterPolicies,omitempty"`
 	// we may support other resource policies in the future, and they could be added separately
 	// OtherResourcePolicies []OtherResourcePolicy
 }
 
 type Policies struct {
-	version              string
-	volumePolicies       []volPolicy
-	includeExcludePolicy *IncludeExcludePolicy
+	version                   string
+	volumePolicies            []volPolicy
+	includeExcludePolicy      *IncludeExcludePolicy
+	clusterScopedFilterPolicy *ClusterScopedFilterPolicy
+	namespacedFilterPolicies  []NamespacedFilterPolicy
 	// OtherPolicies
 }
 
@@ -158,6 +187,8 @@ func (p *Policies) BuildPolicy(resPolicies *ResourcePolicies) error {
 
 	p.version = resPolicies.Version
 	p.includeExcludePolicy = resPolicies.IncludeExcludePolicy
+	p.clusterScopedFilterPolicy = resPolicies.ClusterScopedFilterPolicy
+	p.namespacedFilterPolicies = resPolicies.NamespacedFilterPolicies
 	return nil
 }
 
@@ -235,6 +266,14 @@ func (p *Policies) GetIncludeExcludePolicy() *IncludeExcludePolicy {
 	return p.includeExcludePolicy
 }
 
+func (p *Policies) GetClusterScopedFilterPolicy() *ClusterScopedFilterPolicy {
+	return p.clusterScopedFilterPolicy
+}
+
+func (p *Policies) GetNamespacedFilterPolicies() []NamespacedFilterPolicy {
+	return p.namespacedFilterPolicies
+}
+
 func GetResourcePoliciesFromBackup(
 	backup velerov1api.Backup,
 	client crclient.Client,

pkg/backup/request.go

@@ -19,6 +19,9 @@ package backup
 import (
 	"sync"
 
+	"github.com/gobwas/glob"
+	"k8s.io/apimachinery/pkg/labels"
+
 	"github.com/vmware-tanzu/velero/internal/hook"
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/volume"
@@ -34,6 +37,21 @@ type itemKey struct {
 	name      string
 }
 
+// ResolvedResourceFilter holds the materialized filter state for one kind-group
+// within a namespace.
+type ResolvedResourceFilter struct {
+	LabelSelector    labels.Selector
+	OrLabelSelectors []labels.Selector
+	NameIE           *collections.IncludesExcludes
+}
+
+// ResolvedNamespaceFilter holds the materialized filter state for a namespace.
+// ResourceFilterMap is keyed by the resolved group-resource string.
+type ResolvedNamespaceFilter struct {
+	ResourceFilterMap map[string]*ResolvedResourceFilter
+	CatchAllFilter    *ResolvedResourceFilter
+}
+
 type SynchronizedVSList struct {
 	sync.Mutex
 	VolumeSnapshotList []*volume.Snapshot
@@ -70,6 +88,27 @@ type Request struct {
 	SkippedPVTracker          *skipPVTracker
 	VolumesInformation        volume.BackupVolumesInformation
 	WorkerPool                *ItemBlockWorkerPool
+
+	// ClusterScopedFilterMap holds resolved global filters for cluster-scoped resources.
+	// Key is the resolved group-resource string.
+	ClusterScopedFilterMap map[string]*ResolvedResourceFilter
+
+	// NamespacedFilterMap holds resolved per-namespace filters.
+	// Key is either an exact namespace name or a glob pattern.
+	NamespacedFilterMap map[string]*ResolvedNamespaceFilter
+
+	// NamespacedFilterPatterns preserves the order of patterns for first-match semantics
+	// and caches pre-compiled globs to avoid repeated compilation in the hot path.
+	NamespacedFilterPatterns []NamespacedFilterPattern
+}
+
+// NamespacedFilterPattern pairs a namespace pattern string with its pre-compiled
+// glob so that GetNamespaceFilter does not recompile on every call.
+// Compiled is nil for exact-match (non-glob) patterns, which are looked up
+// directly in NamespacedFilterMap.
+type NamespacedFilterPattern struct {
+	Pattern  string
+	Compiled glob.Glob
 }
 
 // BackupVolumesInformation contains the information needs by generating
@@ -107,3 +146,25 @@ func (r *Request) FillVolumesInformation() {
 func (r *Request) StopWorkerPool() {
 	r.WorkerPool.Stop()
 }
+
+// GetNamespaceFilter returns the resolved filter for a namespace, or nil
+// if the namespace should use global filters. Uses first-match semantics
+// when multiple patterns could match the same namespace.
+func (r *Request) GetNamespaceFilter(namespace string) *ResolvedNamespaceFilter {
+	if r.NamespacedFilterMap == nil {
+		return nil
+	}
+
+	// First check for exact match
+	if f, ok := r.NamespacedFilterMap[namespace]; ok {
+		return f
+	}
+
+	// Walk patterns in definition order using pre-compiled globs (no allocation per call)
+	for _, p := range r.NamespacedFilterPatterns {
+		if p.Compiled != nil && p.Compiled.Match(namespace) {
+			return r.NamespacedFilterMap[p.Pattern]
+		}
+	}
+	return nil
+}

pkg/datamover/dataupload_delete_action.go

@@ -14,6 +14,7 @@ import (
 
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	repotypes "github.com/vmware-tanzu/velero/pkg/repository/types"
 )
@@ -35,6 +36,44 @@ func (d *DataUploadDeleteAction) Execute(input *velero.DeleteItemActionExecuteIn
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(input.Item.UnstructuredContent(), &du); err != nil {
 		return errors.WithStack(errors.Wrapf(err, "failed to convert input.Item from unstructured"))
 	}
+	// Only create a snapshot-info ConfigMap when the DataUpload's owning
+	// backup (its velero.io/backup-name label) matches the backup currently
+	// being deleted. Two other cases reach this code path and must be
+	// skipped, because the resulting CM would be unmatchable and only adds
+	// etcd churn:
+	//
+	//  1. The label is missing. We have no verifiable owner, so a CM created
+	//     with the executing backup's label is a guess that deleteMovedSnapshots
+	//     cannot rely on.
+	//  2. The label names a different backup. Velero does not support
+	//     self-protection, so this almost always means the velero namespace
+	//     was captured in a backup tarball and the DataUpload CR belongs to
+	//     an unrelated backup. Creating a CM labeled with the executing
+	//     backup mislabels the snapshot and causes the real owning backup's
+	//     deleteMovedSnapshots query to miss it, leaking the Kopia snapshot
+	//     in the object store.
+	//
+	// Both cases warn so misconfigured installs surface in logs.
+	owner := du.Labels[velerov1.BackupNameLabel]
+	switch {
+	case owner == "":
+		d.logger.Warnf(
+			"DataUpload %q has no %q label, so its owning backup cannot be verified; "+
+				"skipping snapshot-info ConfigMap creation because a CM without a verifiable owner "+
+				"cannot be matched back to its snapshot at backup deletion time.",
+			du.Name, velerov1.BackupNameLabel,
+		)
+		return nil
+	case owner != label.GetValidName(input.Backup.Name):
+		d.logger.Warnf(
+			"DataUpload %q belongs to backup %q but is being deleted under backup %q; "+
+				"this almost always means the velero namespace was included in a backup tarball. "+
+				"Velero does not support self-protection — exclude the velero namespace from your schedules. "+
+				"Skipping snapshot-info ConfigMap creation to avoid mislabeling.",
+			du.Name, owner, input.Backup.Name,
+		)
+		return nil
+	}
 	cm := genConfigmap(input.Backup, *du)
 	if cm == nil {
 		// will not fail the backup deletion

pkg/datamover/dataupload_delete_action_test.go

@@ -0,0 +1,171 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package datamover
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	logrustest "github.com/sirupsen/logrus/hooks/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func toUnstructured(t *testing.T, du *velerov2alpha1.DataUpload) runtime.Unstructured {
+	t.Helper()
+	m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(du)
+	require.NoError(t, err)
+	return &unstructured.Unstructured{Object: m}
+}
+
+func newCompletedDataUpload(name, ownerBackup string) *velerov2alpha1.DataUpload {
+	du := &velerov2alpha1.DataUpload{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: velerov2alpha1.SchemeGroupVersion.String(),
+			Kind:       "DataUpload",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "velero",
+			Name:      name,
+		},
+		Spec: velerov2alpha1.DataUploadSpec{
+			SnapshotType:          velerov2alpha1.SnapshotTypeCSI,
+			SourcePVC:             "my-pvc",
+			SourceNamespace:       "app",
+			BackupStorageLocation: "default",
+			DataMover:             "velero",
+		},
+		Status: velerov2alpha1.DataUploadStatus{
+			Phase:      velerov2alpha1.DataUploadPhaseCompleted,
+			SnapshotID: "kopia-snapshot-id",
+		},
+	}
+	if ownerBackup != "" {
+		du.Labels = map[string]string{velerov1.BackupNameLabel: ownerBackup}
+	}
+	return du
+}
+
+func TestDataUploadDeleteActionAppliesTo(t *testing.T) {
+	a := NewDataUploadDeleteAction(logrus.StandardLogger(), nil)
+	selector, err := a.AppliesTo()
+	require.NoError(t, err)
+	require.Equal(t, velero.ResourceSelector{IncludedResources: []string{"datauploads.velero.io"}}, selector)
+}
+
+func TestDataUploadDeleteActionExecute(t *testing.T) {
+	tests := []struct {
+		name             string
+		duName           string
+		duOwnerBackup    string // value placed in velero.io/backup-name label on the DataUpload
+		executingBackup  string // name of the Backup being deleted (input.Backup.Name)
+		wantConfigMap    bool
+		wantWarnContains string // substring expected in a warn-level log entry; empty means no warn expected
+	}{
+		{
+			name:             "DataUpload owned by the executing backup creates a snapshot-info ConfigMap",
+			duName:           "daily-backup-abcde",
+			duOwnerBackup:    "daily-backup",
+			executingBackup:  "daily-backup",
+			wantConfigMap:    true,
+			wantWarnContains: "",
+		},
+		{
+			name:             "DataUpload owned by a different backup is skipped and a warning is logged",
+			duName:           "daily-backup-abcde",
+			duOwnerBackup:    "daily-backup",
+			executingBackup:  "hourly-backup",
+			wantConfigMap:    false,
+			wantWarnContains: "velero namespace",
+		},
+		{
+			name:             "DataUpload with no backup-name label is skipped and a warning is logged",
+			duName:           "unlabeled-du",
+			duOwnerBackup:    "",
+			executingBackup:  "some-backup",
+			wantConfigMap:    false,
+			wantWarnContains: "cannot be verified",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			crClient := velerotest.NewFakeControllerRuntimeClient(t)
+			logger, hook := logrustest.NewNullLogger()
+			logger.SetLevel(logrus.DebugLevel)
+			action := NewDataUploadDeleteAction(logger, crClient)
+
+			du := newCompletedDataUpload(tc.duName, tc.duOwnerBackup)
+			backup := builder.ForBackup("velero", tc.executingBackup).StorageLocation("default").Result()
+
+			err := action.Execute(&velero.DeleteItemActionExecuteInput{
+				Item:   toUnstructured(t, du),
+				Backup: backup,
+			})
+			require.NoError(t, err)
+
+			cm := &corev1api.ConfigMap{}
+			getErr := crClient.Get(t.Context(), crclient.ObjectKey{
+				Namespace: backup.Namespace,
+				Name:      fmt.Sprintf("%s-info", du.Name),
+			}, cm)
+
+			if tc.wantConfigMap {
+				require.NoError(t, getErr, "expected snapshot-info ConfigMap to be created")
+				assert.Equal(t, tc.executingBackup, cm.Labels[velerov1.BackupNameLabel])
+				assert.Equal(t, "true", cm.Labels[velerov1.DataUploadSnapshotInfoLabel])
+			} else {
+				require.Error(t, getErr)
+				assert.True(t, apierrors.IsNotFound(getErr),
+					"expected no ConfigMap to be created for foreign DataUpload, but got: %v", getErr)
+			}
+
+			// The action must surface DataUploads it cannot generate a useful
+			// snapshot-info ConfigMap for as warnings, so operators who
+			// accidentally included the velero namespace in a backup (or
+			// otherwise produced DataUploads without a verifiable owner) can
+			// detect the misconfiguration from logs instead of having the
+			// case silently swallowed.
+			var sawWarn bool
+			for _, entry := range hook.AllEntries() {
+				if entry.Level == logrus.WarnLevel &&
+					strings.Contains(entry.Message, tc.duName) &&
+					(tc.wantWarnContains == "" || strings.Contains(entry.Message, tc.wantWarnContains)) {
+					sawWarn = true
+					break
+				}
+			}
+			assert.Equal(t, tc.wantWarnContains != "", sawWarn,
+				"unexpected warn log presence (wantContains=%q, got=%v); entries=%v",
+				tc.wantWarnContains, sawWarn, hook.AllEntries())
+		})
+	}
+}

pkg/datapath/file_system.go

@@ -183,7 +183,7 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, uploaderConfig map[strin
 		}()
 
 		snapshotID, emptySnapshot, totalBytes, incrementalBytes, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, backupParam.RealSource, backupParam.Tags, backupParam.ForceFull,
-			backupParam.ParentSnapshot, source.VolMode, uploaderConfig, fs)
+			backupParam.ParentSnapshot, provider.CBTParam{}, source.VolMode, uploaderConfig, fs)
 
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)

pkg/datapath/file_system_test.go

@@ -96,7 +96,7 @@ func TestAsyncBackup(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			fs := newFileSystemBR("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*fileSystemBR)
 			mockProvider := providerMock.NewProvider(t)
-			mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.result.Backup.IncrementalBytes, test.err)
+			mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.result.Backup.IncrementalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			fs.uploaderProv = mockProvider
 			fs.initialized = true

pkg/podvolume/backupper_test.go

@@ -396,7 +396,7 @@ func TestBackupPodVolumes(t *testing.T) {
 			},
 			uploaderType: "fake-uploader-type",
 			errs: []string{
-				"invalid uploader type 'fake-uploader-type', valid type: 'kopia'",
+				"invalid uploader type 'fake-uploader-type', valid types: 'kopia', 'velero-block'",
 			},
 		},
 		{

pkg/repository/udmrepo/kopialib/lib_repo.go

@@ -33,6 +33,7 @@ import (
 	"github.com/kopia/kopia/repo/manifest"
 	"github.com/kopia/kopia/repo/object"
 	"github.com/kopia/kopia/snapshot"
+	"github.com/kopia/kopia/snapshot/snapshotfs"
 	"github.com/kopia/kopia/snapshot/snapshotmaintenance"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -412,14 +413,74 @@ func (kr *kopiaRepository) NewObjectWriter(ctx context.Context, opt udmrepo.Obje
 	}, nil
 }
 
-// TODO add implementation in following PRs
+const kopiaDirStreamType = "kopia:directory"
+
 func (kr *kopiaRepository) WriteMetadata(ctx context.Context, meta *udmrepo.Metadata, opt udmrepo.ObjectWriteOptions) (udmrepo.ID, error) {
-	return "", errors.New("not supported")
+	if kr.rawWriter == nil {
+		return "", errors.New("repo writer is closed or not open")
+	}
+
+	dirEntries := []*snapshot.DirEntry{}
+	if meta.SubObjects != nil {
+		for _, sub := range meta.SubObjects {
+			rawID, err := object.ParseID(string(sub.ID))
+			if err != nil {
+				return "", errors.Wrapf(err, "error parsing object ID from %v", sub)
+			}
+
+			dirEntries = append(dirEntries, &snapshot.DirEntry{
+				Name:        sub.Name,
+				ObjectID:    rawID,
+				Type:        getKopiaObjectType(sub.Type),
+				FileSize:    sub.Size,
+				Permissions: snapshot.Permissions(sub.Permissions),
+				ModTime:     fs.UTCTimestampFromTime(sub.ModTime),
+				UserID:      sub.UserID,
+				GroupID:     sub.GroupID,
+			})
+		}
+	}
+
+	dirManifest := snapshot.DirManifest{
+		StreamType: kopiaDirStreamType,
+		Entries:    dirEntries,
+	}
+
+	oid, err := snapshotfs.WriteDirManifest(ctx, kr.rawWriter, opt.Description, &dirManifest, getMetadataCompressor())
+	if err != nil {
+		return "", errors.Wrapf(err, "error writing dir manifest: %v", opt.Description)
+	}
+
+	return udmrepo.ID(oid.String()), nil
 }
 
-// TODO add implementation in following PRs
 func (kr *kopiaRepository) ReadMetadata(ctx context.Context, id udmrepo.ID) (*udmrepo.Metadata, error) {
-	return nil, errors.New("not supported")
+	reader, err := kr.OpenObject(ctx, id)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error to open metadata object %v", id)
+	}
+	defer reader.Close()
+
+	dirManifest := snapshot.DirManifest{}
+	if err := json.NewDecoder(reader).Decode(&dirManifest); err != nil {
+		return nil, errors.Wrap(err, "unable to parse directory object")
+	}
+
+	meta := udmrepo.Metadata{}
+	for _, sub := range dirManifest.Entries {
+		meta.SubObjects = append(meta.SubObjects, udmrepo.ObjectMetadata{
+			ID:          udmrepo.ID(sub.ObjectID.String()),
+			Name:        sub.Name,
+			Type:        getObjectDataType(sub.Type),
+			Size:        sub.FileSize,
+			ModTime:     sub.ModTime.ToTime(),
+			Permissions: int(sub.Permissions),
+			UserID:      sub.UserID,
+			GroupID:     sub.GroupID,
+		})
+	}
+
+	return &meta, nil
 }
 
 func (kr *kopiaRepository) PutManifest(ctx context.Context, manifest udmrepo.RepoManifest) (udmrepo.ID, error) {
@@ -775,3 +836,25 @@ func openKopiaRepo(ctx context.Context, configFile string, password string, opti
 
 	return r, nil
 }
+
+func getKopiaObjectType(tp int) snapshot.EntryType {
+	switch tp {
+	case udmrepo.ObjectDataTypeMetadata:
+		return snapshot.EntryTypeDirectory
+	case udmrepo.ObjectDataTypeData:
+		return snapshot.EntryTypeFile
+	default:
+		return snapshot.EntryTypeUnknown
+	}
+}
+
+func getObjectDataType(tp snapshot.EntryType) int {
+	switch tp {
+	case snapshot.EntryTypeDirectory:
+		return udmrepo.ObjectDataTypeMetadata
+	case snapshot.EntryTypeFile:
+		return udmrepo.ObjectDataTypeData
+	default:
+		return udmrepo.ObjectDataTypeUnknown
+	}
+}

pkg/repository/udmrepo/kopialib/lib_repo_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package kopialib
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"math"
@@ -1286,6 +1287,173 @@ func TestIsReady(t *testing.T) {
 	}
 }
 
+type fakeObjectReader struct {
+	*bytes.Reader
+}
+
+func (f *fakeObjectReader) Close() error {
+	return nil
+}
+
+func (f *fakeObjectReader) Length() int64 {
+	return int64(f.Reader.Len())
+}
+
+func TestWriteMetadata(t *testing.T) {
+	testCases := []struct {
+		name            string
+		rawWriter       *repomocks.MockRepositoryWriter
+		rawObjWriter    *repomocks.Writer
+		meta            *udmrepo.Metadata
+		rawWriterRetErr error
+		expectedErr     string
+	}{
+		{
+			name:        "raw writer is nil",
+			expectedErr: "repo writer is closed or not open",
+		},
+		{
+			name:      "invalid object id",
+			rawWriter: repomocks.NewMockRepositoryWriter(t),
+			meta: &udmrepo.Metadata{
+				SubObjects: []udmrepo.ObjectMetadata{
+					{
+						ID: "fake-id",
+					},
+				},
+			},
+			expectedErr: "error parsing object ID from {fake-id  0 0 0001-01-01 00:00:00 +0000 UTC 0 0 0}: malformed content ID: \"fake-id\": invalid content prefix",
+		},
+		{
+			name:         "write dir manifest fail",
+			rawWriter:    repomocks.NewMockRepositoryWriter(t),
+			rawObjWriter: repomocks.NewWriter(t),
+			meta: &udmrepo.Metadata{
+				SubObjects: []udmrepo.ObjectMetadata{
+					{
+						ID: "I123456",
+					},
+				},
+			},
+			rawWriterRetErr: errors.New("fake-write-error"),
+			expectedErr:     "error writing dir manifest: : unable to encode directory JSON: fake-write-error",
+		},
+		{
+			name:         "succeed",
+			rawWriter:    repomocks.NewMockRepositoryWriter(t),
+			rawObjWriter: repomocks.NewWriter(t),
+			meta: &udmrepo.Metadata{
+				SubObjects: []udmrepo.ObjectMetadata{
+					{
+						ID: "I123456",
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			kr := &kopiaRepository{}
+
+			if tc.rawWriter != nil {
+				if tc.rawObjWriter != nil {
+					tc.rawWriter.On("NewObjectWriter", mock.Anything, mock.Anything).Return(tc.rawObjWriter)
+					if tc.rawWriterRetErr != nil {
+						tc.rawObjWriter.On("Write", mock.Anything).Return(0, tc.rawWriterRetErr)
+						tc.rawObjWriter.On("Close").Return(nil)
+					} else {
+						tc.rawObjWriter.On("Write", mock.Anything).Return(10, nil)
+						tc.rawObjWriter.On("Result").Return(object.ID{}, nil)
+						tc.rawObjWriter.On("Close").Return(nil)
+					}
+				}
+				kr.rawWriter = tc.rawWriter
+			}
+
+			_, err := kr.WriteMetadata(t.Context(), tc.meta, udmrepo.ObjectWriteOptions{})
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}
+
+func TestReadMetadata(t *testing.T) {
+	testCases := []struct {
+		name        string
+		rawRepo     *repomocks.MockRepository
+		objectID    udmrepo.ID
+		openErr     error
+		readData    []byte
+		expectedErr string
+		expected    *udmrepo.Metadata
+	}{
+		{
+			name:        "open object fail",
+			rawRepo:     repomocks.NewMockRepository(t),
+			objectID:    "I123456",
+			openErr:     errors.New("fake-open-error"),
+			expectedErr: "error to open metadata object I123456: error to open object: fake-open-error",
+		},
+		{
+			name:        "invalid json",
+			rawRepo:     repomocks.NewMockRepository(t),
+			objectID:    "I123456",
+			readData:    []byte("invalid json"),
+			expectedErr: "unable to parse directory object: invalid character 'i' looking for beginning of value",
+		},
+		{
+			name:     "succeed",
+			rawRepo:  repomocks.NewMockRepository(t),
+			objectID: "I123456",
+			readData: []byte(`{"stream":"kopia:directory","entries":[{"name":"file1","type":"f","mode":"0644","size":100,"uid":1000,"gid":1000,"mtime":"2023-01-01T00:00:00Z","obj":"I123456"}]}`),
+			expected: &udmrepo.Metadata{
+				SubObjects: []udmrepo.ObjectMetadata{
+					{
+						ID:          "I123456",
+						Name:        "file1",
+						Type:        udmrepo.ObjectDataTypeData,
+						Size:        100,
+						ModTime:     time.Date(2023, 1, 1, 0, 0, 0, 0, time.UTC).Local(),
+						Permissions: 420,
+						UserID:      1000,
+						GroupID:     1000,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			kr := &kopiaRepository{}
+
+			if tc.rawRepo != nil {
+				if tc.openErr != nil {
+					tc.rawRepo.On("OpenObject", mock.Anything, mock.Anything).Return(nil, tc.openErr)
+				} else {
+					reader := &fakeObjectReader{Reader: bytes.NewReader(tc.readData)}
+					tc.rawRepo.On("OpenObject", mock.Anything, mock.Anything).Return(reader, nil)
+				}
+				kr.rawRepo = tc.rawRepo
+			}
+
+			meta, err := kr.ReadMetadata(t.Context(), tc.objectID)
+
+			if tc.expectedErr == "" {
+				require.NoError(t, err)
+				assert.Equal(t, tc.expected, meta)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}
+
 func TestSaveSnapshot(t *testing.T) {
 	testCases := []struct {
 		name            string

pkg/repository/udmrepo/repo.go

@@ -78,6 +78,7 @@ type AdvancedFeatureInfo struct {
 
 type ObjectMetadata struct {
 	ID          ID
+	Name        string
 	Type        int // OBJECT_DATA_TYPE_*
 	Size        int64
 	ModTime     time.Time

pkg/uploader/provider/block.go

@@ -0,0 +1,115 @@
+/*
+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 provider
+
+import (
+	"context"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+
+	"github.com/vmware-tanzu/velero/internal/credentials"
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	repokeys "github.com/vmware-tanzu/velero/pkg/repository/keys"
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+)
+
+type blockProvider struct {
+	requestorType string
+	bkRepo        udmrepo.BackupRepo
+	credGetter    *credentials.CredentialGetter
+	log           logrus.FieldLogger
+}
+
+// NewBlockUploaderProvider initialized with open or create a repository
+func NewBlockUploaderProvider(
+	requestorType string,
+	ctx context.Context,
+	credGetter *credentials.CredentialGetter,
+	backupRepo *velerov1api.BackupRepository,
+	log logrus.FieldLogger,
+) (Provider, error) {
+	bp := &blockProvider{
+		requestorType: requestorType,
+		log:           log,
+		credGetter:    credGetter,
+	}
+
+	repoUID := string(backupRepo.GetUID())
+	repoOpt, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(bp, ""),
+		udmrepo.WithConfigFile("", repoUID),
+		udmrepo.WithDescription("Initial velero block uploader provider"),
+	)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error to get repo options")
+	}
+
+	repoSvc := BackupRepoServiceCreateFunc(backupRepo.Spec.RepositoryType, log)
+	log.WithField("repoUID", repoUID).Info("Opening backup repo")
+
+	bp.bkRepo, err = repoSvc.Open(ctx, *repoOpt)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Failed to find backup repository")
+	}
+
+	return bp, nil
+}
+
+func (bp *blockProvider) Close(ctx context.Context) error {
+	return bp.bkRepo.Close(ctx)
+}
+
+func (bp *blockProvider) GetPassword(param any) (string, error) {
+	if bp.credGetter.FromSecret == nil {
+		return "", errors.New("invalid credentials interface")
+	}
+	rawPass, err := bp.credGetter.FromSecret.Get(repokeys.RepoKeySelector())
+	if err != nil {
+		return "", errors.Wrap(err, "error to get password")
+	}
+
+	return strings.TrimSpace(rawPass), nil
+}
+
+// TODO: implement in the following PRs
+func (bp *blockProvider) RunBackup(
+	ctx context.Context,
+	path string,
+	realSource string,
+	tags map[string]string,
+	forceFull bool,
+	parentSnapshot string,
+	cbtParam CBTParam,
+	volMode uploader.PersistentVolumeMode,
+	uploaderCfg map[string]string,
+	updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
+	return "", false, 0, 0, errors.New("block backup not implemented")
+}
+
+// TODO: implement in the following PRs
+func (bp *blockProvider) RunRestore(
+	ctx context.Context,
+	snapshotID string,
+	volumePath string,
+	volMode uploader.PersistentVolumeMode,
+	uploaderCfg map[string]string,
+	updater uploader.ProgressUpdater) (int64, error) {
+	return 0, errors.New("block restore not implemented")
+}

pkg/uploader/provider/block_test.go

@@ -0,0 +1,187 @@
+/*
+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 provider
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+
+	"github.com/vmware-tanzu/velero/internal/credentials"
+	"github.com/vmware-tanzu/velero/internal/credentials/mocks"
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+	udmrepomocks "github.com/vmware-tanzu/velero/pkg/repository/udmrepo/mocks"
+)
+
+func TestNewBlockUploaderProvider(t *testing.T) {
+	requestorType := "testRequestor"
+	ctx := t.Context()
+	backupRepo := repository.NewBackupRepository(velerov1api.DefaultNamespace, repository.BackupRepositoryKey{VolumeNamespace: "fake-volume-ns-02", BackupLocation: "fake-bsl-02", RepositoryType: "fake-repository-type-02"})
+	mockLog := logrus.New()
+
+	testCases := []struct {
+		name                  string
+		mockCredGetter        *mocks.SecretStore
+		mockBackupRepoService udmrepo.BackupRepoService
+		expectedError         string
+	}{
+		{
+			name: "Success",
+			mockCredGetter: func() *mocks.SecretStore {
+				mockCredGetter := &mocks.SecretStore{}
+				mockCredGetter.On("Get", mock.Anything).Return("test", nil)
+				return mockCredGetter
+			}(),
+			mockBackupRepoService: func() udmrepo.BackupRepoService {
+				backupRepoService := &udmrepomocks.BackupRepoService{}
+				var backupRepo udmrepo.BackupRepo
+				backupRepoService.On("Open", t.Context(), mock.Anything).Return(backupRepo, nil)
+				return backupRepoService
+			}(),
+			expectedError: "",
+		},
+		{
+			name: "Error to get repo options",
+			mockCredGetter: func() *mocks.SecretStore {
+				mockCredGetter := &mocks.SecretStore{}
+				mockCredGetter.On("Get", mock.Anything).Return("test", errors.New("failed to get password"))
+				return mockCredGetter
+			}(),
+			mockBackupRepoService: func() udmrepo.BackupRepoService {
+				backupRepoService := &udmrepomocks.BackupRepoService{}
+				var backupRepo udmrepo.BackupRepo
+				backupRepoService.On("Open", t.Context(), mock.Anything).Return(backupRepo, nil)
+				return backupRepoService
+			}(),
+			expectedError: "error to get repo options",
+		},
+		{
+			name: "Error open repository service",
+			mockCredGetter: func() *mocks.SecretStore {
+				mockCredGetter := &mocks.SecretStore{}
+				mockCredGetter.On("Get", mock.Anything).Return("test", nil)
+				return mockCredGetter
+			}(),
+			mockBackupRepoService: func() udmrepo.BackupRepoService {
+				backupRepoService := &udmrepomocks.BackupRepoService{}
+				var backupRepo udmrepo.BackupRepo
+				backupRepoService.On("Open", t.Context(), mock.Anything).Return(backupRepo, errors.New("failed to init repository"))
+				return backupRepoService
+			}(),
+			expectedError: "Failed to find backup repository",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			credGetter := &credentials.CredentialGetter{FromSecret: tc.mockCredGetter}
+			BackupRepoServiceCreateFunc = func(string, logrus.FieldLogger) udmrepo.BackupRepoService {
+				return tc.mockBackupRepoService
+			}
+			_, err := NewBlockUploaderProvider(requestorType, ctx, credGetter, backupRepo, mockLog)
+
+			if tc.expectedError != "" {
+				require.ErrorContains(t, err, tc.expectedError)
+			} else {
+				require.NoError(t, err)
+			}
+
+			tc.mockCredGetter.AssertExpectations(t)
+		})
+	}
+}
+
+func TestBlockProviderClose(t *testing.T) {
+	mockBRepo := udmrepomocks.NewBackupRepo(t)
+	mockBRepo.On("Close", mock.Anything).Return(nil)
+
+	bp := &blockProvider{
+		bkRepo: mockBRepo,
+	}
+
+	err := bp.Close(t.Context())
+	require.NoError(t, err)
+	mockBRepo.AssertExpectations(t)
+}
+
+func TestBlockProviderGetPassword(t *testing.T) {
+	testCases := []struct {
+		name           string
+		emptySecret    bool
+		credGetterFunc func(*mocks.SecretStore, *corev1api.SecretKeySelector)
+		expectError    bool
+		expectedPass   string
+	}{
+		{
+			name: "valid credentials interface",
+			credGetterFunc: func(ss *mocks.SecretStore, selector *corev1api.SecretKeySelector) {
+				ss.On("Get", selector).Return("test", nil)
+			},
+			expectError:  false,
+			expectedPass: "test",
+		},
+		{
+			name:         "empty from secret",
+			emptySecret:  true,
+			expectError:  true,
+			expectedPass: "",
+		},
+		{
+			name: "ErrorGettingPassword",
+			credGetterFunc: func(ss *mocks.SecretStore, selector *corev1api.SecretKeySelector) {
+				ss.On("Get", selector).Return("", errors.New("error getting password"))
+			},
+			expectError:  true,
+			expectedPass: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			credGetter := &credentials.CredentialGetter{}
+			mockCredGetter := &mocks.SecretStore{}
+			if !tc.emptySecret {
+				credGetter.FromSecret = mockCredGetter
+			}
+			repoKeySelector := &corev1api.SecretKeySelector{LocalObjectReference: corev1api.LocalObjectReference{Name: "velero-repo-credentials"}, Key: "repository-password"}
+
+			if tc.credGetterFunc != nil {
+				tc.credGetterFunc(mockCredGetter, repoKeySelector)
+			}
+
+			bp := &blockProvider{
+				credGetter: credGetter,
+			}
+
+			password, err := bp.GetPassword(nil)
+			if tc.expectError {
+				require.Error(t, err, "Expected an error")
+			} else {
+				require.NoError(t, err, "Expected no error")
+			}
+
+			assert.Equal(t, tc.expectedPass, password, "Expected password to match")
+		})
+	}
+}

pkg/uploader/provider/kopia.go

@@ -118,6 +118,7 @@ func (kp *kopiaProvider) RunBackup(
 	tags map[string]string,
 	forceFull bool,
 	parentSnapshot string,
+	_ CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {

pkg/uploader/provider/kopia_test.go

@@ -106,7 +106,7 @@ func TestRunBackup(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			BackupFunc = tc.hookBackupFunc
-			_, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", tc.volMode, map[string]string{}, &updater)
+			_, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", CBTParam{}, tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {

pkg/uploader/provider/mocks/Provider.go

@@ -1,115 +1,17 @@
-// Code generated by mockery v2.53.5. DO NOT EDIT.
+// Code generated by mockery; DO NOT EDIT.
+// github.com/vektra/mockery
+// template: testify
 
 package mocks
 
 import (
-	context "context"
+	"context"
 
 	mock "github.com/stretchr/testify/mock"
-
-	uploader "github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 )
 
-// Provider is an autogenerated mock type for the Provider type
-type Provider struct {
-	mock.Mock
-}
-
-// Close provides a mock function with given fields: ctx
-func (_m *Provider) Close(ctx context.Context) error {
-	ret := _m.Called(ctx)
-
-	if len(ret) == 0 {
-		panic("no return value specified for Close")
-	}
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
-		r0 = rf(ctx)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// RunBackup provides a mock function with given fields: ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater
-func (_m *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
-	ret := _m.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-
-	if len(ret) == 0 {
-		panic("no return value specified for RunBackup")
-	}
-
-	var r0 string
-	var r1 bool
-	var r2 int64
-	var r3 int64
-	var r4 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, error)); ok {
-		return rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	}
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok {
-		r0 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r0 = ret.Get(0).(string)
-	}
-
-	if rf, ok := ret.Get(1).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok {
-		r1 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r1 = ret.Get(1).(bool)
-	}
-
-	if rf, ok := ret.Get(2).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
-		r2 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r2 = ret.Get(2).(int64)
-	}
-
-	if rf, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
-		r3 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r3 = ret.Get(3).(int64)
-	}
-
-	if rf, ok := ret.Get(4).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
-		r4 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r4 = ret.Error(4)
-	}
-
-	return r0, r1, r2, r3, r4
-}
-
-// RunRestore provides a mock function with given fields: ctx, snapshotID, volumePath, volMode, uploaderConfig, updater
-func (_m *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
-	ret := _m.Called(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-
-	if len(ret) == 0 {
-		panic("no return value specified for RunRestore")
-	}
-
-	var r0 int64
-	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
-		return rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-	}
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
-		r0 = rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-	} else {
-		r0 = ret.Get(0).(int64)
-	}
-
-	if rf, ok := ret.Get(1).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
-		r1 = rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // NewProvider creates a new instance of Provider. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewProvider(t interface {
@@ -123,3 +25,289 @@ func NewProvider(t interface {
 
 	return mock
 }
+
+// Provider is an autogenerated mock type for the Provider type
+type Provider struct {
+	mock.Mock
+}
+
+type Provider_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *Provider) EXPECT() *Provider_Expecter {
+	return &Provider_Expecter{mock: &_m.Mock}
+}
+
+// Close provides a mock function for the type Provider
+func (_mock *Provider) Close(ctx context.Context) error {
+	ret := _mock.Called(ctx)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Close")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = returnFunc(ctx)
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// Provider_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close'
+type Provider_Close_Call struct {
+	*mock.Call
+}
+
+// Close is a helper method to define mock.On call
+//   - ctx context.Context
+func (_e *Provider_Expecter) Close(ctx interface{}) *Provider_Close_Call {
+	return &Provider_Close_Call{Call: _e.mock.On("Close", ctx)}
+}
+
+func (_c *Provider_Close_Call) Run(run func(ctx context.Context)) *Provider_Close_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 context.Context
+		if args[0] != nil {
+			arg0 = args[0].(context.Context)
+		}
+		run(
+			arg0,
+		)
+	})
+	return _c
+}
+
+func (_c *Provider_Close_Call) Return(err error) *Provider_Close_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *Provider_Close_Call) RunAndReturn(run func(ctx context.Context) error) *Provider_Close_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// RunBackup provides a mock function for the type Provider
+func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
+	ret := _mock.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RunBackup")
+	}
+
+	var r0 string
+	var r1 bool
+	var r2 int64
+	var r3 int64
+	var r4 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, error)); ok {
+		return returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	}
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok {
+		r0 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok {
+		r1 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r1 = ret.Get(1).(bool)
+	}
+	if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r2 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r2 = ret.Get(2).(int64)
+	}
+	if returnFunc, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r3 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r3 = ret.Get(3).(int64)
+	}
+	if returnFunc, ok := ret.Get(4).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r4 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r4 = ret.Error(4)
+	}
+	return r0, r1, r2, r3, r4
+}
+
+// Provider_RunBackup_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunBackup'
+type Provider_RunBackup_Call struct {
+	*mock.Call
+}
+
+// RunBackup is a helper method to define mock.On call
+//   - ctx context.Context
+//   - path string
+//   - realSource string
+//   - tags map[string]string
+//   - forceFull bool
+//   - parentSnapshot string
+//   - cbtParam provider.CBTParam
+//   - volMode uploader.PersistentVolumeMode
+//   - uploaderCfg map[string]string
+//   - updater uploader.ProgressUpdater
+func (_e *Provider_Expecter) RunBackup(ctx interface{}, path interface{}, realSource interface{}, tags interface{}, forceFull interface{}, parentSnapshot interface{}, cbtParam interface{}, volMode interface{}, uploaderCfg interface{}, updater interface{}) *Provider_RunBackup_Call {
+	return &Provider_RunBackup_Call{Call: _e.mock.On("RunBackup", ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)}
+}
+
+func (_c *Provider_RunBackup_Call) Run(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater)) *Provider_RunBackup_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 context.Context
+		if args[0] != nil {
+			arg0 = args[0].(context.Context)
+		}
+		var arg1 string
+		if args[1] != nil {
+			arg1 = args[1].(string)
+		}
+		var arg2 string
+		if args[2] != nil {
+			arg2 = args[2].(string)
+		}
+		var arg3 map[string]string
+		if args[3] != nil {
+			arg3 = args[3].(map[string]string)
+		}
+		var arg4 bool
+		if args[4] != nil {
+			arg4 = args[4].(bool)
+		}
+		var arg5 string
+		if args[5] != nil {
+			arg5 = args[5].(string)
+		}
+		var arg6 provider.CBTParam
+		if args[6] != nil {
+			arg6 = args[6].(provider.CBTParam)
+		}
+		var arg7 uploader.PersistentVolumeMode
+		if args[7] != nil {
+			arg7 = args[7].(uploader.PersistentVolumeMode)
+		}
+		var arg8 map[string]string
+		if args[8] != nil {
+			arg8 = args[8].(map[string]string)
+		}
+		var arg9 uploader.ProgressUpdater
+		if args[9] != nil {
+			arg9 = args[9].(uploader.ProgressUpdater)
+		}
+		run(
+			arg0,
+			arg1,
+			arg2,
+			arg3,
+			arg4,
+			arg5,
+			arg6,
+			arg7,
+			arg8,
+			arg9,
+		)
+	})
+	return _c
+}
+
+func (_c *Provider_RunBackup_Call) Return(s string, b bool, n int64, n1 int64, err error) *Provider_RunBackup_Call {
+	_c.Call.Return(s, b, n, n1, err)
+	return _c
+}
+
+func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error)) *Provider_RunBackup_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// RunRestore provides a mock function for the type Provider
+func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
+	ret := _mock.Called(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RunRestore")
+	}
+
+	var r0 int64
+	var r1 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
+		return returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+	}
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r0 = returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r1 = returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+	} else {
+		r1 = ret.Error(1)
+	}
+	return r0, r1
+}
+
+// Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore'
+type Provider_RunRestore_Call struct {
+	*mock.Call
+}
+
+// RunRestore is a helper method to define mock.On call
+//   - ctx context.Context
+//   - snapshotID string
+//   - volumePath string
+//   - volMode uploader.PersistentVolumeMode
+//   - uploaderConfig map[string]string
+//   - updater uploader.ProgressUpdater
+func (_e *Provider_Expecter) RunRestore(ctx interface{}, snapshotID interface{}, volumePath interface{}, volMode interface{}, uploaderConfig interface{}, updater interface{}) *Provider_RunRestore_Call {
+	return &Provider_RunRestore_Call{Call: _e.mock.On("RunRestore", ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)}
+}
+
+func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater)) *Provider_RunRestore_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 context.Context
+		if args[0] != nil {
+			arg0 = args[0].(context.Context)
+		}
+		var arg1 string
+		if args[1] != nil {
+			arg1 = args[1].(string)
+		}
+		var arg2 string
+		if args[2] != nil {
+			arg2 = args[2].(string)
+		}
+		var arg3 uploader.PersistentVolumeMode
+		if args[3] != nil {
+			arg3 = args[3].(uploader.PersistentVolumeMode)
+		}
+		var arg4 map[string]string
+		if args[4] != nil {
+			arg4 = args[4].(map[string]string)
+		}
+		var arg5 uploader.ProgressUpdater
+		if args[5] != nil {
+			arg5 = args[5].(uploader.ProgressUpdater)
+		}
+		run(
+			arg0,
+			arg1,
+			arg2,
+			arg3,
+			arg4,
+			arg5,
+		)
+	})
+	return _c
+}
+
+func (_c *Provider_RunRestore_Call) Return(n int64, err error) *Provider_RunRestore_Call {
+	_c.Call.Return(n, err)
+	return _c
+}
+
+func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error)) *Provider_RunRestore_Call {
+	_c.Call.Return(run)
+	return _c
+}

pkg/uploader/provider/provider.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/cbtservice"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
 
@@ -37,6 +38,11 @@ const backupProgressCheckInterval = 10 * time.Second
 
 var ErrorCanceled error = errors.New("uploader is canceled")
 
+type CBTParam struct {
+	Source  cbtservice.SourceInfo
+	Service cbtservice.Service
+}
+
 // Provider which is designed for one pod volume to do the backup or restore
 type Provider interface {
 	// RunBackup which will do backup for one specific volume and return snapshotID, isSnapshotEmpty, error
@@ -48,6 +54,7 @@ type Provider interface {
 		tags map[string]string,
 		forceFull bool,
 		parentSnapshot string,
+		cbtParam CBTParam,
 		volMode uploader.PersistentVolumeMode,
 		uploaderCfg map[string]string,
 		updater uploader.ProgressUpdater) (string, bool, int64, int64, error)
@@ -84,9 +91,13 @@ func NewUploaderProvider(
 	if credGetter.FromFile == nil {
 		return nil, errors.New("uninitialized FileStore credential is not supported")
 	}
-	if uploaderType == uploader.KopiaType {
+
+	switch uploaderType {
+	case uploader.KopiaType:
 		return NewKopiaUploaderProvider(requesterType, ctx, credGetter, backupRepo, log)
-	} else {
+	case uploader.BlockType:
+		return NewBlockUploaderProvider(requesterType, ctx, credGetter, backupRepo, log)
+	default:
 		return nil, errors.Errorf("unsupported uploader type %v", uploaderType)
 	}
 }

pkg/uploader/types.go

@@ -23,6 +23,7 @@ import (
 
 const (
 	KopiaType            = "kopia"
+	BlockType            = "velero-block"
 	SnapshotRequesterTag = "snapshot-requester"
 	SnapshotUploaderTag  = "snapshot-uploader"
 )
@@ -40,8 +41,8 @@ const (
 // It will return an error if it's invalid.
 func ValidateUploaderType(t string) (string, error) {
 	t = strings.TrimSpace(t)
-	if t != KopiaType {
-		return "", fmt.Errorf("invalid uploader type '%s', valid type: '%s'", t, KopiaType)
+	if t != KopiaType && t != BlockType {
+		return "", fmt.Errorf("invalid uploader type '%s', valid types: '%s', '%s'", t, KopiaType, BlockType)
 	}
 
 	return "", nil

pkg/uploader/types_test.go

@@ -23,7 +23,7 @@ func TestValidateUploaderType(t *testing.T) {
 		{
 			"'anything_else' is invalid",
 			"anything_else",
-			"invalid uploader type 'anything_else', valid type: 'kopia'",
+			"invalid uploader type 'anything_else', valid types: 'kopia', 'velero-block'",
 			"",
 		},
 	}

site/assets/_scss/site/objects/_footer.scss

@@ -34,11 +34,20 @@
     height: auto;
     margin-left: 30px;
   }
-  .vm-logo {
-    font-size: .75rem;
-    img {
-      max-width: 75px;
-      margin-left: 30px;
+  .cncf-affiliation {
+    color: $footer-foreground;
+    a {
+      color: $footer-link-color;
+      border-bottom: 0;
+      &:hover {
+        text-decoration: underline;
+      }
+    }
+    .cncf-logo {
+      display: inline-block;
+      margin-top: 0.25rem;
+      max-width: 300px;
+      height: auto;
     }
   }
 }

site/config.yaml

@@ -7,8 +7,8 @@ frontmatter:
   date: [":filename", ":default"]
 params:
   author: Velero Authors
-  vm_logo: vm-logo.png
   logo: Velero.svg
+  cncf_logo: cncf-white.svg
   hero:
     backgroundColor: med-blue
   versioning: true
@@ -47,14 +47,18 @@ params:
     - v0.5.0
     - v0.4.0
     - v0.3.0
-  gh_repo: https://github.com/vmware-tanzu/velero
+  gh_repo: https://github.com/velero-io/velero
   footer:
     title: Getting Started
     content: To help you get started, see the documentation.
     cta_title: ''
     cta_url: /docs
     cta_text: Documentation
-    vm_link: http://vmware.github.io/
+    cncf_link: https://www.cncf.io/
+    # Set to "" once Velero is promoted to incubating or graduated.
+    cncf_status: sandbox
+    lf_policies_link: https://lfprojects.org/policies/
+    copyright_holder: Velero a Series of LF Projects, LLC
   footer_social_links:
     - title: Twitter
       fa_icon: fab fa-twitter
@@ -70,7 +74,7 @@ params:
       url: /blog/index.xml
     - title: GitHub
       fa_icon: fab fa-github
-      url: https://github.com/vmware-tanzu/velero
+      url: https://github.com/velero-io/velero
 minify:
   disableCSS: false
   disableHTML: false

site/layouts/_default/footer.html

@@ -38,17 +38,37 @@
                                                               alt="Homepage"/></a>
             </div>
         </div>
+        {{/* CNCF affiliation block — required by CNCF Website Guidelines.
+             Change `sandbox project` -> `project` when Velero is promoted to
+             incubating or graduated. See
+             https://github.com/cncf/foundation/blob/main/policies-guidance/website-guidelines.md */}}
+        <div class="row align-items-center cncf-affiliation mt-4">
+            <div class="col text-center">
+                <p class="mb-2">
+                    We are a
+                    <a href="{{ .Site.Params.footer.cncf_link }}">Cloud Native Computing Foundation</a>
+                    {{ with .Site.Params.footer.cncf_status }}{{ . }} {{ end }}project.
+                </p>
+                <a href="{{ .Site.Params.footer.cncf_link }}" aria-label="Cloud Native Computing Foundation">
+                    <img src="/img/{{ .Site.Params.cncf_logo }}"
+                         alt="Cloud Native Computing Foundation logo"
+                         class="cncf-logo"
+                         style="max-height: 64px;"/>
+                </a>
+            </div>
+        </div>
+
         <div class="row align-items-center">
-            <div class="col copyright text-center text-md-right mt-4">
-                &copy; {{ now.Format "2006" }} {{ .Site.Params.Author }}.
-                <a href="{{ .Site.Params.footer.vm_link }}" class="vm-logo">A VMware-backed project. <img
-                            src="/img/{{ .Site.Params.vm_logo }}" alt="VMware logo"/></a>
-                <br/>This Website Does Not Use Cookies or Other Tracking Technology.  This site is powered by <a href="https://www.netlify.com">Netlify</a>.<br/><br/>
-                <span>
-                    <a target="_blank" rel="noopener" href='https://www.vmware.com/help/legal.html'>Terms of Use</a> |
-                    <a target="_blank" rel="noopener" href='https://www.vmware.com/help/privacy.html'>Privacy Policy</a> |
-                    <a target="_blank" rel="noopener" href='https://www.vmware.com/help/privacy/california-privacy-rights.html'>Your California Privacy Rights</a>
-                </span>
+            <div class="col copyright text-center mt-4">
+                <p class="mb-1">Copyright {{ .Site.Params.footer.copyright_holder }}.</p>
+                <p class="mb-0">
+                    For website terms of use, trademark policy and other project policies please see
+                    <a href="{{ .Site.Params.footer.lf_policies_link }}">{{ .Site.Params.footer.lf_policies_link }}</a>.
+                </p>
+                <p class="mt-2 small">
+                    This website does not use cookies or other tracking technology.
+                    This site is powered by <a href="https://www.netlify.com">Netlify</a>.
+                </p>
             </div>
         </div>
     </div>