From ea302fc5753971efc8f8249692f83ccbfa9daa58 Mon Sep 17 00:00:00 2001 From: Igor Churmeev Date: Wed, 4 Nov 2026 09:35:40 +0000 Subject: [PATCH] fix(nomad): send the error of a broken file stream before its lines close --- internal/nomad/agentlogs.go | 12 +------ internal/nomad/files.go | 36 +++++++++++++------- internal/nomad/files_test.go | 53 ++++++++++++++++++++++++++++++ internal/nomad/logs_stream_test.go | 47 ++++++++++++++++++++++++++ 4 files changed, 126 insertions(+), 22 deletions(-) diff --git a/internal/nomad/agentlogs.go b/internal/nomad/agentlogs.go index 295ac9a..839a3ae 100644 --- a/internal/nomad/agentlogs.go +++ b/internal/nomad/agentlogs.go @@ -51,17 +51,7 @@ func (c *Client) followAgent(target AgentTarget, level string, lines chan<- stri frames, failed := c.api.Agent().Monitor(cancel, c.agentQuery(target, map[string]string{"log_level": level})) - if frames != nil { - for chunk := range streamLines(frames, notCut, nil, nil, cancel) { - if !sendText(lines, chunk, cancel) { - return - } - } - } - - // The error goes out after the lines and before they close, so a reader - // that sees the end finds it. - passFailures(failed, errs, cancel) + followFrames(frames, failed, notCut, nil, lines, errs, cancel) } // agentPast reads the log file of the agent and keeps its last MiB at the diff --git a/internal/nomad/files.go b/internal/nomad/files.go index 5a47c99..318fa18 100644 --- a/internal/nomad/files.go +++ b/internal/nomad/files.go @@ -68,7 +68,8 @@ var ErrNotText = errors.New("not text") // File follows a file of an allocation as it grows: from its start, or when // it is big, from its last MiB. The caller closes the stream when it stops -// reading, otherwise the request stays open. Do not call it for a pipe: the +// reading, otherwise the request stays open. A failure arrives on its Err after +// the lines before it and before Lines closes. Do not call it for a pipe: the // client reads its first bytes to detect the content type, and waits. func (c *Client) File(ctx context.Context, namespace, allocID, name string) (*LogStream, error) { // The stream is requested from the node that runs the allocation when it @@ -98,22 +99,35 @@ func (c *Client) File(ctx context.Context, namespace, allocID, name string) (*Lo atEnd = func(frame *api.StreamFrame) bool { return frame.Offset >= info.Size } } - lines := streamLines(frames, func(*api.StreamFrame) bool { return from > 0 }, atEnd, nil, cancel) + lines := make(chan string) + out := make(chan error, 1) - return &LogStream{Lines: lines, Err: failures(errs, cancel), Size: info.Size, From: from, cancel: cancel}, nil -} + go func() { + defer close(lines) -// failures passes on the errors of a stream. Its end is not a failure: the -// stream of a file reports it as an error, the stream of a log does not. -func failures(errs <-chan error, cancel <-chan struct{}) <-chan error { - out := make(chan error, 1) + followFrames(frames, errs, func(*api.StreamFrame) bool { return from > 0 }, atEnd, lines, out, cancel) + }() - go passFailures(errs, out, cancel) + return &LogStream{Lines: lines, Err: out, Size: info.Size, From: from, cancel: cancel}, nil +} + +// followFrames sends the lines of the frames, then the first failure. The +// failure goes out after the lines and before they close (the caller closes +// them when this returns), so a reader that sees the end finds it. +func followFrames(frames <-chan *api.StreamFrame, failed <-chan error, startsCut, atEnd func(*api.StreamFrame) bool, lines chan<- string, errs chan<- error, cancel <-chan struct{}) { + if frames != nil { + for chunk := range streamLines(frames, startsCut, atEnd, nil, cancel) { + if !sendText(lines, chunk, cancel) { + return + } + } + } - return out + passFailures(failed, errs, cancel) } -// passFailures sends the first error of errs to out, unless it is an end. +// passFailures sends the first error of errs to out, unless it is an end. The +// stream of a file reports its end as an error, the stream of a log does not. func passFailures(errs <-chan error, out chan<- error, cancel <-chan struct{}) { select { case err := <-errs: diff --git a/internal/nomad/files_test.go b/internal/nomad/files_test.go index 355454a..eef9426 100644 --- a/internal/nomad/files_test.go +++ b/internal/nomad/files_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "io" "net/http" "net/http/httptest" "slices" @@ -259,3 +260,55 @@ func TestFile_WhatIsNotText(t *testing.T) { r.NotEqual("/v1/client/fs/stream/af1f37df", req.path) } } + +func TestFile_ABreakIsItsErrorAfterItsLines(t *testing.T) { + r := require.New(t) + + // The body ends inside a frame: the decoder fails after the last line. + client, _ := clusterServer(t, map[string]string{ + "/v1/allocation/af1f37df": allocInfo, + "/v1/client/fs/stat/af1f37df": `{"Name": "app.env", "Size": 4, "FileMode": "-rw-r--r--", "ContentType": "text/plain"}`, + "/v1/client/fs/stream/af1f37df": streamed(t, 4, "a=1\n") + `{"Data":"`, + }) + + stream, err := client.File(context.Background(), "production", "af1f37df", "/t/local/app.env") + r.NoError(err) + t.Cleanup(stream.Close) + + select { + case err := <-stream.Err: + t.Fatalf("the error came before the last line: %v", err) + case <-time.After(200 * time.Millisecond): + } + + r.Equal("a=1\n", readAll(t, stream)) + + select { + case err := <-stream.Err: + r.ErrorIs(err, io.ErrUnexpectedEOF) + default: + t.Fatal("the lines ended before the error") + } +} + +func TestFile_ARefusedStreamIsItsError(t *testing.T) { + r := require.New(t) + + client, _ := clusterServer(t, map[string]string{ + "/v1/allocation/af1f37df": allocInfo, + "/v1/client/fs/stat/af1f37df": `{"Name": "app.env", "Size": 4, "FileMode": "-rw-r--r--", "ContentType": "text/plain"}`, + }) + + stream, err := client.File(context.Background(), "production", "af1f37df", "/t/local/app.env") + r.NoError(err) + t.Cleanup(stream.Close) + + select { + case err := <-stream.Err: + r.ErrorContains(err, "404") + case <-time.After(2 * time.Second): + t.Fatal("no error was sent") + } + + r.Empty(readAll(t, stream)) +} diff --git a/internal/nomad/logs_stream_test.go b/internal/nomad/logs_stream_test.go index b2a0393..4cdf528 100644 --- a/internal/nomad/logs_stream_test.go +++ b/internal/nomad/logs_stream_test.go @@ -1,6 +1,7 @@ package nomad import ( + "io" "strings" "testing" "time" @@ -291,3 +292,49 @@ func TestStreamLines_HoldALogLineUntilItsBreakWhileFramesStayOpen(t *testing.T) require.Equal(t, []string{"waiting done\n"}, collect(t, lines, 1)) } + +func TestFollowFrames_TheLinesWaitForAnErrorThatComesAfterTheFrames(t *testing.T) { + r := require.New(t) + + frames := make(chan *api.StreamFrame, 1) + failed := make(chan error, 1) + lines := make(chan string) + errs := make(chan error, 1) + cancel := make(chan struct{}) + + t.Cleanup(func() { close(cancel) }) + + go func() { + defer close(lines) + + followFrames(frames, failed, neverCut, nil, lines, errs, cancel) + }() + + frames <- &api.StreamFrame{Data: []byte("a=1\n")} + close(frames) + + select { + case line := <-lines: + r.Equal("a=1\n", line) + case <-time.After(2 * time.Second): + t.Fatal("no line was sent") + } + + // The frames are closed and the error is not here yet: the lines stay open. + select { + case line, ok := <-lines: + t.Fatalf("the lines sent %q (open %v) before the error", line, ok) + case <-time.After(100 * time.Millisecond): + } + + failed <- io.ErrUnexpectedEOF + + r.True(ended(lines)) + + select { + case err := <-errs: + r.ErrorIs(err, io.ErrUnexpectedEOF) + default: + t.Fatal("the lines ended before the error") + } +}