feat(ui): watch volumes, plugins and variables with the event stream, and show the events of one job or allocation

README.md

@@ -29,7 +29,8 @@ One binary, no config to write, no browser.
 - Watches the deploy of a job with `w`: the newest evaluation, the deployment
   and its allocations, until it ends.
 - Updates the screen when the cluster changes, using the Nomad event stream,
-  and shows the stream itself on `:events`.
+  and shows the stream itself on `:events`, for the namespace, or for one job
+  or allocation.
 - Counts on one screen what needs attention: dead jobs, failed allocations,
   tasks that restart or run out of memory, clients that are down or draining,
   blocked evaluations and failed deployments, and opens the list of each with

docs/allocations.md

@@ -28,6 +28,8 @@ A running allocation whose task was restarted or killed for its memory in the
 last hour is painted in the attention color, and `!` keeps it: that is a
 crash loop going on now. A restart of a month ago leaves the row as it is.
 
+`o` opens the events of the allocation under the cursor.
+
 `r` restarts the tasks of an allocation that run, `ctrl-r` restarts all of
 its tasks, the prestart and sidecar ones too, and `ctrl-k` stops it, after
 you confirm. With several allocations marked, they act on all of them. When you stop an
@@ -61,7 +63,8 @@ From here:
 
 - `c` opens the client the allocation runs on;
 - `p` opens the previous allocation, `n` the next one;
-- `f` opens the follow-up evaluation.
+- `f` opens the follow-up evaluation;
+- `o` opens the events of the allocation.
 
 ## Checks
 

docs/configuration.md

@@ -149,8 +149,10 @@ without saving.
 Most screens, such as jobs, allocations and clients, follow the Nomad event
 stream: urga reloads the screen when something on it changes. It also reloads
 every 30 seconds in case an event is missed. Screens that the event stream
-does not cover, such as namespaces, variables and servers, reload every 2
-seconds.
+does not cover, such as namespaces and servers, reload every 2 seconds.
+Volumes and CSI plugins follow the stream on Nomad 1.10 and later, variables
+on Nomad 2.0 and later. On an older cluster, or when urga cannot read the
+version of the cluster, they are read every 2 seconds.
 
 If the event stream is not available, which usually means the ACL token does
 not allow it, urga reloads every screen every 2 seconds instead. The status

docs/jobs.md

@@ -19,7 +19,8 @@ From the list:
 - `t` opens its task groups;
 - `d` shows the job as the cluster describes it;
 - `h` shows the job file it was submitted with;
-- `v` opens its versions.
+- `v` opens its versions;
+- `o` shows the events of the job.
 
 ## Periodic and parameterized jobs
 

docs/keys.md

@@ -147,13 +147,19 @@ the events that report a failure.
 
 With the cursor on the top row, it stays there and shows each event as it
 arrives. Moved down, it stays on its event. `enter` opens what the event is
-about: a job, an allocation, a deployment, an evaluation, a client or a
-service. `esc` comes back to the events as they were, and the ones that
-happened meanwhile are added on top.
+about: a job, an allocation, a deployment, an evaluation, a client, a
+service, a volume, a CSI plugin or a variable. `esc` comes back to the
+events as they were, and the ones that happened meanwhile are added on top.
 
-The list follows the namespace of the session. Events of clients belong to no
-namespace and show in every one. In another namespace it starts again from
-what the cluster keeps.
+The list follows the namespace of the session. Events of clients and CSI
+plugins belong to no namespace and show in every one. In another namespace it
+starts again from what the cluster keeps.
+
+`o` on a job or an allocation opens the same screen with only the events of
+it. The cluster sends only those, from what it still keeps. The title names
+the job or the allocation, and the screen stays on it when you switch
+namespace. These need only a token that can read the job, while the events
+of the whole namespace need a management token.
 
 ## Sort
 
@@ -216,6 +222,7 @@ start, stop, restart or drain then applies to every marked row.
 | `v` | Versions | |
 | `l` | Logs of a task in every allocation of the job. Not shown for a periodic or parameterized job. | |
 | `w` | Watch the deploy of the job at the version it runs. Not shown for a periodic or parameterized job. | |
