csi: check returned volume capability validation (#7831)

This changeset corrects handling of the `ValidationVolumeCapabilities`
response:

* The CSI spec for the `ValidationVolumeCapabilities` requires that
  plugins only set the `Confirmed` field if they've validated all
  capabilities. The Nomad client improperly assumes that the lack of a
  `Confirmed` field should be treated as a failure. This breaks the
  Azure and Linode block storage plugins, which don't set this
  optional field.

* The CSI spec also requires that the orchestrator check the validation
  responses to guard against older versions of a plugin reporting
  "valid" for newer fields it doesn't understand.

plugins/csi/client.go

@@ -82,6 +82,7 @@ type client struct {
 	identityClient   csipbv1.IdentityClient
 	controllerClient CSIControllerClient
 	nodeClient       CSINodeClient
+	logger           hclog.Logger
 }
 
 func (c *client) Close() error {
@@ -106,6 +107,7 @@ func NewClient(addr string, logger hclog.Logger) (CSIPlugin, error) {
 		identityClient:   csipbv1.NewIdentityClient(conn),
 		controllerClient: csipbv1.NewControllerClient(conn),
 		nodeClient:       csipbv1.NewNodeClient(conn),
+		logger:           logger,
 	}, nil
 }
 
@@ -318,17 +320,50 @@ func (c *client) ControllerValidateCapabilities(ctx context.Context, volumeID st
 		return err
 	}
 
-	if resp.Confirmed == nil {
-		if resp.Message != "" {
-			return fmt.Errorf("Volume validation failed, message: %s", resp.Message)
-		}
+	if resp.Message != "" {
+		// this should only ever be set if Confirmed isn't set, but
+		// it's not a validation failure.
+		c.logger.Debug(resp.Message)
+	}
 
-		return fmt.Errorf("Volume validation failed")
+	// The protobuf accessors below safely handle nil pointers.
+	// The CSI spec says we can only assert the plugin has
+	// confirmed the volume capabilities, not that it hasn't
+	// confirmed them, so if the field is nil we have to assume
+	// the volume is ok.
+	confirmedCaps := resp.GetConfirmed().GetVolumeCapabilities()
+	if confirmedCaps != nil {
+		for _, requestedCap := range req.VolumeCapabilities {
+			if !compareCapabilities(requestedCap, confirmedCaps) {
+				return fmt.Errorf("volume capability validation failed: missing %v", req)
+			}
+		}
 	}
 
 	return nil
 }
 
+// compareCapabilities returns true if the 'got' capabilities contains
+// the 'expected' capability
+func compareCapabilities(expected *csipbv1.VolumeCapability, got []*csipbv1.VolumeCapability) bool {
+	for _, cap := range got {
+		if expected.GetAccessMode().GetMode() != cap.GetAccessMode().GetMode() {
+			continue
+		}
+		// AccessType Block is an empty struct even if set, so the
+		// only way to test for it is to check that the AccessType
+		// isn't Mount.
+		if expected.GetMount() == nil && cap.GetMount() != nil {
+			continue
+		}
+		if expected.GetMount() != cap.GetMount() {
+			continue
+		}
+		return true
+	}
+	return false
+}
+
 //
 // Node Endpoints
 //

plugins/csi/client_test.go

@@ -8,6 +8,7 @@ import (
 
 	csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/golang/protobuf/ptypes/wrappers"
+	"github.com/hashicorp/nomad/nomad/structs"
 	fake "github.com/hashicorp/nomad/plugins/csi/testing"
 	"github.com/stretchr/testify/require"
 )
@@ -473,6 +474,95 @@ func TestClient_RPC_ControllerUnpublishVolume(t *testing.T) {
 	}
 }
 
