cri: exclude cached layer bytes from image_pulling_throughput

The image_pulling_throughput histogram divided the full image size (all
layers plus config) by wall-clock pull duration. Layers already present
in the content store were counted in the numerator, so the reported
MB/s came out way higher than what was actually fetched. Fully-cached
pulls were the worst case: they "pulled" in milliseconds but reported
the full image size, showing up as huge outliers in the histogram.

Both pull paths (local client.Pull and the transfer service) already
maintain a totalBytesRead counter in pullRequestReporter for progress
and timeout checks. Cached blobs never trigger a fetch, so the counter
naturally excludes them. Plumb that value out of both helpers and use
it in place of image.Size(ctx). Fully-cached pulls (bytesPulled == 0)
are not observed, so they don't produce near-infinite samples.

Also updates the metric's Help text to say the denominator is the
end-to-end pull duration including layer extraction and snapshotter
unpack, not just network transfer.

Addresses #13244

Signed-off-by: Ahmet Alp Balkan <ahmet@linkedin.com>

internal/cri/server/images/image_pull.go

@@ -180,11 +180,14 @@ func (c *CRIImageService) PullImage(ctx context.Context, name string, credential
 	//
 	// Transfer service does not currently support all the CRI image config options.
 	// TODO: Add support for DisableSnapshotAnnotations, DiscardUnpackedLayers, ImagePullWithSyncFs and unpackDuplicationSuppressor
-	var image containerd.Image
+	var (
+		image       containerd.Image
+		bytesPulled uint64
+	)
 	if c.config.UseLocalImagePull {
-		image, err = c.pullImageWithLocalPull(ctx, ref, credentials, snapshotter, labels, imagePullProgressTimeout)
+		image, bytesPulled, err = c.pullImageWithLocalPull(ctx, ref, credentials, snapshotter, labels, imagePullProgressTimeout)
 	} else {
-		image, err = c.pullImageWithTransferService(ctx, ref, credentials, snapshotter, labels, imagePullProgressTimeout)
+		image, bytesPulled, err = c.pullImageWithTransferService(ctx, ref, credentials, snapshotter, labels, imagePullProgressTimeout)
 	}
 
 	if err != nil {
@@ -216,11 +219,9 @@ func (c *CRIImageService) PullImage(ctx context.Context, name string, credential
 		}
 	}
 
-	const mbToByte = 1024 * 1024
-	size, _ := image.Size(ctx)
-	imagePullingSpeed := float64(size) / mbToByte / time.Since(startTime).Seconds()
-	imagePullThroughput.Observe(imagePullingSpeed)
+	recordImagePullThroughput(imagePullThroughput, bytesPulled, time.Since(startTime))
 
+	size, _ := image.Size(ctx)
 	log.G(ctx).Infof("Pulled image %q with image id %q, repo tag %q, repo digest %q, size %q in %s", name, imageID,
 		repoTag, repoDigest, strconv.FormatInt(size, 10), time.Since(startTime))
 	// NOTE(random-liu): the actual state in containerd is the source of truth, even we maintain
@@ -232,6 +233,10 @@ func (c *CRIImageService) PullImage(ctx context.Context, name string, credential
 }
 
 // pullImageWithLocalPull handles image pulling using the local client.
+//
+// The returned bytesPulled is the number of bytes actually fetched from the
+// registry — cached layers are not counted because they never trigger an
+// HTTP request.
 func (c *CRIImageService) pullImageWithLocalPull(
 	ctx context.Context,
 	ref string,
@@ -239,7 +244,7 @@ func (c *CRIImageService) pullImageWithLocalPull(
 	snapshotter string,
 	labels map[string]string,
 	imagePullProgressTimeout time.Duration,
-) (containerd.Image, error) {
+) (containerd.Image, uint64, error) {
 	pctx, pcancel := context.WithCancel(ctx)
 	defer pcancel()
 	pullReporter := newPullProgressReporter(ref, pcancel, imagePullProgressTimeout)
@@ -279,12 +284,17 @@ func (c *CRIImageService) pullImageWithLocalPull(
 	image, err := c.client.Pull(pctx, ref, pullOpts...)
 	pcancel()
 	if err != nil {
-		return nil, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err)
+		return nil, 0, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err)
 	}
-	return image, nil
+	_, bytesPulled := pullReporter.reqReporter.status()
+	return image, bytesPulled, nil
 }
 
 // pullImageWithTransferService handles image pulling using the transfer service.
+//
+// The returned bytesPulled is the number of bytes actually fetched from the
+// registry, accumulated from transfer progress events; cached layers emit an
+// "already exists" event that does not increment the counter.
 func (c *CRIImageService) pullImageWithTransferService(
 	ctx context.Context,
 	ref string,
@@ -292,7 +302,7 @@ func (c *CRIImageService) pullImageWithTransferService(
 	snapshotter string,
 	labels map[string]string,
 	imagePullProgressTimeout time.Duration,
-) (containerd.Image, error) {
+) (containerd.Image, uint64, error) {
 	log.G(ctx).Debugf("PullImage %q with snapshotter %s using transfer service", ref, snapshotter)
 	rctx, rcancel := context.WithCancel(ctx)
 	defer rcancel()
@@ -318,7 +328,7 @@ func (c *CRIImageService) pullImageWithTransferService(
 
 	reg, err := registry.NewOCIRegistry(ctx, ref, opts...)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create OCI registry: %w", err)
+		return nil, 0, fmt.Errorf("failed to create OCI registry: %w", err)
 	}
 
 	transferProgressReporter.start(rctx)
@@ -326,15 +336,16 @@ func (c *CRIImageService) pullImageWithTransferService(
 	err = c.transferrer.Transfer(rctx, reg, is, transfer.WithProgress(transferProgressReporter.createProgressFunc(rctx)))
 	rcancel()
 	if err != nil {
-		return nil, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err)
+		return nil, 0, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err)
 	}
 
 	// Image should be pulled, unpacked and present in containerd image store at this moment
 	image, err := c.client.GetImage(ctx, ref)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get image %q from containerd image store: %w", ref, err)
+		return nil, 0, fmt.Errorf("failed to get image %q from containerd image store: %w", ref, err)
 	}
-	return image, nil
+	_, bytesPulled := transferProgressReporter.reqReporter.status()
+	return image, bytesPulled, nil
 }
 
 // ParseAuth parses AuthConfig and returns username and password/secret required by containerd.

