feat(ui): show a full client on the client list as soon as it fills up

docs/clients.md

@@ -15,8 +15,11 @@ A client that holds as many allocations as its limit allows (`node_max_allocs`,
 Nomad 1.10.2 and later) is full: the scheduler places nothing more on it, and
 its row has the attention color, like a client that is draining or
 ineligible. With the event stream on, the list shows a client that fills up
-or frees a place within 30 seconds: Nomad streams a change of the client,
-not of the allocations placed on it.
+or frees a place as soon as an allocation is placed on it or stops. For that
+it follows the allocations of every namespace, which needs a token that may
+read the jobs of every namespace, such as a management token. With any other
+token the list follows the clients alone, says nothing about it, and shows a
+full client within 30 seconds.
 
 - `ctrl-d` drains a client, or stops draining it, after you confirm.
 - `i` marks a client as eligible or ineligible for new work.

docs/configuration.md

@@ -166,6 +166,12 @@ on Nomad 2.0 and later and on Nomad Enterprise 1.11.4 and later. On an older
 cluster, or when urga cannot read the version of the cluster, they are read
 every 2 seconds.
 
+The client list also follows the allocations of every namespace, so a client
+that fills up shows as full at once. That needs a token that may read the jobs
+of every namespace. When the cluster refuses that stream, or ends it within 5
+seconds with no event, which is how Nomad 2.0 refuses it, the list follows the
+clients alone, with no warning.
+
 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
 line says so.

internal/ui/app_test.go

