Add additional BlobCacheMetrics, expose BlobCacheMetrics via SharedBlobCacheService

docs/changelog/111730.yaml

@@ -0,0 +1,5 @@
+pr: 111730
+summary: "Add callback for copy-to-cache metrics, additional `BlobCacheMetrics`"
+area: Store
+type: enhancement
+issues: []

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/BlobCacheMetrics.java

@@ -7,15 +7,37 @@
 
 package org.elasticsearch.blobcache;
 
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.telemetry.TelemetryProvider;
+import org.elasticsearch.telemetry.metric.DoubleHistogram;
 import org.elasticsearch.telemetry.metric.LongCounter;
 import org.elasticsearch.telemetry.metric.LongHistogram;
 import org.elasticsearch.telemetry.metric.MeterRegistry;
 
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
 public class BlobCacheMetrics {
+    private static final String CACHE_POPULATION_REASON_ATTRIBUTE_KEY = "cachePopulationReason";
+    private static final String SHARD_ID_ATTRIBUTE_KEY = "shardId";
+
     private final LongCounter cacheMissCounter;
     private final LongCounter evictedCountNonZeroFrequency;
     private final LongHistogram cacheMissLoadTimes;
+    private final DoubleHistogram cachePopulateThroughput;
+    private final LongCounter cachePopulationBytes;
+    private final LongCounter cachePopulationTime;
+
+    public enum CachePopulationReason {
+        /**
+         * When warming the cache
+         */
+        Warming,
+        /**
+         * When the data we need is not in the cache
+         */
+        CacheMiss
+    }
 
     public BlobCacheMetrics(MeterRegistry meterRegistry) {
         this(
@@ -33,14 +55,39 @@ public BlobCacheMetrics(MeterRegistry meterRegistry) {
                 "es.blob_cache.cache_miss_load_times.histogram",
                 "The time in milliseconds for populating entries in the blob store resulting from a cache miss, expressed as a histogram.",
                 "ms"
+            ),
+            meterRegistry.registerDoubleHistogram(
+                "es.blob_cache.populate_throughput.histogram",
+                "The throughput when populating the blob store from the cache",
+                "MB/second"
+            ),
+            meterRegistry.registerLongCounter(
+                "es.blob_cache.populate_bytes.total",
+                "The number of bytes that have been loaded into the cache",
+                "bytes"
+            ),
+            meterRegistry.registerLongCounter(
+                "es.blob_cache.populate_time.total",
+                "The time spent copying data into the cache",
+                "milliseconds"
             )
         );
     }
 
-    BlobCacheMetrics(LongCounter cacheMissCounter, LongCounter evictedCountNonZeroFrequency, LongHistogram cacheMissLoadTimes) {
+    BlobCacheMetrics(
+        LongCounter cacheMissCounter,
+        LongCounter evictedCountNonZeroFrequency,
+        LongHistogram cacheMissLoadTimes,
+        DoubleHistogram cachePopulateThroughput,
+        LongCounter cachePopulationBytes,
+        LongCounter cachePopulationTime
+    ) {
         this.cacheMissCounter = cacheMissCounter;
         this.evictedCountNonZeroFrequency = evictedCountNonZeroFrequency;
         this.cacheMissLoadTimes = cacheMissLoadTimes;
+        this.cachePopulateThroughput = cachePopulateThroughput;
+        this.cachePopulationBytes = cachePopulationBytes;
+        this.cachePopulationTime = cachePopulationTime;
     }
 
     public static BlobCacheMetrics NOOP = new BlobCacheMetrics(TelemetryProvider.NOOP.getMeterRegistry());
@@ -56,4 +103,47 @@ public LongCounter getEvictedCountNonZeroFrequency() {
     public LongHistogram getCacheMissLoadTimes() {
         return cacheMissLoadTimes;
     }
+
+    /**
+     * Record the various cache population metrics after a chunk is copied to the cache
+     *
+     * @param totalBytesCopied The total number of bytes copied
+     * @param totalCopyTimeNanos The time taken to copy the bytes in nanoseconds
+     * @param shardId The shard ID to which the chunk belonged
+     * @param cachePopulationReason The reason for the cache being populated
+     */
+    public void recordCachePopulationMetrics(
+        int totalBytesCopied,
+        long totalCopyTimeNanos,
+        ShardId shardId,
+        CachePopulationReason cachePopulationReason
+    ) {
+        Map<String, Object> metricAttributes = Map.of(
+            SHARD_ID_ATTRIBUTE_KEY,
+            shardId,
+            CACHE_POPULATION_REASON_ATTRIBUTE_KEY,
+            cachePopulationReason
+        );
+        assert totalBytesCopied > 0 : "We shouldn't be recording zero-sized copies";
+        cachePopulationBytes.incrementBy(totalBytesCopied, metricAttributes);
+
+        // This is almost certainly paranoid, but if we had a very fast/small copy with a very coarse nanosecond timer it might happen?
+        if (totalCopyTimeNanos > 0) {
+            cachePopulateThroughput.record(toMegabytesPerSecond(totalBytesCopied, totalCopyTimeNanos), metricAttributes);
+            cachePopulationTime.incrementBy(TimeUnit.NANOSECONDS.toMillis(totalCopyTimeNanos), metricAttributes);
+        }
+    }
+
+    /**
+     * Calculate throughput as megabytes/second
+     *
+     * @param totalBytes The total number of bytes transferred
+     * @param totalNanoseconds The time to transfer in nanoseconds
+     * @return The throughput as megabytes/second
+     */
+    private double toMegabytesPerSecond(int totalBytes, long totalNanoseconds) {
+        double totalSeconds = totalNanoseconds / 1_000_000_000.0;
+        double totalMegabytes = totalBytes / 1_000_000.0;
+        return totalMegabytes / totalSeconds;
+    }
 }

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/BlobCachePopulationListener.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.blobcache.shared;
+
+public interface BlobCachePopulationListener {
+
+    BlobCachePopulationListener NOOP = (bc, ctn) -> {};
+
+    /**
+     * Notify of a blob cache population that occurred
+     *
+     * Note that <code>bytesCopied</code> is approximate, there are cases where we write
+     * more than we read, due to page alignment, and there are times when we read more
+     * than we write, e.g. when filling multiple gaps. Notifiers should try and record
+     * the larger of those two numbers when invoking this method.
+     *
+     * @param bytesCopied The number of bytes copied into the cache
+     * @param copyTimeNanos The time in nanoseconds taken to copy those bytes
+     */
+    void onCachePopulation(int bytesCopied, long copyTimeNanos);
+}

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBytes.java

@@ -172,6 +172,7 @@ static void preallocate(Path cacheFile, long fileSize) throws IOException {
      * @param buf bytebuffer to use for writing
      * @throws IOException on failure
      */
+    // TODO: Support BlobCachePopulationListener when required
     public static void copyToCacheFileAligned(
         IO fc,
         InputStream input,
@@ -212,11 +213,19 @@ public static void copyToCacheFileAligned(
      * @param fileChannelPos position in {@code fc} to write to
      * @param progressUpdater callback to invoke with the number of copied bytes as they are copied
      * @param buffer bytebuffer to use for writing
+     * @param populationListener a listener that will be notified upon completion of the copy
      * @return the number of bytes copied
      * @throws IOException on failure
      */
-    public static int copyToCacheFileAligned(IO fc, InputStream input, int fileChannelPos, IntConsumer progressUpdater, ByteBuffer buffer)
-        throws IOException {
+    public static int copyToCacheFileAligned(
+        IO fc,
+        InputStream input,
+        int fileChannelPos,
+        IntConsumer progressUpdater,
+        ByteBuffer buffer,
+        BlobCachePopulationListener populationListener
+    ) throws IOException {
+        final long copyStartTime = System.nanoTime();
         int bytesCopied = 0;
         while (true) {
             final int bytesRead = Streams.read(input, buffer, buffer.remaining());
@@ -226,6 +235,10 @@ public static int copyToCacheFileAligned(IO fc, InputStream input, int fileChann
             bytesCopied += copyBufferToCacheFileAligned(fc, fileChannelPos + bytesCopied, buffer);
             progressUpdater.accept(bytesCopied);
         }
+        if (bytesCopied > 0) {
+            long elapsedTimeNanos = System.nanoTime() - copyStartTime;
+            populationListener.onCachePopulation(bytesCopied, elapsedTimeNanos);
+        }
         return bytesCopied;
     }