diff --git a/docs/cinder-csi-plugin/using-cinder-csi-plugin.md b/docs/cinder-csi-plugin/using-cinder-csi-plugin.md
index c3a3656ffd..20638ea8a0 100644
--- a/docs/cinder-csi-plugin/using-cinder-csi-plugin.md
+++ b/docs/cinder-csi-plugin/using-cinder-csi-plugin.md
@@ -267,7 +267,9 @@ helm install --namespace kube-system --name cinder-csi ./charts/cinder-csi-plugi
 | StorageClass `parameters`  | `availability`          | `nova`          | String. Volume Availability Zone |
 | StorageClass `parameters`  | `type`                  | Empty String    | String. Name/ID of Volume type. Corresponding volume type should exist in cinder     |
 | VolumeSnapshotClass `parameters` | `force-create`    | `false`         | Enable to support creating snapshot for a volume in in-use status |
-| Inline Volume `volumeAttributes`   | `capacity`              | `1Gi`       | volume size for creating inline volumes|
+| VolumeSnapshotClass `parameters` | `type`            | Empty String    | `snapshot` creates a VolumeSnapshot object linked to a Cinder volume snapshot. `backup` creates a VolumeSnapshot object linked to a cinder volume backup. Defaults to `snapshot` if not defined |
+| VolumeSnapshotClass `parameters` | `backup-max-duration-seconds-per-gb`  | `20`    | Defines the amount of time to wait for a backup to complete in seconds per GB of volume size |
+| Inline Volume `volumeAttributes`   | `capacity`              | `1Gi`       | volume size for creating inline volumes| 
 | Inline Volume `VolumeAttributes`   | `type`              | Empty String  | Name/ID of Volume type. Corresponding volume type should exist in cinder |
 
 ## Local Development
diff --git a/pkg/csi/cinder/controllerserver.go b/pkg/csi/cinder/controllerserver.go
index 110d3e3ec8..4768705ab8 100644
--- a/pkg/csi/cinder/controllerserver.go
+++ b/pkg/csi/cinder/controllerserver.go
@@ -17,9 +17,11 @@ limitations under the License.
 package cinder
 
 import (
+	"fmt"
 	"strconv"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/snapshots"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/volumes"
 	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
@@ -110,36 +112,71 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	}
 	content := req.GetVolumeContentSource()
 	var snapshotID string
