Log destination-caused failures at warn instead of error (#1341)

Failures caused by the user-configured destination (storage 4xx, DNS
resolution of user-provided hosts, stream endpoint failures) dominate
error-level logs while carrying no operational signal. Classify them
via IsDestinationError and log at warn in Controller.OnError and the
image sink; internal failures keep error level.

Stream endpoint and websocket write failures carry no matchable error
type, so they are tagged with MarkDestinationError at the call site,
preserving the user-facing message and psrpc code.

Backup storage is operator-owned, so backup upload failures flatten
their error chains to stay out of the destination classification.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: milos-lk

pkg/errors/errors.go

@@ -1,4 +1,4 @@
-// Copyright 2023 LiveKit, Inc.
+// Copyright 2026 LiveKit, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -16,9 +16,11 @@ package errors
 
 import (
 	"errors"
+	"net"
 	"strings"
 
 	"github.com/livekit/psrpc"
+	"github.com/livekit/storage"
 )
 
 func New(err string) error {
@@ -33,6 +35,46 @@ func As(err error, target any) bool {
 	return errors.As(err, target)
 }
 
+var errDestinationMarker = errors.New("destination error")
+
+type destinationError struct {
+	err error
+}
+
+func (e *destinationError) Error() string {
+	return e.err.Error()
+}
+
+func (e *destinationError) Unwrap() []error {
+	return []error{e.err, errDestinationMarker}
+}
+
+// MarkDestinationError tags err as caused by the user-configured destination,
+// preserving its message and unwrap chain.
+func MarkDestinationError(err error) error {
+	if err == nil {
+		return nil
+	}
+	return &destinationError{err: err}
+}
+
+// IsDestinationError reports whether a failure was caused by the user-configured
+// destination rather than by the egress service: a marked error, a 4xx response
+// from storage, or a DNS resolution failure for a user-provided host.
+func IsDestinationError(err error) bool {
+	if errors.Is(err, errDestinationMarker) {
+		return true
+	}
+
+	var statusErr *storage.ErrorWithStatusCode
+	if errors.As(err, &statusErr) && statusErr.StatusCode >= 400 && statusErr.StatusCode < 500 {
+		return true
+	}
+
+	var dnsErr *net.DNSError
+	return errors.As(err, &dnsErr) && dnsErr.IsNotFound
+}
+
 type ErrArray struct {
 	errs []error
 }

pkg/pipeline/controller.go

@@ -491,7 +491,7 @@ func (c *Controller) streamFailed(ctx context.Context, stream *config.Stream, st
 
 	// fail egress if no outputs remaining
 	if c.OutputCount.Load() == 0 {
-		return psrpc.NewError(psrpc.Unavailable, streamErr)
+		return psrpc.NewError(psrpc.Unavailable, errors.MarkDestinationError(streamErr))
 	}
 
 	logger.Infow("stream failed",
@@ -612,7 +612,11 @@ func (c *Controller) sendEOS() {
 }
 
 func (c *Controller) OnError(err error) {
-	logger.Errorw("controller onError invoked", err)
+	if errors.IsDestinationError(err) {
+		logger.Warnw("controller onError invoked", err)
+	} else {
+		logger.Errorw("controller onError invoked", err)
+	}
 	if errors.Is(err, errors.ErrPipelineFrozen) && c.Debug.EnableProfiling {
 		c.generateDotFile("error")
 		c.generatePProf()

pkg/pipeline/sink/image.go

@@ -108,7 +108,11 @@ func (s *ImageSink) Start() error {
 				continue
 			}
 			if err := s.handleNewImage(update); err != nil {
-				logger.Errorw("new image handling failed", err)
+				if errors.IsDestinationError(err) {
+					logger.Warnw("new image handling failed", err)
+				} else {
+					logger.Errorw("new image handling failed", err)
+				}
 				failed = true
 				s.callbacks.OnError(err)
 			}

pkg/pipeline/sink/uploader/uploader.go

@@ -177,7 +177,7 @@ func (u *Uploader) Upload(
 			return "", 0, psrpc.NewErrorf(psrpc.InvalidArgument,
 				"primary: %s\nbackup: %s", primaryErr.Error(), backupErr.Error())
 		}
-		return "", 0, psrpc.NewError(psrpc.InvalidArgument, backupErr)
+		return "", 0, psrpc.NewErrorf(psrpc.InvalidArgument, "%s", backupErr.Error())
 	}
 
 	return "", 0, primaryErr

pkg/pipeline/sink/websocket.go

@@ -99,7 +99,7 @@ func newWebsocketSink(
 				if err == io.EOF {
 					return gst.FlowEOS
 				}
-				callbacks.OnError(psrpc.NewError(psrpc.Unavailable, err))
+				callbacks.OnError(psrpc.NewError(psrpc.Unavailable, errors.MarkDestinationError(err)))
 			}
 
 			return gst.FlowOK