Add ListVolumes with PublishedNodes

Author: davidz627

Gopkg.lock

@@ -71,7 +71,7 @@
 
 [[constraint]]
   name = "github.com/container-storage-interface/spec"
-  version = "1.1.0"
+  revision = "9e773d2dd46a9b113d94fb3c245231d231bcfeda"
 
 [[override]]
   name = "google.golang.org/grpc"

Gopkg.toml

@@ -10,3 +10,6 @@ images:
 - name: gke.gcr.io/gcp-compute-persistent-disk-csi-driver
   newName: gcr.io/dyzz-csi-staging/csi/gce-pd-driver
   newTag: "latest"
+- name: gke.gcr.io/csi-attacher
+  newName: gcr.io/dyzz-csi-staging/csi/csi-attacher
+  newTag: "latest"

deploy/kubernetes/overlays/dev/kustomization.yaml

@@ -95,6 +95,10 @@ func (cloud *FakeCloudProvider) ListZones(ctx context.Context, region string) ([
 	return []string{cloud.zone, "country-region-fakesecondzone"}, nil
 }
 
+func (cloud *FakeCloudProvider) ListDisks(ctx context.Context, maxEntries int64, pageToken string) ([]*compute.Disk, string, error) {
+	return nil, "", nil
+}
+
 func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string, maxEntries int64, pageToken string) ([]*compute.Snapshot, string, error) {
 	var sourceDisk string
 	snapshots := []*compute.Snapshot{}

pkg/gce-cloud-provider/compute/fake-gce.go

@@ -51,6 +51,7 @@ type GCECompute interface {
 	GetDiskTypeURI(volKey *meta.Key, diskType string) string
 	WaitForAttach(ctx context.Context, volKey *meta.Key, instanceZone, instanceName string) error
 	ResizeDisk(ctx context.Context, volKey *meta.Key, requestBytes int64) (int64, error)
+	ListDisks(ctx context.Context, maxEntries int64, pageToken string) ([]*compute.Disk, string, error)
 	// Regional Disk Methods
 	GetReplicaZoneURI(zone string) string
 	// Instance Methods
@@ -63,6 +64,24 @@ type GCECompute interface {
 	DeleteSnapshot(ctx context.Context, snapshotName string) error
 }
 
+// ListDisks lists disks based on maxEntries and pageToken only in the project
+// and zone that the driver is running in.
+func (cloud *CloudProvider) ListDisks(ctx context.Context, maxEntries int64, pageToken string) ([]*compute.Disk, string, error) {
+	lCall := cloud.service.Disks.List(cloud.project, cloud.zone)
+	if maxEntries != 0 {
+		lCall = lCall.MaxResults(maxEntries)
+	}
+	if len(pageToken) != 0 {
+		lCall = lCall.PageToken(pageToken)
+	}
+	diskList, err := lCall.Do()
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to list disks from cloud project %s, zone %s, max results %v, page token %s: %v",
+			cloud.project, cloud.zone, maxEntries, pageToken, err)
+	}
+	return diskList.Items, diskList.NextPageToken, nil
+}
+
 // RepairUnderspecifiedVolumeKey will query the cloud provider and check each zone for the disk specified
 // by the volume key and return a volume key with a correct zone
 func (cloud *CloudProvider) RepairUnderspecifiedVolumeKey(ctx context.Context, volumeKey *meta.Key) (*meta.Key, error) {

pkg/gce-cloud-provider/compute/gce-compute.go

@@ -515,8 +515,34 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 
 func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (*csi.ListVolumesResponse, error) {
 	// https://cloud.google.com/compute/docs/reference/beta/disks/list
-	// List volumes in the whole region? In only the zone that this controller is running?
-	return nil, status.Error(codes.Unimplemented, "")
+	if req.MaxEntries > 500 || req.MaxEntries < 0 {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf(
+			"ListVolumes got max entries request %v. GCE only supports values between 0-500", req.MaxEntries))
+	}
+	diskList, nextToken, err := gceCS.CloudProvider.ListDisks(ctx, int64(req.MaxEntries), req.StartingToken)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown list disk error: %v", err))
+	}
+	entries := []*csi.ListVolumesResponse_Entry{}
+	for _, d := range diskList {
+		users := []string{}
+		for _, u := range d.Users {
+			users = append(users, cleanSelfLink(u))
+		}
+		entries = append(entries, &csi.ListVolumesResponse_Entry{
+			Volume: &csi.Volume{
+				VolumeId: cleanSelfLink(d.SelfLink),
+			},
+			Status: &csi.ListVolumesResponse_VolumeStatus{
+				PublishedNodeIds: users,
+			},
+		})
+	}
+
+	return &csi.ListVolumesResponse{
+		Entries:   entries,
+		NextToken: nextToken,
+	}, nil
 }
 
 func (gceCS *GCEControllerServer) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest) (*csi.GetCapacityResponse, error) {

pkg/gce-pd-csi-driver/controller.go

@@ -714,6 +714,49 @@ func TestCreateVolumeArguments(t *testing.T) {
 	}
 }
 
