gantry: attribute origin transfer timeouts

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,28 +2134,47 @@ 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
 // FailureTransient: see runOriginPull's docs for why we still record
 // 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
 	if errors.As(err, &oe) && oe.Class != ifaces.FailureUnspecified {
 		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)

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)

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.

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
 

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)
+			}
+		})
+	}
+}