feat(storage): rationalize read path OTEL

packages/orchestrator/cmd/inspect-build/validate.go

@@ -190,7 +190,7 @@ func openChunker(ctx context.Context, storagePath, buildID, artifact string, h *
 		return nil, nil, 0, nil, err
 	}
 
-	chunker, err := block.NewChunker(flags, size, int64(h.Metadata.BlockSize), filepath.Join(cacheDir, "cache"), m)
+	chunker, err := block.NewChunker(flags, size, int64(h.Metadata.BlockSize), filepath.Join(cacheDir, "cache"), m, seekableType(artifact))
 	if err != nil {
 		os.RemoveAll(cacheDir)
 

packages/orchestrator/pkg/sandbox/block/fetch_session.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"sync"
 	"sync/atomic"
+
+	"github.com/e2b-dev/infra/packages/shared/pkg/storage"
 )
 
 type fetchSession struct {
@@ -21,12 +23,23 @@ type fetchSession struct {
 	fetchErr error
 	done     bool // true once terminated (success or error)
 
+	// source is the backend that served the fetch; set by runFetch.
+	source atomic.Int32
+
 	// bytesReady is the byte count (from chunkOff) up to which all blocks
 	// are fully written and marked cached. Atomic so registerAndWait can
 	// do a lock-free fast-path check: bytesReady only increases.
 	bytesReady atomic.Int64
 }
 
+func (s *fetchSession) setSource(src storage.Source) {
+	s.source.Store(int32(src))
+}
+
+func (s *fetchSession) Source() storage.Source {
+	return storage.Source(s.source.Load())
+}
+
 // contains reports whether the session covers the byte range [off, off+length).
 func (s *fetchSession) contains(off, length int64) bool {
 	return s.chunkOff <= off && s.chunkOff+s.chunkLen >= off+length

packages/orchestrator/pkg/sandbox/block/metrics/main.go

@@ -12,17 +12,20 @@ const (
 	orchestratorBlockSlices      = "orchestrator.blocks.slices"
 	orchestratorBlockChunksFetch = "orchestrator.blocks.chunks.fetch"
 	orchestratorBlockChunksStore = "orchestrator.blocks.chunks.store"
+	orchestratorChunkSlice       = "orchestrator.chunk.slice"
 )
 
 type Metrics struct {
 	// SlicesMetric is used to measure page faulting performance.
 	SlicesTimerFactory telemetry.TimerFactory
 
-	// WriteChunksMetric is used to measure the time taken to download chunks from remote storage
+	// RemoteReadsTimerFactory is used to measure the time taken to download chunks from remote storage
 	RemoteReadsTimerFactory telemetry.TimerFactory
 
 	// WriteChunksMetric is used to measure performance of writing chunks to disk.
 	WriteChunksTimerFactory telemetry.TimerFactory
+
+	ChunkSliceTimerFactory telemetry.FloatTimerFactory
 }
 
 func NewMetrics(meterProvider metric.MeterProvider) (Metrics, error) {
@@ -58,5 +61,13 @@ func NewMetrics(meterProvider metric.MeterProvider) (Metrics, error) {
 		return m, fmt.Errorf("failed to get stored chunks metric: %w", err)
 	}
 
+	if m.ChunkSliceTimerFactory, err = telemetry.NewFloatTimerFactory(
+		blocksMeter, orchestratorChunkSlice,
+		"Time taken by Chunker to serve a Slice() (source=mmap when served from cache)",
+		"Bytes returned",
+	); err != nil {
+		return m, fmt.Errorf("error creating chunk slice timer factory: %w", err)
+	}
+
 	return m, nil
 }

packages/orchestrator/pkg/sandbox/block/streaming_chunk.go

@@ -30,6 +30,7 @@ type Chunker struct {
 	metrics      metrics.Metrics
 	fetchTimeout time.Duration
 	featureFlags *featureflags.Client
+	objType      storage.SeekableObjectType
 
 	size int64
 
@@ -42,6 +43,7 @@ func NewChunker(
 	size, blockSize int64,
 	cachePath string,
 	metrics metrics.Metrics,
+	objType storage.SeekableObjectType,
 ) (*Chunker, error) {
 	cache, err := NewCache(size, blockSize, cachePath, false)
 	if err != nil {
@@ -54,6 +56,7 @@ func NewChunker(
 		metrics:      metrics,
 		featureFlags: ff,
 		fetchTimeout: defaultFetchTimeout,
+		objType:      objType,
 	}, nil
 }
 
@@ -69,54 +72,66 @@ func (c *Chunker) ReadAt(ctx context.Context, b []byte, off int64, upstream stor
 }
 
 func (c *Chunker) Slice(ctx context.Context, off, length int64, upstream storage.RangeOpener, ft *storage.FrameTable) ([]byte, error) {
+	ct := ft.CompressionType()
 	attrs := chunkerAttrs
 	if ft.IsCompressed() {
 		attrs = chunkerAttrsCompressed
 	}
+
+	sliceStart := time.Now()
 	timer := c.metrics.SlicesTimerFactory.Begin()
 
-	// Fast path: already cached
+	// Fast path: already cached.
 	b, err := c.cache.Slice(off, length)
 	if err == nil {
 		timer.RecordRaw(ctx, length, attrs.successFromCache)
+		c.metrics.ChunkSliceTimerFactory.Record(ctx, time.Since(sliceStart), length, storage.OKAttrs(c.objType, storage.SourceMmap, ct))
 
 		return b, nil
 	}
 
 	if !errors.As(err, &BytesNotAvailableError{}) {
 		timer.RecordRaw(ctx, length, attrs.failCacheRead)
+		c.metrics.ChunkSliceTimerFactory.Record(ctx, time.Since(sliceStart), 0, storage.ErrAttrs(c.objType, storage.SourceMmap, ct, err))
 
 		return nil, fmt.Errorf("failed read from cache at offset %d: %w", off, err)
 	}
 
 	// Fetch every chunk the range spans (one fetch session per chunk).
+	var src storage.Source
 	end := off + length
 	for cur := off; cur < end; {
 		chunkOff, chunkLen, lerr := c.locateChunk(cur, ft)
 		if lerr != nil {
 			timer.RecordRaw(ctx, length, attrs.failRemoteFetch)
+			c.metrics.ChunkSliceTimerFactory.Record(ctx, time.Since(sliceStart), 0, storage.ErrAttrs(c.objType, src, ct, lerr))
 
 			return nil, fmt.Errorf("failed to locate chunk for offset %d: %w", cur, lerr)
 		}
 		chunkEnd := chunkOff + chunkLen
 		rangeEnd := min(end, chunkEnd)
-		if err := c.fetch(ctx, cur, rangeEnd-cur, upstream, ft); err != nil {
+		s, err := c.fetch(ctx, cur, rangeEnd-cur, upstream, ft)
+		if err != nil {
 			timer.RecordRaw(ctx, length, attrs.failRemoteFetch)
+			c.metrics.ChunkSliceTimerFactory.Record(ctx, time.Since(sliceStart), 0, storage.ErrAttrs(c.objType, s, ct, err))
 
 			return nil, fmt.Errorf("failed to ensure data at %d-%d: %w", cur, rangeEnd, err)
 		}
+		src = max(src, s)
 		cur = chunkEnd
 	}
 
 	// sliceDirect skips isCached — the waiter already confirmed the data is in the mmap.
 	b, cacheErr := c.cache.sliceDirect(off, length)
 	if cacheErr != nil {
 		timer.RecordRaw(ctx, length, attrs.failLocalReadAgain)
+		c.metrics.ChunkSliceTimerFactory.Record(ctx, time.Since(sliceStart), 0, storage.ErrAttrs(c.objType, src, ct, cacheErr))
 
 		return nil, fmt.Errorf("failed to read from cache after ensuring data at %d-%d: %w", off, off+length, cacheErr)
 	}
 
 	timer.RecordRaw(ctx, length, attrs.successFromRemote)
+	c.metrics.ChunkSliceTimerFactory.Record(ctx, time.Since(sliceStart), length, storage.OKAttrs(c.objType, src, ct))
 
 	return b, nil
 }
@@ -156,38 +171,54 @@ func (c *Chunker) getOrCreateSession(ctx context.Context, off, length int64, ups
 // fetch ensures the chunk for [off, off+length) is fetched and waits
 // for every block the range spans (a span can cross block boundaries
 // after dedup; waiting only on the start block leaves the tail unfetched).
-func (c *Chunker) fetch(ctx context.Context, off, length int64, upstream storage.RangeOpener, ft *storage.FrameTable) error {
+func (c *Chunker) fetch(ctx context.Context, off, length int64, upstream storage.RangeOpener, ft *storage.FrameTable) (storage.Source, error) {
 	chunkOff, chunkLen, err := c.locateChunk(off, ft)
 	if err != nil {
-		return fmt.Errorf("failed to locate chunk for offset %d: %w", off, err)
+		return storage.UnknownSource, fmt.Errorf("failed to locate chunk for offset %d: %w", off, err)
 	}
 
 	session, justGotCached := c.getOrCreateSession(ctx, chunkOff, chunkLen, upstream, ft)
 	if justGotCached {
-		return nil
+		return storage.SourceMmap, nil
 	}
 
 	blockSize := c.cache.BlockSize()
 	startBlock := (off / blockSize) * blockSize
 	endBlock := ((off + length - 1) / blockSize) * blockSize
 	chunkEnd := chunkOff + chunkLen
+
+	// If the session has already streamed past every byte
+	// we need, the data is in the mmap and we never have to wait. Report
+	// source=mmap.
+	endByte := min(endBlock+blockSize, chunkEnd) - chunkOff
+	if session.bytesReady.Load() >= endByte {
+		return storage.SourceMmap, nil
+	}
+
 	for b := startBlock; b <= endBlock; b += blockSize {
 		if b >= chunkEnd {
 			break // tail belongs to the caller's next chunk fetch.
 		}
 		if err := session.registerAndWait(ctx, b); err != nil {
-			return err
+			return session.Source(), err
 		}
 	}
 
-	return nil
+	return session.Source(), nil
 }
 
 // runFetch fetches data from storage into the mmap cache. Runs in a background goroutine.
 func (c *Chunker) runFetch(ctx context.Context, s *fetchSession, upstream storage.RangeOpener, ft *storage.FrameTable) {
 	ctx, cancel := context.WithTimeout(ctx, c.fetchTimeout)
 	defer cancel()
 
+	ctx, span := tracer.Start(ctx, "chunk.fetch")
+	defer span.End()
+	span.SetAttributes(
+		attribute.Int64("off", s.chunkOff),
+		attribute.Int64("len", s.chunkLen),
+	)
+
 	defer c.releaseSession(s)
 
 	// Unconditionally terminate the session on exit so registerAndWait
@@ -211,16 +242,26 @@ func (c *Chunker) runFetch(ctx context.Context, s *fetchSession, upstream storag
 	}
 	defer releaseLock()
 
+	ct := ft.CompressionType()
+
 	attrs := chunkerAttrs
 	if ft.IsCompressed() {
 		attrs = chunkerAttrsCompressed
 	}
 	fetchTimer := c.metrics.RemoteReadsTimerFactory.Begin()
 
-	readBytes, err := c.progressiveRead(ctx, s, mmapSlice, upstream, ft)
+	fetchStart := time.Now()
+
+	stats, src, open, err := c.progressiveRead(ctx, s, mmapSlice, upstream, ft)
+	var readBytes int64
+	if stats != nil {
+		readBytes = stats.UncompressedBytes
+	}
 	if err != nil {
 		fetchTimer.RecordRaw(ctx, readBytes, attrs.remoteFailure)
 
+		storage.RecordReadFetch(ctx, time.Since(fetchStart), readBytes, storage.ErrAttrs(c.objType, src, ct, err))
+
 		s.fail(err)
 
 		return
@@ -231,24 +272,56 @@ func (c *Chunker) runFetch(ctx context.Context, s *fetchSession, upstream storag
 	// closing the TOCTOU window in getOrCreateSession.
 	c.cache.setIsCached(s.chunkOff, s.chunkLen)
 
+	fetchWall := time.Since(fetchStart)
 	fetchTimer.RecordRaw(ctx, readBytes, attrs.remoteSuccess)
+	storage.RecordReadFetch(ctx, fetchWall, readBytes, storage.OKAttrs(c.objType, src, ct))
+
+	// fetch wall / (open + read + decompress); >1 = unaccounted overhead.
+	if stats != nil {
+		if work := open + stats.Read + stats.Decompress; work > 0 {
+			ratio := fetchWall.Seconds() / work.Seconds()
+			storage.RecordPipelineEfficiency(ctx, ratio, storage.OKAttrs(c.objType, src, ct))
+		}
+	}
+
 	s.setDone()
 }
 
-func (c *Chunker) progressiveRead(ctx context.Context, s *fetchSession, mmapSlice []byte, upstream storage.RangeOpener, ft *storage.FrameTable) (totalRead int64, err error) {
-	reader, err := upstream.OpenRangeReader(ctx, s.chunkOff, s.chunkLen, ft)
+func (c *Chunker) progressiveRead(ctx context.Context, s *fetchSession, mmapSlice []byte, upstream storage.RangeOpener, ft *storage.FrameTable) (stats *storage.ReadStats, source storage.Source, open time.Duration, err error) {
+	doneInflight := storage.StartInflight(ctx, storage.InflightFetchAttrs(c.objType))
+	defer doneInflight()
+
+	openStart := time.Now()
+	reader, source, err := upstream.OpenRangeReader(ctx, s.chunkOff, s.chunkLen, ft)
+	open = time.Since(openStart)
+	s.setSource(source)
 	if err != nil {
-		return 0, fmt.Errorf("failed to open range reader at %d: %w", s.chunkOff, err)
+		return nil, source, open, fmt.Errorf("failed to open range reader at %d: %w", s.chunkOff, err)
 	}
+
 	defer func() {
-		if closeErr := reader.Close(context.WithoutCancel(ctx)); closeErr != nil && err == nil {
+		var closeErr error
+		stats, closeErr = reader.Close(context.WithoutCancel(ctx))
+		if closeErr != nil && err == nil {
 			err = closeErr
 		}
+		if err != nil {
+			return
+		}
+
+		ct := ft.CompressionType()
+		okAttrs := storage.OKAttrs(c.objType, source, ct)
+
+		storage.RecordReadOpen(ctx, open, 0, okAttrs)
+		if stats != nil {
+			storage.RecordReadRead(ctx, stats.Read, stats.CompressedBytes, okAttrs)
+		}
 	}()
 
 	blockSize := c.cache.BlockSize()
 	readBatch := max(blockSize, int64(c.featureFlags.IntFlag(ctx, featureflags.MinChunkerReadSizeKB))*1024)
 
+	var totalRead int64
 	for totalRead < s.chunkLen {
 		// Read in batches of max(blockSize, minReadBatchSize) to align notification
 		// granularity with the read size and minimize lock/notify overhead.
@@ -273,11 +346,11 @@ func (c *Chunker) progressiveRead(ctx context.Context, s *fetchSession, mmapSlic
 				break // all bytes received; trailing EOF is expected
 			}
 
-			return totalRead, fmt.Errorf("failed reading at offset %d after %d bytes: %w", s.chunkOff, totalRead, readErr)
+			return stats, source, open, fmt.Errorf("failed reading at offset %d after %d bytes: %w", s.chunkOff, totalRead, readErr)
 		}
 	}
 
-	return totalRead, nil
+	return stats, source, open, nil
 }
 
 // releaseSession removes s from the active list (swap-delete).