+func TestListVolumeArgs(t *testing.T) {
+	testCases := []struct {
+		name       string
+		maxEntries int32
+		expErr     bool
+	}{
+		{
+			name: "normal",
+		},
+		{
+			name:       "fine amount of entries",
+			maxEntries: 420,
+		},
+		{
+			name:       "too many entries",
+			maxEntries: 501,
+			expErr:     true,
+		},
+		{
+			name:       "negative entries",
+			maxEntries: -1,
+			expErr:     true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		// Setup new driver each time so no interference
+		gceDriver := initGCEDriver(t, nil)
+		lvr := &csi.ListVolumesRequest{
+			MaxEntries: tc.maxEntries,
+		}
+		_, err := gceDriver.cs.ListVolumes(context.TODO(), lvr)
+		if tc.expErr && err == nil {
+			t.Fatalf("Got no error when expecting an error")
+		}
+		if !tc.expErr && err != nil {
+			t.Fatalf("Got error %v, expecting none", err)
+		}
+
+	}
+}
+
 func TestCreateVolumeWithVolumeSource(t *testing.T) {
 	// Define test cases
 	testCases := []struct {

pkg/gce-pd-csi-driver/controller_test.go

@@ -67,6 +67,7 @@ func (gceDriver *GCEDriver) SetupGCEDriver(cloudProvider gce.GCECompute, mounter
 		csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
 		csi.ControllerServiceCapability_RPC_PUBLISH_READONLY,
 		csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
+		csi.ControllerServiceCapability_RPC_LIST_VOLUMES_PUBLISHED_NODES,
 	}
 	gceDriver.AddControllerServiceCapabilities(csc)
 	ns := []csi.NodeServiceCapability_RPC_Type{

pkg/gce-pd-csi-driver/gce-pd-driver.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/test/remote"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	. "github.com/onsi/ginkgo"
@@ -466,6 +467,36 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle after restoring CMEK key")
 	})
 
+	It("Should create disks, attach them places, and verify List returns correct results", func() {
+		Expect(testContexts).ToNot(BeEmpty())
+		testContext := getRandomTestContext()
+
+		p, z, _ := testContext.Instance.GetIdentity()
+		client := testContext.Client
+
+		nodeID := testContext.Instance.GetNodeID()
+
+		volID := createUniqueVolumeOrError(client, p, z)
+		defer deleteVolumeOrError(client, volID, p)
+
+		secondVolID := createUniqueVolumeOrError(client, p, z)
+		defer deleteVolumeOrError(client, secondVolID, p)
+
+		// Attach volID to current instance
+		err := client.ControllerPublishVolume(volID, nodeID)
+		Expect(err).To(BeNil(), "Failed ControllerPublishVolume")
+		defer client.ControllerUnpublishVolume(volID, nodeID)
+
+		// List Volumes
+		volsToNodes, err := client.ListVolumes()
+		Expect(err).To(BeNil(), "Failed ListVolumes")
+
+		// Verify
+		Expect(volsToNodes[volID]).ToNot(BeNil(), "Couldn't find attached nodes for vol")
+		Expect(volsToNodes[volID]).To(ContainElement(nodeID), "Couldn't find node in attached nodes for vol")
+		Expect(volsToNodes[secondVolID]).To(BeNil(), "Second vol ID attached nodes not nil")
+	})
+
 	It("Should create and delete snapshot for RePD in two zones ", func() {
 		Expect(testContexts).ToNot(BeEmpty())
 		testContext := getRandomTestContext()
@@ -540,3 +571,32 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}()
 	})
 })
+
+func createUniqueVolumeOrError(client *remote.CsiClient, project, zone string) string {
+	// Create Disk
+	volName := testNamePrefix + string(uuid.NewUUID())
+	volID, err := client.CreateVolume(volName, nil, defaultSizeGb, nil)
+	Expect(err).To(BeNil(), "CreateVolume failed with error: %v", err)
+
+	// Validate Disk Created
+	cloudDisk, err := computeService.Disks.Get(project, zone, volName).Do()
+	Expect(err).To(BeNil(), "Could not get disk from cloud directly")
+	Expect(cloudDisk.Type).To(ContainSubstring(standardDiskType))
+	Expect(cloudDisk.Status).To(Equal(readyState))
+	Expect(cloudDisk.SizeGb).To(Equal(defaultSizeGb))
+	Expect(cloudDisk.Name).To(Equal(volName))
+
+	return volID
+}
+
+func deleteVolumeOrError(client *remote.CsiClient, volID, project string) {
+	// Delete Disk
+	err := client.DeleteVolume(volID)
+	Expect(err).To(BeNil(), "DeleteVolume failed")
+
+	// Validate Disk Deleted
+	key, err := common.VolumeIDToKey(volID)
+	Expect(err).To(BeNil(), "Failed to conver volume ID To key")
+	_, err = computeService.Disks.Get(project, key.Zone, key.Name).Do()
+	Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+}

test/e2e/tests/single_zone_e2e_test.go

@@ -138,6 +138,18 @@ func (c *CsiClient) ControllerPublishVolume(volId, nodeId string) error {
 	return err
 }
 
+func (c *CsiClient) ListVolumes() (map[string]([]string), error) {
+	resp, err := c.ctrlClient.ListVolumes(context.Background(), &csipb.ListVolumesRequest{})
+	if err != nil {
+		return nil, err
+	}
+	vols := map[string]([]string){}
+	for _, e := range resp.Entries {
+		vols[e.Volume.VolumeId] = e.Status.PublishedNodeIds
+	}
+	return vols, nil
+}
+
 func (c *CsiClient) ControllerUnpublishVolume(volId, nodeId string) error {
 	cupreq := &csipb.ControllerUnpublishVolumeRequest{
 		VolumeId: volId,

test/remote/client-wrappers.go

@@ -18,6 +18,7 @@ package remote
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -137,9 +138,9 @@ func (i *InstanceInfo) CreateOrGetInstance(serviceAccount string) error {
 		if err != nil {
 			ret := fmt.Sprintf("could not create instance %s: API error: %v", i.name, err)
 			if op != nil {
-				ret = fmt.Sprintf("%s: %v", ret, op.Error)
+				ret = fmt.Sprintf("%s. op error: %v", ret, op.Error)
 			}
-			return fmt.Errorf(ret)
+			return errors.New(ret)
 		} else if op.Error != nil {
 			return fmt.Errorf("could not create instance %s: %+v", i.name, op.Error)
 		}