Add support to restore snapshots to different namespaces

- When creating snapshots users need to provide comma seperated regexes
with "stork/snapshot-restore-namespaces" annotaion to specify which
namespaces the snapshot can be restored to
- When creating PVC from snapshots, if a snapshot exists in another
namespace, the snapshot namespace should be specified with
"stork/snapshot-source-namespace" annotation

Issue #71

drivers/volume/portworx/portworx.go

@@ -678,15 +678,20 @@ func (p *portworx) SnapshotRestore(
 
 	switch snapshotData.Spec.PortworxSnapshot.SnapshotType {
 	case crdv1.PortworxSnapshotTypeLocal:
+		snapshotNamespace, ok := pvc.Annotations[snapshotcontroller.StorkSnapshotSourceNamespaceAnnotation]
+		if !ok {
+			snapshotNamespace = pvc.GetNamespace()
+		}
+
 		// Check if this is a group snapshot
-		snap, err := k8s.Instance().GetSnapshot(snapshotName, pvc.GetNamespace())
+		snap, err := k8s.Instance().GetSnapshot(snapshotName, snapshotNamespace)
 		if err != nil {
 			return nil, nil, err
 		}
 
 		if isGroupSnap(snap) {
 			return nil, nil, fmt.Errorf("volumesnapshot: [%s] %s was used to create group snapshots. "+
-				"To restore, use the volumesnapshots that were created by this group snapshot.", pvc.GetNamespace(), snapshotName)
+				"To restore, use the volumesnapshots that were created by this group snapshot.", snapshotNamespace, snapshotName)
 		}
 
 		locator := &api.VolumeLocator{

pkg/snapshot/snapshotprovisioner.go

@@ -1,8 +1,11 @@
 package snapshotcontroller
 
 import (
+	"encoding/csv"
 	"errors"
 	"fmt"
+	"regexp"
+	"strings"
 
 	"github.com/kubernetes-incubator/external-storage/lib/controller"
 	crdv1 "github.com/kubernetes-incubator/external-storage/snapshot/pkg/apis/crd/v1"
@@ -19,6 +22,13 @@ import (
 // provisioner with some changes. It is a part of the main package there,
 // so can't vendor it in here.
 
+const (
+	storkSnapshotRestoreNamespacesAnnotation = "stork/snapshot-restore-namespaces"
+	// StorkSnapshotSourceNamespaceAnnotation Annotation used to specify the
+	// source of the snapshot when creating a PVC
+	StorkSnapshotSourceNamespaceAnnotation = "stork/snapshot-source-namespace"
+)
+
 type snapshotProvisioner struct {
 	// Kubernetes Client.
 	client kubernetes.Interface
@@ -74,6 +84,42 @@ func (p *snapshotProvisioner) snapshotRestore(
 	return pvSrc, labels, err
 }
 
+func (p *snapshotProvisioner) isSnapshotAllowed(
+	snapshot crdv1.VolumeSnapshot,
+	namespace string,
+) bool {
+	allowedNamespaces, ok := snapshot.Metadata.Annotations[storkSnapshotRestoreNamespacesAnnotation]
+	if !ok {
+		return false
+	}
+
+	csvReader := csv.NewReader(strings.NewReader(allowedNamespaces))
+	namespaces, err := csvReader.ReadAll()
+	if err != nil {
+		log.Errorf("Error parsing allowed namespaces: %v", allowedNamespaces)
+		return false
+	}
+
+	if len(namespaces) != 1 {
+		log.Errorf("Invalid allowed namespaces: %v", allowedNamespaces)
+		return false
+	}
+
+	for _, namespaceRegEx := range namespaces[0] {
+		// Add start and end delimiters to match complete strings
+		// Works even if the input already had the delimiters
+		regex, err := regexp.Compile("^" + strings.TrimSpace(namespaceRegEx) + "$")
+		if err != nil {
+			log.Errorf("Invalid regex for allowed namespaces: %v", namespaceRegEx)
+			return false
+		}
+		if regex.MatchString(namespace) {
+			return true
+		}
+	}
+	return false
+}
+
 // Provision creates a storage asset and returns a PV object representing it.
 func (p *snapshotProvisioner) Provision(options controller.VolumeOptions) (*v1.PersistentVolume, error) {
 	if options.PVC.Spec.Selector != nil {
@@ -83,17 +129,26 @@ func (p *snapshotProvisioner) Provision(options controller.VolumeOptions) (*v1.P
 	if !ok {
 		return nil, fmt.Errorf("snapshot annotation not found on PV")
 	}
+	snapshotNamespace, ok := options.PVC.Annotations[StorkSnapshotSourceNamespaceAnnotation]
+	if !ok {
+		snapshotNamespace = options.PVC.Namespace
+	}
 
 	var snapshot crdv1.VolumeSnapshot
 	err := p.crdclient.Get().
 		Resource(crdv1.VolumeSnapshotResourcePlural).
-		Namespace(options.PVC.Namespace).
+		Namespace(snapshotNamespace).
 		Name(snapshotName).
 		Do().Into(&snapshot)
-
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve VolumeSnapshot %s: %v", snapshotName, err)
 	}
+
+	if snapshotNamespace != options.PVC.Namespace &&
+		!p.isSnapshotAllowed(snapshot, options.PVC.Namespace) {
+		return nil, fmt.Errorf("Snapshot %v cannot be used in namespace %v", snapshotName, options.PVC.Namespace)
+	}
+
 	// FIXME: should also check if any VolumeSnapshotData points
 	// to this VolumeSnapshot
 	if len(snapshot.Spec.SnapshotDataName) == 0 {