From 678b3873657a1f49e99b3cb24889c4834836f814 Mon Sep 17 00:00:00 2001 From: Samuel Karp Date: Thu, 15 Oct 2026 21:37:50 +0000 Subject: [PATCH] cri: record pull progress on request events pullProgressReporter samples activeReqs and totalBytesRead on a periodic ticker. A response with an empty body, such as a manifest Resolve HEAD request, does not increment totalBytesRead. In addition, an activeReqs transition to zero and back to one between ticker ticks is not observed by the ticker. In both cases, lastSeenTimestamp is not updated and a subsequent request can be cancelled before it has run for timeout. Record the timestamp of the most recent request start, response header receipt, or body byte read in pullRequestReporter, and advance lastSeenTimestamp from that timestamp on each tick. Assisted-by: Antigravity Signed-off-by: Samuel Karp --- internal/cri/server/images/image_pull.go | 49 ++++--- internal/cri/server/images/image_pull_test.go | 132 ++++++++++++++++-- 2 files changed, 150 insertions(+), 31 deletions(-) diff --git a/internal/cri/server/images/image_pull.go b/internal/cri/server/images/image_pull.go index 5edf976ab6..c4d0afac0b 100644 --- a/internal/cri/server/images/image_pull.go +++ b/internal/cri/server/images/image_pull.go @@ -299,7 +299,7 @@ func (c *CRIImageService) pullImageWithLocalPull( if err != nil { return nil, 0, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err) } - _, bytesPulled := pullReporter.reqReporter.status() + _, bytesPulled, _ := pullReporter.reqReporter.status() return image, bytesPulled, nil } @@ -357,7 +357,7 @@ func (c *CRIImageService) pullImageWithTransferService( if err != nil { return nil, 0, fmt.Errorf("failed to get image %q from containerd image store: %w", ref, err) } - _, bytesPulled := transferProgressReporter.reqReporter.status() + _, bytesPulled, _ := transferProgressReporter.reqReporter.status() return image, bytesPulled, nil } @@ -719,7 +719,6 @@ func (reporter *pullProgressReporter) start(ctx context.Context) { lastSeenBytesRead = uint64(0) lastSeenTimestamp = time.Now() - prevActiveReqs = int32(0) ) // check progress more frequently if timeout < default internal @@ -733,7 +732,10 @@ func (reporter *pullProgressReporter) start(ctx context.Context) { for { select { case <-ticker.C: - activeReqs, bytesRead := reporter.reqReporter.status() + activeReqs, bytesRead, lastProgress := reporter.reqReporter.status() + if lastProgress.After(lastSeenTimestamp) { + lastSeenTimestamp = lastProgress + } log.G(ctx).WithField("ref", reporter.ref). WithField("activeReqs", activeReqs). @@ -743,19 +745,11 @@ func (reporter *pullProgressReporter) start(ctx context.Context) { WithField("reportInterval", reportInterval). Debugf("progress for image pull") - // Reset the no-progress timer when idle, when bytes were - // read, or when a request has just started after an idle - // period. Without the idle→active reset, a request can be - // cancelled less than `timeout` after it was issued, - // because lastSeenTimestamp was captured on an earlier - // tick while activeReqs was still 0. - if activeReqs == 0 || bytesRead > lastSeenBytesRead || prevActiveReqs == 0 { - lastSeenBytesRead = bytesRead + lastSeenBytesRead = bytesRead + if activeReqs == 0 { lastSeenTimestamp = time.Now() - prevActiveReqs = activeReqs continue } - prevActiveReqs = activeReqs if time.Since(lastSeenTimestamp) > reporter.timeout { log.G(ctx).Errorf("cancel pulling image %s because of no progress in %v", reporter.ref, reporter.timeout) @@ -763,7 +757,7 @@ func (reporter *pullProgressReporter) start(ctx context.Context) { return } case <-ctx.Done(): - activeReqs, bytesRead := reporter.reqReporter.status() + activeReqs, bytesRead, _ := reporter.reqReporter.status() log.G(ctx).Infof("stop pulling image %s: active requests=%v, bytes read=%v", reporter.ref, activeReqs, bytesRead) return } @@ -804,9 +798,17 @@ type pullRequestReporter struct { // totalBytesRead indicates that the total bytes has been read from // remote registry. totalBytesRead atomic.Uint64 + // lastProgressNano is the unix nanosecond timestamp of the most recent + // request start, response header arrival, or body byte read. + lastProgressNano atomic.Int64 +} + +func (reporter *pullRequestReporter) touch() { + reporter.lastProgressNano.Store(time.Now().UnixNano()) } func (reporter *pullRequestReporter) incRequest() { + reporter.touch() reporter.activeReqs.Add(1) } @@ -815,13 +817,19 @@ func (reporter *pullRequestReporter) decRequest() { } func (reporter *pullRequestReporter) incByteRead(nr uint64) { - reporter.totalBytesRead.Add(nr) + if nr > 0 { + reporter.totalBytesRead.Add(nr) + reporter.touch() + } } -func (reporter *pullRequestReporter) status() (currentReqs int32, totalBytesRead uint64) { +func (reporter *pullRequestReporter) status() (currentReqs int32, totalBytesRead uint64, lastProgress time.Time) { currentReqs = reporter.activeReqs.Load() totalBytesRead = reporter.totalBytesRead.Load() - return currentReqs, totalBytesRead + if lastProgressNano := reporter.lastProgressNano.Load(); lastProgressNano > 0 { + lastProgress = time.Unix(0, lastProgressNano) + } + return currentReqs, totalBytesRead, lastProgress } // pullRequestReporterRoundTripper wraps http.RoundTripper with pull request @@ -853,6 +861,7 @@ func (rt *pullRequestReporterRoundTripper) RoundTrip(req *http.Request) (*http.R rt.reqReporter.decRequest() return nil, err } + rt.reqReporter.touch() resp.Body = &countingReadCloser{ rc: resp.Body, @@ -1045,7 +1054,7 @@ func (reporter *transferProgressReporter) start(ctx context.Context) { reporter.checkProgress(ctx, reportInterval) continue case <-ctx.Done(): - activeReqs, bytesRead := reporter.reqReporter.status() + activeReqs, bytesRead, _ := reporter.reqReporter.status() log.G(ctx).Infof("stop pulling image %s: active requests=%v, bytes read=%v", reporter.ref, activeReqs, bytesRead) return } @@ -1054,7 +1063,7 @@ func (reporter *transferProgressReporter) start(ctx context.Context) { } func (reporter *transferProgressReporter) checkProgress(ctx context.Context, reportInterval time.Duration) { - activeReqs, bytesRead := reporter.reqReporter.status() + activeReqs, bytesRead, _ := reporter.reqReporter.status() lastSeenBytesRead := reporter.lastSeenBytesRead lastSeenTimestamp := reporter.lastSeenTimestamp diff --git a/internal/cri/server/images/image_pull_test.go b/internal/cri/server/images/image_pull_test.go index e532adbd82..a987b88465 100644 --- a/internal/cri/server/images/image_pull_test.go +++ b/internal/cri/server/images/image_pull_test.go @@ -19,11 +19,16 @@ package images import ( "context" "encoding/base64" + "io" + "net/http" + "strings" "testing" + "testing/synctest" "time" "github.com/containerd/errdefs" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" runtime "k8s.io/cri-api/pkg/apis/runtime/v1" "github.com/containerd/platforms" @@ -557,7 +562,7 @@ func TestTransferProgressReporter(t *testing.T) { }, }, check: func(t *testing.T, r *transferProgressReporter, cancelCalled <-chan struct{}) { - activeReqs, totalBytesRead := r.reqReporter.status() + activeReqs, totalBytesRead, _ := r.reqReporter.status() assert.Equal(t, int32(0), activeReqs, "Expected 0 active requests") assert.Equal(t, uint64(1000), totalBytesRead, "Expected 1000 bytes read") }, @@ -589,7 +594,7 @@ func TestTransferProgressReporter(t *testing.T) { }, }, check: func(t *testing.T, r *transferProgressReporter, cancelCalled <-chan struct{}) { - activeReqs, totalBytesRead := r.reqReporter.status() + activeReqs, totalBytesRead, _ := r.reqReporter.status() assert.Equal(t, int32(0), activeReqs, "Expected 0 active requests") assert.Equal(t, uint64(1000), totalBytesRead, "Expected 1000 bytes read") }, @@ -625,7 +630,7 @@ func TestTransferProgressReporter(t *testing.T) { }, }, check: func(t *testing.T, r *transferProgressReporter, cancelCalled <-chan struct{}) { - activeReqs, totalBytesRead := r.reqReporter.status() + activeReqs, totalBytesRead, _ := r.reqReporter.status() assert.Equal(t, int32(0), activeReqs, "Expected zero active request") assert.Equal(t, uint64(0), totalBytesRead, "Expected zero bytes read") }, @@ -726,7 +731,7 @@ func TestTransferProgressReporter(t *testing.T) { }, }, check: func(t *testing.T, r *transferProgressReporter, cancelCalled <-chan struct{}) { - activeReqs, totalBytesRead := r.reqReporter.status() + activeReqs, totalBytesRead, _ := r.reqReporter.status() assert.Equal(t, int32(1), activeReqs, "Expected one active request") assert.Equal(t, uint64(2000), totalBytesRead, "Expected 2000 bytes read") }, @@ -769,13 +774,6 @@ func TestTransferProgressReporter(t *testing.T) { // TestPullProgressReporter covers the core no-progress cancellation // behavior of pullProgressReporter: a stuck request (active, no bytes) // is eventually cancelled, while a progressing request is not. -// -// The flaky failure in TestCRIImagePullTimeout/HoldingContentOpenWriterWithLocalPull -// — which this fix addresses — is a timing race that's not cleanly -// expressible as a unit test: the boundary between the buggy and fixed -// cancel times coincides at 1.5*timeout, so any check near that -// threshold is scheduler-jitter prone. The semantic regression is -// covered by the existing integration test. func TestPullProgressReporter(t *testing.T) { t.Run("StuckRequestStillGetsCancelled", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) @@ -841,6 +839,118 @@ func TestPullProgressReporter(t *testing.T) { case <-done: } }) + + t.Run("ZeroBodyResponseFollowedByRequestIsNotCancelled", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + cancelCalled := make(chan struct{}) + reporter := newPullProgressReporter("test-image:latest", func() { + select { + case <-cancelCalled: + default: + close(cancelCalled) + } + }, 200*time.Millisecond) + + client := &http.Client{ + Transport: roundTripperFunc(func(req *http.Request) (*http.Response, error) { + switch req.URL.Path { + case "/token": + time.Sleep(190 * time.Millisecond) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("token")), + }, nil + case "/v2/test/manifests/latest": + time.Sleep(160 * time.Millisecond) + return &http.Response{ + StatusCode: http.StatusOK, + Body: http.NoBody, + }, nil + case "/v2/test/manifests/sha256:1234": + time.Sleep(160 * time.Millisecond) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("manifest")), + }, nil + default: + t.Fatalf("unexpected request path: %s", req.URL.Path) + return nil, nil + } + }), + } + assert.NoError(t, reporter.optionUpdateClient(client)) + reporter.start(ctx) + + doReq := func(method, url string) { + req, err := http.NewRequestWithContext(ctx, method, url, nil) + require.NoError(t, err) + resp, err := client.Do(req) + require.NoError(t, err) + _, err = io.Copy(io.Discard, resp.Body) + assert.NoError(t, err) + assert.NoError(t, resp.Body.Close()) + } + + doReq(http.MethodGet, "https://example.com/token") + doReq(http.MethodHead, "https://example.com/v2/test/manifests/latest") + doReq(http.MethodGet, "https://example.com/v2/test/manifests/sha256:1234") + + select { + case <-cancelCalled: + t.Fatal("pull was cancelled during active GET following a 0-byte HEAD response") + default: + } + }) + }) + + t.Run("RequestAfterMissedIdleTransitionIsNotCancelled", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + cancelCalled := make(chan struct{}) + reporter := newPullProgressReporter("test-image:latest", func() { + select { + case <-cancelCalled: + default: + close(cancelCalled) + } + }, 200*time.Millisecond) + + // With a 200ms timeout, the reporter ticks every 100ms. Each + // request finishes within 200ms, and the idle periods fall + // between ticks. + reporter.reqReporter.incRequest() + reporter.reqReporter.incByteRead(1024) + reporter.start(ctx) + + time.Sleep(120 * time.Millisecond) + reporter.reqReporter.decRequest() + + time.Sleep(60 * time.Millisecond) + reporter.reqReporter.incRequest() + time.Sleep(90 * time.Millisecond) + reporter.reqReporter.decRequest() + + time.Sleep(20 * time.Millisecond) + reporter.reqReporter.incRequest() + + select { + case <-cancelCalled: + t.Fatal("pull was cancelled before any request exceeded timeout") + case <-time.After(130 * time.Millisecond): + } + }) + }) +} + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (fn roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return fn(req) } // fakeObserver records every Observe call so tests can assert both the