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