diff --git a/docs/changelog/134374.yaml b/docs/changelog/134374.yaml new file mode 100644 index 0000000000000..8d0ddc665f3b9 --- /dev/null +++ b/docs/changelog/134374.yaml @@ -0,0 +1,5 @@ +pr: 134374 +summary: Add file extension metadata to cache miss counter from `SharedBlobCacheService` +area: Search +type: enhancement +issues: [] diff --git a/qa/evil-tests/src/test/java/org/elasticsearch/index/store/LuceneFilesExtensionsTests.java b/qa/evil-tests/src/test/java/org/elasticsearch/index/store/LuceneFilesExtensionsTests.java index 9f547a47305e9..238c256cdd62b 100644 --- a/qa/evil-tests/src/test/java/org/elasticsearch/index/store/LuceneFilesExtensionsTests.java +++ b/qa/evil-tests/src/test/java/org/elasticsearch/index/store/LuceneFilesExtensionsTests.java @@ -9,10 +9,13 @@ package org.elasticsearch.index.store; +import org.apache.lucene.index.IndexFileNames; import org.elasticsearch.core.Assertions; import org.elasticsearch.core.SuppressForbidden; import org.elasticsearch.test.ESTestCase; +import java.util.Locale; + import static org.hamcrest.Matchers.containsString; public class LuceneFilesExtensionsTests extends ESTestCase { @@ -21,6 +24,7 @@ public void testUnknownFileExtension() { if (Assertions.ENABLED) { AssertionError e = expectThrows(AssertionError.class, () -> LuceneFilesExtensions.fromExtension("abc")); assertThat(e.getMessage(), containsString("unknown Lucene file extension [abc]")); + assertFalse(LuceneFilesExtensions.isLuceneExtension("abc")); setEsAllowUnknownLuceneFileExtensions("true"); try { @@ -41,4 +45,19 @@ public void setEsAllowUnknownLuceneFileExtensions(final String value) { System.setProperty("es.allow_unknown_lucene_file_extensions", value); } } + + public void testIsLuceneExtension() { + assertFalse(LuceneFilesExtensions.isLuceneExtension(null)); + assertFalse(LuceneFilesExtensions.isLuceneExtension("bcde")); + String randomStringWithLuceneExtension = randomAlphanumericOfLength(10) + + "." + + LuceneFilesExtensions.values()[randomInt(LuceneFilesExtensions.values().length) - 1].getExtension(); + String extension = IndexFileNames.getExtension(randomStringWithLuceneExtension); + assertTrue(extension + " should be considered a Lucene extension", LuceneFilesExtensions.isLuceneExtension(extension)); + String upperCaseExtension = extension.toUpperCase(Locale.ROOT); + assertFalse( + upperCaseExtension + " (uppercase) should not be considered a Lucene extension", + LuceneFilesExtensions.isLuceneExtension(upperCaseExtension) + ); + } } diff --git a/server/src/main/java/org/elasticsearch/index/store/LuceneFilesExtensions.java b/server/src/main/java/org/elasticsearch/index/store/LuceneFilesExtensions.java index 5fe85222e2597..522db5bb7bbbe 100644 --- a/server/src/main/java/org/elasticsearch/index/store/LuceneFilesExtensions.java +++ b/server/src/main/java/org/elasticsearch/index/store/LuceneFilesExtensions.java @@ -165,6 +165,11 @@ public static LuceneFilesExtensions fromExtension(String ext) { return null; } + @Nullable + public static boolean isLuceneExtension(String ext) { + return extensions.containsKey(ext); + } + @Nullable public static LuceneFilesExtensions fromFile(String fileName) { return fromExtension(IndexFileNames.getExtension(fileName)); diff --git a/x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java b/x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java index 8c04afba0c1c8..37474a2bb2154 100644 --- a/x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java +++ b/x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java @@ -9,6 +9,7 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import org.apache.lucene.index.IndexFileNames; import org.apache.lucene.store.AlreadyClosedException; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.ActionRunnable; @@ -37,6 +38,7 @@ import org.elasticsearch.core.TimeValue; import org.elasticsearch.env.Environment; import org.elasticsearch.env.NodeEnvironment; +import org.elasticsearch.index.store.LuceneFilesExtensions; import org.elasticsearch.monitor.fs.FsProbe; import org.elasticsearch.node.NodeRoleSettings; import org.elasticsearch.threadpool.ThreadPool; @@ -67,6 +69,9 @@ import java.util.function.Predicate; import java.util.stream.Collectors; +import static org.elasticsearch.blobcache.BlobCacheMetrics.LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY; +import static org.elasticsearch.blobcache.BlobCacheMetrics.NON_LUCENE_EXTENSION_TO_RECORD; + /** * A caching layer on a local node to minimize network roundtrips to the remote blob store. */ @@ -1254,7 +1259,8 @@ public int populateAndRead( final ByteRange rangeToWrite, final ByteRange rangeToRead, final RangeAvailableHandler reader, - final RangeMissingHandler writer + final RangeMissingHandler writer, + String resourceDescription ) throws Exception { // some cache files can grow after being created, so rangeToWrite can be larger than the initial {@code length} assert rangeToWrite.start() >= 0 : rangeToWrite; @@ -1273,6 +1279,7 @@ public void fillCacheRange( IntConsumer progressUpdater, ActionListener completionListener ) throws IOException { + String blobFileExtension = getFileExtension(resourceDescription); writer.fillCacheRange( channel, channelPos, @@ -1283,7 +1290,8 @@ public void fillCacheRange( completionListener.map(unused -> { var elapsedTime = TimeUnit.NANOSECONDS.toMillis(relativeTimeInNanosSupplier.getAsLong() - startTime); blobCacheMetrics.getCacheMissLoadTimes().record(elapsedTime); - blobCacheMetrics.getCacheMissCounter().increment(); + blobCacheMetrics.getCacheMissCounter() + .incrementBy(1L, Map.of(LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY, blobFileExtension)); return null; }) ); @@ -2075,4 +2083,17 @@ public void close() { } } } + + private static String getFileExtension(String resourceDescription) { + // TODO: consider introspecting resourceDescription for compound files + if (resourceDescription.endsWith(LuceneFilesExtensions.CFS.getExtension())) { + return LuceneFilesExtensions.CFS.getExtension(); + } + String extension = IndexFileNames.getExtension(resourceDescription); + if (LuceneFilesExtensions.isLuceneExtension(extension)) { + return extension; + } else { + return NON_LUCENE_EXTENSION_TO_RECORD; + } + } } diff --git a/x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBlobCacheServiceTests.java b/x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBlobCacheServiceTests.java index 51161e0f5f454..49da1cb6d0259 100644 --- a/x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBlobCacheServiceTests.java +++ b/x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBlobCacheServiceTests.java @@ -43,6 +43,9 @@ import java.io.IOException; import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; @@ -51,6 +54,8 @@ import java.util.Set; import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -171,6 +176,78 @@ public void testBasicEviction() throws IOException { } } + public void testCacheMissOnPopulate() throws Exception { + Settings settings = Settings.builder() + .put(NODE_NAME_SETTING.getKey(), "node") + .put(SharedBlobCacheService.SHARED_CACHE_SIZE_SETTING.getKey(), ByteSizeValue.ofBytes(size(50)).getStringRep()) + .put(SharedBlobCacheService.SHARED_CACHE_REGION_SIZE_SETTING.getKey(), ByteSizeValue.ofBytes(size(10)).getStringRep()) + .put("path.home", createTempDir()) + .build(); + final DeterministicTaskQueue taskQueue = new DeterministicTaskQueue(); + RecordingMeterRegistry recordingMeterRegistry = new RecordingMeterRegistry(); + BlobCacheMetrics metrics = new BlobCacheMetrics(recordingMeterRegistry); + ExecutorService ioExecutor = Executors.newCachedThreadPool(); + try ( + NodeEnvironment environment = new NodeEnvironment(settings, TestEnvironment.newEnvironment(settings)); + var cacheService = new SharedBlobCacheService<>(environment, settings, taskQueue.getThreadPool(), ioExecutor, metrics) + ) { + ByteRange rangeRead = ByteRange.of(0L, 1L); + ByteRange rangeWrite = ByteRange.of(0L, 1L); + Path tempFile = createTempFile("test", "other"); + String resourceDescription = tempFile.toAbsolutePath().toString(); + final var cacheKey = generateCacheKey(); + SharedBlobCacheService.CacheFile cacheFile = cacheService.getCacheFile(cacheKey, 1L); + + ByteBuffer writeBuffer = ByteBuffer.allocate(1); + + final int bytesRead = cacheFile.populateAndRead( + rangeRead, + rangeWrite, + (channel, pos, relativePos, len) -> len, + (channel, channelPos, streamFactory, relativePos, len, progressUpdater, completionListener) -> { + try (var in = Files.newInputStream(tempFile)) { + SharedBytes.copyToCacheFileAligned(channel, in, channelPos, progressUpdater, writeBuffer.clear()); + } + ActionListener.completeWith(completionListener, () -> null); + }, + resourceDescription + ); + assertThat(bytesRead, is(1)); + List measurements = recordingMeterRegistry.getRecorder() + .getMeasurements(InstrumentType.LONG_COUNTER, "es.blob_cache.miss_that_triggered_read.total"); + Measurement first = measurements.getFirst(); + assertThat(first.attributes().get("file_extension"), is("other")); + assertThat(first.value(), is(1L)); + + Path tempFile2 = createTempFile("test", "cfs"); + resourceDescription = tempFile2.toAbsolutePath().toString(); + cacheFile = cacheService.getCacheFile(generateCacheKey(), 1L); + + ByteBuffer writeBuffer2 = ByteBuffer.allocate(1); + + final int bytesRead2 = cacheFile.populateAndRead( + rangeRead, + rangeWrite, + (channel, pos, relativePos, len) -> len, + (channel, channelPos, streamFactory, relativePos, len, progressUpdater, completionListener) -> { + try (var in = Files.newInputStream(tempFile2)) { + SharedBytes.copyToCacheFileAligned(channel, in, channelPos, progressUpdater, writeBuffer2.clear()); + } + ActionListener.completeWith(completionListener, () -> null); + }, + resourceDescription + ); + assertThat(bytesRead2, is(1)); + + measurements = recordingMeterRegistry.getRecorder() + .getMeasurements(InstrumentType.LONG_COUNTER, "es.blob_cache.miss_that_triggered_read.total"); + Measurement measurement = measurements.get(1); + assertThat(measurement.attributes().get("file_extension"), is("cfs")); + assertThat(measurement.value(), is(1L)); + } + ioExecutor.shutdown(); + } + private static boolean tryEvict(SharedBlobCacheService.CacheFileRegion region1) { if (randomBoolean()) { return region1.tryEvict(); diff --git a/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/input/FrozenIndexInput.java b/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/input/FrozenIndexInput.java index d7cf22a05981f..dc0199f38b4f2 100644 --- a/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/input/FrozenIndexInput.java +++ b/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/input/FrozenIndexInput.java @@ -177,7 +177,8 @@ private void readWithoutBlobCacheSlow(ByteBuffer b, long position, int length) t return null; } } - ) + ), + fileInfo.physicalName() ); assert bytesRead == length : bytesRead + " vs " + length; byteBufferReference.finish(bytesRead);