@@ -166,6 +166,10 @@ type fakeClient struct {
 	revertedTo       uint64
 	diffErr          error
 
+	// streamOf, when set, answers each ask for a stream in place of changes
+	// and watchErr, so a test can refuse some topics and not others.
+	streamOf func(namespace string, topics []string) (*nomad.Changes, error)
+
 	calls      int
 	allocCalls int
 
@@ -1294,6 +1298,10 @@ func (f *fakeClient) BatchQueueStatus(_ context.Context, pool string) ([]nomad.Q
 func (f *fakeClient) Events(_ context.Context, namespace string, topics []string) (*nomad.Changes, error) {
 	f.watchedNamespace, f.watchedTopics = namespace, topics
 
+	if f.streamOf != nil {
+		return f.streamOf(namespace, topics)
+	}
+
 	if f.watchErr != nil {
 		return nil, f.watchErr
 	}

internal/ui/node.go

@@ -38,7 +38,17 @@ func (p nodesPage) title(_ env, count int) string {
 }
 
 func (nodesPage) titles() []string { return nodeTitles }
-func (nodesPage) topics() []string { return []string{nomad.TopicNode} }
+
+// topics: a client fills up when an allocation is placed on it, which is a
+// change of the allocation, not of the client.
+func (nodesPage) topics() []string { return []string{nomad.TopicNode, nomad.TopicAllocation} }
+
+// where: the places a client holds are counted over every namespace.
+func (nodesPage) where(env) string { return nomad.AllNamespaces }
+
+// fewerTopics: allocations of every namespace need a token that may read the
+// jobs of each one. Without it the list follows the clients alone.
+func (nodesPage) fewerTopics() []string { return []string{nomad.TopicNode} }
 
 // fetch reads the clients, of the pool when the page has one: the cluster
 // says which clients a pool holds. The list does not carry the allocation

internal/ui/node_test.go

@@ -2,7 +2,12 @@ package ui
 
 import (
 	"errors"
+	"io"
+	"slices"
 	"testing"
+	"time"
+
+	tea "charm.land/bubbletea/v2"
 
 	"github.com/stretchr/testify/require"
 
@@ -50,14 +55,195 @@ func TestClients_TheList(t *testing.T) {
 
 	r.Contains(plain(m.render()), "Clients [1]")
 	r.Equal([]string{"ID", "Name", "Datacenter", "Pool", "Version", "Status", "Eligibility", "Drain", "CPU", "MEM", "Address"}, m.screen.page.titles())
-	r.Equal([]string{nomad.TopicNode}, m.screen.page.topics())
+	r.Equal([]string{nomad.TopicNode, nomad.TopicAllocation}, m.screen.page.topics())
 
 	// The CPU and memory use of the machine is unknown until it is read.
 	r.Regexp(`^\s*node-1\s+nomad-server-01\s+dc1\s+default\s+1\.11\.1\s+ready\s+eligible\s+false\s+-\s+-\s+10\.0\.0\.2`, fileRow(t, m, 0))
 
 	r.Equal([]string{"enter Allocations", "ctrl+l Agent Logs", "ctrl+d Drain", "i Toggle Eligibility", "ctrl+g GC", "space Mark", "ctrl+a Mark All"}, keyNames(m))
 }
 
+func TestClients_FollowTheWorkOfEveryNamespace(t *testing.T) {
+	r := require.New(t)
+
+	client := &fakeClient{nodes: busyClient(), changes: newChanges()}
+
+	m, _ := nodeModelOf(client)
+	r.Equal("production", m.namespace)
+
+	m.update(m.watchScreen()())
+
+	r.Equal(nomad.AllNamespaces, client.watchedNamespace)
+	r.Equal([]string{nomad.TopicNode, nomad.TopicAllocation}, client.watchedTopics)
+}
+
+// endedAtOnce is a stream that ends as soon as it opens, with no event: how
+// Nomad 2.0 refuses one.
+func endedAtOnce() *nomad.Changes {
+	ended := newChanges()
+	ended.errs <- io.EOF
+	close(ended.c)
+
+	return ended.stream()
+}
+
+// refusals are the two ways a cluster refuses a stream: a 403, or a stream
+// that ends at once.
+func refusals(t *testing.T) map[string]func() (*nomad.Changes, error) {
+	t.Helper()
+
+	return map[string]func() (*nomad.Changes, error){
+		"refused":       func() (*nomad.Changes, error) { return nil, forbidden(t) },
+		"ended at once": func() (*nomad.Changes, error) { return endedAtOnce(), nil },
+	}
+}
+
+// throughRefusal runs the answer to an ask for a stream through the model. A
+// stream that opened is read first, to its end.
+func throughRefusal(m Model, msg tea.Msg) (Model, tea.Cmd) {
+	if opened, ok := msg.(watchingMsg); ok {
+		var read tea.Cmd
+		m, read = m.update(opened)
+		msg = read()
+	}
+
+	return m.update(msg)
+}
+
+func TestClients_ARefusedStreamOfTheWorkFollowsTheClientsAlone(t *testing.T) {
+	for name, refuse := range refusals(t) {
+		t.Run(name, func(t *testing.T) {
+			r := require.New(t)
+
+			var asked [][]string
+
+			client := &fakeClient{jobs: twoJobs(), nodes: busyClient(), changes: newChanges()}
+
+			// The jobs screen streams, so a refusal shown on the client list
+			// would be a warning. It hears a change, which the stream of the
+			// list does not inherit.
+			m := watching(t, client)
+			m, _ = m.update(changeMsg{id: m.watch.id})
+
+			client.streamOf = func(_ string, topics []string) (*nomad.Changes, error) {
+				asked = append(asked, topics)
+				if slices.Contains(topics, nomad.TopicAllocation) {
+					return refuse()
+				}
+
+				return newChanges().stream(), nil
+			}
+
+			m, _ = m.update(key(':'))
+			m = typeIn(m, "nodes")
+			m, _ = m.update(enter())
+			m, _ = m.update(nodesMsg(busyClient()))
+			tick := m.pollTick
+
+			msg := m.watchScreen()()
+			r.Equal([]string{nomad.TopicNode, nomad.TopicAllocation}, client.watchedTopics)
+
+			m, cmd := throughRefusal(m, msg)
+			r.NotNil(cmd)
+			m, _ = m.update(cmd())
+
+			// The list follows what the token may read: the clients, with no
+			// warning and no faster poll.
+			r.Equal([][]string{{nomad.TopicNode, nomad.TopicAllocation}, {nomad.TopicNode}}, asked)
+			r.Equal(nomad.AllNamespaces, client.watchedNamespace)
+			r.True(m.watch.live())
+			r.Empty(m.flash.text)
+			r.Equal(slowPoll, m.pollEvery())
+			r.Equal(tick, m.pollTick)
+		})
+	}
+}
+
+func TestClients_ARefusalOfTheClientsAloneSaysSo(t *testing.T) {
+	for name, refuse := range refusals(t) {
+		t.Run(name, func(t *testing.T) {
+			r := require.New(t)
+
+			client := &fakeClient{nodes: busyClient()}
+			client.streamOf = func(string, []string) (*nomad.Changes, error) { return refuse() }
+
+			m, _ := nodeModelOf(client)
+
+			m, cmd := throughRefusal(m, m.watchScreen()())
+			r.NotNil(cmd)
+			m, _ = throughRefusal(m, cmd())
+
+			// Refused the clients too, the list is polled the way it was.
+			r.Equal([]string{nomad.TopicNode}, client.watchedTopics)
+			r.False(m.watch.live())
+			r.Contains(m.flash.text, "asking every")
+			r.Equal(time.Millisecond, m.pollEvery())
+		})
+	}
+}
+
+func TestClients_AStreamOfTheWorkThatDropsLaterSaysSo(t *testing.T) {
+	for name, before := range map[string]func(Model) Model{
+		"after a change": func(m Model) Model {
+			m, _ = m.update(changeMsg{id: m.watch.id})
+
+			return m
+		},
+		"after a while": func(m Model) Model {
+			m.watch.opened = m.watch.opened.Add(-refusedWithin)
+
+			return m
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			r := require.New(t)
+
+			client := &fakeClient{nodes: busyClient(), changes: newChanges()}
+
+			m, _ := nodeModelOf(client)
+			m, _ = m.update(m.watchScreen()())
+			r.True(m.watch.live())
+
+			m = before(m)
+			m, _ = m.update(watchEndedMsg{id: m.watch.id, err: io.EOF})
+
+			// A stream that worked and dropped is not a refusal: the list does
+			// not ask for the clients alone.
+			r.Equal([]string{nomad.TopicNode, nomad.TopicAllocation}, client.watchedTopics)
+			r.False(m.watch.live())
+			r.Contains(m.flash.text, "EOF")
+			r.Equal(time.Millisecond, m.pollEvery())
+		})
+	}
+}
+
+func TestClients_OpenedAgainAsksForTheWorkAgain(t *testing.T) {
+	r := require.New(t)
+
+	client := &fakeClient{jobs: twoJobs(), nodes: busyClient()}
+	client.streamOf = func(_ string, topics []string) (*nomad.Changes, error) {
+		if slices.Contains(topics, nomad.TopicAllocation) {
+			return nil, forbidden(t)
+		}
+
+		return newChanges().stream(), nil
+	}
+
+	m, _ := nodeModelOf(client)
+	m, cmd := throughRefusal(m, m.watchScreen()())
+	m, _ = m.update(cmd())
+	r.Equal([]string{nomad.TopicNode}, client.watchedTopics)
+
+	// The token may be another by the time the list opens again.
+	m, _ = m.update(escape())
+	m, _ = m.update(key(':'))
+	m = typeIn(m, "nodes")
+	m, _ = m.update(enter())
+
+	m.update(m.watchScreen()())
+	r.Equal([]string{nomad.TopicNode, nomad.TopicAllocation}, client.watchedTopics)
+}
+
 func TestClients_OnlyThoseThatTakeNoWork(t *testing.T) {
 	r := require.New(t)
 

internal/ui/note.go

@@ -16,10 +16,6 @@ const noteKept = 1000
 // dropped.
 const noteAgain = 30 * time.Second
 
-// noteRefusedWithin is how soon a stream the cluster refuses ends: Nomad 2.0
-// answers 200 and ends the stream at once, with no event.
-const noteRefusedWithin = 5 * time.Second
-
 // noteAgainMsg asks to follow the failures again, for the connection with
 // that id.
 type noteAgainMsg struct{ id int }
@@ -48,10 +44,9 @@ type noteState struct {
 	// failing is what the note knows failed and has not left that state.
 	failing map[string]bool
 
-	// heard says the open stream sent an event, and opened is when it
-	// opened.
-	heard  bool
-	opened time.Time
+	// streamStart is when the open stream opened and whether it sent an
+	// event.
+	streamStart
 
 	// from is where the next stream starts: past the newest event read, or
 	// where the last stream started. 0 is past the newest event the cluster
@@ -207,7 +202,7 @@ func (m Model) takeNoteEvents(msg feedEventsMsg) (Model, tea.Cmd) {
 func (m Model) takeNoteEnded(feedEndedMsg) (Model, tea.Cmd) {
 	m.note.stop()
 
-	if !m.note.heard && m.now.Sub(m.note.opened) < noteRefusedWithin {
+	if m.note.endedAtOnce(m.now) {
 		return m, nil
 	}
 

internal/ui/note_test.go

@@ -503,7 +503,7 @@ func TestNote_FollowsAgainAQuietStreamThatDrops(t *testing.T) {
 
 	// The cluster sent nothing, and the stream was open too long to be a
 	// refusal: a server restarted, or the network dropped.
-	m.note.opened = m.note.opened.Add(-noteRefusedWithin)
+	m.note.opened = m.note.opened.Add(-refusedWithin)
 
 	m, cmd := m.update(feedEndedMsg{feed: feed, err: io.ErrUnexpectedEOF})
 	requireTriesLater(t, m, cmd)

internal/ui/page.go

@@ -508,6 +508,12 @@ func namespaceOf(p page, e env) string {
 	return e.namespace
 }
 
+// fallback is a page that still follows part of what it shows when the
+// cluster refuses the stream of all its topics.
+type fallback interface {
+	fewerTopics() []string
+}
+
 // ofTheSession is a list the session opens by name, of what it looks at.
 type ofTheSession struct{}
 

internal/ui/session_test.go

@@ -134,7 +134,7 @@ func TestSession_ANamespaceGivenOnTheCommandLineWins(t *testing.T) {
 	cfg := sessionConfig(t, filepath.Join(t.TempDir(), "config.json"))
 
 	cfg.UseNamespace("staging")
-	cfg.Screen = "nodes"
+	cfg.Screen = "deployments"
 
 	client := &fakeClient{changes: newChanges()}
 	m := New(client, Options{
@@ -149,7 +149,7 @@ func TestSession_ANamespaceGivenOnTheCommandLineWins(t *testing.T) {
 	// Typing a namespace on the command line is asking for it now. The
 	// rest of the session still comes back.
 	r.Equal("production", m.namespace)
-	r.IsType(nodesPage{}, m.screen.page)
+	r.IsType(deploymentsPage{}, m.screen.page)
 
 	fire(t, m.watchScreen())
 	r.Equal("production", client.watchedNamespace)