From c392fef44949b8a119127582f0f5b3929a314c6a Mon Sep 17 00:00:00 2001 From: Vaibhav Patel Date: Wed, 16 Sep 2026 23:13:49 -0400 Subject: [PATCH] gantry: attribute origin transfer timeouts --- cmd/gantry/main.go | 69 ++++++++++++++++--- cmd/gantry/origin_pull_test.go | 34 +++++++++ designs/gantry-128gib-single-layer-pulls.md | 9 +-- internal/gantry/mirror/mirror.go | 26 +++++++ .../origin_diagnostics_internal_test.go | 40 +++++++++++ 5 files changed, 165 insertions(+), 13 deletions(-) create mode 100644 internal/gantry/mirror/origin_diagnostics_internal_test.go diff --git a/cmd/gantry/main.go b/cmd/gantry/main.go index 5a7afb4a1..a2d933847 100644 --- a/cmd/gantry/main.go +++ b/cmd/gantry/main.go @@ -1837,6 +1837,8 @@ func newPullerPump(infl *inflight.Map, originClient ifaces.OriginPuller, cstore func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cstore ifaces.LocalContentStore, neg *negcache.Cache, lg *slog.Logger, h *inflight.Handle, registry, repository string, d digest.Digest, kind ifaces.OriginRefKind, progressTimeout time.Duration, markPresent func(ctx context.Context, d digest.Digest) bool, onOriginSuccess func(kind string, bytes int64), onDownstreamFailure func(kind, class string), leaseHooks leaseMetricHooks) { defer h.Done() + pullStartedAt := time.Now() + // The requesting peer's stream has closed, so this context is detached from // that request. Bound inactivity rather than total duration: progressing // large layers may run for hours, while a stalled body must still release @@ -1851,11 +1853,18 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs Kind: kind, } - rc, _, err := originClient.Pull(ctx, ref) + rc, expectedSize, err := originClient.Pull(ctx, ref) if err != nil { // A delegated credential is requester-specific. Its origin failure // must not poison the digest-wide cache for another requester. - recordOriginFailure(neg, d, err, lg, "origin pull failed", registry, repository, registryauth.Authorization(ctx) == "") + recordOriginFailure(neg, d, err, lg, "origin pull failed", registry, repository, registryauth.Authorization(ctx) == "", + slog.String("pull_mode", "detached"), + slog.String("deadline_owner", originPullDeadlineOwner(ctx, err)), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", -1), + slog.Int64("written", 0), + ) + return } @@ -1905,9 +1914,17 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs } w, err := cstore.Writer(ctx, d) + deadlineOwner := originPullDeadlineOwner(ctx, err) + if err != nil { releaseLeaseOnFailure() - recordOriginFailure(neg, d, err, lg, "cache writer open failed", registry, repository, true) + recordOriginFailure(neg, d, err, lg, "cache writer open failed", registry, repository, true, + slog.String("pull_mode", "detached"), + slog.String("deadline_owner", deadlineOwner), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", expectedSize), + slog.Int64("written", 0), + ) // Origin returned 2xx (we got past originClient.Pull above) // but the cache writer couldn't open - terminal downstream // failure. Bump p2p_origin_pull_failure_total{class=transient} @@ -1932,7 +1949,13 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs written, err := copyWithOriginProgressTimeout(ctx, cancel, w, rc, progressTimeout) if err != nil { releaseLeaseOnFailure() - recordOriginFailure(neg, d, err, lg, "origin pull copy failed", registry, repository, true) + recordOriginFailure(neg, d, err, lg, "origin pull copy failed", registry, repository, true, + slog.String("pull_mode", "detached"), + slog.String("deadline_owner", originPullDeadlineOwner(ctx, err)), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", expectedSize), + slog.Int64("written", written), + ) // io.Copy could have failed because origin truncated the // stream OR because the local cache writer errored. We // can't easily distinguish - but we already passed origin's @@ -1948,9 +1971,18 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs return } - if err := w.Commit(ctx); err != nil { + commitErr := w.Commit(ctx) + deadlineOwner = originPullDeadlineOwner(ctx, commitErr) + + if commitErr != nil { releaseLeaseOnFailure() - recordOriginFailure(neg, d, err, lg, "cache commit failed (digest mismatch or io error)", registry, repository, true) + recordOriginFailure(neg, d, commitErr, lg, "cache commit failed (digest mismatch or io error)", registry, repository, true, + slog.String("pull_mode", "detached"), + slog.String("deadline_owner", deadlineOwner), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", expectedSize), + slog.Int64("written", written), + ) // Commit failure means EITHER the cache's internal // digestpipe caught a content mismatch (origin lied) OR // the local cache had an I/O error at finalize. Either @@ -2102,6 +2134,22 @@ func copyWithOriginProgressTimeout(ctx context.Context, cancel context.CancelCau return written, err } +func originPullDeadlineOwner(ctx context.Context, err error) string { + switch { + case errors.Is(context.Cause(ctx), errOriginPullNoProgress): + return "progress" + case errors.Is(context.Cause(ctx), context.Canceled), errors.Is(context.Cause(ctx), context.DeadlineExceeded): + return "caller" + } + + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return "transport" + } + + return "none" +} + // recordOriginFailure classifies err and records the failure into the // per-puller the design doc negative cache. Non-the design doc callers (e.g. cache I/O // errors not covered by *ifaces.OriginError) are bucketed as @@ -2109,7 +2157,7 @@ func copyWithOriginProgressTimeout(ctx context.Context, cancel context.CancelCau // them. The log is emitted at WARN regardless of class. recordCooldown is // false for requester-specific origin failures that are unsafe to store in a // digest-only shared cache. -func recordOriginFailure(neg *negcache.Cache, d digest.Digest, err error, lg *slog.Logger, msg, registry, repository string, recordCooldown bool) { +func recordOriginFailure(neg *negcache.Cache, d digest.Digest, err error, lg *slog.Logger, msg, registry, repository string, recordCooldown bool, details ...any) { class := ifaces.FailureTransient var oe *ifaces.OriginError @@ -2117,13 +2165,16 @@ func recordOriginFailure(neg *negcache.Cache, d digest.Digest, err error, lg *sl class = oe.Class } - lg.Warn(msg, + attrs := []any{ slog.String("digest", d.String()), slog.String("registry", registry), slog.String("repository", repository), slog.String("failure_class", string(class)), slog.Any("err", err), - ) + } + attrs = append(attrs, details...) + + lg.Warn(msg, attrs...) if neg != nil && recordCooldown { neg.RecordFailure(d, class) diff --git a/cmd/gantry/origin_pull_test.go b/cmd/gantry/origin_pull_test.go index b6679bf88..6b6b7fee9 100644 --- a/cmd/gantry/origin_pull_test.go +++ b/cmd/gantry/origin_pull_test.go @@ -178,6 +178,12 @@ type pacedReader struct { delay time.Duration } +type originTimeoutError struct{} + +func (originTimeoutError) Error() string { return "timed out" } +func (originTimeoutError) Timeout() bool { return true } +func (originTimeoutError) Temporary() bool { return true } + func (r *pacedReader) Read(p []byte) (int, error) { if r.remaining == 0 { return 0, io.EOF @@ -254,6 +260,34 @@ func TestRunOriginPullKeepsWriterContextAlive(t *testing.T) { } } +func TestOriginPullDeadlineOwner(t *testing.T) { + progressCtx, progressCancel := context.WithCancelCause(context.Background()) + progressCancel(errOriginPullNoProgress) + + callerCtx, callerCancel := context.WithCancel(context.Background()) + callerCancel() + + tests := []struct { + name string + ctx context.Context + err error + want string + }{ + {name: "progress", ctx: progressCtx, err: context.Canceled, want: "progress"}, + {name: "caller", ctx: callerCtx, err: context.Canceled, want: "caller"}, + {name: "transport", ctx: context.Background(), err: originTimeoutError{}, want: "transport"}, + {name: "none", ctx: context.Background(), err: errors.New("failed"), want: "none"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := originPullDeadlineOwner(test.ctx, test.err); got != test.want { + t.Fatalf("owner = %q; want %q", got, test.want) + } + }) + } +} + func TestRunOriginPull_ReopenFailurePreventsAdvertiseAndSuccess(t *testing.T) { body := []byte("committed-but-not-reopenable") d := trackerDigestOf(body) diff --git a/designs/gantry-128gib-single-layer-pulls.md b/designs/gantry-128gib-single-layer-pulls.md index 968c03f38..03c54611d 100644 --- a/designs/gantry-128gib-single-layer-pulls.md +++ b/designs/gantry-128gib-single-layer-pulls.md @@ -122,7 +122,7 @@ downloading its prefix from origin again. | ID | Issue | Evidence | Owning PR | Status | |---|---|---|---|---| -| L7 | Logs expose a context deadline, but do not identify the deadline owner, bytes transferred, elapsed time, or whether the pull was live or detached. | `cmd/gantry/main.go:1963-1977`, `internal/gantry/mirror/mirror.go:1151-1160` | PR 3 | Open | +| L7 | Logs expose a context deadline, but do not identify the deadline owner, bytes transferred, elapsed time, or whether the pull was live or detached. | `cmd/gantry/main.go:1963-1977`, `internal/gantry/mirror/mirror.go:1151-1160` | PR 3 | Addressed | ### Constraints, not open Gantry issues @@ -269,10 +269,11 @@ watchdog. **Purpose:** Make field failures attributable without changing transfer semantics. -- [ ] Log the pull mode (`live` or `detached`), deadline owner, expected size, +- [x] Log the pull mode (`live` or `detached`), deadline owner, expected size, bytes transferred, and elapsed time. -- [ ] Add bounded-cardinality metrics for deadline owner and pull mode. -- [ ] Distinguish caller cancellation, connection/header timeout, body idle +- [x] Use the existing bounded byte and failure metrics alongside the new log + fields instead of adding a second counter for the same failures. +- [x] Distinguish caller cancellation, connection/header timeout, body idle timeout, and downstream writer failure. **Addresses:** L7. diff --git a/internal/gantry/mirror/mirror.go b/internal/gantry/mirror/mirror.go index ddf945141..2ffd9cb6b 100644 --- a/internal/gantry/mirror/mirror.go +++ b/internal/gantry/mirror/mirror.go @@ -1120,6 +1120,7 @@ func (s *Server) serveHeadMiss(ctx context.Context, w http.ResponseWriter, d dig // short-circuited in serveHeadMiss above so it never reaches this // section. func (s *Server) serveFromOrigin(ctx context.Context, w http.ResponseWriter, d digest.Digest, kind ifaces.OriginRefKind, upstream, repo string, logger *slog.Logger) { + pullStartedAt := time.Now() pRef := ifaces.OriginRef{Registry: upstream, Repository: repo, Digest: d, Kind: kind} if s.liveStreamThrough { @@ -1128,6 +1129,14 @@ func (s *Server) serveFromOrigin(ctx context.Context, w http.ResponseWriter, d d pr, psize, perr := s.origin.Pull(ctx, pRef) if perr != nil { + logger.Debug("mirror: live origin pull failed", + slog.String("pull_mode", "live"), + slog.String("deadline_owner", originDeadlineOwner(ctx, perr)), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", -1), + slog.Int64("written", 0), + slog.Any("err", perr), + ) // the design doc negative-cache: classify and record the origin-side // failure so the next direct-origin attempt for the same // digest on this node short-circuits on the recently_failed @@ -1160,6 +1169,10 @@ func (s *Server) serveFromOrigin(ctx context.Context, w http.ResponseWriter, d d if streamErr != nil { logger.Debug("mirror: live origin stream failed", + slog.String("pull_mode", "live"), + slog.String("deadline_owner", originDeadlineOwner(ctx, streamErr)), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", psize), slog.Int64("written", written), slog.Any("err", streamErr), ) @@ -1331,6 +1344,19 @@ func (s *Server) serveFromOrigin(ctx context.Context, w http.ResponseWriter, d d } } +func originDeadlineOwner(ctx context.Context, err error) string { + if errors.Is(context.Cause(ctx), context.Canceled) || errors.Is(context.Cause(ctx), context.DeadlineExceeded) { + return "caller" + } + + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return "transport" + } + + return "none" +} + // peerFallbackResult is the outcome of tryPeerFallback. type peerFallbackResult int diff --git a/internal/gantry/mirror/origin_diagnostics_internal_test.go b/internal/gantry/mirror/origin_diagnostics_internal_test.go new file mode 100644 index 000000000..5fcc8361d --- /dev/null +++ b/internal/gantry/mirror/origin_diagnostics_internal_test.go @@ -0,0 +1,40 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package mirror + +import ( + "context" + "errors" + "testing" +) + +type originTimeoutError struct{} + +func (originTimeoutError) Error() string { return "timed out" } +func (originTimeoutError) Timeout() bool { return true } +func (originTimeoutError) Temporary() bool { return true } + +func TestOriginDeadlineOwner(t *testing.T) { + callerCtx, callerCancel := context.WithCancel(context.Background()) + callerCancel() + + tests := []struct { + name string + ctx context.Context + err error + want string + }{ + {name: "caller", ctx: callerCtx, err: context.Canceled, want: "caller"}, + {name: "transport", ctx: context.Background(), err: originTimeoutError{}, want: "transport"}, + {name: "none", ctx: context.Background(), err: errors.New("failed"), want: "none"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := originDeadlineOwner(test.ctx, test.err); got != test.want { + t.Fatalf("owner = %q; want %q", got, test.want) + } + }) + } +}