(monitoring) Added GCS/NFS in-flight read counters (#2672)

Author: levb

packages/shared/pkg/storage/storage_cache_seekable.go

@@ -14,6 +14,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/launchdarkly/go-sdk-common/v3/ldcontext"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 
@@ -52,6 +53,11 @@ var (
 		"Total bytes written to NFS",
 		"Total writes to NFS",
 	))
+	nfsCacheConcurrentReads = utils.Must(meter.Int64UpDownCounter(
+		"orchestrator.storage.slab.nfs.read.concurrent",
+		metric.WithDescription("Number of NFS cache range readers currently open"),
+		metric.WithUnit("{read}"),
+	))
 )
 
 type featureFlagsClient interface {
@@ -119,7 +125,7 @@ func (c *cachedSeekable) OpenRangeReader(ctx context.Context, off int64, length
 		rc := io.ReadCloser(&fsRangeReadCloser{Reader: io.NewSectionReader(fp, 0, length), file: fp})
 		rc = withSpan(rc, span)
 
-		return rc, nil
+		return withNFSGauge(ctx, rc), nil
 	}
 
 	if !os.IsNotExist(err) {
@@ -169,6 +175,26 @@ func (r *spanReadCloser) Close() error {
 	return err
 }
 
+// nfsGaugeReadCloser wraps a reader and decrements the NFS concurrent reads
+// gauge on Close.
+type nfsGaugeReadCloser struct {
+	io.ReadCloser
+
+	ctx context.Context //nolint:containedctx // needed for gauge decrement in Close
+}
+
+func (r *nfsGaugeReadCloser) Close() error {
+	nfsCacheConcurrentReads.Add(r.ctx, -1)
+
+	return r.ReadCloser.Close()
+}
+
+func withNFSGauge(ctx context.Context, rc io.ReadCloser) io.ReadCloser {
+	nfsCacheConcurrentReads.Add(ctx, 1)
+
+	return &nfsGaugeReadCloser{ReadCloser: rc, ctx: ctx}
+}
+
 // newCacheWriteThroughReader wraps a reader, buffering all data read through it.
 // On Close, it asynchronously writes the buffered data to the NFS cache only
 // if the total bytes read match the expected length (to avoid caching truncated data).

packages/shared/pkg/storage/storage_cache_seekable_compressed.go

@@ -44,7 +44,7 @@ func (c *cachedSeekable) openReaderCompressed(ctx context.Context, offsetU int64
 			return nil, fmt.Errorf("decompress cached frame: %w", err)
 		}
 
-		return decompressed, nil
+		return withNFSGauge(ctx, decompressed), nil
 	case !os.IsNotExist(err):
 		recordCacheReadError(ctx, cacheTypeSeekable, cacheOpOpenRangeReader, err)
 	}

packages/shared/pkg/storage/storage_google.go

@@ -17,6 +17,7 @@ import (
 	"cloud.google.com/go/storage"
 	"github.com/googleapis/gax-go/v2"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
@@ -68,6 +69,11 @@ var (
 		"Total bytes written to GCS",
 		"Total writes to GCS",
 	))
+	gcsConcurrentReads = utils.Must(meter.Int64UpDownCounter(
+		"orchestrator.storage.gcs.read.concurrent",
+		metric.WithDescription("Number of GCS range readers currently open"),
+		metric.WithUnit("{read}"),
+	))
 )
 
 type gcpStorage struct {
@@ -559,6 +565,8 @@ func (o *gcpObject) OpenRangeReader(ctx context.Context, offsetU int64, length i
 			return nil, err
 		}
 
+		gcsConcurrentReads.Add(ctx, 1)
+
 		return &timedReadCloser{inner: rc, timer: timer, ctx: ctx}, nil
 	}
 
@@ -584,6 +592,8 @@ func (o *gcpObject) OpenRangeReader(ctx context.Context, offsetU int64, length i
 		return nil, err
 	}
 
+	gcsConcurrentReads.Add(ctx, 1)
+
 	return &timedReadCloser{inner: decompressed, timer: timer, ctx: ctx}, nil
 }
 
@@ -609,6 +619,8 @@ func (r *timedReadCloser) Read(p []byte) (int, error) {
 }
 
 func (r *timedReadCloser) Close() error {
+	gcsConcurrentReads.Add(r.ctx, -1)
+
 	err := r.inner.Close()
 
 	if r.closeErr != nil || err != nil {