event stream: add events for CSI volumes and plugins (#24724)

Adds new topics to the event stream for CSI volumes and CSI plugins. We'll emit
event when either is created or deleted, and when CSI volumes are claimed.

.changelog/24724.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+csi: Added CSI volume and plugin events to the event stream
+```

nomad/state/events.go

@@ -43,6 +43,9 @@ var MsgTypeEvents = map[structs.MessageType]string{
 	structs.ServiceRegistrationDeleteByNodeIDRequestType: structs.TypeServiceDeregistration,
 	structs.HostVolumeRegisterRequestType:                structs.TypeHostVolumeRegistered,
 	structs.HostVolumeDeleteRequestType:                  structs.TypeHostVolumeDeleted,
+	structs.CSIVolumeRegisterRequestType:                 structs.TypeCSIVolumeRegistered,
+	structs.CSIVolumeDeregisterRequestType:               structs.TypeCSIVolumeDeregistered,
+	structs.CSIVolumeClaimRequestType:                    structs.TypeCSIVolumeClaim,
 }
 
 func eventsFromChanges(tx ReadTxn, changes Changes) *structs.Events {
@@ -190,7 +193,6 @@ func eventFromChange(change memdb.Change) (structs.Event, bool) {
 			}
 			return structs.Event{
 				Topic: structs.TopicHostVolume,
-				Key:   before.ID,
 				FilterKeys: []string{
 					before.ID,
 					before.Name,
@@ -201,6 +203,40 @@ func eventFromChange(change memdb.Change) (structs.Event, bool) {
 					Volume: before,
 				},
 			}, true
+		case TableCSIVolumes:
+			before, ok := change.Before.(*structs.CSIVolume)
+			if !ok {
+				return structs.Event{}, false
+			}
+			return structs.Event{
+				Topic: structs.TopicCSIVolume,
+				Key:   before.ID,
+				FilterKeys: []string{
+					before.ID,
+					before.Name,
+					before.PluginID,
+				},
+				Namespace: before.Namespace,
+				Payload: &structs.CSIVolumeEvent{
+					Volume: before,
+				},
+			}, true
+		case TableCSIPlugins:
+			// note: there is no CSIPlugin event type, because CSI plugins don't
+			// have their own write RPCs; they are always created/removed via
+			// node updates
+			before, ok := change.Before.(*structs.CSIPlugin)
+			if !ok {
+				return structs.Event{}, false
+			}
+			return structs.Event{
+				Topic:      structs.TopicCSIPlugin,
+				Key:        before.ID,
+				FilterKeys: []string{before.ID},
+				Payload: &structs.CSIPluginEvent{
+					Plugin: before,
+				},
+			}, true
 		}
 		return structs.Event{}, false
 	}
@@ -396,6 +432,40 @@ func eventFromChange(change memdb.Change) (structs.Event, bool) {
 				Volume: after,
 			},
 		}, true
+	case TableCSIVolumes:
+		after, ok := change.After.(*structs.CSIVolume)
+		if !ok {
+			return structs.Event{}, false
+		}
+		return structs.Event{
+			Topic: structs.TopicCSIVolume,
+			Key:   after.ID,
+			FilterKeys: []string{
+				after.ID,
+				after.Name,
+				after.PluginID,
+			},
+			Namespace: after.Namespace,
+			Payload: &structs.CSIVolumeEvent{
+				Volume: after,
+			},
+		}, true
+	case TableCSIPlugins:
+		// note: there is no CSIPlugin event type, because CSI plugins don't
+		// have their own write RPCs; they are always created/removed via
+		// node updates
+		after, ok := change.After.(*structs.CSIPlugin)
+		if !ok {
+			return structs.Event{}, false
+		}
+		return structs.Event{
+			Topic:      structs.TopicCSIPlugin,
+			Key:        after.ID,
+			FilterKeys: []string{after.ID},
+			Payload: &structs.CSIPluginEvent{
+				Plugin: after,
+			},
+		}, true
 	}
 
 	return structs.Event{}, false

nomad/state/events_test.go

@@ -1216,7 +1216,6 @@ func Test_eventsFromChanges_ACLBindingRule(t *testing.T) {
 }
 
 func TestEvents_HostVolumes(t *testing.T) {
-
 	ci.Parallel(t)
 	store := TestStateStoreCfg(t, TestStateStorePublisher(t))
 	defer store.StopEventBroker()
@@ -1258,6 +1257,84 @@ func TestEvents_HostVolumes(t *testing.T) {
 	must.Eq(t, "HostVolumeDeleted", events[4].Type)
 }
 
+func TestEvents_CSIVolumes(t *testing.T) {
+	ci.Parallel(t)
+	store := TestStateStoreCfg(t, TestStateStorePublisher(t))
+	defer store.StopEventBroker()
+
+	index, err := store.LatestIndex()
+	must.NoError(t, err)
+
+	plugin := mock.CSIPlugin()
+	vol := mock.CSIVolume(plugin)
+
+	index++
+	must.NoError(t, store.UpsertCSIVolume(index, []*structs.CSIVolume{vol}))
+
+	alloc := mock.Alloc()
+	index++
+	store.UpsertAllocs(structs.MsgTypeTestSetup, index, []*structs.Allocation{alloc})
+
+	claim := &structs.CSIVolumeClaim{
+		AllocationID:   alloc.ID,
+		NodeID:         uuid.Generate(),
+		Mode:           structs.CSIVolumeClaimGC,
+		AccessMode:     structs.CSIVolumeAccessModeSingleNodeWriter,
+		AttachmentMode: structs.CSIVolumeAttachmentModeFilesystem,
+		State:          structs.CSIVolumeClaimStateReadyToFree,
+	}
+	index++
+	must.NoError(t, store.CSIVolumeClaim(index, time.Now().UnixNano(), vol.Namespace, vol.ID, claim))
+
+	index++
+	must.NoError(t, store.CSIVolumeDeregister(index, vol.Namespace, []string{vol.ID}, false))
+
+	events := WaitForEvents(t, store, 0, 3, 1*time.Second)
+	must.Len(t, 3, events)
+	must.Eq(t, "CSIVolume", events[0].Topic)
+	must.Eq(t, "CSIVolumeRegistered", events[0].Type)
+	must.Eq(t, "CSIVolume", events[1].Topic)
+	must.Eq(t, "CSIVolumeClaim", events[1].Type)
+	must.Eq(t, "CSIVolume", events[2].Topic)
+	must.Eq(t, "CSIVolumeDeregistered", events[2].Type)
+
+}
+
+func TestEvents_CSIPlugins(t *testing.T) {
+	ci.Parallel(t)
+	store := TestStateStoreCfg(t, TestStateStorePublisher(t))
+	defer store.StopEventBroker()
+
+	index, err := store.LatestIndex()
+	must.NoError(t, err)
+
+	node := mock.Node()
+	plugin := mock.CSIPlugin()
+
+	index++
+	must.NoError(t, store.UpsertNode(structs.NodeRegisterRequestType, index, node))
+
+	node = node.Copy()
+	node.CSINodePlugins = map[string]*structs.CSIInfo{
+		plugin.ID: {
+			PluginID:   plugin.ID,
+			Healthy:    true,
+			UpdateTime: time.Now(),
+		},
+	}
+	index++
+	must.NoError(t, store.UpsertNode(structs.NodeRegisterRequestType, index, node))
+
+	events := WaitForEvents(t, store, 0, 3, 1*time.Second)
+	must.Len(t, 3, events)
+	must.Eq(t, "Node", events[0].Topic)
+	must.Eq(t, "NodeRegistration", events[0].Type)
+	must.Eq(t, "Node", events[1].Topic)
+	must.Eq(t, "NodeRegistration", events[1].Type)
+	must.Eq(t, "CSIPlugin", events[2].Topic)
+	must.Eq(t, "NodeRegistration", events[2].Type)
+}
+
 func requireNodeRegistrationEventEqual(t *testing.T, want, got structs.Event) {
 	t.Helper()
 

nomad/state/schema.go

@@ -27,6 +27,8 @@ const (
 	TableAllocs               = "allocs"
 	TableJobSubmission        = "job_submission"
 	TableHostVolumes          = "host_volumes"
+	TableCSIVolumes           = "csi_volumes"
+	TableCSIPlugins           = "csi_plugins"
 )
 
 const (
@@ -1150,7 +1152,7 @@ func clusterMetaTableSchema() *memdb.TableSchema {
 // CSIVolumes are identified by id globally, and searchable by driver
 func csiVolumeTableSchema() *memdb.TableSchema {
 	return &memdb.TableSchema{
-		Name: "csi_volumes",
+		Name: TableCSIVolumes,
 		Indexes: map[string]*memdb.IndexSchema{
 			"id": {
 				Name:         "id",
@@ -1182,7 +1184,7 @@ func csiVolumeTableSchema() *memdb.TableSchema {
 // CSIPlugins are identified by id globally, and searchable by driver
 func csiPluginTableSchema() *memdb.TableSchema {
 	return &memdb.TableSchema{
-		Name: "csi_plugins",
+		Name: TableCSIPlugins,
 		Indexes: map[string]*memdb.IndexSchema{
 			"id": {
 				Name:         "id",