csi: node unmount from the client before unpublish RPC

When an allocation stops, the `csi_hook` makes an unpublish RPC to the
servers to unpublish via the CSI RPCs: first to the node plugins and
then the controller plugins. The controller RPCs must happen after the
node RPCs so that the node has had a chance to unmount the volume
before the controller tries to detach the associated device.

But the client has local access to the node plugins and can
independently determine if it's safe to send unpublish RPC to those
plugins. This will allow the server to treat the node plugin as
abandoned if a client is disconnected and `stop_on_client_disconnect`
is set. This will let the server try to send unpublish RPCs to the
controller plugins, under the assumption that the client will be
trying to unmount the volume on its end first.

client/allocrunner/csi_hook.go

@@ -107,6 +107,11 @@ func (c *csiHook) Postrun() error {
 
 	for _, pair := range c.volumeRequests {
 
+		err := c.unmount(pair)
+		if err != nil {
+			mErr = multierror.Append(mErr, err)
+		}
+
 		mode := structs.CSIVolumeClaimRead
 		if !pair.request.ReadOnly {
 			mode = structs.CSIVolumeClaimWrite
@@ -132,7 +137,7 @@ func (c *csiHook) Postrun() error {
 				AuthToken: c.nodeSecret,
 			},
 		}
-		err := c.rpcClient.RPC("CSIVolume.Unpublish",
+		err = c.rpcClient.RPC("CSIVolume.Unpublish",
 			req, &structs.CSIVolumeUnpublishResponse{})
 		if err != nil {
 			mErr = multierror.Append(mErr, err)
@@ -231,3 +236,25 @@ func (c *csiHook) shouldRun() bool {
 
 	return false
 }
+
+func (c *csiHook) unmount(pair *volumeAndRequest) error {
+
+	mounter, err := c.csimanager.MounterForPlugin(context.TODO(), pair.volume.PluginID)
+	if err != nil {
+		return err
+	}
+
+	usageOpts := &csimanager.UsageOptions{
+		ReadOnly:       pair.request.ReadOnly,
+		AttachmentMode: pair.request.AttachmentMode,
+		AccessMode:     pair.request.AccessMode,
+		MountOptions:   pair.request.MountOptions,
+	}
+
+	err = mounter.UnmountVolume(context.TODO(),
+		pair.volume.ID, pair.volume.RemoteID(), c.alloc.ID, usageOpts)
+	if err != nil {
+		return err
+	}
+	return nil
+}

client/allocrunner/csi_hook_test.go

@@ -59,7 +59,7 @@ func TestCSIHook(t *testing.T) {
 					"test-alloc-dir/%s/testvolume0/ro-file-system-single-node-reader-only", alloc.ID)},
 			},
 			expectedMountCalls:     1,
-			expectedUnmountCalls:   0, // not until this is done client-side
+			expectedUnmountCalls:   1,
 			expectedClaimCalls:     1,
 			expectedUnpublishCalls: 1,
 		},
@@ -83,7 +83,7 @@ func TestCSIHook(t *testing.T) {
 					"test-alloc-dir/%s/testvolume0/ro-file-system-single-node-reader-only", alloc.ID)},
 			},
 			expectedMountCalls:     1,
-			expectedUnmountCalls:   0, // not until this is done client-side
+			expectedUnmountCalls:   1,
 			expectedClaimCalls:     1,
 			expectedUnpublishCalls: 1,
 		},
@@ -122,7 +122,7 @@ func TestCSIHook(t *testing.T) {
 		// 			"test-alloc-dir/%s/testvolume0/ro-file-system-multi-node-reader-only", alloc.ID)},
 		// 	},
 		// 	expectedMountCalls:     1,
-		// 	expectedUnmountCalls:   0, // not until this is done client-side
+		// 	expectedUnmountCalls:   1,
 		// 	expectedClaimCalls:     1,
 		// 	expectedUnpublishCalls: 1,
 		// },