+| `o` | Events of the job, its allocations, evaluations and deployments. Not shown for a periodic or parameterized job. | |
 | `e` | Edit the job in your editor | yes |
 | `n` | New job: opens a job template in your editor | yes |
 | `ctrl-s` | Start a stopped job, or stop a running one | yes |
@@ -281,6 +288,7 @@ On what was done to the count of a group, `d` describes its policy.
 | `ctrl-r` | Restart every task, the prestart and sidecar ones too | yes |
 | `ctrl-k` | Stop | yes |
 | `l` | Logs of a task in every allocation of the list | |
+| `o` | Events of the allocation | |
 | `space`, `ctrl-a` | Mark | |
 
 ### Tasks
@@ -290,6 +298,7 @@ On what was done to the count of a group, `d` describes its policy.
 | `enter` | Logs, stdout | |
 | `ctrl-e` | Logs, stderr | |
 | `e` | Events of the task | |
+| `o` | Events of the allocation | |
 | `b` | Files of the task | |
 | `s` | Shell in the task | yes |
 | `r` | Restart the task. Shown when it is running. | yes |

docs/storage.md

@@ -29,8 +29,9 @@ still being made (`pending`) is yellow. A CSI volume whose plugin has fewer
 healthy controllers or node plugins than expected has the attention color:
 attaching it may fail. `!` shows only those volumes.
 
-The cluster reports no change of a CSI volume in its event stream, so the
-storage screens are polled, not watched.
+The storage screens follow the event stream on Nomad 1.10 and later. On an
+older cluster, or when urga cannot read the version of the cluster, they are
+read every 2 seconds.
 
 ## A volume
 

internal/nomad/events.go

