From faffbf240450751e962259e7efcc18eb421e9126 Mon Sep 17 00:00:00 2001 From: Tim Gross Date: Thu, 19 Dec 2024 13:49:09 -0500 Subject: [PATCH] event stream: add events for CSI volumes and plugins 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 | 3 ++ nomad/state/events.go | 70 +++++++++++++++++++++++++ nomad/state/events_test.go | 25 +++++++++ nomad/state/schema.go | 6 ++- nomad/state/state_store.go | 72 +++++++++++++------------- nomad/state/state_store_restore.go | 4 +- nomad/stream/event_broker.go | 8 +++ nomad/structs/event.go | 16 ++++++ nomad/volumewatcher/volumes_watcher.go | 6 +-- website/content/api-docs/events.mdx | 69 +++++++++++++----------- 10 files changed, 205 insertions(+), 74 deletions(-) create mode 100644 .changelog/24724.txt diff --git a/.changelog/24724.txt b/.changelog/24724.txt new file mode 100644 index 00000000000..20775747c80 --- /dev/null +++ b/.changelog/24724.txt @@ -0,0 +1,3 @@ +```release-note:improvement +csi: Added CSI volume and plugin events to the event stream +``` diff --git a/nomad/state/events.go b/nomad/state/events.go index c338133c418..ed32648a82f 100644 --- a/nomad/state/events.go +++ b/nomad/state/events.go @@ -41,6 +41,8 @@ var MsgTypeEvents = map[structs.MessageType]string{ structs.ServiceRegistrationUpsertRequestType: structs.TypeServiceRegistration, structs.ServiceRegistrationDeleteByIDRequestType: structs.TypeServiceDeregistration, structs.ServiceRegistrationDeleteByNodeIDRequestType: structs.TypeServiceDeregistration, + structs.CSIVolumeRegisterRequestType: structs.TypeCSIVolumeRegistered, + structs.CSIVolumeDeregisterRequestType: structs.TypeCSIVolumeDeregistered, } func eventsFromChanges(tx ReadTxn, changes Changes) *structs.Events { @@ -181,6 +183,40 @@ func eventFromChange(change memdb.Change) (structs.Event, bool) { Service: 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 } @@ -358,6 +394,40 @@ func eventFromChange(change memdb.Change) (structs.Event, bool) { Service: 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 diff --git a/nomad/state/events_test.go b/nomad/state/events_test.go index 8e15e27fb9d..588cde9ba16 100644 --- a/nomad/state/events_test.go +++ b/nomad/state/events_test.go @@ -1215,6 +1215,31 @@ func Test_eventsFromChanges_ACLBindingRule(t *testing.T) { must.Eq(t, bindingRule, receivedDeleteChange.Events[0].Payload.(*structs.ACLBindingRuleEvent).ACLBindingRule) } +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})) + + index++ + must.NoError(t, store.CSIVolumeDeregister(index, vol.Namespace, []string{vol.ID}, false)) + + events := WaitForEvents(t, store, 0, 2, 1*time.Second) + must.Len(t, 2, 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, "CSIVolumeDeregistered", events[1].Type) +} + func requireNodeRegistrationEventEqual(t *testing.T, want, got structs.Event) { t.Helper() diff --git a/nomad/state/schema.go b/nomad/state/schema.go index 7ee83e074eb..0aae6c1fcf9 100644 --- a/nomad/state/schema.go +++ b/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", diff --git a/nomad/state/state_store.go b/nomad/state/state_store.go index 65ce87813db..bf2aca6ed0b 100644 --- a/nomad/state/state_store.go +++ b/nomad/state/state_store.go @@ -1398,7 +1398,7 @@ func appendNodeEvents(index uint64, node *structs.Node, events []*structs.NodeEv func upsertCSIPluginsForNode(txn *txn, node *structs.Node, index uint64) error { upsertFn := func(info *structs.CSIInfo) error { - raw, err := txn.First("csi_plugins", "id", info.PluginID) + raw, err := txn.First(TableCSIPlugins, "id", info.PluginID) if err != nil { return fmt.Errorf("csi_plugin lookup error: %s %v", info.PluginID, err) } @@ -1429,7 +1429,7 @@ func upsertCSIPluginsForNode(txn *txn, node *structs.Node, index uint64) error { plug.ModifyIndex = index - err = txn.Insert("csi_plugins", plug) + err = txn.Insert(TableCSIPlugins, plug) if err != nil { return fmt.Errorf("csi_plugins insert error: %v", err) } @@ -1458,7 +1458,7 @@ func upsertCSIPluginsForNode(txn *txn, node *structs.Node, index uint64) error { // remove the client node from any plugin that's not // running on it. - iter, err := txn.Get("csi_plugins", "id") + iter, err := txn.Get(TableCSIPlugins, "id") if err != nil { return fmt.Errorf("csi_plugins lookup failed: %v", err) } @@ -1503,7 +1503,7 @@ func upsertCSIPluginsForNode(txn *txn, node *structs.Node, index uint64) error { } } - if err := txn.Insert("index", &IndexEntry{"csi_plugins", index}); err != nil { + if err := txn.Insert("index", &IndexEntry{TableCSIPlugins, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } @@ -1525,7 +1525,7 @@ func deleteNodeCSIPlugins(txn *txn, node *structs.Node, index uint64) error { } for id := range names { - raw, err := txn.First("csi_plugins", "id", id) + raw, err := txn.First(TableCSIPlugins, "id", id) if err != nil { return fmt.Errorf("csi_plugins lookup error %s: %v", id, err) } @@ -1546,7 +1546,7 @@ func deleteNodeCSIPlugins(txn *txn, node *structs.Node, index uint64) error { } } - if err := txn.Insert("index", &IndexEntry{"csi_plugins", index}); err != nil { + if err := txn.Insert("index", &IndexEntry{TableCSIPlugins, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } @@ -1556,13 +1556,13 @@ func deleteNodeCSIPlugins(txn *txn, node *structs.Node, index uint64) error { // updateOrGCPlugin updates a plugin but will delete it if the plugin is empty func updateOrGCPlugin(index uint64, txn Txn, plug *structs.CSIPlugin) error { if plug.IsEmpty() { - err := txn.Delete("csi_plugins", plug) + err := txn.Delete(TableCSIPlugins, plug) if err != nil { return fmt.Errorf("csi_plugins delete error: %v", err) } } else { plug.ModifyIndex = index - err := txn.Insert("csi_plugins", plug) + err := txn.Insert(TableCSIPlugins, plug) if err != nil { return fmt.Errorf("csi_plugins update error %s: %v", plug.ID, err) } @@ -1661,7 +1661,7 @@ func (s *StateStore) deleteJobFromPlugins(index uint64, txn Txn, job *structs.Jo } if len(plugins) > 0 { - if err = txn.Insert("index", &IndexEntry{"csi_plugins", index}); err != nil { + if err = txn.Insert("index", &IndexEntry{TableCSIPlugins, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } } @@ -2561,7 +2561,7 @@ func (s *StateStore) JobSummaryByPrefix(ws memdb.WatchSet, namespace, id string) // UpsertCSIVolume inserts a volume in the state store. func (s *StateStore) UpsertCSIVolume(index uint64, volumes []*structs.CSIVolume) error { - txn := s.db.WriteTxn(index) + txn := s.db.WriteTxnMsgT(structs.CSIVolumeRegisterRequestType, index) defer txn.Abort() for _, v := range volumes { @@ -2571,7 +2571,7 @@ func (s *StateStore) UpsertCSIVolume(index uint64, volumes []*structs.CSIVolume) return fmt.Errorf("volume %s is in nonexistent namespace %s", v.ID, v.Namespace) } - obj, err := txn.First("csi_volumes", "id", v.Namespace, v.ID) + obj, err := txn.First(TableCSIVolumes, "id", v.Namespace, v.ID) if err != nil { return fmt.Errorf("volume existence check error: %v", err) } @@ -2600,13 +2600,13 @@ func (s *StateStore) UpsertCSIVolume(index uint64, volumes []*structs.CSIVolume) v.WriteAllocs[allocID] = nil } - err = txn.Insert("csi_volumes", v) + err = txn.Insert(TableCSIVolumes, v) if err != nil { return fmt.Errorf("volume insert: %v", err) } } - if err := txn.Insert("index", &IndexEntry{"csi_volumes", index}); err != nil { + if err := txn.Insert("index", &IndexEntry{TableCSIVolumes, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } @@ -2619,7 +2619,7 @@ func (s *StateStore) CSIVolumes(ws memdb.WatchSet) (memdb.ResultIterator, error) txn := s.db.ReadTxn() defer txn.Abort() - iter, err := txn.Get("csi_volumes", "id") + iter, err := txn.Get(TableCSIVolumes, "id") if err != nil { return nil, fmt.Errorf("csi_volumes lookup failed: %v", err) } @@ -2635,7 +2635,7 @@ func (s *StateStore) CSIVolumes(ws memdb.WatchSet) (memdb.ResultIterator, error) func (s *StateStore) CSIVolumeByID(ws memdb.WatchSet, namespace, id string) (*structs.CSIVolume, error) { txn := s.db.ReadTxn() - watchCh, obj, err := txn.FirstWatch("csi_volumes", "id", namespace, id) + watchCh, obj, err := txn.FirstWatch(TableCSIVolumes, "id", namespace, id) if err != nil { return nil, fmt.Errorf("volume lookup failed for %s: %v", id, err) } @@ -2656,7 +2656,7 @@ func (s *StateStore) CSIVolumeByID(ws memdb.WatchSet, namespace, id string) (*st func (s *StateStore) CSIVolumesByPluginID(ws memdb.WatchSet, namespace, prefix, pluginID string) (memdb.ResultIterator, error) { txn := s.db.ReadTxn() - iter, err := txn.Get("csi_volumes", "plugin_id", pluginID) + iter, err := txn.Get(TableCSIVolumes, "plugin_id", pluginID) if err != nil { return nil, fmt.Errorf("volume lookup failed: %v", err) } @@ -2684,7 +2684,7 @@ func (s *StateStore) CSIVolumesByIDPrefix(ws memdb.WatchSet, namespace, volumeID txn := s.db.ReadTxn() - iter, err := txn.Get("csi_volumes", "id_prefix", namespace, volumeID) + iter, err := txn.Get(TableCSIVolumes, "id_prefix", namespace, volumeID) if err != nil { return nil, err } @@ -2698,7 +2698,7 @@ func (s *StateStore) csiVolumeByIDPrefixAllNamespaces(ws memdb.WatchSet, prefix txn := s.db.ReadTxn() // Walk the entire csi_volumes table - iter, err := txn.Get("csi_volumes", "id") + iter, err := txn.Get(TableCSIVolumes, "id") if err != nil { return nil, err @@ -2750,7 +2750,7 @@ func (s *StateStore) CSIVolumesByNodeID(ws memdb.WatchSet, prefix, nodeID string txn := s.db.ReadTxn() for id, namespace := range ids { if strings.HasPrefix(id, prefix) { - watchCh, raw, err := txn.FirstWatch("csi_volumes", "id", namespace, id) + watchCh, raw, err := txn.FirstWatch(TableCSIVolumes, "id", namespace, id) if err != nil { return nil, fmt.Errorf("volume lookup failed: %s %v", id, err) } @@ -2771,7 +2771,7 @@ func (s *StateStore) CSIVolumesByNamespace(ws memdb.WatchSet, namespace, prefix func (s *StateStore) csiVolumesByNamespaceImpl(txn *txn, ws memdb.WatchSet, namespace, prefix string) (memdb.ResultIterator, error) { - iter, err := txn.Get("csi_volumes", "id_prefix", namespace, prefix) + iter, err := txn.Get(TableCSIVolumes, "id_prefix", namespace, prefix) if err != nil { return nil, fmt.Errorf("volume lookup failed: %v", err) } @@ -2786,7 +2786,7 @@ func (s *StateStore) CSIVolumeClaim(index uint64, now int64, namespace, id strin txn := s.db.WriteTxn(index) defer txn.Abort() - row, err := txn.First("csi_volumes", "id", namespace, id) + row, err := txn.First(TableCSIVolumes, "id", namespace, id) if err != nil { return fmt.Errorf("volume lookup failed: %s: %v", id, err) } @@ -2844,11 +2844,11 @@ func (s *StateStore) CSIVolumeClaim(index uint64, now int64, namespace, id strin volume.WriteAllocs[allocID] = nil } - if err = txn.Insert("csi_volumes", volume); err != nil { + if err = txn.Insert(TableCSIVolumes, volume); err != nil { return fmt.Errorf("volume update failed: %s: %v", id, err) } - if err = txn.Insert("index", &IndexEntry{"csi_volumes", index}); err != nil { + if err = txn.Insert("index", &IndexEntry{TableCSIVolumes, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } @@ -2857,11 +2857,11 @@ func (s *StateStore) CSIVolumeClaim(index uint64, now int64, namespace, id strin // CSIVolumeDeregister removes the volume from the server func (s *StateStore) CSIVolumeDeregister(index uint64, namespace string, ids []string, force bool) error { - txn := s.db.WriteTxn(index) + txn := s.db.WriteTxnMsgT(structs.CSIVolumeDeregisterRequestType, index) defer txn.Abort() for _, id := range ids { - existing, err := txn.First("csi_volumes", "id", namespace, id) + existing, err := txn.First(TableCSIVolumes, "id", namespace, id) if err != nil { return fmt.Errorf("volume lookup failed: %s: %v", id, err) } @@ -2885,12 +2885,12 @@ func (s *StateStore) CSIVolumeDeregister(index uint64, namespace string, ids []s } } - if err = txn.Delete("csi_volumes", existing); err != nil { + if err = txn.Delete(TableCSIVolumes, existing); err != nil { return fmt.Errorf("volume delete failed: %s: %v", id, err) } } - if err := txn.Insert("index", &IndexEntry{"csi_volumes", index}); err != nil { + if err := txn.Insert("index", &IndexEntry{TableCSIVolumes, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } @@ -3072,7 +3072,7 @@ func (s *StateStore) CSIPlugins(ws memdb.WatchSet) (memdb.ResultIterator, error) txn := s.db.ReadTxn() defer txn.Abort() - iter, err := txn.Get("csi_plugins", "id") + iter, err := txn.Get(TableCSIPlugins, "id") if err != nil { return nil, fmt.Errorf("csi_plugins lookup failed: %v", err) } @@ -3086,7 +3086,7 @@ func (s *StateStore) CSIPlugins(ws memdb.WatchSet) (memdb.ResultIterator, error) func (s *StateStore) CSIPluginsByIDPrefix(ws memdb.WatchSet, pluginID string) (memdb.ResultIterator, error) { txn := s.db.ReadTxn() - iter, err := txn.Get("csi_plugins", "id_prefix", pluginID) + iter, err := txn.Get(TableCSIPlugins, "id_prefix", pluginID) if err != nil { return nil, err } @@ -3110,7 +3110,7 @@ func (s *StateStore) CSIPluginByID(ws memdb.WatchSet, id string) (*structs.CSIPl // CSIPluginByIDTxn returns a named CSIPlugin func (s *StateStore) CSIPluginByIDTxn(txn Txn, ws memdb.WatchSet, id string) (*structs.CSIPlugin, error) { - watchCh, obj, err := txn.FirstWatch("csi_plugins", "id", id) + watchCh, obj, err := txn.FirstWatch(TableCSIPlugins, "id", id) if err != nil { return nil, fmt.Errorf("csi_plugin lookup failed: %s %v", id, err) } @@ -3167,7 +3167,7 @@ func (s *StateStore) UpsertCSIPlugin(index uint64, plug *structs.CSIPlugin) erro txn := s.db.WriteTxn(index) defer txn.Abort() - existing, err := txn.First("csi_plugins", "id", plug.ID) + existing, err := txn.First(TableCSIPlugins, "id", plug.ID) if err != nil { return fmt.Errorf("csi_plugin lookup error: %s %v", plug.ID, err) } @@ -3178,11 +3178,11 @@ func (s *StateStore) UpsertCSIPlugin(index uint64, plug *structs.CSIPlugin) erro plug.CreateTime = existing.(*structs.CSIPlugin).CreateTime } - err = txn.Insert("csi_plugins", plug) + err = txn.Insert(TableCSIPlugins, plug) if err != nil { return fmt.Errorf("csi_plugins insert error: %v", err) } - if err := txn.Insert("index", &IndexEntry{"csi_plugins", index}); err != nil { + if err := txn.Insert("index", &IndexEntry{TableCSIPlugins, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } return txn.Commit() @@ -3242,7 +3242,7 @@ func (s *StateStore) DeleteCSIPlugin(index uint64, id string) error { return structs.ErrCSIPluginInUse } - err = txn.Delete("csi_plugins", plug) + err = txn.Delete(TableCSIPlugins, plug) if err != nil { return fmt.Errorf("csi_plugins delete error: %v", err) } @@ -5900,13 +5900,13 @@ func (s *StateStore) updateJobCSIPlugins(index uint64, job, prev *structs.Job, t } for _, plugIn := range plugIns { - err = txn.Insert("csi_plugins", plugIn) + err = txn.Insert(TableCSIPlugins, plugIn) if err != nil { return fmt.Errorf("csi_plugins insert error: %v", err) } } - if err := txn.Insert("index", &IndexEntry{"csi_plugins", index}); err != nil { + if err := txn.Insert("index", &IndexEntry{TableCSIPlugins, index}); err != nil { return fmt.Errorf("index update failed: %v", err) } diff --git a/nomad/state/state_store_restore.go b/nomad/state/state_store_restore.go index 0a1638422f2..41a571b71d9 100644 --- a/nomad/state/state_store_restore.go +++ b/nomad/state/state_store_restore.go @@ -181,7 +181,7 @@ func (r *StateRestore) ScalingPolicyRestore(scalingPolicy *structs.ScalingPolicy // CSIPluginRestore is used to restore a CSI plugin func (r *StateRestore) CSIPluginRestore(plugin *structs.CSIPlugin) error { - if err := r.txn.Insert("csi_plugins", plugin); err != nil { + if err := r.txn.Insert(TableCSIPlugins, plugin); err != nil { return fmt.Errorf("csi plugin insert failed: %v", err) } return nil @@ -189,7 +189,7 @@ func (r *StateRestore) CSIPluginRestore(plugin *structs.CSIPlugin) error { // CSIVolumeRestore is used to restore a CSI volume func (r *StateRestore) CSIVolumeRestore(volume *structs.CSIVolume) error { - if err := r.txn.Insert("csi_volumes", volume); err != nil { + if err := r.txn.Insert(TableCSIVolumes, volume); err != nil { return fmt.Errorf("csi volume insert failed: %v", err) } return nil diff --git a/nomad/stream/event_broker.go b/nomad/stream/event_broker.go index 41d52f064e4..7895f294478 100644 --- a/nomad/stream/event_broker.go +++ b/nomad/stream/event_broker.go @@ -363,6 +363,14 @@ func aclAllowsSubscription(aclObj *acl.ACL, subReq *SubscribeRequest) bool { if ok := aclObj.AllowNsOp(subReq.Namespace, acl.NamespaceCapabilityReadJob); !ok { return false } + case structs.TopicCSIVolume: + if ok := aclObj.AllowNsOp(subReq.Namespace, acl.NamespaceCapabilityCSIReadVolume); !ok { + return false + } + case structs.TopicCSIPlugin: + if ok := aclObj.AllowNsOp(subReq.Namespace, acl.NamespaceCapabilityReadJob); !ok { + return false + } case structs.TopicNode: if ok := aclObj.AllowNodeRead(); !ok { return false diff --git a/nomad/structs/event.go b/nomad/structs/event.go index 466421d38a5..fba87b0612e 100644 --- a/nomad/structs/event.go +++ b/nomad/structs/event.go @@ -31,6 +31,8 @@ const ( TopicACLAuthMethod Topic = "ACLAuthMethod" TopicACLBindingRule Topic = "ACLBindingRule" TopicService Topic = "Service" + TopicCSIVolume Topic = "CSIVolume" + TopicCSIPlugin Topic = "CSIPlugin" TopicAll Topic = "*" TypeNodeRegistration = "NodeRegistration" @@ -63,6 +65,8 @@ const ( TypeACLBindingRuleDeleted = "ACLBindingRuleDeleted" TypeServiceRegistration = "ServiceRegistration" TypeServiceDeregistration = "ServiceDeregistration" + TypeCSIVolumeRegistered = "CSIVolumeRegistered" + TypeCSIVolumeDeregistered = "CSIVolumeDeregistered" ) // Event represents a change in Nomads state. @@ -188,3 +192,15 @@ type ACLAuthMethodEvent struct { type ACLBindingRuleEvent struct { ACLBindingRule *ACLBindingRule } + +// CSIVolumeEvent holds a newly updated or deleted CSI volume to be +// used as an event in the event stream +type CSIVolumeEvent struct { + Volume *CSIVolume +} + +// CSIPluginEvent holds a newly updated or deleted CSI plugin to be +// used as an event in the event stream +type CSIPluginEvent struct { + Plugin *CSIPlugin +} diff --git a/nomad/volumewatcher/volumes_watcher.go b/nomad/volumewatcher/volumes_watcher.go index 2da49c7d1f6..a9cf7ae3524 100644 --- a/nomad/volumewatcher/volumes_watcher.go +++ b/nomad/volumewatcher/volumes_watcher.go @@ -141,9 +141,9 @@ func (w *Watcher) getVolumes(ctx context.Context, minIndex uint64) ([]*structs.C } // getVolumesImpl retrieves all volumes from the passed state store. -func (w *Watcher) getVolumesImpl(ws memdb.WatchSet, state *state.StateStore) (interface{}, uint64, error) { +func (w *Watcher) getVolumesImpl(ws memdb.WatchSet, store *state.StateStore) (interface{}, uint64, error) { - iter, err := state.CSIVolumes(ws) + iter, err := store.CSIVolumes(ws) if err != nil { return nil, 0, err } @@ -159,7 +159,7 @@ func (w *Watcher) getVolumesImpl(ws memdb.WatchSet, state *state.StateStore) (in } // Use the last index that affected the volume table - index, err := state.Index("csi_volumes") + index, err := store.Index(state.TableCSIVolumes) if err != nil { return nil, 0, err } diff --git a/website/content/api-docs/events.mdx b/website/content/api-docs/events.mdx index 0cc34c76a73..8b5d85fc507 100644 --- a/website/content/api-docs/events.mdx +++ b/website/content/api-docs/events.mdx @@ -28,19 +28,21 @@ the nature of this endpoint individual topics require specific policies. Note that if you do not include a `topic` parameter all topics will be included by default, requiring a management token. -| Topic | ACL Required | -| ------------ | -------------------- | -| `*` | `management` | -| `ACLToken` | `management` | -| `ACLPolicy` | `management` | -| `ACLRole` | `management` | -| `Job` | `namespace:read-job` | -| `Allocation` | `namespace:read-job` | -| `Deployment` | `namespace:read-job` | -| `Evaluation` | `namespace:read-job` | -| `Node` | `node:read` | -| `NodePool` | `management` | -| `Service` | `namespace:read-job` | +| Topic | ACL Required | +|--------------|-----------------------------| +| `*` | `management` | +| `ACLPolicy` | `management` | +| `ACLRole` | `management` | +| `ACLToken` | `management` | +| `Allocation` | `namespace:read-job` | +| `CSIPlugin` | `namespace:read-job` | +| `CSIVolume` | `namespace:csi-read-volume` | +| `Deployment` | `namespace:read-job` | +| `Evaluation` | `namespace:read-job` | +| `Job` | `namespace:read-job` | +| `NodePool` | `management` | +| `Node` | `node:read` | +| `Service` | `namespace:read-job` | ### Parameters @@ -66,14 +68,16 @@ by default, requiring a management token. ### Event Topics | Topic | Output | -| ---------- | ------------------------------- | -| ACLToken | ACLToken | +|------------|---------------------------------| | ACLPolicy | ACLPolicy | | ACLRoles | ACLRole | +| ACLToken | ACLToken | | Allocation | Allocation (no job information) | -| Job | Job | -| Evaluation | Evaluation | +| CSIPlugin | CSIPlugin | +| CSIVolume | CSIVolume | | Deployment | Deployment | +| Evaluation | Evaluation | +| Job | Job | | Node | Node | | NodeDrain | Node | | NodePool | NodePool | @@ -82,33 +86,36 @@ by default, requiring a management token. ### Event Types | Type | -| ----------------------------- | -| ACLTokenUpserted | -| ACLTokenDeleted | -| ACLPolicyUpserted | +|-------------------------------| | ACLPolicyDeleted | -| ACLRoleUpserted | +| ACLPolicyUpserted | | ACLRoleDeleted | +| ACLRoleUpserted | +| ACLTokenDeleted | +| ACLTokenUpserted | | AllocationCreated | -| AllocationUpdated | | AllocationUpdateDesiredStatus | -| DeploymentStatusUpdate | -| DeploymentPromotion | +| AllocationUpdated | +| CSIVolumeDeregistered | +| CSIVolumeRegistered | | DeploymentAllocHealth | +| DeploymentPromotion | +| DeploymentStatusUpdate | | EvaluationUpdated | -| JobRegistered | -| JobDeregistered | | JobBatchDeregistered | -| NodeRegistration | +| JobDeregistered | +| JobRegistered | | NodeDeregistration | -| NodeEligibility | | NodeDrain | +| NodeEligibility | | NodeEvent | -| NodePoolUpserted | | NodePoolDeleted | +| NodePoolUpserted | +| NodeRegistration | | PlanResult | -| ServiceRegistration | | ServiceDeregistration | +| ServiceRegistration | + ### Sample Request