Reduce Overhead of RepositoryData Cache for Large Repositories (#66587) (#66609)

This adds caching the fact that `RepositoryData` was too large to be
cached so we don't serialize it over and over just to find out we can't
cache it.

Author: original-brownbear

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -678,10 +678,10 @@ public void onFailure(Exception e) {
      * @param rootBlobs         Blobs at the repository root
      * @return RepositoryData
      */
-    private RepositoryData safeRepositoryData(long repositoryStateId, Map<String, BlobMetadata> rootBlobs) throws IOException {
+    private RepositoryData safeRepositoryData(long repositoryStateId, Map<String, BlobMetadata> rootBlobs) {
         final long generation = latestGeneration(rootBlobs.keySet());
         final long genToLoad;
-        final Tuple<Long, BytesReference> cached;
+        final CachedRepositoryData cached;
         if (bestEffortConsistency) {
             genToLoad = latestKnownRepoGen.updateAndGet(known -> Math.max(known, repositoryStateId));
             cached = null;
@@ -700,8 +700,8 @@ private RepositoryData safeRepositoryData(long repositoryStateId, Map<String, Bl
             throw new RepositoryException(metadata.name(), "concurrent modification of the index-N file, expected current generation [" +
                 repositoryStateId + "], actual current generation [" + genToLoad + "]");
         }
-        if (cached != null && cached.v1() == genToLoad) {
-            return repositoryDataFromCachedEntry(cached);
+        if (cached != null && cached.generation() == genToLoad && cached.hasData()) {
+            return cached.repositoryData();
         }
         return getRepositoryData(genToLoad);
     }
@@ -1288,7 +1288,6 @@ public String startVerification() {
                 String seed = UUIDs.randomBase64UUID();
                 byte[] testBytes = Strings.toUTF8Bytes(seed);
                 BlobContainer testContainer = blobStore().blobContainer(basePath().add(testBlobPrefix(seed)));
-                BytesArray bytes = new BytesArray(testBytes);
                 testContainer.writeBlobAtomic("master.dat", new BytesArray(testBytes), true);
                 return seed;
             }
@@ -1314,20 +1313,62 @@ public void endVerification(String seed) {
     private final AtomicLong latestKnownRepoGen = new AtomicLong(RepositoryData.UNKNOWN_REPO_GEN);
 
     // Best effort cache of the latest known repository data and its generation, cached serialized as compressed json
-    private final AtomicReference<Tuple<Long, BytesReference>> latestKnownRepositoryData = new AtomicReference<>();
+    private final AtomicReference<CachedRepositoryData> latestKnownRepositoryData =
+            new AtomicReference<>(new CachedRepositoryData(RepositoryData.EMPTY_REPO_GEN, null));
+
+    /**
+     * Cached serialized repository data or placeholder to keep track of the fact that data for a generation was too large to be cached.
+     */
+    private static final class CachedRepositoryData {
+
+        private final long generation;
+
+        @Nullable
+        private final BytesReference repositoryData;
+
+        CachedRepositoryData(long generation, @Nullable BytesReference repositoryData) {
+            this.generation = generation;
+            this.repositoryData = repositoryData;
+        }
+
+        long generation() {
+            return generation;
+        }
+
+        boolean hasData() {
+            return generation == RepositoryData.EMPTY_REPO_GEN || repositoryData != null;
+        }
+
+        @Nullable
+        RepositoryData repositoryData() {
+            if (generation == RepositoryData.EMPTY_REPO_GEN) {
+                return RepositoryData.EMPTY;
+            }
+            if (repositoryData == null) {
+                return null;
+            }
+            try (InputStream input = CompressorFactory.COMPRESSOR.threadLocalInputStream(repositoryData.streamInput())) {
+                return RepositoryData.snapshotsFromXContent(
+                        XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, input),
+                        generation, false);
+            } catch (IOException e) {
+                throw new AssertionError("no actual IO happens here", e);
+            }
+        }
+    }
 
     @Override
     public void getRepositoryData(ActionListener<RepositoryData> listener) {
         if (latestKnownRepoGen.get() == RepositoryData.CORRUPTED_REPO_GEN) {
             listener.onFailure(corruptedStateException(null));
             return;
         }
-        final Tuple<Long, BytesReference> cached = latestKnownRepositoryData.get();
+        final CachedRepositoryData cached = latestKnownRepositoryData.get();
         // Fast path loading repository data directly from cache if we're in fully consistent mode and the cache matches up with
         // the latest known repository generation
-        if (bestEffortConsistency == false && cached != null && cached.v1() == latestKnownRepoGen.get()) {
+        if (bestEffortConsistency == false && cached.generation() == latestKnownRepoGen.get() && cached.hasData()) {
             try {
-                listener.onResponse(repositoryDataFromCachedEntry(cached));
+                listener.onResponse(cached.repositoryData());
             } catch (Exception e) {
                 listener.onFailure(e);
             }
@@ -1357,26 +1398,28 @@ private void doGetRepositoryData(ActionListener<RepositoryData> listener) {
                 }
                 genToLoad = latestKnownRepoGen.updateAndGet(known -> Math.max(known, generation));
                 if (genToLoad > generation) {
-                    logger.info("Determined repository generation [" + generation
-                        + "] from repository contents but correct generation must be at least [" + genToLoad + "]");
+                    logger.info("Determined repository generation [{}] from repository contents but correct generation must be at " +
+                            "least [{}]", generation, genToLoad);
                 }
             } else {
                 // We only rely on the generation tracked in #latestKnownRepoGen which is exclusively updated from the cluster state
                 genToLoad = latestKnownRepoGen.get();
             }
             try {
-                final Tuple<Long, BytesReference> cached = latestKnownRepositoryData.get();
+                final CachedRepositoryData cached = latestKnownRepositoryData.get();
                 final RepositoryData loaded;
                 // Caching is not used with #bestEffortConsistency see docs on #cacheRepositoryData for details
-                if (bestEffortConsistency == false && cached != null && cached.v1() == genToLoad) {
-                    loaded = repositoryDataFromCachedEntry(cached);
+                if (bestEffortConsistency == false && cached.generation() == genToLoad && cached.hasData()) {
+                    loaded = cached.repositoryData();
                 } else {
                     loaded = getRepositoryData(genToLoad);
-                    // We can cache serialized in the most recent version here without regard to the actual repository metadata version
-                    // since we're only caching the information that we just wrote and thus won't accidentally cache any information that
-                    // isn't safe
-                    cacheRepositoryData(compressRepoDataForCache(BytesReference.bytes(
-                            loaded.snapshotsToXContent(XContentFactory.jsonBuilder(), Version.CURRENT))), genToLoad);
+                    if (cached == null || cached.generation() < genToLoad) {
+                        // We can cache serialized in the most recent version here without regard to the actual repository metadata version
+                        // since we're only caching the information that we just wrote and thus won't accidentally cache any information
+                        // that isn't safe
+                        cacheRepositoryData(compressRepoDataForCache(BytesReference.bytes(
+                                loaded.snapshotsToXContent(XContentFactory.jsonBuilder(), Version.CURRENT))), genToLoad);
+                    }
                 }
                 listener.onResponse(loaded);
                 return;
@@ -1416,11 +1459,12 @@ private void doGetRepositoryData(ActionListener<RepositoryData> listener) {
      * @param generation repository generation of the given repository data
      */
     private void cacheRepositoryData(@Nullable BytesReference serialized, long generation) {
+        assert generation >= 0 : "No need to cache abstract generations but attempted to cache [" + generation + "]";
         latestKnownRepositoryData.updateAndGet(known -> {
-            if (known != null && known.v1() > generation) {
+            if (known.generation() > generation) {
                 return known;
             }
-            return serialized == null ? null : new Tuple<>(generation, serialized);
+            return new CachedRepositoryData(generation, serialized);
         });
     }
 
@@ -1456,14 +1500,6 @@ private BytesReference compressRepoDataForCache(BytesReference uncompressed) {
         }
     }
 
-    private RepositoryData repositoryDataFromCachedEntry(Tuple<Long, BytesReference> cacheEntry) throws IOException {
-        try (InputStream input = CompressorFactory.COMPRESSOR.threadLocalInputStream(cacheEntry.v2().streamInput())) {
-            return RepositoryData.snapshotsFromXContent(
-                    XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
-                            LoggingDeprecationHandler.INSTANCE, input), cacheEntry.v1(), false);
-        }
-    }
-
     private RepositoryException corruptedStateException(@Nullable Exception cause) {
         return new RepositoryException(metadata.name(),
             "Could not read repository data because the contents of the repository do not match its " +