internal/cri/server/images/image_pull_test.go

@@ -834,3 +834,66 @@ func TestPullProgressReporter(t *testing.T) {
 		}
 	})
 }
+
+// fakeObserver records every Observe call so tests can assert both the
+// presence and the value of observations without touching the real histogram.
+type fakeObserver struct {
+	samples []float64
+}
+
+func (f *fakeObserver) Observe(v float64) {
+	f.samples = append(f.samples, v)
+}
+
+func TestRecordImagePullThroughput(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		bytesPulled uint64
+		duration    time.Duration
+		wantSamples int
+		wantValue   float64 // only checked when wantSamples == 1
+	}{
+		{
+			name:        "fully cached pull is not observed",
+			bytesPulled: 0,
+			duration:    2 * time.Second,
+			wantSamples: 0,
+		},
+		{
+			name:        "zero duration is not observed",
+			bytesPulled: 10 * mbToByte,
+			duration:    0,
+			wantSamples: 0,
+		},
+		{
+			name:        "cold pull observes MB/s from fetched bytes",
+			bytesPulled: 10 * mbToByte,
+			duration:    2 * time.Second,
+			wantSamples: 1,
+			wantValue:   5.0,
+		},
+		{
+			name:        "partial cache hit observes only fetched bytes",
+			// 200 MB image, 150 MB cached, 50 MB actually fetched over 1s.
+			bytesPulled: 50 * mbToByte,
+			duration:    1 * time.Second,
+			wantSamples: 1,
+			wantValue:   50.0,
+		},
+		{
+			name:        "sub-second pull observes correctly",
+			bytesPulled: 25 * mbToByte,
+			duration:    500 * time.Millisecond,
+			wantSamples: 1,
+			wantValue:   50.0,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			obs := &fakeObserver{}
+			recordImagePullThroughput(obs, tc.bytesPulled, tc.duration)
+			if assert.Len(t, obs.samples, tc.wantSamples) && tc.wantSamples == 1 {
+				assert.InDelta(t, tc.wantValue, obs.samples[0], 0.001)
+			}
+		})
+	}
+}

internal/cri/server/images/metrics.go

@@ -17,14 +17,20 @@
 package images
 
 import (
+	"time"
+
 	"github.com/docker/go-metrics"
 	prom "github.com/prometheus/client_golang/prometheus"
 )
 
+const mbToByte = 1024 * 1024
+
 var (
 	imagePulls           metrics.LabeledCounter
 	inProgressImagePulls metrics.Gauge
-	// image size in MB / image pull duration in seconds
+	// imagePullThroughput observes one MB/s sample per image pull (not per
+	// layer): fetched bytes over end-to-end pull duration (includes
+	// extraction). Fully-cached pulls are skipped.
 	imagePullThroughput prom.Histogram
 )
 
@@ -44,10 +50,24 @@ func init() {
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "image_pulling_throughput",
-			Help:      "image pull throughput",
+			Help:      "image pull throughput in MB/s (fetched bytes / pull duration; cached layers excluded)",
 			Buckets:   prom.DefBuckets,
 		},
 	)
 	ns.Add(imagePullThroughput)
 	metrics.Register(ns)
 }
+
+// recordImagePullThroughput observes a single pull throughput sample in MB/s.
+// bytesPulled is the bytes actually fetched from the registry — cached layers
+// never trigger a fetch, so they're naturally excluded. duration is the
+// end-to-end pull duration (includes extraction, not just network transfer).
+//
+// Pulls that fetched nothing (fully cached) are skipped rather than recorded
+// as zero or near-infinite samples, which would pollute the histogram.
+func recordImagePullThroughput(obs prom.Observer, bytesPulled uint64, duration time.Duration) {
+	if bytesPulled == 0 || duration <= 0 {
+		return
+	}
+	obs.Observe(float64(bytesPulled) / mbToByte / duration.Seconds())
+}