From d68f98fb1ef38fa17a1d3630aeb89b22b72e56f7 Mon Sep 17 00:00:00 2001 From: Wilson Silva Date: Tue, 8 Sep 2026 18:30:08 +0100 Subject: [PATCH] fix(serverless): keep a log follow alive until the stream opens Following an application that has not written a log entry yet ended by itself after about thirty seconds. The log store holds the tail request open without writing its response headers, so the edge times the request out before the first entry arrives, and any non-200 status on the tail route was fatal. A gateway status is now retryable: the follow reports it once and waits for the stream to open, the way it already waits after the server ends a stream. Nothing has been delivered when a gateway answers, so a reconnect cannot repeat an entry. Jira Ticket -> https://runware.atlassian.net/browse/RUNSERV-516 --- docs/runware_serverless_apps_logs.md | 10 +-- internal/api/serverless/logs.go | 42 +++++++++++- internal/api/serverless/logs_test.go | 47 +++++++++++++ internal/cmd/serverless/apps_logs.go | 34 +++++++--- internal/cmd/serverless/apps_logs_test.go | 83 +++++++++++++++++++++++ 5 files changed, 200 insertions(+), 16 deletions(-) diff --git a/docs/runware_serverless_apps_logs.md b/docs/runware_serverless_apps_logs.md index c60f219..b696cea 100644 --- a/docs/runware_serverless_apps_logs.md +++ b/docs/runware_serverless_apps_logs.md @@ -9,10 +9,12 @@ Show recent application logs, oldest first, and optionally follow new ones. The recent page is read from the runtime log query over --window (default 1h), and --limit and --cursor page through it. With --follow the command prints the recent page, then streams new entries until interrupted; the stream reconnects -when the server ends it. The live stream has no window, so --window, --limit -and --cursor apply to the recent page only, and --cursor cannot be combined -with --follow. Entries written between the recent page and the start of the -stream, or while the stream reconnects, can be missed or repeated. +when the server ends it, and waits for the stream to open when a gateway +answers first, which is what an application that has written nothing does. The +live stream has no window, so --window, --limit and --cursor apply to the +recent page only, and --cursor cannot be combined with --follow. Entries +written between the recent page and the start of the stream, or while the +stream reconnects, can be missed or repeated. In table format each entry is one line: time, level and message. In json or yaml format the recent page is printed as one document; with --follow every diff --git a/internal/api/serverless/logs.go b/internal/api/serverless/logs.go index 6f8b23b..9598951 100644 --- a/internal/api/serverless/logs.go +++ b/internal/api/serverless/logs.go @@ -53,6 +53,34 @@ func (e *TailStreamError) Error() string { return "log stream failed: " + e.Detail } +// TailUnavailableError reports that a gateway answered the tail request instead +// of the log store, so the stream never opened. No entry was delivered, which +// makes a reconnect free of repeated output. An app that has not written an +// entry yet answers this way: the store holds the request without writing its +// response headers, and the edge times it out before the first entry arrives. +type TailUnavailableError struct { + StatusCode int + Err error +} + +func (e *TailUnavailableError) Error() string { + return "log stream unavailable: " + e.Err.Error() +} + +func (e *TailUnavailableError) Unwrap() error { return e.Err } + +// isGatewayStatus reports whether a proxy in front of the log store, rather +// than the store itself, ended the request. These are transient, so a tail +// reconnects on them instead of failing. +func isGatewayStatus(statusCode int) bool { + switch statusCode { + case http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout: + return true + default: + return false + } +} + // GetLogEntries returns one page of a named log query. func (c *Client) GetLogEntries(ctx context.Context, queryID string, params GetLogEntriesParams) (LogEntryPage, error) { if c.apiKey == "" { @@ -105,8 +133,9 @@ func (c *Client) GetLogEntries(ctx context.Context, queryID string, params GetLo // TailLogs follows a named log query for one app and hands every new entry to // emit. It returns ErrTailEnded when the server closes the stream cleanly, a -// *TailStreamError when the server reports a failure, ctx.Err() when the -// caller stops, and any error emit returns. +// *TailStreamError when the server reports a failure, a *TailUnavailableError +// when a gateway answers before the stream opens, ctx.Err() when the caller +// stops, and any error emit returns. func (c *Client) TailLogs(ctx context.Context, queryID, appID string, emit func(LogEntry) error) error { if c.apiKey == "" { return transport.ErrNoAPIKey @@ -127,7 +156,14 @@ func (c *Client) TailLogs(ctx context.Context, queryID, appID string, emit func( if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(io.LimitReader(resp.Body, maxSSEFrameBytes)) c.logResponse(ctx, resp, body) - return problemFromBody(body, resp.StatusCode) + problem := problemFromBody(body, resp.StatusCode) + if isGatewayStatus(resp.StatusCode) { + return &TailUnavailableError{ + StatusCode: resp.StatusCode, + Err: problem, + } + } + return problem } c.logResponse(ctx, resp, nil) if mediaType, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type")); mediaType != "text/event-stream" { diff --git a/internal/api/serverless/logs_test.go b/internal/api/serverless/logs_test.go index 3d01813..1d4d775 100644 --- a/internal/api/serverless/logs_test.go +++ b/internal/api/serverless/logs_test.go @@ -247,6 +247,53 @@ func TestTailLogs_NonStreamStatusIsAProblem(t *testing.T) { } } +func TestTailLogs_GatewayStatusIsRetryable(t *testing.T) { + for _, status := range []int{http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout} { + t.Run(http.StatusText(status), func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + // What the edge answers with: plain text, not a problem document. + w.Header().Set("Content-Type", "text/plain; charset=UTF-8") + w.WriteHeader(status) + _, _ = fmt.Fprintf(w, "error code: %d", status) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + err := c.TailLogs(context.Background(), LogQueryRuntimeTail, testAppID, func(LogEntry) error { return nil }) + unavailable, ok := errors.AsType[*TailUnavailableError](err) + if !ok || unavailable.StatusCode != status { + t.Fatalf("err = %#v", err) + } + if !strings.HasPrefix(unavailable.Error(), "log stream unavailable: ") { + t.Fatalf("message = %q", unavailable.Error()) + } + re, ok := errors.AsType[*transport.RunwareError](err) + if !ok || re.StatusCode != status { + t.Fatalf("unwrapped err = %#v", err) + } + }) + } +} + +func TestTailLogs_NonGatewayServerErrorStaysFatal(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", testLogsProblemJSON) + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(`{"title":"Internal Server Error","status":500,"detail":"the log store failed"}`)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + err := c.TailLogs(context.Background(), LogQueryRuntimeTail, testAppID, func(LogEntry) error { return nil }) + if _, ok := errors.AsType[*TailUnavailableError](err); ok { + t.Fatalf("500 must not be retryable, err = %#v", err) + } + re, ok := errors.AsType[*transport.RunwareError](err) + if !ok || re.StatusCode != http.StatusInternalServerError { + t.Fatalf("err = %#v", err) + } +} + func TestTailLogs_RejectsNonEventStreamBody(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") diff --git a/internal/cmd/serverless/apps_logs.go b/internal/cmd/serverless/apps_logs.go index e75f996..2d4a8cc 100644 --- a/internal/cmd/serverless/apps_logs.go +++ b/internal/cmd/serverless/apps_logs.go @@ -20,8 +20,9 @@ import ( ) // tailReconnectDelay separates two connection attempts after the server -// reports a stream failure, or after a clean end of a stream that lived -// shorter than this, so a server that closes at once is not hammered. +// reports a stream failure, after a gateway answers before the stream opens, +// or after a clean end of a stream that lived shorter than this, so a server +// that closes at once is not hammered. const tailReconnectDelay = 2 * time.Second // logWindows lists the accepted --window values, in the order they are documented. @@ -50,10 +51,12 @@ func newAppsLogsCmd(logger *log.Logger) *cobra.Command { The recent page is read from the runtime log query over --window (default 1h), and --limit and --cursor page through it. With --follow the command prints the recent page, then streams new entries until interrupted; the stream reconnects -when the server ends it. The live stream has no window, so --window, --limit -and --cursor apply to the recent page only, and --cursor cannot be combined -with --follow. Entries written between the recent page and the start of the -stream, or while the stream reconnects, can be missed or repeated. +when the server ends it, and waits for the stream to open when a gateway +answers first, which is what an application that has written nothing does. The +live stream has no window, so --window, --limit and --cursor apply to the +recent page only, and --cursor cannot be combined with --follow. Entries +written between the recent page and the start of the stream, or while the +stream reconnects, can be missed or repeated. In table format each entry is one line: time, level and message. In json or yaml format the recent page is printed as one document; with --follow every @@ -187,10 +190,12 @@ func logEmitter(format output.Format, out io.Writer) func(serverlessapi.LogEntry } // followLogs keeps a live stream open until ctx is cancelled. A clean end of a -// long-lived stream reconnects at once; a reported stream failure, or a clean -// end of a short-lived stream, reconnects after tailReconnectDelay. Any other -// error is returned. Cancellation is a normal exit. +// long-lived stream reconnects at once; a reported stream failure, a gateway +// answering before the stream opens, or a clean end of a short-lived stream, +// reconnects after tailReconnectDelay. Any other error is returned. +// Cancellation is a normal exit. func followLogs(ctx context.Context, tail logTailer, emit func(serverlessapi.LogEntry) error, errOut io.Writer) error { + reportedUnavailable := false for { started := time.Now() err := tail(ctx, emit) @@ -198,10 +203,21 @@ func followLogs(ctx context.Context, tail logTailer, emit func(serverlessapi.Log return nil //nolint:nilerr // Cancellation is the normal way a follow ends. } _, streamFailed := errors.AsType[*serverlessapi.TailStreamError](err) + unavailable, gatewayAnswered := errors.AsType[*serverlessapi.TailUnavailableError](err) switch { case streamFailed: + reportedUnavailable = false _, _ = fmt.Fprintf(errOut, "%v; reconnecting\n", err) + case gatewayAnswered: + // Every attempt on an app that has written nothing answers this + // way, so the notice would repeat for as long as the app stays + // quiet. Say it once, then wait in silence for the stream to open. + if !reportedUnavailable { + reportedUnavailable = true + _, _ = fmt.Fprintf(errOut, "%v; waiting for the stream to open\n", unavailable) + } case errors.Is(err, serverlessapi.ErrTailEnded): + reportedUnavailable = false if time.Since(started) >= tailReconnectDelay { continue } diff --git a/internal/cmd/serverless/apps_logs_test.go b/internal/cmd/serverless/apps_logs_test.go index 37a4f9e..4044279 100644 --- a/internal/cmd/serverless/apps_logs_test.go +++ b/internal/cmd/serverless/apps_logs_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "net/http" "strings" "testing" "testing/synctest" @@ -231,6 +232,88 @@ func TestFollowLogs_WaitsBeforeReconnectingAfterStreamFailure(t *testing.T) { }) } +func TestFollowLogs_WaitsForTheStreamToOpenWhenAGatewayAnswers(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + var attempts []time.Time + var emitted []string + tail := func(_ context.Context, emit func(serverlessapi.LogEntry) error) error { + attempts = append(attempts, time.Now()) + // Three gateway answers, as a quiet application produces, then the + // stream opens and carries one entry. + if len(attempts) <= 3 { + return &serverlessapi.TailUnavailableError{ + StatusCode: http.StatusGatewayTimeout, + Err: errors.New("the origin did not respond in time"), + } + } + if err := emit(serverlessapi.LogEntry{Body: testLogBodyReady}); err != nil { + return err + } + cancel() + return context.Canceled + } + var errOut bytes.Buffer + emit := func(entry serverlessapi.LogEntry) error { + emitted = append(emitted, entry.Body) + return nil + } + if err := followLogs(ctx, tail, emit, &errOut); err != nil { + t.Fatalf("followLogs: %v", err) + } + if len(attempts) != 4 { + t.Fatalf("attempts = %v", attempts) + } + for i := 1; i < len(attempts); i++ { + if attempts[i].Sub(attempts[i-1]) != tailReconnectDelay { + t.Fatalf("attempt %d waited %v", i, attempts[i].Sub(attempts[i-1])) + } + } + if len(emitted) != 1 || emitted[0] != testLogBodyReady { + t.Fatalf("emitted = %v", emitted) + } + // The notice names the condition once, however long the wait lasts. + if got := strings.Count(errOut.String(), "waiting for the stream to open"); got != 1 { + t.Fatalf("notice count = %d, stderr = %q", got, errOut.String()) + } + if !strings.Contains(errOut.String(), "log stream unavailable: the origin did not respond in time") { + t.Fatalf("stderr = %q", errOut.String()) + } + }) +} + +func TestFollowLogs_NoticesTheGatewayAgainAfterAStreamOpened(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + attempts := 0 + unavailable := func() error { + return &serverlessapi.TailUnavailableError{ + StatusCode: http.StatusBadGateway, + Err: errors.New("bad gateway"), + } + } + tail := func(context.Context, func(serverlessapi.LogEntry) error) error { + attempts++ + switch attempts { + case 1, 3: + return unavailable() + case 2: + return serverlessapi.ErrTailEnded + default: + cancel() + return context.Canceled + } + } + var errOut bytes.Buffer + if err := followLogs(ctx, tail, func(serverlessapi.LogEntry) error { return nil }, &errOut); err != nil { + t.Fatalf("followLogs: %v", err) + } + if got := strings.Count(errOut.String(), "waiting for the stream to open"); got != 2 { + t.Fatalf("notice count = %d, stderr = %q", got, errOut.String()) + } + }) +} + func TestFollowLogs_ReturnsOtherErrors(t *testing.T) { boom := errors.New("boom") tail := func(context.Context, func(serverlessapi.LogEntry) error) error { return boom }