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 {
+    public static final String CACHE_POPULATION_REASON_ATTRIBUTE_KEY = "cachePopulationReason";
+    public 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.toString(),
+            CACHE_POPULATION_REASON_ATTRIBUTE_KEY,
+            cachePopulationReason.name()
+        );
+        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;
     }
 

x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBytesTests.java

@@ -8,15 +8,24 @@
 package org.elasticsearch.blobcache.shared;
 
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.env.TestEnvironment;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.test.ESTestCase;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.file.Files;
 
+import static org.elasticsearch.blobcache.shared.SharedBytes.PAGE_SIZE;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+
 public class SharedBytesTests extends ESTestCase {
 
     public void testReleasesFileCorrectly() throws Exception {
@@ -41,4 +50,78 @@ public void testReleasesFileCorrectly() throws Exception {
             assertFalse(Files.exists(sharedBytesPath));
         }
     }
+
+    public void testPopulationListenerIsCalledOnSuccess() throws IOException {
+        doWithSharedBytes(sharedBytes -> {
+            // position + length must be <= region size
+            final int position = randomIntBetween(0, sharedBytes.regionSize - 1);
+            final int alignedPosition = position - position % PAGE_SIZE;
+            final int streamLength = randomIntBetween(1, sharedBytes.regionSize - alignedPosition);
+            logger.info("Copying {} bytes to position {} (region size={})", streamLength, alignedPosition, sharedBytes.regionSize);
+            final int[] bytesCopiedHolder = new int[1];
+            final long[] copyTimeNanosHolder = new long[1];
+            try (ByteArrayInputStream inputStream = new ByteArrayInputStream(randomByteArrayOfLength(streamLength))) {
+                SharedBytes.copyToCacheFileAligned(
+                    sharedBytes.getFileChannel(0),
+                    inputStream,
+                    alignedPosition,
+                    p -> {},
+                    ByteBuffer.allocate(randomIntBetween(1, 4) * PAGE_SIZE),
+                    (bytesCopied, copyTimeNanos) -> {
+                        bytesCopiedHolder[0] = bytesCopied;
+                        copyTimeNanosHolder[0] = copyTimeNanos;
+                    }
+                );
+                final int partialPage = streamLength % PAGE_SIZE;
+                final int pageAlignedCopySize = partialPage == 0 ? streamLength : streamLength + PAGE_SIZE - partialPage;
+                assertEquals(pageAlignedCopySize, bytesCopiedHolder[0]);
+                assertThat(copyTimeNanosHolder[0], greaterThan(0L));
+            }
+        });
+    }
+
+    public void testPopulationListenerIsNotCalledForZeroLengthStream() throws IOException {
+        doWithSharedBytes(sharedBytes -> {
+            final BlobCachePopulationListener listener = mock(BlobCachePopulationListener.class);
+            final int position = randomIntBetween(0, sharedBytes.regionSize);
+            final int alignedPosition = position - position % PAGE_SIZE;
+            logger.info("Copying nothing to position {} (region size={})", alignedPosition, sharedBytes.regionSize);
+            try (ByteArrayInputStream inputStream = new ByteArrayInputStream(new byte[] {})) {
+                SharedBytes.copyToCacheFileAligned(
+                    sharedBytes.getFileChannel(0),
+                    inputStream,
+                    alignedPosition,
+                    p -> {},
+                    ByteBuffer.allocate(randomIntBetween(1, 4) * PAGE_SIZE),
+                    listener
+                );
+                verifyNoInteractions(listener);
+            }
+        });
+    }
+
+    private <E extends Exception> void doWithSharedBytes(CheckedConsumer<SharedBytes, E> consumer) throws E, IOException {
+        int regions = randomIntBetween(1, 10);
+        var nodeSettings = Settings.builder()
+            .put(Node.NODE_NAME_SETTING.getKey(), "node")
+            .put("path.home", createTempDir())
+            .putList(Environment.PATH_DATA_SETTING.getKey(), createTempDir().toString())
+            .build();
+        try (var nodeEnv = new NodeEnvironment(nodeSettings, TestEnvironment.newEnvironment(nodeSettings))) {
+            SharedBytes sharedBytes = null;
+            try {
+                sharedBytes = new SharedBytes(
+                    regions,
+                    randomIntBetween(1, 16) * 4096,
+                    nodeEnv,
+                    ignored -> {},
+                    ignored -> {},
+                    IOUtils.WINDOWS == false && randomBoolean()
+                );
+                consumer.accept(sharedBytes);
+            } finally {
+                sharedBytes.decRef();
+            }
+        }
+    }
 }