Wrap all GRPC errors in status, fix semantics of NotFound errors

Author: davidz627

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

@@ -17,13 +17,14 @@ package gcecloudprovider
 import (
 	"context"
 	"fmt"
-	"golang.org/x/oauth2/google"
-	"gopkg.in/gcfg.v1"
 	"net/http"
 	"os"
 	"runtime"
 	"time"
 
+	"golang.org/x/oauth2/google"
+	"gopkg.in/gcfg.v1"
+
 	"cloud.google.com/go/compute/metadata"
 	"golang.org/x/oauth2"
 	beta "google.golang.org/api/compute/v0.beta"
@@ -242,3 +243,9 @@ func IsGCEError(err error, reason string) bool {
 	}
 	return false
 }
+
+// IsGCENotFoundError returns true if the error is a googleapi.Error with
+// notFound reason
+func IsGCENotFoundError(err error) bool {
+	return IsGCEError(err, "notFound")
+}

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

@@ -224,14 +224,14 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 
 	volKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		// Cannot find volume associated with this ID because can't even get the name or zone
-		// This is a success according to the spec
+		klog.Warningf("Treating volume as deleted because volume id %s is invalid: %v", volumeID, err)
 		return &csi.DeleteVolumeResponse{}, nil
 	}
 
 	volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, volKey)
 	if err != nil {
-		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
+		klog.Warningf("Treating volume as deleted because cannot find volume %v: %v", volKey.String(), err)
+		return &csi.DeleteVolumeResponse{}, nil
 	}
 
 	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
@@ -267,7 +267,7 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 
 	volKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerPublishVolume volume ID is invalid: %v", err))
 	}
 
 	volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, volKey)
@@ -294,7 +294,7 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 
 	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey)
 	if err != nil {
-		if gce.IsGCEError(err, "notFound") {
+		if gce.IsGCENotFoundError(err) {
 			return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find disk %v: %v", volKey.String(), err))
 		}
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown get disk error: %v", err))
@@ -305,7 +305,7 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 	}
 	instance, err := gceCS.CloudProvider.GetInstanceOrError(ctx, instanceZone, instanceName)
 	if err != nil {
-		if gce.IsGCEError(err, "notFound") {
+		if gce.IsGCENotFoundError(err) {
 			return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find instance %v: %v", nodeID, err))
 		}
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown get instance error: %v", err))
@@ -365,7 +365,7 @@ func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context,
 
 	volKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		return nil, err
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerUnpublishVolume Volume ID is invalid: %v", err))
 	}
 
 	// Acquires the lock for the volume on that node only, because we need to support the ability
@@ -382,7 +382,12 @@ func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context,
 	}
 	instance, err := gceCS.CloudProvider.GetInstanceOrError(ctx, instanceZone, instanceName)
 	if err != nil {
-		return nil, err
+		if gce.IsGCENotFoundError(err) {
+			// Node not existing on GCE means that disk has been detached
+			klog.Warningf("Treating volume %v as unpublished because node %v could not be found", volKey.String(), instanceName)
+			return &csi.ControllerUnpublishVolumeResponse{}, nil
+		}
+		return nil, status.Error(codes.Internal, fmt.Sprintf("error getting instance: %v", err))
 	}
 
 	deviceName, err := common.GetDeviceName(volKey)
@@ -420,7 +425,7 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 	}
 	volKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		return nil, status.Error(codes.NotFound, fmt.Sprintf("Volume ID is of improper format, got %v", volumeID))
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ValidateVolumeCapabilities Volume ID is invalid: %v", err))
 	}
 
 	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
@@ -430,7 +435,7 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 
 	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey)
 	if err != nil {
-		if gce.IsGCEError(err, "notFound") {
+		if gce.IsGCENotFoundError(err) {
 			return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find disk %v: %v", volKey.Name, err))
 		}
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown get disk error: %v", err))
@@ -532,7 +537,7 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 	}
 	volKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("CreateSnapshot Volume ID is invalid: %v", err))
 	}
 
 	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
@@ -670,7 +675,7 @@ func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, re
 
 	volKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerExpandVolume volume ID is invalid: %v", err))
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerExpandVolume Volume ID is invalid: %v", err))
 	}
 
 	resizedGb, err := gceCS.CloudProvider.ResizeDisk(ctx, volKey, reqBytes)

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

@@ -90,8 +90,7 @@ func (ns *GCENodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePub
 
 	notMnt, err := ns.Mounter.Interface.IsLikelyNotMountPoint(targetPath)
 	if err != nil && !os.IsNotExist(err) {
-		klog.Errorf("cannot validate mount point: %s %v", targetPath, err)
-		return nil, err
+		return nil, status.Error(codes.Internal, fmt.Sprintf("cannot validate mount point: %s %v", targetPath, err))
 	}
 	if !notMnt {
 		// TODO(#95): check if mount is compatible. Return OK if it is, or appropriate error.
@@ -129,8 +128,7 @@ func (ns *GCENodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePub
 		sourcePath = stagingTargetPath
 
 		if err := ns.Mounter.Interface.MakeDir(targetPath); err != nil {
-			klog.Errorf("mkdir failed on disk %s (%v)", targetPath, err)
-			return nil, err
+			return nil, status.Error(codes.Internal, fmt.Sprintf("mkdir failed on disk %s (%v)", targetPath, err))
 		}
 	} else if blk := volumeCapability.GetBlock(); blk != nil {
 		klog.V(4).Infof("NodePublishVolume with block volume mode")
@@ -245,7 +243,7 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 
 	volumeKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		return nil, err
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("NodeStageVolume Volume ID is invalid: %v", err))
 	}
 
 	// Part 1: Get device path of attached device
@@ -387,7 +385,7 @@ func (ns *GCENodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpa
 
 	volKey, err := common.VolumeIDToKey(volumeID)
 	if err != nil {
-		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerExpandVolume volume ID is invalid: %v", err))
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerExpandVolume Volume ID is invalid: %v", err))
 	}
 
 	devicePath, err := ns.getDevicePath(volumeID, "")