-	var sourcevolID string
+	var sourceVolID string
+	var sourceBackupID string
+	var backupsAreEnabled bool
+	backupsAreEnabled, err = cloud.BackupsAreEnabled()
+	klog.V(4).Infof("Backups enabled: %v", backupsAreEnabled)
+	if err != nil {
+		klog.Errorf("Failed to check if backups are enabled: %v", err)
+	}
 
 	if content != nil && content.GetSnapshot() != nil {
 		snapshotID = content.GetSnapshot().GetSnapshotId()
-		_, err := cloud.GetSnapshotByID(snapshotID)
-		if err != nil {
-			if cpoerrors.IsNotFound(err) {
-				return nil, status.Errorf(codes.NotFound, "VolumeContentSource Snapshot %s not found", snapshotID)
+
+		snap, err := cloud.GetSnapshotByID(snapshotID)
+		if err != nil && !cpoerrors.IsNotFound(err) {
+			return nil, err
+		}
+		// If the snapshot exists but is not yet available, fail.
+		if err == nil && snap.Status != "available" {
+			return nil, status.Errorf(codes.Unavailable, "VolumeContentSource Snapshot %s is not yet available. status: %s", snapshotID, snap.Status)
+		}
+
+		// In case a snapshot is not found
+		// check if a Backup with the same ID exists
+		if backupsAreEnabled && cpoerrors.IsNotFound(err) {
+			back, err := cloud.GetBackupByID(snapshotID)
+			if err != nil {
+				//If there is an error getting the backup as well, fail.
+				return nil, status.Errorf(codes.NotFound, "VolumeContentSource Snapshot or Backup with ID %s not found", snapshotID)
+			}
+			if back.Status != "available" {
+				// If the backup exists but is not yet available, fail.
+				return nil, status.Errorf(codes.Unavailable, "VolumeContentSource Backup %s is not yet available. status: %s", snapshotID, back.Status)
 			}
-			return nil, status.Errorf(codes.Internal, "Failed to retrieve the snapshot %s: %v", snapshotID, err)
+			// If an available backup is found, create the volume from the backup
+			sourceBackupID = snapshotID
+			snapshotID = ""
 		}
+		// In case GetSnapshotByID has error IsNotFound and backups are not enabled
+		// TODO: Change 'snapshotID == ""' to '!backupsAreEnabled' when cloud.BackupsAreEnabled() is correctly implemented
+		if cpoerrors.IsNotFound(err) && snapshotID == "" {
+			return nil, err
+		}
+
 	}
 
 	if content != nil && content.GetVolume() != nil {
-		sourcevolID = content.GetVolume().GetVolumeId()
-		_, err := cloud.GetVolume(sourcevolID)
+		sourceVolID = content.GetVolume().GetVolumeId()
+		_, err := cloud.GetVolume(sourceVolID)
 		if err != nil {
 			if cpoerrors.IsNotFound(err) {
-				return nil, status.Errorf(codes.NotFound, "Source Volume %s not found", sourcevolID)
+				return nil, status.Errorf(codes.NotFound, "Source Volume %s not found", sourceVolID)
 			}
-			return nil, status.Errorf(codes.Internal, "Failed to retrieve the source volume %s: %v", sourcevolID, err)
+			return nil, status.Errorf(codes.Internal, "Failed to retrieve the source volume %s: %v", sourceVolID, err)
 		}
 	}
 
-	vol, err := cloud.CreateVolume(volName, volSizeGB, volType, volAvailability, snapshotID, sourcevolID, &properties)
+	vol, err := cloud.CreateVolume(volName, volSizeGB, volType, volAvailability, snapshotID, sourceVolID, sourceBackupID, properties)
+	// When creating a volume from a backup, the response does not include the backupID.
+	if sourceBackupID != "" {
+		vol.BackupID = &sourceBackupID
+	}
 
 	if err != nil {
 		klog.Errorf("Failed to CreateVolume: %v", err)
 		return nil, status.Errorf(codes.Internal, "CreateVolume failed with error %v", err)
-
 	}
 
 	klog.V(4).Infof("CreateVolume: Successfully created volume %s in Availability Zone: %s of size %d GiB", vol.ID, vol.AvailabilityZone, vol.Size)
@@ -326,6 +363,25 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 
 	name := req.Name
 	volumeID := req.GetSourceVolumeId()
+	snapshotType := req.Parameters[openstack.SnapshotType]
+	filters := map[string]string{"Name": name}
+	backupMaxDurationSecondsPerGB := openstack.BackupMaxDurationSecondsPerGBDefault
+
+	// Current time, used for CreatedAt
+	var ctime *timestamppb.Timestamp
+	// Size of the created snapshot, used to calculate the amount of time to wait for the backup to finish
+	var snapSize int
+	// If true, skips creating a snapshot because a backup already exists
+	var backupAlreadyExists bool
+	var snap *snapshots.Snapshot
+	var backup *backups.Backup
+	var backups []backups.Backup
+	var err error
+
+	// Set snapshot type to 'snapshot' by default
+	if snapshotType == "" {
+		snapshotType = "snapshot"
+	}
 
 	if name == "" {
 		return nil, status.Error(codes.InvalidArgument, "Snapshot name must be provided in CreateSnapshot request")
@@ -335,73 +391,216 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		return nil, status.Error(codes.InvalidArgument, "VolumeID must be provided in CreateSnapshot request")
 	}
 
-	// Verify a snapshot with the provided name doesn't already exist for this tenant
-	var snap *snapshots.Snapshot
+	// Verify snapshot type has a valid value
+	if snapshotType != "snapshot" && snapshotType != "backup" {
+		return nil, status.Error(codes.InvalidArgument, "Snapshot type must be 'backup', 'snapshot' or not defined")
+	}
+	var backupsAreEnabled bool
+	backupsAreEnabled, err = cs.Cloud.BackupsAreEnabled()
+	klog.V(4).Infof("Backups enabled: %v", backupsAreEnabled)
+	if err != nil {
+		klog.Errorf("Failed to check if backups are enabled: %v", err)
+	}
+
+	// Prechecks in case of a backup
+	if snapshotType == "backup" {
+		if !backupsAreEnabled {
+			return nil, status.Error(codes.FailedPrecondition, "Backups are not enabled in Cinder")
+		}
+		// Get a list of backups with the provided name
+		backups, err = cs.Cloud.ListBackups(filters)
+		if err != nil {
+			klog.Errorf("Failed to query for existing Backup during CreateSnapshot: %v", err)
+			return nil, status.Error(codes.Internal, "Failed to get backups")
+		}
+		// If more than one backup with the provided name exists, fail
+		if len(backups) > 1 {
+			klog.Errorf("found multiple existing backups with selected name (%s) during create", name)
+			return nil, status.Error(codes.Internal, "Multiple backups reported by Cinder with same name")
+		}
+
+		if len(backups) == 1 {
+			backup = &backups[0]
+
+			// Verify the existing backup has the same VolumeID, otherwise it belongs to another volume
+			if backup.VolumeID != volumeID {
+				return nil, status.Error(codes.AlreadyExists, "Backup with given name already exists, with different source volume ID")
+			}
+
+			// If a backup of the volume already exists, skip creating the snapshot
+			backupAlreadyExists = true
+			klog.V(3).Infof("Found existing backup %s from volume with ID: %s", name, volumeID)
+		}
+
+		// Get the max duration to wait in seconds per GB of snapshot and fail if parsing fails
+		if item, ok := (req.Parameters)[openstack.BackupMaxDurationPerGB]; ok {
+			backupMaxDurationSecondsPerGB, err = strconv.Atoi(item)
+			if err != nil {
+				klog.Errorf("Setting backup-max-duration-seconds-per-gb failed due to a parsing error: %v", err)
+				return nil, status.Error(codes.Internal, "Failed to parse backup-max-duration-seconds-per-gb")
+			}
+		}
+	}
+
+	// Create the snapshot if the backup does not already exist and wait for it to be ready
+	if !backupAlreadyExists {
+		snap, err = cs.createSnapshot(name, volumeID, req.Parameters)
+		if err != nil {
+			return nil, err
+		}
+
+		ctime = timestamppb.New(snap.CreatedAt)
+		if err = ctime.CheckValid(); err != nil {
+			klog.Errorf("Error to convert time to timestamp: %v", err)
+		}
+
+		snap.Status, err = cs.Cloud.WaitSnapshotReady(snap.ID)
+		if err != nil {
+			klog.Errorf("Failed to WaitSnapshotReady: %v", err)
+			return nil, status.Errorf(codes.Internal, "CreateSnapshot failed with error: %v. Current snapshot status: %v", err, snap.Status)
+		}
+
+		snapSize = snap.Size
+	}
+
+	if snapshotType == "snapshot" {
+		return &csi.CreateSnapshotResponse{
+			Snapshot: &csi.Snapshot{
+				SnapshotId:     snap.ID,
+				SizeBytes:      int64(snap.Size * 1024 * 1024 * 1024),
+				SourceVolumeId: snap.VolumeID,
+				CreationTime:   ctime,
+				ReadyToUse:     true,
+			},
+		}, nil
+	}
+
+	// If snapshotType is 'backup', create a backup from the snapshot and delete the snapshot.
+	if snapshotType == "backup" {
+
+		if !backupAlreadyExists {
+			backup, err = cs.createBackup(name, volumeID, snap, req.Parameters)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		ctime = timestamppb.New(backup.CreatedAt)
+		if err := ctime.CheckValid(); err != nil {
+			klog.Errorf("Error to convert time to timestamp: %v", err)
+		}
+
+		backup.Status, err = cs.Cloud.WaitBackupReady(backup.ID, snapSize, backupMaxDurationSecondsPerGB)
+		if err != nil {
+			klog.Errorf("Failed to WaitBackupReady: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateBackup failed with error %v. Current backups status: %s", err, backup.Status))
+		}
+
+		// Necessary to get all the backup information, including size.
+		backup, err = cs.Cloud.GetBackupByID(backup.ID)
+		if err != nil {
+			klog.Errorf("Failed to GetBackupByID after backup creation: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("GetBackupByID failed with error %v", err))
+		}
+
+		err = cs.Cloud.DeleteSnapshot(snap.ID)
+		if err != nil && !cpoerrors.IsNotFound(err) {
+			klog.Errorf("Failed to DeleteSnapshot: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("DeleteSnapshot failed with error %v", err))
+		}
+	}
+
+	return &csi.CreateSnapshotResponse{
+		Snapshot: &csi.Snapshot{
+			SnapshotId:     backup.ID,
+			SizeBytes:      int64(backup.Size * 1024 * 1024 * 1024),
+			SourceVolumeId: backup.VolumeID,
+			CreationTime:   ctime,
+			ReadyToUse:     true,
+		},
+	}, nil
+
+}
+
+func (cs *controllerServer) createSnapshot(name string, volumeID string, parameters map[string]string) (snap *snapshots.Snapshot, err error) {
+
 	filters := map[string]string{}
 	filters["Name"] = name
+
+	// List existing snapshots with the same name
 	snapshots, _, err := cs.Cloud.ListSnapshots(filters)
 	if err != nil {
 		klog.Errorf("Failed to query for existing Snapshot during CreateSnapshot: %v", err)
 		return nil, status.Error(codes.Internal, "Failed to get snapshots")
 	}
 
+	// If more than one snapshot with the provided name exists, fail
+	if len(snapshots) > 1 {
+		klog.Errorf("found multiple existing snapshots with selected name (%s) during create", name)
+
+		return nil, status.Error(codes.Internal, "Multiple snapshots reported by Cinder with same name")
+	}
+
+	// Verify a snapshot with the provided name doesn't already exist for this tenant
 	if len(snapshots) == 1 {
 		snap = &snapshots[0]
-
 		if snap.VolumeID != volumeID {
 			return nil, status.Error(codes.AlreadyExists, "Snapshot with given name already exists, with different source volume ID")
 		}
 
+		// If the snapshot for the correct volume already exists, return it
 		klog.V(3).Infof("Found existing snapshot %s from volume with ID: %s", name, volumeID)
+		return snap, nil
+	}
 
-	} else if len(snapshots) > 1 {
-		klog.Errorf("found multiple existing snapshots with selected name (%s) during create", name)
-		return nil, status.Error(codes.Internal, "Multiple snapshots reported by Cinder with same name")
-
-	} else {
-		// Add cluster ID to the snapshot metadata
-		properties := map[string]string{cinderCSIClusterIDKey: cs.Driver.cluster}
-
-		// see https://github.com/kubernetes-csi/external-snapshotter/pull/375/
-		// Also, we don't want to tag every param but we still want to send the
-		// 'force-create' flag to openstack layer so that we will honor the
-		// force create functions
-		for _, mKey := range []string{"csi.storage.k8s.io/volumesnapshot/name", "csi.storage.k8s.io/volumesnapshot/namespace", "csi.storage.k8s.io/volumesnapshotcontent/name", openstack.SnapshotForceCreate} {
-			if v, ok := req.Parameters[mKey]; ok {
-				properties[mKey] = v
-			}
-		}
+	// Add cluster ID to the snapshot metadata
+	properties := map[string]string{cinderCSIClusterIDKey: cs.Driver.cluster}
 
-		// TODO: Delegate the check to openstack itself and ignore the conflict
-		snap, err = cs.Cloud.CreateSnapshot(name, volumeID, &properties)
-		if err != nil {
-			klog.Errorf("Failed to Create snapshot: %v", err)
-			return nil, status.Errorf(codes.Internal, "CreateSnapshot failed with error %v", err)
+	// see https://github.com/kubernetes-csi/external-snapshotter/pull/375/
+	// Also, we don't want to tag every param but we still want to send the
+	// 'force-create' flag to openstack layer so that we will honor the
+	// force create functions
+	for _, mKey := range []string{"csi.storage.k8s.io/volumesnapshot/name", "csi.storage.k8s.io/volumesnapshot/namespace", "csi.storage.k8s.io/volumesnapshotcontent/name", openstack.SnapshotForceCreate} {
+		if v, ok := parameters[mKey]; ok {
+			properties[mKey] = v
 		}
+	}
 
-		klog.V(3).Infof("CreateSnapshot %s from volume with ID: %s", name, volumeID)
+	// TODO: Delegate the check to openstack itself and ignore the conflict
+	snap, err = cs.Cloud.CreateSnapshot(name, volumeID, properties)
+	if err != nil {
+		klog.Errorf("Failed to Create snapshot: %v", err)
+		return nil, status.Errorf(codes.Internal, "CreateSnapshot failed with error %v", err)
 	}
 
-	ctime := timestamppb.New(snap.CreatedAt)
-	if err := ctime.CheckValid(); err != nil {
-		klog.Errorf("Error to convert time to timestamp: %v", err)
+	klog.V(3).Infof("CreateSnapshot %s from volume with ID: %s", name, volumeID)
+
+	return snap, nil
+}
+
+func (cs *controllerServer) createBackup(name string, volumeID string, snap *snapshots.Snapshot, parameters map[string]string) (*backups.Backup, error) {
+
+	// Add cluster ID to the snapshot metadata
+	properties := map[string]string{cinderCSIClusterIDKey: cs.Driver.cluster}
+
+	// see https://github.com/kubernetes-csi/external-snapshotter/pull/375/
+	// Also, we don't want to tag every param but we still want to send the
+	// 'force-create' flag to openstack layer so that we will honor the
+	// force create functions
+	for _, mKey := range []string{"csi.storage.k8s.io/volumesnapshot/name", "csi.storage.k8s.io/volumesnapshot/namespace", "csi.storage.k8s.io/volumesnapshotcontent/name", openstack.SnapshotForceCreate, openstack.SnapshotType} {
+		if v, ok := parameters[mKey]; ok {
+			properties[mKey] = v
+		}
 	}
 
-	err = cs.Cloud.WaitSnapshotReady(snap.ID)
+	backup, err := cs.Cloud.CreateBackup(name, volumeID, snap.ID, properties)
 	if err != nil {
-		klog.Errorf("Failed to WaitSnapshotReady: %v", err)
-		return nil, status.Errorf(codes.Internal, "CreateSnapshot failed with error %v", err)
+		klog.Errorf("Failed to Create backup: %v", err)
+		return nil, status.Error(codes.Internal, fmt.Sprintf("CreateBackup failed with error %v", err))
 	}
+	klog.V(4).Infof("Backup created: %+v", backup)
 
-	return &csi.CreateSnapshotResponse{
-		Snapshot: &csi.Snapshot{
-			SnapshotId:     snap.ID,
-			SizeBytes:      int64(snap.Size * 1024 * 1024 * 1024),
-			SourceVolumeId: snap.VolumeID,
-			CreationTime:   ctime,
-			ReadyToUse:     true,
-		},
-	}, nil
+	return backup, nil
 }
 
 func (cs *controllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
@@ -413,8 +612,18 @@ func (cs *controllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 		return nil, status.Error(codes.InvalidArgument, "Snapshot ID must be provided in DeleteSnapshot request")
 	}
 
+	// If volumeSnapshot object was linked to a cinder backup, delete the backup.
+	back, err := cs.Cloud.GetBackupByID(id)
+	if err == nil && back != nil {
+		err = cs.Cloud.DeleteBackup(id)
+		if err != nil {
+			klog.Errorf("Failed to Delete backup: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("DeleteBackup failed with error %v", err))
+		}
+	}
+
 	// Delegate the check to openstack itself
-	err := cs.Cloud.DeleteSnapshot(id)
+	err = cs.Cloud.DeleteSnapshot(id)
 	if err != nil {
 		if cpoerrors.IsNotFound(err) {
 			klog.V(3).Infof("Snapshot %s is already deleted.", id)
@@ -675,6 +884,16 @@ func getCreateVolumeResponse(vol *volumes.Volume, ignoreVolumeAZ bool, accessibl
 		}
 	}
 
+	if vol.BackupID != nil && *vol.BackupID != "" {
+		volsrc = &csi.VolumeContentSource{
+			Type: &csi.VolumeContentSource_Snapshot{
+				Snapshot: &csi.VolumeContentSource_SnapshotSource{
+					SnapshotId: *vol.BackupID,
+				},
+			},
+		}
+	}
+
 	var accessibleTopology []*csi.Topology
 	// If ignore-volume-az is true , dont set the accessible topology to volume az,
 	// use from preferred topologies instead.
diff --git a/pkg/csi/cinder/controllerserver_test.go b/pkg/csi/cinder/controllerserver_test.go
index b0a712cca2..f8d7bcd541 100644
--- a/pkg/csi/cinder/controllerserver_test.go
+++ b/pkg/csi/cinder/controllerserver_test.go
@@ -44,8 +44,8 @@ func init() {
 func TestCreateVolume(t *testing.T) {
 	// mock OpenStack
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", &properties).Return(&FakeVol, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", "", properties).Return(&FakeVol, nil)
 
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 	// Init assert
@@ -90,9 +90,9 @@ func TestCreateVolume(t *testing.T) {
 func TestCreateVolumeWithParam(t *testing.T) {
 	// mock OpenStack
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (string, string, int, error)
 	// Vol type and availability comes from CreateVolumeRequest.Parameters
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), "dummyVolType", "cinder", "", "", &properties).Return(&FakeVol, nil)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), "dummyVolType", "cinder", "", "", "", properties).Return(&FakeVol, nil)
 
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 	// Init assert
@@ -146,8 +146,8 @@ func TestCreateVolumeWithExtraMetadata(t *testing.T) {
 		"csi.storage.k8s.io/pvc/name":      FakePVCName,
 		"csi.storage.k8s.io/pvc/namespace": FakePVCNamespace,
 	}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", &properties).Return(&FakeVol, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", "", properties).Return(&FakeVol, nil)
 
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 
@@ -186,8 +186,8 @@ func TestCreateVolumeWithExtraMetadata(t *testing.T) {
 
 func TestCreateVolumeFromSnapshot(t *testing.T) {
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", FakeSnapshotID, "", &properties).Return(&FakeVolFromSnapshot, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", FakeSnapshotID, "", "", properties).Return(&FakeVolFromSnapshot, nil)
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 
 	// Init assert
@@ -233,8 +233,8 @@ func TestCreateVolumeFromSnapshot(t *testing.T) {
 
 func TestCreateVolumeFromSourceVolume(t *testing.T) {
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", "", FakeVolID, &properties).Return(&FakeVolFromSourceVolume, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", "", FakeVolID, "", properties).Return(&FakeVolFromSourceVolume, nil)
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 
 	// Init assert
@@ -451,10 +451,12 @@ func TestListVolumes(t *testing.T) {
 
 // Test CreateSnapshot
 func TestCreateSnapshot(t *testing.T) {
-	osmock.On("CreateSnapshot", FakeSnapshotName, FakeVolID, &map[string]string{cinderCSIClusterIDKey: "cluster"}).Return(&FakeSnapshotRes, nil)
-	osmock.On("ListSnapshots", map[string]string{"Name": FakeSnapshotName}).Return(FakeSnapshotListEmpty, "", nil)
-	osmock.On("WaitSnapshotReady", FakeSnapshotID).Return(nil)
 
+	osmock.On("CreateSnapshot", FakeSnapshotName, FakeVolID, map[string]string{cinderCSIClusterIDKey: "cluster"}).Return(&FakeSnapshotRes, nil)
+	osmock.On("ListSnapshots", map[string]string{"Name": FakeSnapshotName}).Return(FakeSnapshotListEmpty, "", nil)
+	osmock.On("WaitSnapshotReady", FakeSnapshotID).Return(FakeSnapshotRes.Status, nil)
+	osmock.On("ListBackups", map[string]string{"Name": FakeSnapshotName}).Return(FakeBackupListEmpty, nil)
+	osmock.On("GetSnapshotByID", FakeVolID).Return(&FakeSnapshotRes, nil)
 	// Init assert
 	assert := assert.New(t)
 
@@ -487,7 +489,7 @@ func TestCreateSnapshotWithExtraMetadata(t *testing.T) {
 		openstack.SnapshotForceCreate:                   "true",
 	}
 
-	osmock.On("CreateSnapshot", FakeSnapshotName, FakeVolID, &properties).Return(&FakeSnapshotRes, nil)
+	osmock.On("CreateSnapshot", FakeSnapshotName, FakeVolID, properties).Return(&FakeSnapshotRes, nil)
 	osmock.On("ListSnapshots", map[string]string{"Name": FakeSnapshotName}).Return(FakeSnapshotListEmpty, "", nil)
 	osmock.On("WaitSnapshotReady", FakeSnapshotID).Return(nil)
 
@@ -522,6 +524,7 @@ func TestCreateSnapshotWithExtraMetadata(t *testing.T) {
 func TestDeleteSnapshot(t *testing.T) {
 	// DeleteSnapshot(volumeID string) error
 	osmock.On("DeleteSnapshot", FakeSnapshotID).Return(nil)
+	osmock.On("DeleteBackup", FakeSnapshotID).Return(nil)
 
 	// Init assert
 	assert := assert.New(t)
diff --git a/pkg/csi/cinder/fake.go b/pkg/csi/cinder/fake.go
index dfed9220bc..3d3efbe28d 100644
--- a/pkg/csi/cinder/fake.go
+++ b/pkg/csi/cinder/fake.go
@@ -17,6 +17,7 @@ limitations under the License.
 package cinder
 
 import (
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/snapshots"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/volumes"
 	"golang.org/x/net/context"
@@ -94,6 +95,7 @@ var FakeSnapshotRes = snapshots.Snapshot{
 	Name:     "fake-snapshot",
 	VolumeID: FakeVolID,
 	Size:     1,
+	Status:   "available",
 }
 
 var FakeSnapshotsRes = []snapshots.Snapshot{FakeSnapshotRes}
@@ -102,6 +104,7 @@ var FakeVolListMultiple = []volumes.Volume{FakeVol1, FakeVol3}
 var FakeVolList = []volumes.Volume{FakeVol1}
 var FakeVolListEmpty = []volumes.Volume{}
 var FakeSnapshotListEmpty = []snapshots.Snapshot{}
+var FakeBackupListEmpty = []backups.Backup{}
 
 var FakeInstanceID = "321a8b81-3660-43e5-bab8-6470b65ee4e8"
 
diff --git a/pkg/csi/cinder/nodeserver.go b/pkg/csi/cinder/nodeserver.go
index 23058fcef9..3c1faf1fe9 100644
--- a/pkg/csi/cinder/nodeserver.go
+++ b/pkg/csi/cinder/nodeserver.go
@@ -150,7 +150,7 @@ func nodePublishEphemeral(req *csi.NodePublishVolumeRequest, ns *nodeServer) (*c
 		volumeType = ""
 	}
 
-	evol, err := ns.Cloud.CreateVolume(volName, size, volumeType, volAvailability, "", "", &properties)
+	evol, err := ns.Cloud.CreateVolume(volName, size, volumeType, volAvailability, "", "", "", properties)
 
 	if err != nil {
 		klog.V(3).Infof("Failed to Create Ephemeral Volume: %v", err)
diff --git a/pkg/csi/cinder/nodeserver_test.go b/pkg/csi/cinder/nodeserver_test.go
index 8e67d6f70c..dbfeb6144b 100644
--- a/pkg/csi/cinder/nodeserver_test.go
+++ b/pkg/csi/cinder/nodeserver_test.go
@@ -129,7 +129,7 @@ func TestNodePublishVolumeEphermeral(t *testing.T) {
 	fvolName := fmt.Sprintf("ephemeral-%s", FakeVolID)
 	tState := []string{"available"}
 
-	omock.On("CreateVolume", fvolName, 2, "test", "nova", "", "", &properties).Return(&FakeVol, nil)
+	omock.On("CreateVolume", fvolName, 2, "test", "nova", "", "", "", properties).Return(&FakeVol, nil)
 
 	omock.On("AttachVolume", FakeNodeID, FakeVolID).Return(FakeVolID, nil)
 	omock.On("WaitDiskAttached", FakeNodeID, FakeVolID).Return(nil)
diff --git a/pkg/csi/cinder/openstack/openstack.go b/pkg/csi/cinder/openstack/openstack.go
index e3a76a167f..4782e43545 100644
--- a/pkg/csi/cinder/openstack/openstack.go
+++ b/pkg/csi/cinder/openstack/openstack.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/gophercloud/gophercloud"
 	"github.com/gophercloud/gophercloud/openstack"
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/snapshots"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/volumes"
 	"github.com/gophercloud/gophercloud/openstack/compute/v2/servers"
@@ -44,7 +45,7 @@ func AddExtraFlags(fs *pflag.FlagSet) {
 }
 
 type IOpenStack interface {
-	CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, tags *map[string]string) (*volumes.Volume, error)
+	CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (*volumes.Volume, error)
 	DeleteVolume(volumeID string) error
 	AttachVolume(instanceID, volumeID string) (string, error)
 	ListVolumes(limit int, startingToken string) ([]volumes.Volume, string, error)
@@ -55,11 +56,17 @@ type IOpenStack interface {
 	GetAttachmentDiskPath(instanceID, volumeID string) (string, error)
 	GetVolume(volumeID string) (*volumes.Volume, error)
 	GetVolumesByName(name string) ([]volumes.Volume, error)
-	CreateSnapshot(name, volID string, tags *map[string]string) (*snapshots.Snapshot, error)
+	CreateSnapshot(name, volID string, tags map[string]string) (*snapshots.Snapshot, error)
 	ListSnapshots(filters map[string]string) ([]snapshots.Snapshot, string, error)
 	DeleteSnapshot(snapID string) error
 	GetSnapshotByID(snapshotID string) (*snapshots.Snapshot, error)
-	WaitSnapshotReady(snapshotID string) error
+	WaitSnapshotReady(snapshotID string) (string, error)
+	CreateBackup(name, volID string, snapshotID string, tags map[string]string) (*backups.Backup, error)
+	ListBackups(filters map[string]string) ([]backups.Backup, error)
+	DeleteBackup(backupID string) error
+	GetBackupByID(backupID string) (*backups.Backup, error)
+	BackupsAreEnabled() (bool, error)
+	WaitBackupReady(backupID string, snapshotSize int, backupMaxDurationSecondsPerGB int) (string, error)
 	GetInstanceByID(instanceID string) (*servers.Server, error)
 	ExpandVolume(volumeID string, status string, size int) error
 	GetMaxVolLimit() int64
diff --git a/pkg/csi/cinder/openstack/openstack_backups.go b/pkg/csi/cinder/openstack/openstack_backups.go
new file mode 100644
index 0000000000..8ddc05cff0
--- /dev/null
+++ b/pkg/csi/cinder/openstack/openstack_backups.go
@@ -0,0 +1,221 @@
+/*
+Copyright 2023 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 openstack backups provides an implementation of Cinder Backup features
+// cinder functions using Gophercloud.
+package openstack
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/gophercloud/gophercloud/openstack"
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
+	"golang.org/x/net/context"
+	"k8s.io/cloud-provider-openstack/pkg/metrics"
+	"k8s.io/klog/v2"
+)
+
+const (
+	backupReadyStatus                    = "available"
+	backupErrorStatus                    = "error"
+	backupBinary                         = "cinder-backup"
+	backupDescription                    = "Created by OpenStack Cinder CSI driver"
+	BackupMaxDurationSecondsPerGBDefault = 20
+	BackupMaxDurationPerGB               = "backup-max-duration-seconds-per-gb"
+	backupBaseDurationSeconds            = 30
+	backupReadyCheckIntervalSeconds      = 7
+)
+
+// CreateBackup issues a request to create a Backup from the specified Snapshot with the corresponding ID and
+// returns the resultant gophercloud Backup Item upon success.
+func (os *OpenStack) CreateBackup(name, volID string, snapshotID string, tags map[string]string) (*backups.Backup, error) {
+	blockstorageServiceClient, err := openstack.NewBlockStorageV3(os.blockstorage.ProviderClient, os.epOpts)
+	if err != nil {
+		return &backups.Backup{}, err
+	}
+
+	force := false
+	// if no flag given, then force will be false by default
+	// if flag it given , check it
+	if item, ok := (tags)[SnapshotForceCreate]; ok {
+		var err error
+		force, err = strconv.ParseBool(item)
+		if err != nil {
+			klog.V(5).Infof("Make force create flag to false due to: %v", err)
+		}
+		delete(tags, SnapshotForceCreate)
+	}
+
+	opts := &backups.CreateOpts{
+		VolumeID:    volID,
+		SnapshotID:  snapshotID,
+		Name:        name,
+		Force:       force,
+		Description: backupDescription,
+	}
+
+	if tags != nil {
+		// Set openstack microversion to 3.43 to send metadata along with the backup
+		blockstorageServiceClient.Microversion = "3.43"
+		opts.Metadata = tags
+	}
+
+	// TODO: Do some check before really call openstack API on the input
+	mc := metrics.NewMetricContext("backup", "create")
+	backup, err := backups.Create(blockstorageServiceClient, opts).Extract()
+	if mc.ObserveRequest(err) != nil {
+		return &backups.Backup{}, err
+	}
+	// There's little value in rewrapping these gophercloud types into yet another abstraction/type, instead just
+	// return the gophercloud item
+	return backup, nil
+}
+
+// ListBackups retrieves a list of active backups from Cinder for the corresponding Tenant.  We also
+// provide the ability to provide limit and offset to enable the consumer to provide accurate pagination.
+// In addition the filters argument provides a mechanism for passing in valid filter strings to the list
+// operation.  Valid filter keys are:  Name, Status, VolumeID, Limit, Marker (TenantID has no effect).
+func (os *OpenStack) ListBackups(filters map[string]string) ([]backups.Backup, error) {
+	var allBackups []backups.Backup
+
+	// Build the Opts
+	opts := backups.ListOpts{}
+	for key, val := range filters {
+		switch key {
+		case "Status":
+			opts.Status = val
+		case "Name":
+			opts.Name = val
+		case "VolumeID":
+			opts.VolumeID = val
+		case "Marker":
+			opts.Marker = val
+		case "Limit":
+			opts.Limit, _ = strconv.Atoi(val)
+		default:
+			klog.V(3).Infof("Not a valid filter key %s", key)
+		}
+	}
+	mc := metrics.NewMetricContext("backup", "list")
+
+	allPages, err := backups.List(os.blockstorage, opts).AllPages()
+	if err != nil {
+		return nil, err
+	}
+	allBackups, err = backups.ExtractBackups(allPages)
+	if err != nil {
+		return nil, err
+	}
+
+	if mc.ObserveRequest(err) != nil {
+		return nil, err
+	}
+
+	return allBackups, nil
+}
+
+// DeleteBackup issues a request to delete the Backup with the specified ID from the Cinder backend.
+func (os *OpenStack) DeleteBackup(backupID string) error {
+	mc := metrics.NewMetricContext("backup", "delete")
+	err := backups.Delete(os.blockstorage, backupID).ExtractErr()
+	if mc.ObserveRequest(err) != nil {
+		klog.Errorf("Failed to delete backup: %v", err)
+	}
+	return err
+}
+
+// GetBackupByID returns backup details by id.
+func (os *OpenStack) GetBackupByID(backupID string) (*backups.Backup, error) {
+	mc := metrics.NewMetricContext("backup", "get")
+	backup, err := backups.Get(os.blockstorage, backupID).Extract()
+	if mc.ObserveRequest(err) != nil {
+		klog.Errorf("Failed to get backup: %v", err)
+		return nil, err
+	}
+	return backup, nil
+}
+
+func (os *OpenStack) BackupsAreEnabled() (bool, error) {
+	// TODO: Check if the backup service is enabled
+	return true, nil
+}
+
+// WaitBackupReady waits until backup is ready. It waits longer depending on
+// the size of the corresponding snapshot.
+func (os *OpenStack) WaitBackupReady(backupID string, snapshotSize int, backupMaxDurationSecondsPerGB int) (string, error) {
+	var err error
+
+	duration := time.Duration(backupMaxDurationSecondsPerGB*snapshotSize + backupBaseDurationSeconds)
+
+	err = os.waitBackupReadyWithContext(backupID, duration)
+	if err == context.DeadlineExceeded {
+		err = fmt.Errorf("timeout, Backup %s is still not Ready: %v", backupID, err)
+	}
+
+	back, _ := os.GetBackupByID(backupID)
+
+	if back != nil {
+		return back.Status, err
+	} else {
+		return "Failed to get backup status", err
+	}
+}
+
+// Supporting function for WaitBackupReady().
+// Allows for a timeout while waiting for the backup to be ready.
+func (os *OpenStack) waitBackupReadyWithContext(backupID string, duration time.Duration) error {
+	ctx, cancel := context.WithTimeout(context.Background(), duration*time.Second)
+	defer cancel()
+	var done bool
+	var err error
+	ticker := time.NewTicker(backupReadyCheckIntervalSeconds * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			done, err = os.backupIsReady(backupID)
+			if err != nil {
+				return err
+			}
+
+			if done {
+				return nil
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+
+}
+
+// Supporting function for waitBackupReadyWithContext().
+// Returns true when the backup is ready.
+func (os *OpenStack) backupIsReady(backupID string) (bool, error) {
+	backup, err := os.GetBackupByID(backupID)
+	if err != nil {
+		return false, err
+	}
+
+	if backup.Status == backupErrorStatus {
+		return false, errors.New("backup is in error state")
+	}
+
+	return backup.Status == backupReadyStatus, nil
+}
diff --git a/pkg/csi/cinder/openstack/openstack_mock.go b/pkg/csi/cinder/openstack/openstack_mock.go
index 5c52cf0fbf..481e0157fc 100644
--- a/pkg/csi/cinder/openstack/openstack_mock.go
+++ b/pkg/csi/cinder/openstack/openstack_mock.go
@@ -17,6 +17,7 @@ limitations under the License.
 package openstack
 
 import (
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/snapshots"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/volumes"
 	"github.com/gophercloud/gophercloud/openstack/compute/v2/servers"
@@ -41,6 +42,18 @@ var fakeSnapshot = snapshots.Snapshot{
 	Metadata: make(map[string]string),
 }
 
+var fakemap = make(map[string]string)
+
+var fakeBackup = backups.Backup{
+	ID:         "eb5e4e9a-a4e5-4728-a748-04f9e2868573",
+	Name:       "fake-snapshot",
+	Status:     "available",
+	Size:       1,
+	VolumeID:   "CSIVolumeID",
+	SnapshotID: "261a8b81-3660-43e5-bab8-6470b65ee4e8",
+	Metadata:   &fakemap,
+}
+
 // revive:disable:exported
 // OpenStackMock is an autogenerated mock type for the IOpenStack type
 // ORIGINALLY GENERATED BY mockery with hand edits
@@ -72,19 +85,19 @@ func (_m *OpenStackMock) AttachVolume(instanceID string, volumeID string) (strin
 }
 
 // CreateVolume provides a mock function with given fields: name, size, vtype, availability, tags
-func (_m *OpenStackMock) CreateVolume(name string, size int, vtype string, availability string, snapshotID string, sourceVolID string, tags *map[string]string) (*volumes.Volume, error) {
-	ret := _m.Called(name, size, vtype, availability, snapshotID, sourceVolID, tags)
+func (_m *OpenStackMock) CreateVolume(name string, size int, vtype string, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (*volumes.Volume, error) {
+	ret := _m.Called(name, size, vtype, availability, snapshotID, sourceVolID, sourceBackupID, tags)
 
 	var r0 *volumes.Volume
-	if rf, ok := ret.Get(0).(func(string, int, string, string, string, string, *map[string]string) *volumes.Volume); ok {
-		r0 = rf(name, size, vtype, availability, snapshotID, sourceVolID, tags)
+	if rf, ok := ret.Get(0).(func(string, int, string, string, string, string, string, map[string]string) *volumes.Volume); ok {
+		r0 = rf(name, size, vtype, availability, snapshotID, sourceVolID, sourceBackupID, tags)
 	} else {
 		r0 = ret.Get(0).(*volumes.Volume)
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(string, int, string, string, string, string, *map[string]string) error); ok {
-		r1 = rf(name, size, vtype, availability, snapshotID, sourceVolID, tags)
+	if rf, ok := ret.Get(1).(func(string, int, string, string, string, string, string, map[string]string) error); ok {
+		r1 = rf(name, size, vtype, availability, snapshotID, sourceVolID, sourceBackupID, tags)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -245,11 +258,11 @@ func (_m *OpenStackMock) ListSnapshots(filters map[string]string) ([]snapshots.S
 }
 
 // CreateSnapshot provides a mock function with given fields: name, volID, tags
-func (_m *OpenStackMock) CreateSnapshot(name string, volID string, tags *map[string]string) (*snapshots.Snapshot, error) {
+func (_m *OpenStackMock) CreateSnapshot(name string, volID string, tags map[string]string) (*snapshots.Snapshot, error) {
 	ret := _m.Called(name, volID, tags)
 
 	var r0 *snapshots.Snapshot
-	if rf, ok := ret.Get(0).(func(string, string, *map[string]string) *snapshots.Snapshot); ok {
+	if rf, ok := ret.Get(0).(func(string, string, map[string]string) *snapshots.Snapshot); ok {
 		r0 = rf(name, volID, tags)
 	} else {
 		if ret.Get(0) != nil {
@@ -258,7 +271,7 @@ func (_m *OpenStackMock) CreateSnapshot(name string, volID string, tags *map[str
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(string, string, *map[string]string) error); ok {
+	if rf, ok := ret.Get(1).(func(string, string, map[string]string) error); ok {
 		r1 = rf(name, volID, tags)
 	} else {
 		r1 = ret.Error(1)
@@ -281,6 +294,62 @@ func (_m *OpenStackMock) DeleteSnapshot(snapID string) error {
 	return r0
 }
 
+func (_m *OpenStackMock) ListBackups(filters map[string]string) ([]backups.Backup, error) {
+	ret := _m.Called(filters)
+
+	var r0 []backups.Backup
+	if rf, ok := ret.Get(0).(func(map[string]string) []backups.Backup); ok {
+		r0 = rf(filters)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]backups.Backup)
+		}
+	}
+	var r1 error
+	if rf, ok := ret.Get(1).(func(map[string]string) error); ok {
+		r1 = rf(filters)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+func (_m *OpenStackMock) CreateBackup(name, volID string, snapshotID string, tags map[string]string) (*backups.Backup, error) {
+	ret := _m.Called(name, volID, snapshotID, tags)
+
+	var r0 *backups.Backup
+	if rf, ok := ret.Get(0).(func(string, string, string, map[string]string) *backups.Backup); ok {
+		r0 = rf(name, volID, snapshotID, tags)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*backups.Backup)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, string, string, map[string]string) error); ok {
+		r1 = rf(name, volID, snapshotID, tags)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+func (_m *OpenStackMock) DeleteBackup(backupID string) error {
+	ret := _m.Called(backupID)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string) error); ok {
+		r0 = rf(backupID)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // ListVolumes provides a mock function without param
 func (_m *OpenStackMock) ListVolumes(limit int, marker string) ([]volumes.Volume, string, error) {
 	ret := _m.Called(limit, marker)
@@ -342,23 +411,59 @@ func (_m *OpenStackMock) GetSnapshotByID(snapshotID string) (*snapshots.Snapshot
 	return &fakeSnapshot, nil
 }
 
-func (_m *OpenStackMock) WaitSnapshotReady(snapshotID string) error {
+func (_m *OpenStackMock) WaitSnapshotReady(snapshotID string) (string, error) {
 	ret := _m.Called(snapshotID)
 
-	var r0 error
-	if rf, ok := ret.Get(0).(func(string) error); ok {
+	var r0 string
+	if rf, ok := ret.Get(0).(func(string) string); ok {
 		r0 = rf(snapshotID)
 	} else {
-		r0 = ret.Error(0)
+		r0 = ret.String(0)
 	}
 
-	return r0
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(snapshotID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+func (_m *OpenStackMock) GetBackupByID(backupID string) (*backups.Backup, error) {
+
+	return &fakeBackup, nil
+}
+
+func (_m *OpenStackMock) WaitBackupReady(backupID string, snapshotSize int, backupMaxDurationSecondsPerGB int) (string, error) {
+	ret := _m.Called(backupID)
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func(string) string); ok {
+		r0 = rf(backupID)
+	} else {
+		r0 = ret.String(0)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(backupID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
 }
 
 func (_m *OpenStackMock) GetMaxVolLimit() int64 {
 	return 256
 }
 
+func (_m *OpenStackMock) BackupsAreEnabled() (bool, error) {
+	return true, nil
+}
+
 func (_m *OpenStackMock) GetInstanceByID(instanceID string) (*servers.Server, error) {
 	return nil, nil
 }
diff --git a/pkg/csi/cinder/openstack/openstack_snapshots.go b/pkg/csi/cinder/openstack/openstack_snapshots.go
index 6337e5313c..bd2f94365f 100644
--- a/pkg/csi/cinder/openstack/openstack_snapshots.go
+++ b/pkg/csi/cinder/openstack/openstack_snapshots.go
@@ -39,23 +39,24 @@ const (
 
 	snapshotDescription = "Created by OpenStack Cinder CSI driver"
 	SnapshotForceCreate = "force-create"
+	SnapshotType        = "type"
 )
 
 // CreateSnapshot issues a request to take a Snapshot of the specified Volume with the corresponding ID and
 // returns the resultant gophercloud Snapshot Item upon success
-func (os *OpenStack) CreateSnapshot(name, volID string, tags *map[string]string) (*snapshots.Snapshot, error) {
+func (os *OpenStack) CreateSnapshot(name, volID string, tags map[string]string) (*snapshots.Snapshot, error) {
 
 	force := false
 	// if no flag given, then force will be false by default
 	// if flag it given , check it
-	if item, ok := (*tags)[SnapshotForceCreate]; ok {
+	if item, ok := (tags)[SnapshotForceCreate]; ok {
 		var err error
 		force, err = strconv.ParseBool(item)
 		if err != nil {
 			klog.V(5).Infof("Make force create flag to false due to: %v", err)
 		}
 
-		delete(*tags, SnapshotForceCreate)
+		delete(tags, SnapshotForceCreate)
 	}
 	// Force the creation of snapshot even the Volume is in in-use state
 	opts := &snapshots.CreateOpts{
@@ -65,7 +66,7 @@ func (os *OpenStack) CreateSnapshot(name, volID string, tags *map[string]string)
 		Force:       force,
 	}
 	if tags != nil {
-		opts.Metadata = *tags
+		opts.Metadata = tags
 	}
 	// TODO: Do some check before really call openstack API on the input
 	mc := metrics.NewMetricContext("snapshot", "create")
@@ -157,7 +158,7 @@ func (os *OpenStack) GetSnapshotByID(snapshotID string) (*snapshots.Snapshot, er
 }
 
 // WaitSnapshotReady waits till snapshot is ready
-func (os *OpenStack) WaitSnapshotReady(snapshotID string) error {
+func (os *OpenStack) WaitSnapshotReady(snapshotID string) (string, error) {
 	backoff := wait.Backoff{
 		Duration: snapReadyDuration,
 		Factor:   snapReadyFactor,
@@ -173,10 +174,16 @@ func (os *OpenStack) WaitSnapshotReady(snapshotID string) error {
 	})
 
 	if wait.Interrupted(err) {
-		err = fmt.Errorf("timeout, Snapshot  %s is still not Ready %v", snapshotID, err)
+		err = fmt.Errorf("timeout, Snapshot %s is still not Ready %v", snapshotID, err)
 	}
 
-	return err
+	snap, _ := os.GetSnapshotByID(snapshotID)
+
+	if snap != nil {
+		return snap.Status, err
+	} else {
+		return "Failed to get snapshot status", err
+	}
 }
 
 func (os *OpenStack) snapshotIsReady(snapshotID string) (bool, error) {
diff --git a/pkg/csi/cinder/openstack/openstack_volumes.go b/pkg/csi/cinder/openstack/openstack_volumes.go
index 9ef45b455a..96550975d3 100644
--- a/pkg/csi/cinder/openstack/openstack_volumes.go
+++ b/pkg/csi/cinder/openstack/openstack_volumes.go
@@ -51,7 +51,7 @@ const (
 var volumeErrorStates = [...]string{"error", "error_extending", "error_deleting"}
 
 // CreateVolume creates a volume of given size
-func (os *OpenStack) CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, tags *map[string]string) (*volumes.Volume, error) {
+func (os *OpenStack) CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (*volumes.Volume, error) {
 
 	opts := &volumes.CreateOpts{
 		Name:             name,
@@ -60,14 +60,26 @@ func (os *OpenStack) CreateVolume(name string, size int, vtype, availability str
 		AvailabilityZone: availability,
 		Description:      volumeDescription,
 		SnapshotID:       snapshotID,
-		SourceVolID:      sourcevolID,
+		SourceVolID:      sourceVolID,
+		BackupID:         sourceBackupID,
 	}
 	if tags != nil {
-		opts.Metadata = *tags
+		opts.Metadata = tags
+	}
+
+	blockstorageClient, err := openstack.NewBlockStorageV3(os.blockstorage.ProviderClient, os.epOpts)
+	if err != nil {
+		return nil, err
+	}
+
+	// creating volumes from backups is available since 3.47 microversion
+	// https://docs.openstack.org/cinder/latest/contributor/api_microversion_history.html#id47
+	if !os.bsOpts.IgnoreVolumeMicroversion && sourceBackupID != "" {
+		blockstorageClient.Microversion = "3.47"
 	}
 
 	mc := metrics.NewMetricContext("volume", "create")
-	vol, err := volumes.Create(os.blockstorage, opts).Extract()
+	vol, err := volumes.Create(blockstorageClient, opts).Extract()
 	if mc.ObserveRequest(err) != nil {
 		return nil, err
 	}
diff --git a/tests/sanity/cinder/fakecloud.go b/tests/sanity/cinder/fakecloud.go
index d1bbed949f..f04f21fc67 100644
--- a/tests/sanity/cinder/fakecloud.go
+++ b/tests/sanity/cinder/fakecloud.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/gophercloud/gophercloud"
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/snapshots"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/volumes"
 	"github.com/gophercloud/gophercloud/openstack/compute/v2/servers"
@@ -19,6 +20,7 @@ type cloud struct {
 	volumes   map[string]*volumes.Volume
 	snapshots map[string]*snapshots.Snapshot
 	instances map[string]*servers.Server
+	backups   map[string]*backups.Backup
 }
 
 func getfakecloud() *cloud {
@@ -32,7 +34,7 @@ func getfakecloud() *cloud {
 var _ openstack.IOpenStack = &cloud{}
 
 // Fake Cloud
-func (cloud *cloud) CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, tags *map[string]string) (*volumes.Volume, error) {
+func (cloud *cloud) CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourceVolID string, sourceBackupID string, tags map[string]string) (*volumes.Volume, error) {
 
 	vol := &volumes.Volume{
 		ID:               randString(10),
@@ -43,6 +45,7 @@ func (cloud *cloud) CreateVolume(name string, size int, vtype, availability stri
 		AvailabilityZone: availability,
 		SnapshotID:       snapshotID,
 		SourceVolID:      sourceVolID,
+		BackupID:         &sourceBackupID,
 	}
 
 	cloud.volumes[vol.ID] = vol
@@ -149,7 +152,7 @@ func invalidError() error {
 	return gophercloud.ErrDefault400{}
 }
 
-func (cloud *cloud) CreateSnapshot(name, volID string, tags *map[string]string) (*snapshots.Snapshot, error) {
+func (cloud *cloud) CreateSnapshot(name, volID string, tags map[string]string) (*snapshots.Snapshot, error) {
 
 	snap := &snapshots.Snapshot{
 		ID:        randString(10),
@@ -220,10 +223,84 @@ func (cloud *cloud) GetSnapshotByID(snapshotID string) (*snapshots.Snapshot, err
 	return snap, nil
 }
 
-func (cloud *cloud) WaitSnapshotReady(snapshotID string) error {
+func (cloud *cloud) WaitSnapshotReady(snapshotID string) (string, error) {
+	return "available", nil
+}
+
+func (cloud *cloud) CreateBackup(name, volID string, snapshotID string, tags map[string]string) (*backups.Backup, error) {
+
+	backup := &backups.Backup{
+		ID:         randString(10),
+		Name:       name,
+		Status:     "available",
+		VolumeID:   volID,
+		SnapshotID: snapshotID,
+		CreatedAt:  time.Now(),
+	}
+
+	cloud.backups[backup.ID] = backup
+	return backup, nil
+}
+
+func (cloud *cloud) ListBackups(filters map[string]string) ([]backups.Backup, error) {
+	var backuplist []backups.Backup
+	startingToken := filters["Marker"]
+	limitfilter := filters["Limit"]
+	limit, _ := strconv.Atoi(limitfilter)
+	name := filters["Name"]
+	volumeID := filters["VolumeID"]
+
+	for _, value := range cloud.backups {
+		if volumeID != "" {
+			if value.VolumeID == volumeID {
+				backuplist = append(backuplist, *value)
+				break
+			}
+		} else if name != "" {
+			if value.Name == name {
+				backuplist = append(backuplist, *value)
+				break
+			}
+		} else {
+			backuplist = append(backuplist, *value)
+		}
+	}
+
+	if startingToken != "" {
+		t, _ := strconv.Atoi(startingToken)
+		backuplist = backuplist[t:]
+	}
+	if limit != 0 {
+		backuplist = backuplist[:limit]
+	}
+
+	return backuplist, nil
+}
+
+func (cloud *cloud) DeleteBackup(backupID string) error {
+	delete(cloud.backups, backupID)
+
 	return nil
 }
 
+func (cloud *cloud) GetBackupByID(backupID string) (*backups.Backup, error) {
+	backup, ok := cloud.backups[backupID]
+
+	if !ok {
+		return nil, notFoundError()
+	}
+
+	return backup, nil
+}
+
+func (cloud *cloud) BackupsAreEnabled() (bool, error) {
+	return true, nil
+}
+
+func (cloud *cloud) WaitBackupReady(backupID string, snapshotSize int, backupMaxDurationSecondsPerGB int) (string, error) {
+	return "", nil
+}
+
 func randString(n int) string {
 	const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
 	b := make([]byte, n)