@@ -2,6 +2,9 @@ package nomad
 
 import (
 	"context"
+	"encoding/json"
+	"regexp"
+	"strconv"
 	"time"
 
 	"github.com/hashicorp/nomad/api"
@@ -16,8 +19,87 @@ const (
 	TopicNode       = "Node"
 	TopicNodePool   = "NodePool"
 	TopicService    = "Service"
+
+	// Nomad's api module has no constant for these.
+	TopicCSIVolume  = "CSIVolume"
+	TopicCSIPlugin  = "CSIPlugin"
+	TopicHostVolume = "HostVolume"
+	TopicVariable   = "Variable"
 )
 
+// since is the first version of Nomad that sends the events of a topic. A
+// topic not in it is older than any cluster urga reads.
+var since = map[string]string{
+	TopicCSIVolume:  "1.10.0",
+	TopicCSIPlugin:  "1.10.0",
+	TopicHostVolume: "1.10.0",
+	TopicVariable:   "2.0.0",
+}
+
+// Streams says a cluster of that version sends the events of every one of
+// the topics. A cluster that does not know a topic opens the stream anyway
+// and never sends anything for it, so the version is checked before it is
+// asked.
+func Streams(version string, topics []string) bool {
+	for _, topic := range topics {
+		if minimum, ok := since[topic]; ok && !AtLeast(version, minimum) {
+			return false
+		}
+	}
+
+	return true
+}
+
+var numbers = regexp.MustCompile(`^(\d+)\.(\d+)\.(\d+)`)
+
+// AtLeast says version reads as major.minor.patch at its start, and is at or
+// above the minimum. What follows the patch (-beta.1, +ent) is not compared.
+func AtLeast(version, minimum string) bool {
+	have, ok := triple(version)
+	if !ok {
+		return false
+	}
+
+	need, ok := triple(minimum)
+	if !ok {
+		return false
+	}
+
+	return have.compare(need) >= 0
+}
+
+type release [3]int
+
+func (a release) compare(b release) int {
+	for i := range a {
+		if a[i] != b[i] {
+			return a[i] - b[i]
+		}
+	}
+
+	return 0
+}
+
+func triple(version string) (release, bool) {
+	parts := numbers.FindStringSubmatch(version)
+	if parts == nil {
+		return release{}, false
+	}
+
+	var r release
+
+	for i := range r {
+		n, err := strconv.Atoi(parts[i+1])
+		if err != nil {
+			return release{}, false
+		}
+
+		r[i] = n
+	}
+
+	return r, true
+}
+
 // Change reports that something in the cluster changed. It carries no new
 // state: the screen reads that with its usual request.
 type Change struct {
@@ -86,13 +168,36 @@ type Event struct {
 // Feed is what happens in the cluster, event by event.
 type Feed = Stream[Event]
 
-// Feed follows every topic of the cluster in a namespace, from the event of
-// an index on: 1 is the oldest the cluster still keeps. The caller closes it
-// when it stops reading.
-func (c *Client) Feed(ctx context.Context, namespace string, from uint64) (*Feed, error) {
+// Subject is what a feed follows: one job, one allocation, or, when it is
+// empty, the whole namespace.
+type Subject struct{ JobID, AllocID string }
+
+// topics are what the cluster is asked for. An allocation, an evaluation and
+// a deployment carry the ID of their job as a filter key, so the cluster
+// sends the work of a job by the job's ID. Services are left out: their
+// filter keys hold the service name, which another job's service can share.
+func (s Subject) topics() map[api.Topic][]string {
+	switch {
+	case s.AllocID != "":
+		return map[api.Topic][]string{api.TopicAllocation: {s.AllocID}}
+	case s.JobID != "":
+		key := []string{s.JobID}
+
+		return map[api.Topic][]string{
+			api.TopicJob: key, api.TopicAllocation: key, api.TopicEvaluation: key, api.TopicDeployment: key,
+		}
+	}
+
+	return map[api.Topic][]string{api.TopicAll: nil}
+}
+
+// Feed follows the subject in a namespace, from the event of an index on: 1
+// is the oldest the cluster still keeps. The caller closes it when it stops
+// reading.
+func (c *Client) Feed(ctx context.Context, namespace string, subject Subject, from uint64) (*Feed, error) {
 	// Asked for an index past its newest event, the cluster starts at that
 	// event again: what came before the index was read already.
-	return follow(ctx, c, namespace, map[api.Topic][]string{api.TopicAll: nil}, from, func(event api.Event) (Event, bool) {
+	return follow(ctx, c, namespace, subject.topics(), from, func(event api.Event) (Event, bool) {
 		return newEvent(event), event.Index >= from
 	})
 }
@@ -133,11 +238,64 @@ func newEvent(e api.Event) Event {
 		if sr, err := e.Service(); err == nil && sr != nil {
 			event.Namespace, event.Name = sr.Namespace, sr.ServiceName
 		}
+	default:
+		event.readStored(e)
 	}
 
 	return event
 }
 
+// readStored names an event of a topic the api module cannot decode.
+func (event *Event) readStored(e api.Event) {
+	switch e.Topic {
+	case TopicCSIVolume:
+		if v, ok := payloadOf(e, "Volume"); ok {
+			event.Namespace, event.Name, event.State, event.At = v.Namespace, v.Name, csiState(v.Schedulable), unixTime(v.ModifyTime)
+		}
+	case TopicHostVolume:
+		if v, ok := payloadOf(e, "Volume"); ok {
+			event.Namespace, event.Name, event.State, event.At = v.Namespace, v.Name, v.State, unixTime(v.ModifyTime)
+
+			// A deleted volume has no key of its own.
+			if event.Key == "" {
+				event.Key = v.ID
+			}
+		}
+	case TopicCSIPlugin:
+		if v, ok := payloadOf(e, "Plugin"); ok {
+			event.Name, event.At = v.ID, unixTime(v.ModifyTime)
+		}
+	case TopicVariable:
+		if v, ok := payloadOf(e, "Metadata"); ok {
+			event.Namespace, event.Name, event.At = v.Namespace, v.Path, unixTime(v.ModifyTime)
+		}
+	}
+}
+
+// stored is the fields urga reads of an object the api module cannot decode.
+type stored struct {
+	ID, Name, Namespace, Path, State string
+	Schedulable                      bool
+	ModifyTime                       int64
+}
+
+// payloadOf reads the object an event carries under a key of its payload.
+func payloadOf(e api.Event, key string) (stored, bool) {
+	raw, ok := e.Payload[key]
+	if !ok {
+		return stored{}, false
+	}
+
+	data, err := json.Marshal(raw)
+	if err != nil {
+		return stored{}, false
+	}
+
+	var v stored
+
+	return v, json.Unmarshal(data, &v) == nil
+}
+
 // streamBuffer is how many events of a stream wait for the reader.
 const streamBuffer = 256