+func TestClient_RPC_ControllerValidateVolume(t *testing.T) {
+
+	cases := []struct {
+		Name        string
+		ResponseErr error
+		Response    *csipbv1.ValidateVolumeCapabilitiesResponse
+		ExpectedErr error
+	}{
+		{
+			Name:        "handles underlying grpc errors",
+			ResponseErr: fmt.Errorf("some grpc error"),
+			ExpectedErr: fmt.Errorf("some grpc error"),
+		},
+		{
+			Name:        "handles empty success",
+			Response:    &csipbv1.ValidateVolumeCapabilitiesResponse{},
+			ResponseErr: nil,
+			ExpectedErr: nil,
+		},
+		{
+			Name: "handles validate success",
+			Response: &csipbv1.ValidateVolumeCapabilitiesResponse{
+				Confirmed: &csipbv1.ValidateVolumeCapabilitiesResponse_Confirmed{
+					VolumeContext: map[string]string{},
+					VolumeCapabilities: []*csipbv1.VolumeCapability{
+						{
+							AccessType: &csipbv1.VolumeCapability_Block{
+								Block: &csipbv1.VolumeCapability_BlockVolume{},
+							},
+							AccessMode: &csipbv1.VolumeCapability_AccessMode{
+								Mode: csipbv1.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
+							},
+						},
+					},
+				},
+			},
+			ResponseErr: nil,
+			ExpectedErr: nil,
+		},
+		{
+			Name: "handles validation failure",
+			Response: &csipbv1.ValidateVolumeCapabilitiesResponse{
+				Confirmed: &csipbv1.ValidateVolumeCapabilitiesResponse_Confirmed{
+					VolumeContext: map[string]string{},
+					VolumeCapabilities: []*csipbv1.VolumeCapability{
+						{
+							AccessType: &csipbv1.VolumeCapability_Block{
+								Block: &csipbv1.VolumeCapability_BlockVolume{},
+							},
+							AccessMode: &csipbv1.VolumeCapability_AccessMode{
+								Mode: csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+							},
+						},
+					},
+				},
+			},
+			ResponseErr: nil,
+			ExpectedErr: fmt.Errorf("volume capability validation failed"),
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.Name, func(t *testing.T) {
+			_, cc, _, client := newTestClient()
+			defer client.Close()
+
+			requestedCaps := &VolumeCapability{
+				AccessType: VolumeAccessTypeBlock,
+				AccessMode: VolumeAccessModeMultiNodeMultiWriter,
+				MountVolume: &structs.CSIMountOptions{ // should be ignored
+					FSType:     "ext4",
+					MountFlags: []string{"noatime", "errors=remount-ro"},
+				},
+			}
+			cc.NextValidateVolumeCapabilitiesResponse = c.Response
+			cc.NextErr = c.ResponseErr
+
+			err := client.ControllerValidateCapabilities(
+				context.TODO(), "volumeID", requestedCaps)
+			if c.ExpectedErr != nil {
+				require.Error(t, c.ExpectedErr, err, c.Name)
+			} else {
+				require.NoError(t, err, c.Name)
+			}
+		})
+	}
+
+}
+
 func TestClient_RPC_NodeStageVolume(t *testing.T) {
 	cases := []struct {
 		Name        string

plugins/csi/testing/client.go

@@ -44,10 +44,11 @@ func (f *IdentityClient) Probe(ctx context.Context, in *csipbv1.ProbeRequest, op
 
 // ControllerClient is a CSI controller client used for testing
 type ControllerClient struct {
-	NextErr                     error
-	NextCapabilitiesResponse    *csipbv1.ControllerGetCapabilitiesResponse
-	NextPublishVolumeResponse   *csipbv1.ControllerPublishVolumeResponse
-	NextUnpublishVolumeResponse *csipbv1.ControllerUnpublishVolumeResponse
+	NextErr                                error
+	NextCapabilitiesResponse               *csipbv1.ControllerGetCapabilitiesResponse
+	NextPublishVolumeResponse              *csipbv1.ControllerPublishVolumeResponse
+	NextUnpublishVolumeResponse            *csipbv1.ControllerUnpublishVolumeResponse
+	NextValidateVolumeCapabilitiesResponse *csipbv1.ValidateVolumeCapabilitiesResponse
 }
 
 // NewControllerClient returns a new ControllerClient
@@ -60,6 +61,7 @@ func (f *ControllerClient) Reset() {
 	f.NextCapabilitiesResponse = nil
 	f.NextPublishVolumeResponse = nil
 	f.NextUnpublishVolumeResponse = nil
+	f.NextValidateVolumeCapabilitiesResponse = nil
 }
 
 func (c *ControllerClient) ControllerGetCapabilities(ctx context.Context, in *csipbv1.ControllerGetCapabilitiesRequest, opts ...grpc.CallOption) (*csipbv1.ControllerGetCapabilitiesResponse, error) {
@@ -75,7 +77,7 @@ func (c *ControllerClient) ControllerUnpublishVolume(ctx context.Context, in *cs
 }
 
 func (c *ControllerClient) ValidateVolumeCapabilities(ctx context.Context, in *csipbv1.ValidateVolumeCapabilitiesRequest, opts ...grpc.CallOption) (*csipbv1.ValidateVolumeCapabilitiesResponse, error) {
-	panic("not implemented") // TODO: Implement
+	return c.NextValidateVolumeCapabilitiesResponse, c.NextErr
 }
 
 // NodeClient is a CSI Node client used for testing