diff --git a/packages/envd/internal/logs/interceptor.go b/packages/envd/internal/logs/interceptor.go index 92b99c0e39..37b1bdf150 100644 --- a/packages/envd/internal/logs/interceptor.go +++ b/packages/envd/internal/logs/interceptor.go @@ -2,6 +2,7 @@ package logs import ( "context" + "errors" "fmt" "strconv" "strings" @@ -66,7 +67,7 @@ func NewUnaryLogInterceptor(logger *zerolog.Logger) connect.UnaryInterceptorFunc Str(string(OperationIDKey), ctx.Value(OperationIDKey).(string)) if err != nil { - l = l.Int("error_code", int(connect.CodeOf(err))) + l = l.Int("error_code", int(codeOf(err))) } if req != nil { @@ -116,7 +117,7 @@ func LogServerStreamWithoutEvents[T any, R any]( Str(string(OperationIDKey), ctx.Value(OperationIDKey).(string)) if err != nil { - logEvent = logEvent.Int("error_code", int(connect.CodeOf(err))) + logEvent = logEvent.Int("error_code", int(codeOf(err))) } else { logEvent = logEvent.Interface("response", nil) } @@ -146,7 +147,7 @@ func LogClientStreamWithoutEvents[T any, R any]( Str(string(OperationIDKey), ctx.Value(OperationIDKey).(string)) if err != nil { - logEvent = logEvent.Int("error_code", int(connect.CodeOf(err))) + logEvent = logEvent.Int("error_code", int(codeOf(err))) } if res != nil && err == nil { @@ -162,11 +163,39 @@ func LogClientStreamWithoutEvents[T any, R any]( return res, err } -// Return logger with error level if err is not nil, otherwise return logger with debug level +// getErrDebugLogEvent picks the log level by the nature of err, so a routine +// client-side cancellation of a streaming call (the client stopped reading the +// stream — it timed out, was cancelled by a sibling task, or the caller went +// away) is not logged at ERROR next to genuine failures. +// +// - nil -> Debug (normal completion) +// - context.Canceled -> Info (client went away; expected, not a fault) +// - context.DeadlineExceeded -> Warn (the call outran its deadline) +// - anything else -> Error (a real failure) func getErrDebugLogEvent(logger *zerolog.Logger, err error) *zerolog.Event { - if err != nil { + switch { + case err == nil: + return logger.Debug() //nolint:zerologlint // this builds an event, it is not expected to return it + case errors.Is(err, context.Canceled): + return logger.Info().Err(err) //nolint:zerologlint // this builds an event, it is not expected to return it + case errors.Is(err, context.DeadlineExceeded): + return logger.Warn().Err(err) //nolint:zerologlint // this builds an event, it is not expected to return it + default: return logger.Error().Err(err) //nolint:zerologlint // this builds an event, it is not expected to return it } +} - return logger.Debug() //nolint:zerologlint // this builds an event, it is not expected to return it +// codeOf maps an error to a connect code, recognizing the standard context +// sentinels that connect.CodeOf otherwise reports as CodeUnknown. This keeps +// error_code meaningful for downstream aggregation: a client cancellation is +// CodeCanceled and a deadline is CodeDeadlineExceeded, not Unknown. +func codeOf(err error) connect.Code { + switch { + case errors.Is(err, context.Canceled): + return connect.CodeCanceled + case errors.Is(err, context.DeadlineExceeded): + return connect.CodeDeadlineExceeded + default: + return connect.CodeOf(err) + } } diff --git a/packages/envd/internal/logs/interceptor_test.go b/packages/envd/internal/logs/interceptor_test.go new file mode 100644 index 0000000000..1f4edfae28 --- /dev/null +++ b/packages/envd/internal/logs/interceptor_test.go @@ -0,0 +1,89 @@ +package logs + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "testing" + + "connectrpc.com/connect" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// levelOf runs getErrDebugLogEvent against err and returns the zerolog level +// the emitted event carried. +func levelOf(t *testing.T, err error) string { + t.Helper() + + var buf bytesBuffer + logger := zerolog.New(&buf) + getErrDebugLogEvent(&logger, err).Msg("test") + + var parsed map[string]any + require.NoError(t, json.Unmarshal(buf.b, &parsed), "log line: %s", string(buf.b)) + + lvl, _ := parsed["level"].(string) + + return lvl +} + +// bytesBuffer is a tiny io.Writer so the test needs no extra deps beyond what +// the package already uses. +type bytesBuffer struct{ b []byte } + +func (w *bytesBuffer) Write(p []byte) (int, error) { + w.b = append(w.b, p...) + + return len(p), nil +} + +func TestGetErrDebugLogEvent_Level(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + err error + want string + }{ + {"nil is debug", nil, "debug"}, + {"context canceled is info", context.Canceled, "info"}, + {"wrapped canceled is info", fmt.Errorf("stream canceled before start event: %w", context.Canceled), "info"}, + {"deadline is warn", context.DeadlineExceeded, "warn"}, + {"wrapped deadline is warn", fmt.Errorf("x: %w", context.DeadlineExceeded), "warn"}, + {"real failure is error", errors.New("boom"), "error"}, + {"connect internal is error", connect.NewError(connect.CodeInternal, errors.New("boom")), "error"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tc.want, levelOf(t, tc.err)) + }) + } +} + +func TestCodeOf(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + err error + want connect.Code + }{ + {"canceled -> Canceled (not Unknown)", context.Canceled, connect.CodeCanceled}, + {"wrapped canceled -> Canceled", fmt.Errorf("stream canceled: %w", context.Canceled), connect.CodeCanceled}, + {"deadline -> DeadlineExceeded", context.DeadlineExceeded, connect.CodeDeadlineExceeded}, + {"plain error -> Unknown", errors.New("boom"), connect.CodeUnknown}, + {"connect code preserved", connect.NewError(connect.CodeInvalidArgument, errors.New("x")), connect.CodeInvalidArgument}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tc.want, codeOf(tc.err)) + }) + } +} diff --git a/packages/envd/internal/services/process/start.go b/packages/envd/internal/services/process/start.go index bad18ce3e5..c610832478 100644 --- a/packages/envd/internal/services/process/start.go +++ b/packages/envd/internal/services/process/start.go @@ -86,7 +86,11 @@ func (s *Service) handleStart(ctx context.Context, req *connect.Request[rpc.Star select { case <-ctx.Done(): - cancel(ctx.Err()) + // Client went away before the start event was sent (it stopped + // reading the stream). Record the cause so the "server stream end" + // log line can tell a client cancel from a deadline; the process + // itself keeps running on its own context (see procCtx above). + cancel(fmt.Errorf("stream canceled before start event: %w", context.Cause(ctx))) return case event := <-start: @@ -122,7 +126,9 @@ func (s *Service) handleStart(ctx context.Context, req *connect.Request[rpc.Star return } case <-ctx.Done(): - cancel(ctx.Err()) + // Client stopped reading mid-stream (timeout / sibling cancel / + // caller exit). The process is unaffected; only the stream ends. + cancel(fmt.Errorf("stream canceled while streaming output: %w", context.Cause(ctx))) return case event, ok := <-data: @@ -147,7 +153,8 @@ func (s *Service) handleStart(ctx context.Context, req *connect.Request[rpc.Star select { case <-ctx.Done(): - cancel(ctx.Err()) + // Client went away while waiting for the terminal exit event. + cancel(fmt.Errorf("stream canceled before end event: %w", context.Cause(ctx))) return case event, ok := <-end: