Merge pull request #14391 from samuelkarp/cri-pull-progress-timestamp

cri: record pull progress on request events

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

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