csi: plugins track jobs in addition to allocations, and use job infor…

…mation to set expected counts (#8699)

* nomad/structs/csi: add explicit job support
* nomad/state/state_store: capture job updates directly
* api/nodes: CSIInfo needs the AllocID
* command/agent/csi_endpoint: AllocID was missing
Co-authored-by: Tim Gross <tgross@hashicorp.com>

api/nodes.go

@@ -568,6 +568,7 @@ type CSIControllerInfo struct {
 // as plugin health changes on the node.
 type CSIInfo struct {
 	PluginID                 string
+	AllocID                  string
 	Healthy                  bool
 	HealthDescription        string
 	UpdateTime               time.Time

command/agent/csi_endpoint.go

@@ -278,10 +278,10 @@ func structsCSIPluginToApi(plug *structs.CSIPlugin) *api.CSIPlugin {
 		Allocations:         make([]*api.AllocationListStub, 0, len(plug.Allocations)),
 		ControllerRequired:  plug.ControllerRequired,
 		ControllersHealthy:  plug.ControllersHealthy,
-		ControllersExpected: len(plug.Controllers),
+		ControllersExpected: plug.ControllersExpected,
 		Controllers:         make(map[string]*api.CSIInfo, len(plug.Controllers)),
 		NodesHealthy:        plug.NodesHealthy,
-		NodesExpected:       len(plug.Nodes),
+		NodesExpected:       plug.NodesExpected,
 		Nodes:               make(map[string]*api.CSIInfo, len(plug.Nodes)),
 		CreateIndex:         plug.CreateIndex,
 		ModifyIndex:         plug.ModifyIndex,
@@ -364,6 +364,7 @@ func structsCSIInfoToApi(info *structs.CSIInfo) *api.CSIInfo {
 	}
 	out := &api.CSIInfo{
 		PluginID:                 info.PluginID,
+		AllocID:                  info.AllocID,
 		Healthy:                  info.Healthy,
 		HealthDescription:        info.HealthDescription,
 		UpdateTime:               info.UpdateTime,

command/agent/csi_endpoint_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/hashicorp/nomad/nomad/mock"
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/structs"
-	"github.com/kr/pretty"
 	"github.com/stretchr/testify/require"
 )
 
@@ -33,11 +32,13 @@ func TestHTTP_CSIEndpointPlugin(t *testing.T) {
 		out, ok := obj.(*api.CSIPlugin)
 		require.True(t, ok)
 
-		require.Equal(t, 1, out.ControllersExpected)
+		// ControllersExpected is 0 because this plugin was created without a job,
+		// which sets expected
+		require.Equal(t, 0, out.ControllersExpected)
 		require.Equal(t, 1, out.ControllersHealthy)
 		require.Len(t, out.Controllers, 1)
 
-		require.Equal(t, 2, out.NodesExpected)
+		require.Equal(t, 0, out.NodesExpected)
 		require.Equal(t, 2, out.NodesHealthy)
 		require.Len(t, out.Nodes, 2)
 	})
@@ -92,11 +93,7 @@ func TestHTTP_CSIEndpointVolume(t *testing.T) {
 		out, ok := raw.(*api.CSIVolume)
 		require.True(t, ok)
 
-		pretty.Log(out)
-
-		require.Equal(t, 1, out.ControllersExpected)
 		require.Equal(t, 1, out.ControllersHealthy)
-		require.Equal(t, 2, out.NodesExpected)
 		require.Equal(t, 2, out.NodesHealthy)
 	})
 }

nomad/csi_endpoint_test.go

@@ -689,6 +689,83 @@ func TestCSIPluginEndpoint_RegisterViaFingerprint(t *testing.T) {
 	require.Nil(t, resp2.Plugin)
 }
 
+func TestCSIPluginEndpoint_RegisterViaJob(t *testing.T) {
+	t.Parallel()
+	srv, shutdown := TestServer(t, nil)
+	defer shutdown()
+	testutil.WaitForLeader(t, srv.RPC)
+
+	codec := rpcClient(t, srv)
+
+	// Register a job that creates the plugin
+	job := mock.Job()
+	job.TaskGroups[0].Tasks[0].CSIPluginConfig = &structs.TaskCSIPluginConfig{
+		ID:   "foo",
+		Type: structs.CSIPluginTypeNode,
+	}
+
+	req1 := &structs.JobRegisterRequest{
+		Job:          job,
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+	resp1 := &structs.JobRegisterResponse{}
+	err := msgpackrpc.CallWithCodec(codec, "Job.Register", req1, resp1)
+	require.NoError(t, err)
+
+	// Verify that the plugin exists and is unhealthy
+	req2 := &structs.CSIPluginGetRequest{
+		ID:           "foo",
+		QueryOptions: structs.QueryOptions{Region: "global"},
+	}
+	resp2 := &structs.CSIPluginGetResponse{}
+	err = msgpackrpc.CallWithCodec(codec, "CSIPlugin.Get", req2, resp2)
+	require.NoError(t, err)
+	require.NotNil(t, resp2.Plugin)
+	require.Zero(t, resp2.Plugin.ControllersHealthy)
+	require.Zero(t, resp2.Plugin.NodesHealthy)
+	require.Equal(t, job.ID, resp2.Plugin.NodeJobs[structs.DefaultNamespace][job.ID].ID)
+
+	// Health depends on node fingerprints
+	deleteNodes := state.CreateTestCSIPlugin(srv.fsm.State(), "foo")
+	defer deleteNodes()
+
+	resp2.Plugin = nil
+	err = msgpackrpc.CallWithCodec(codec, "CSIPlugin.Get", req2, resp2)
+	require.NoError(t, err)
+	require.NotNil(t, resp2.Plugin)
+	require.NotZero(t, resp2.Plugin.ControllersHealthy)
+	require.NotZero(t, resp2.Plugin.NodesHealthy)
+	require.Equal(t, job.ID, resp2.Plugin.NodeJobs[structs.DefaultNamespace][job.ID].ID)
+
+	// All fingerprints failing makes the plugin unhealthy, but does not delete it
+	deleteNodes()
+	err = msgpackrpc.CallWithCodec(codec, "CSIPlugin.Get", req2, resp2)
+	require.NoError(t, err)
+	require.NotNil(t, resp2.Plugin)
+	require.Zero(t, resp2.Plugin.ControllersHealthy)
+	require.Zero(t, resp2.Plugin.NodesHealthy)
+	require.Equal(t, job.ID, resp2.Plugin.NodeJobs[structs.DefaultNamespace][job.ID].ID)
+
+	// Job deregistration is necessary to gc the plugin
+	req3 := &structs.JobDeregisterRequest{
+		JobID: job.ID,
+		Purge: true,
+		WriteRequest: structs.WriteRequest{
+			Region:    "global",
+			Namespace: structs.DefaultNamespace,
+		},
+	}
+	resp3 := &structs.JobDeregisterResponse{}
+	err = msgpackrpc.CallWithCodec(codec, "Job.Deregister", req3, resp3)
+	require.NoError(t, err)
+
+	// Plugin has been gc'ed
+	resp2.Plugin = nil
+	err = msgpackrpc.CallWithCodec(codec, "CSIPlugin.Get", req2, resp2)
+	require.NoError(t, err)
+	require.Nil(t, resp2.Plugin)
+}
+
 func TestCSIPluginEndpoint_DeleteViaGC(t *testing.T) {
 	t.Parallel()
 	srv, shutdown := TestServer(t, func(c *Config) {