PWX-37650: Supports adding the pod name as a locator label for clone/…

…snap restore (#2453)

Signed-off-by: Adam Krpan <[email protected]>

api/api.go

@@ -1329,6 +1329,14 @@ func (v *VolumeSpec) IsNFSProxyVolume() bool {
 	return v.GetProxySpec() != nil && v.GetProxySpec().NfsSpec != nil
 }
 
+// GetFADAPodName returns the FlashArray Pod name specified in the Pure Block spec, or empty if any fields are unspecified
+func (v *VolumeSpec) GetFADAPodName() string {
+	if v.GetProxySpec() != nil && v.GetProxySpec().PureBlockSpec != nil {
+		return v.GetProxySpec().PureBlockSpec.PodName
+	}
+	return ""
+}
+
 // GetCloneCreatorOwnership returns the appropriate ownership for the
 // new snapshot and if an update is required
 func (v *VolumeSpec) GetCloneCreatorOwnership(ctx context.Context) (*Ownership, bool) {

api/server/sdk/volume_ops.go

@@ -35,6 +35,9 @@ import (
 	"github.com/libopenstorage/openstorage/volume"
 )
 
+// FADAPodLabelKey is a label added to volume locators in the case of FADA volume clone/snap restore
+const FADAPodLabelKey = "pure-pod-name" // Used to plumb in the pod name for volume cloning
+
 // When create is called for an existing volume, this function is called to make sure
 // the SDK only returns that the volume is ready when the status is UP
 func (s *VolumeServer) waitForVolumeReady(ctx context.Context, id string) (*api.Volume, error) {
@@ -105,7 +108,6 @@ func (s *VolumeServer) create(
 	spec *api.VolumeSpec,
 	additionalCloneLabels map[string]string,
 ) (string, error) {
-
 	// Check if the volume has already been created or is in process of creation
 	volName := locator.GetName()
 	v, err := util.VolumeFromName(s.driver(ctx), volName)
@@ -170,8 +172,18 @@ func (s *VolumeServer) create(
 		}
 
 		// Create a snapshot from the parent
+		// Only include the FADA pod label
+		var labels map[string]string = nil
+		if locator.GetVolumeLabels() != nil {
+			if pod, ok := locator.GetVolumeLabels()[FADAPodLabelKey]; ok {
+				labels = map[string]string{
+					FADAPodLabelKey: pod,
+				}
+			}
+		}
 		id, err = s.driver(ctx).Snapshot(parent.GetId(), false, &api.VolumeLocator{
-			Name: volName,
+			Name:         volName,
+			VolumeLabels: labels,
 		}, false)
 		if err != nil {
 			if err == kvdb.ErrNotFound {
@@ -334,7 +346,8 @@ func (s *VolumeServer) Clone(
 	}
 
 	locator := &api.VolumeLocator{
-		Name: req.GetName(),
+		Name:         req.GetName(),
+		VolumeLabels: req.GetAdditionalLabels(),
 	}
 	source := &api.Source{
 		Parent: req.GetParentId(),

csi/controller.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/libopenstorage/openstorage/api"
+	"github.com/libopenstorage/openstorage/api/server/sdk"
 	"github.com/libopenstorage/openstorage/pkg/grpcutil"
 	"github.com/libopenstorage/openstorage/pkg/units"
 	"github.com/libopenstorage/openstorage/pkg/util"
@@ -586,10 +587,14 @@ func (s *OsdCsiServer) CreateVolume(
 		}
 		newVolumeId = createResp.VolumeId
 	} else {
+		clonedMetadata := getClonedPVCMetadata(locator)
+		if spec.GetFADAPodName() != "" {
+			clonedMetadata[sdk.FADAPodLabelKey] = spec.GetFADAPodName()
+		}
 		cloneResp, err := volumes.Clone(ctx, &api.SdkVolumeCloneRequest{
 			Name:             req.GetName(),
 			ParentId:         source.Parent,
-			AdditionalLabels: getClonedPVCMetadata(locator),
+			AdditionalLabels: clonedMetadata,
 		})
 		if err != nil {
 			return nil, err

csi/controller_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/golang/protobuf/ptypes/timestamp"
 
 	"github.com/libopenstorage/openstorage/api"
+	"github.com/libopenstorage/openstorage/api/server/sdk"
 	authsecrets "github.com/libopenstorage/openstorage/pkg/auth/secrets"
 	"github.com/libopenstorage/openstorage/pkg/units"
 
@@ -1973,6 +1974,122 @@ func TestControllerCreateVolumeFromSnapshot(t *testing.T) {
 	assert.Equal(t, mockParentID, volumeInfo.GetVolumeContext()[api.SpecParent])
 }
 
+func TestControllerCreateVolumeFromSnapshotFADAPod(t *testing.T) {
+	// Create server and client connection
+	s := newTestServer(t)
+	defer s.Stop()
+	c := csi.NewControllerClient(s.Conn())
+	s.mockClusterEnumerateNode(t, "node-1")
+	// Setup request
+	mockParentID := "parendId"
+	name := "myvol"
+	pod := "mypod"
+	size := int64(1234)
+	req := &csi.CreateVolumeRequest{
+		Name: name,
+		VolumeCapabilities: []*csi.VolumeCapability{
+			{},
+		},
+		CapacityRange: &csi.CapacityRange{
+			RequiredBytes: size,
+		},
+		VolumeContentSource: &csi.VolumeContentSource{
+			Type: &csi.VolumeContentSource_Snapshot{
+				Snapshot: &csi.VolumeContentSource_SnapshotSource{
+					SnapshotId: mockParentID,
+				},
+			},
+		},
+		Secrets: map[string]string{authsecrets.SecretTokenKey: systemUserToken},
+		Parameters: map[string]string{
+			api.SpecPurePodName: pod,
+		},
+	}
+
+	// Setup mock functions
+	id := "myid"
+	snapID := id + "-snap"
+	gomock.InOrder(
+
+		// First check on parent
+		s.MockDriver().
+			EXPECT().
+			Enumerate(&api.VolumeLocator{
+				VolumeIds: []string{mockParentID},
+			}, nil).
+			Return([]*api.Volume{{Id: mockParentID}}, nil).
+			Times(1),
+
+		// VolFromName (name)
+		s.MockDriver().
+			EXPECT().
+			Inspect([]string{name}).
+			Return(nil, fmt.Errorf("not found")).
+			Times(1),
+
+		s.MockDriver().
+			EXPECT().
+			Enumerate(gomock.Any(), nil).
+			Return(nil, fmt.Errorf("not found")).
+			Times(1),
+
+		//VolFromName parent
+		s.MockDriver().
+			EXPECT().
+			Inspect(gomock.Any()).
+			Return(
+				[]*api.Volume{{
+					Id: mockParentID,
+				}}, nil).
+			Times(1),
+
+		// create
+		s.MockDriver().
+			EXPECT().
+			Snapshot(gomock.Any(), gomock.Any(), &api.VolumeLocator{Name: name, VolumeLabels: map[string]string{sdk.FADAPodLabelKey: pod}}, gomock.Any()).
+			Return(snapID, nil).
+			Times(1),
+		s.MockDriver().
+			EXPECT().
+			Enumerate(&api.VolumeLocator{
+				VolumeIds: []string{snapID},
+			}, nil).
+			Return([]*api.Volume{
+				{
+					Id:     id,
+					Source: &api.Source{Parent: mockParentID},
+				},
+			}, nil).
+			Times(2),
+
+		s.MockDriver().
+			EXPECT().
+			Set(gomock.Any(), gomock.Any(), gomock.Any()).
+			Return(nil).
+			Times(1),
+
+		s.MockDriver().
+			EXPECT().
+			Enumerate(gomock.Any(), nil).
+			Return([]*api.Volume{
+				{
+					Id:     id,
+					Source: &api.Source{Parent: mockParentID},
+				},
+			}, nil).
+			Times(1),
+	)
+
+	r, err := c.CreateVolume(context.Background(), req)
+	assert.Nil(t, err)
+	assert.NotNil(t, r)
+	volumeInfo := r.GetVolume()
+
+	assert.Equal(t, id, volumeInfo.GetVolumeId())
+	assert.NotEqual(t, "true", volumeInfo.GetVolumeContext()[api.SpecSharedv4])
+	assert.Equal(t, mockParentID, volumeInfo.GetVolumeContext()[api.SpecParent])
+}
+
 func TestControllerCreateVolumeSnapshotThroughParameters(t *testing.T) {
 	// Create server and client connection
 	s := newTestServer(t)