From d55e5ce055e793e8b1f8281f794e5609684e0b8e Mon Sep 17 00:00:00 2001 From: Igor Churmeev Date: Wed, 4 Nov 2026 09:58:57 +0000 Subject: [PATCH 1/2] refactor(ui): share the rule that tells a refused stream from one that dropped --- internal/ui/note.go | 13 ++++--------- internal/ui/note_test.go | 2 +- internal/ui/watch.go | 16 ++++++++++++++++ 3 files changed, 21 insertions(+), 10 deletions(-) diff --git a/internal/ui/note.go b/internal/ui/note.go index a01756f..38fe01e 100644 --- a/internal/ui/note.go +++ b/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 } diff --git a/internal/ui/note_test.go b/internal/ui/note_test.go index 382bb87..089eb1b 100644 --- a/internal/ui/note_test.go +++ b/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) diff --git a/internal/ui/watch.go b/internal/ui/watch.go index 0d5ea51..74f3da5 100644 --- a/internal/ui/watch.go +++ b/internal/ui/watch.go @@ -19,8 +19,24 @@ const ( // slowPoll is how often a watched screen polls anyway. The stream // reports changes; this poll only catches one the stream missed. slowPoll = 30 * time.Second + + // refusedWithin is how soon a stream the cluster refuses ends: Nomad 2.0 + // answers 200 and ends the stream at once, with no event. + refusedWithin = 5 * time.Second ) +// streamStart is when a stream opened, and whether it sent an event since. +type streamStart struct { + opened time.Time + heard bool +} + +// endedAtOnce says a stream that ends at now was refused: it sent nothing +// and ends within refusedWithin of opening. +func (s streamStart) endedAtOnce(now time.Time) bool { + return !s.heard && now.Sub(s.opened) < refusedWithin +} + // Messages of the stream. Each carries the id of its stream: a screen that // was left closes its stream, and messages still in flight from it must // not change the screen that is open. From d51e3d26e84579792046134478d34a5780be2d71 Mon Sep 17 00:00:00 2001 From: Igor Churmeev Date: Wed, 4 Nov 2026 10:03:12 +0000 Subject: [PATCH 2/2] feat(ui): show a full client on the client list as soon as it fills up --- docs/clients.md | 7 +- docs/configuration.md | 6 ++ internal/ui/app_test.go | 8 ++ internal/ui/node.go | 12 ++- internal/ui/node_test.go | 188 +++++++++++++++++++++++++++++++++++- internal/ui/page.go | 6 ++ internal/ui/session_test.go | 4 +- internal/ui/watch.go | 37 ++++++- internal/ui/watch_test.go | 2 +- 9 files changed, 260 insertions(+), 10 deletions(-) diff --git a/docs/clients.md b/docs/clients.md index 5560c9f..8c33f24 100644 --- a/docs/clients.md +++ b/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. diff --git a/docs/configuration.md b/docs/configuration.md index 94ab178..e8f94aa 100644 --- a/docs/configuration.md +++ b/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. diff --git a/internal/ui/app_test.go b/internal/ui/app_test.go index 68645d0..cb5cc49 100644 --- a/internal/ui/app_test.go +++ b/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 } diff --git a/internal/ui/node.go b/internal/ui/node.go index cbadbb8..9c719e8 100644 --- a/internal/ui/node.go +++ b/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 diff --git a/internal/ui/node_test.go b/internal/ui/node_test.go index 16c2c71..5613d3b 100644 --- a/internal/ui/node_test.go +++ b/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,7 +55,7 @@ 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)) @@ -58,6 +63,187 @@ func TestClients_TheList(t *testing.T) { 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) diff --git a/internal/ui/page.go b/internal/ui/page.go index 4af85a7..dac4275 100644 --- a/internal/ui/page.go +++ b/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{} diff --git a/internal/ui/session_test.go b/internal/ui/session_test.go index b4a64e9..904363b 100644 --- a/internal/ui/session_test.go +++ b/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) diff --git a/internal/ui/watch.go b/internal/ui/watch.go index 74f3da5..a48ff08 100644 --- a/internal/ui/watch.go +++ b/internal/ui/watch.go @@ -79,13 +79,23 @@ type watchState struct { // refused says the cluster has already refused the stream and the warning // was shown: every screen asks again, and every screen is refused. refused bool + + // fewer says the screen follows its fewer topics: the cluster refused the + // stream of all of them. + fewer bool + + // streamStart is when the open stream opened and whether it sent an event. + streamStart } // watchScreen opens an event stream for what the screen shows. A screen // that watches nothing, or a cluster that does not stream, is polled the -// way it always was. +// way it always was. A screen whose topics were all refused may follow fewer. func (m Model) watchScreen() tea.Cmd { topics := m.screen.page.topics() + if f, ok := m.screen.page.(fallback); ok && m.watch.fewer { + topics = f.fewerTopics() + } if len(topics) == 0 || !nomad.Streams(m.nomadVersion, topics) { return nil } @@ -150,7 +160,7 @@ func reasonOf[T any](stream *nomad.Stream[T]) error { // start keeps the stream and starts reading it. A stream that opens after // the screen that asked for it is gone is closed instead. Once a stream // opens, a later refusal is worth a warning again. -func (w watchState) start(msg watchingMsg) (watchState, tea.Cmd) { +func (w watchState) start(msg watchingMsg, now time.Time) (watchState, tea.Cmd) { if msg.id != w.id { msg.changes.Close() @@ -161,6 +171,7 @@ func (w watchState) start(msg watchingMsg) (watchState, tea.Cmd) { w.changes = msg.changes w.refused = false + w.streamStart = streamStart{opened: now} return w, w.waitForChange() } @@ -173,6 +184,7 @@ func (w watchState) keepChange(msg changeMsg) (watchState, tea.Cmd) { return w, nil } + w.heard = true next := w.waitForChange() if w.settling { @@ -199,13 +211,31 @@ func (m Model) watchEnded(msg watchEndedMsg) (Model, tea.Cmd) { return m, nil } + refused := m.refusedAll() + m.watch = m.watch.end() + // A screen that can follow fewer topics does, with no warning, and its + // poll timer stays as it was. + if refused { + m.watch.fewer = true + + return m, m.watchScreen() + } + // The timer set while the stream was up is slow: the screen polls at // its own pace from now on. return m.noteWatchEnded(msg.err).restartPoll().schedulePoll() } +// refusedAll says the cluster refused the stream of all the topics of a +// screen that can follow fewer: it did not open it, or ended it at once. +func (m Model) refusedAll() bool { + _, ok := m.screen.page.(fallback) + + return ok && !m.watch.fewer && (!m.watch.live() || m.watch.endedAtOnce(m.now)) +} + // noteWatchEnded warns that the screen is no longer live, when urga did not // stop the stream itself. The rows are still there and still polled, so it // shows as a warning. @@ -231,6 +261,7 @@ func (m Model) noteWatchEnded(err error) Model { func (w watchState) end() watchState { w.stop() w.id++ + w.fewer = false return w } @@ -266,7 +297,7 @@ func (m Model) pollEvery() time.Duration { // startWatch keeps the stream the cluster opened. func (m Model) startWatch(msg watchingMsg) (Model, tea.Cmd) { var cmd tea.Cmd - m.watch, cmd = m.watch.start(msg) + m.watch, cmd = m.watch.start(msg, m.now) return m, cmd } diff --git a/internal/ui/watch_test.go b/internal/ui/watch_test.go index 4e59f32..5aacae4 100644 --- a/internal/ui/watch_test.go +++ b/internal/ui/watch_test.go @@ -318,7 +318,7 @@ func TestWatch_EveryListTheClusterTalksAboutIsWatched(t *testing.T) { allocationsView: {nomad.TopicAllocation}, deploymentsView: {nomad.TopicDeployment}, evaluationsView: {nomad.TopicEvaluation}, - nodesView: {nomad.TopicNode}, + nodesView: {nomad.TopicNode, nomad.TopicAllocation}, servicesView: {nomad.TopicService}, nodePoolsView: {nomad.TopicNodePool}, volumesView: {nomad.TopicCSIVolume, nomad.TopicHostVolume, nomad.TopicCSIPlugin},