ES-10936 Include project ID in enrich cache key (#128908)

This prevents collisions in the enrich cache for different projects.

Collisions would be unlikely without this chance, since the key
includes the index name, and the index name includes a
millisecond-precision timestamp, so it would only happen if two
projects executed enrich policies with the same name at the same time,
or if one project manipulated the alias to match the timestamp of the
other project.

However, collisions would have serious consequences (allowing
read-across of data between projects). This change closes that gap.

Author: PeteGillinElastic

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichCache.java

@@ -9,11 +9,11 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.cache.Cache;
 import org.elasticsearch.common.cache.CacheBuilder;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.Maps;
-import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.xpack.core.enrich.action.EnrichStatsAction;
@@ -84,14 +84,15 @@ private Cache<CacheKey, CacheValue> createCache(long maxWeight, ToLongBiFunction
      * This method notifies the given listener of the value in this cache for the given search parameters. If there is no value in the cache
      * for these search parameters, then the new cache value is computed using searchResponseFetcher.
      *
-     * @param enrichIndex The enrich index from which the results will be retrieved
-     * @param lookupValue The value that will be used in the search
-     * @param maxMatches The max number of matches that the search will return
+     * @param projectId             The ID of the project
+     * @param enrichIndex           The enrich index from which the results will be retrieved
+     * @param lookupValue           The value that will be used in the search
+     * @param maxMatches            The max number of matches that the search will return
      * @param searchResponseFetcher The function used to compute the value to be put in the cache, if there is no value in the cache already
-     * @param listener A listener to be notified of the value in the cache
+     * @param listener              A listener to be notified of the value in the cache
      */
-    @FixForMultiProject(description = "The enrich cache will currently leak data between projects. We need to either disable or fix it.")
     public void computeIfAbsent(
+        ProjectId projectId,
         String enrichIndex,
         Object lookupValue,
         int maxMatches,
@@ -100,7 +101,7 @@ public void computeIfAbsent(
     ) {
         // intentionally non-locking for simplicity...it's OK if we re-put the same key/value in the cache during a race condition.
         long cacheStart = relativeNanoTimeProvider.getAsLong();
-        var cacheKey = new CacheKey(enrichIndex, lookupValue, maxMatches);
+        var cacheKey = new CacheKey(projectId, enrichIndex, lookupValue, maxMatches);
         List<Map<?, ?>> response = get(cacheKey);
         long cacheRequestTime = relativeNanoTimeProvider.getAsLong() - cacheStart;
         if (response != null) {
@@ -200,11 +201,11 @@ private static Object innerDeepCopy(Object value, boolean unmodifiable) {
      *
      * @param enrichIndex The enrich <i>index</i> (i.e. not the alias, but the concrete index that the alias points to)
      * @param lookupValue The value that is used to find matches in the enrich index
-     * @param maxMatches The max number of matches that the enrich lookup should return. This changes the size of the search response and
-     * should thus be included in the cache key
+     * @param maxMatches  The max number of matches that the enrich lookup should return. This changes the size of the search response and
+     *                    should thus be included in the cache key
      */
     // Visibility for testing
-    record CacheKey(String enrichIndex, Object lookupValue, int maxMatches) {
+    record CacheKey(ProjectId projectId, String enrichIndex, Object lookupValue, int maxMatches) {
         /**
          * In reality, the size in bytes of the cache key is a function of the {@link CacheKey#lookupValue} field plus some constant for
          * the object itself, the string reference for the enrich index (but not the string itself because it's taken from the metadata),

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichProcessorFactory.java

@@ -138,6 +138,7 @@ private SearchRunner createSearchRunner(ProjectMetadata project, String indexAli
         return (value, maxMatches, reqSupplier, handler) -> {
             // intentionally non-locking for simplicity...it's OK if we re-put the same key/value in the cache during a race condition.
             enrichCache.computeIfAbsent(
+                project.id(),
                 getEnrichIndexKey(project, indexAlias),
                 value,
                 maxMatches,

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichCacheTests.java

@@ -7,6 +7,7 @@
 package org.elasticsearch.xpack.enrich;
 
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.search.SearchHit;
@@ -38,10 +39,11 @@ public class EnrichCacheTests extends ESTestCase {
     public void testCaching() {
         // Emulated search requests that an enrich processor could generate:
         // (two unique searches for two enrich policies)
-        var cacheKey1 = new EnrichCache.CacheKey("policy1-1", "1", 1);
-        var cacheKey2 = new EnrichCache.CacheKey("policy1-1", "2", 1);
-        var cacheKey3 = new EnrichCache.CacheKey("policy2-1", "1", 1);
-        var cacheKey4 = new EnrichCache.CacheKey("policy2-1", "2", 1);
+        var projectId = randomProjectIdOrDefault();
+        var cacheKey1 = new EnrichCache.CacheKey(projectId, "policy1-1", "1", 1);
+        var cacheKey2 = new EnrichCache.CacheKey(projectId, "policy1-1", "2", 1);
+        var cacheKey3 = new EnrichCache.CacheKey(projectId, "policy2-1", "1", 1);
+        var cacheKey4 = new EnrichCache.CacheKey(projectId, "policy2-1", "2", 1);
         // Emulated search response (content doesn't matter, since it isn't used, it just a cache entry)
         EnrichCache.CacheValue searchResponse = new EnrichCache.CacheValue(List.of(Map.of("test", "entry")), 1L);
 
@@ -75,10 +77,10 @@ public void testCaching() {
         assertThat(cacheStats.evictions(), equalTo(1L));
         assertThat(cacheStats.cacheSizeInBytes(), equalTo(3L));
 
-        cacheKey1 = new EnrichCache.CacheKey("policy1-2", "1", 1);
-        cacheKey2 = new EnrichCache.CacheKey("policy1-2", "2", 1);
-        cacheKey3 = new EnrichCache.CacheKey("policy2-2", "1", 1);
-        cacheKey4 = new EnrichCache.CacheKey("policy2-2", "2", 1);
+        cacheKey1 = new EnrichCache.CacheKey(projectId, "policy1-2", "1", 1);
+        cacheKey2 = new EnrichCache.CacheKey(projectId, "policy1-2", "2", 1);
+        cacheKey3 = new EnrichCache.CacheKey(projectId, "policy2-2", "1", 1);
+        cacheKey4 = new EnrichCache.CacheKey(projectId, "policy2-2", "2", 1);
 
         // Because enrich index has changed, cache can't serve cached entries
         assertThat(enrichCache.get(cacheKey1), nullValue());
@@ -115,10 +117,13 @@ public void testComputeIfAbsent() throws InterruptedException {
         // We use a relative time provider that increments 1ms every time it is called. So each operation appears to take 1ms
         EnrichCache enrichCache = new EnrichCache(3, () -> testNanoTime.addAndGet(TimeValue.timeValueMillis(1).getNanos()));
 
+        ProjectId projectId = randomProjectIdOrDefault();
+        long expectedMisses = 0L;
         {
+            // Do initial computeIfAbsent, assert that it is a cache miss and the search is performed:
             CountDownLatch queriedDatabaseLatch = new CountDownLatch(1);
             CountDownLatch notifiedOfResultLatch = new CountDownLatch(1);
-            enrichCache.computeIfAbsent("policy1-1", "1", 1, (searchResponseActionListener) -> {
+            enrichCache.computeIfAbsent(projectId, "policy1-1", "1", 1, (searchResponseActionListener) -> {
                 SearchResponse searchResponse = convertToSearchResponse(searchResponseMap);
                 searchResponseActionListener.onResponse(searchResponse);
                 searchResponse.decRef();
@@ -132,26 +137,103 @@ public void testComputeIfAbsent() throws InterruptedException {
             EnrichStatsAction.Response.CacheStats cacheStats = enrichCache.getStats(randomAlphaOfLength(10));
             assertThat(cacheStats.count(), equalTo(1L));
             assertThat(cacheStats.hits(), equalTo(0L));
-            assertThat(cacheStats.misses(), equalTo(1L));
+            assertThat(cacheStats.misses(), equalTo(++expectedMisses));
             assertThat(cacheStats.evictions(), equalTo(0L));
             assertThat(cacheStats.hitsTimeInMillis(), equalTo(0L));
             assertThat(cacheStats.missesTimeInMillis(), equalTo(2L)); // cache query and enrich query + cache put
         }
 
         {
+            // Do the same call, assert that it is a cache hit and no search is performed:
             CountDownLatch notifiedOfResultLatch = new CountDownLatch(1);
-            enrichCache.computeIfAbsent("policy1-1", "1", 1, (searchResponseActionListener) -> {
+            enrichCache.computeIfAbsent(projectId, "policy1-1", "1", 1, (searchResponseActionListener) -> {
                 fail("Expected no call to the database because item should have been in the cache");
             }, assertNoFailureListener(r -> notifiedOfResultLatch.countDown()));
             assertThat(notifiedOfResultLatch.await(5, TimeUnit.SECONDS), equalTo(true));
             EnrichStatsAction.Response.CacheStats cacheStats = enrichCache.getStats(randomAlphaOfLength(10));
             assertThat(cacheStats.count(), equalTo(1L));
             assertThat(cacheStats.hits(), equalTo(1L));
-            assertThat(cacheStats.misses(), equalTo(1L));
+            assertThat(cacheStats.misses(), equalTo(expectedMisses));
             assertThat(cacheStats.evictions(), equalTo(0L));
             assertThat(cacheStats.hitsTimeInMillis(), equalTo(1L));
             assertThat(cacheStats.missesTimeInMillis(), equalTo(2L));
         }
+
+        {
+            // Do a computeIfAbsent with a different index, assert that it is a cache miss and the search is performed:
+            CountDownLatch queriedDatabaseLatch = new CountDownLatch(1);
+            CountDownLatch notifiedOfResultLatch = new CountDownLatch(1);
+            enrichCache.computeIfAbsent(projectId, "policy1-2", "1", 1, (searchResponseActionListener) -> {
+                SearchResponse searchResponse = convertToSearchResponse(searchResponseMap);
+                searchResponseActionListener.onResponse(searchResponse);
+                searchResponse.decRef();
+                queriedDatabaseLatch.countDown();
+            }, assertNoFailureListener(response -> {
+                assertThat(response, equalTo(searchResponseMap));
+                notifiedOfResultLatch.countDown();
+            }));
+            assertThat(queriedDatabaseLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            assertThat(notifiedOfResultLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            EnrichStatsAction.Response.CacheStats cacheStats = enrichCache.getStats(randomAlphaOfLength(10));
+            assertThat(cacheStats.misses(), equalTo(++expectedMisses));
+        }
+
+        {
+            // Do a computeIfAbsent with a different project, assert that it is a cache miss and the search is performed:
+            CountDownLatch queriedDatabaseLatch = new CountDownLatch(1);
+            CountDownLatch notifiedOfResultLatch = new CountDownLatch(1);
+            enrichCache.computeIfAbsent(randomUniqueProjectId(), "policy1-1", "1", 1, (searchResponseActionListener) -> {
+                SearchResponse searchResponse = convertToSearchResponse(searchResponseMap);
+                searchResponseActionListener.onResponse(searchResponse);
+                searchResponse.decRef();
+                queriedDatabaseLatch.countDown();
+            }, assertNoFailureListener(response -> {
+                assertThat(response, equalTo(searchResponseMap));
+                notifiedOfResultLatch.countDown();
+            }));
+            assertThat(queriedDatabaseLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            assertThat(notifiedOfResultLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            EnrichStatsAction.Response.CacheStats cacheStats = enrichCache.getStats(randomAlphaOfLength(10));
+            assertThat(cacheStats.misses(), equalTo(++expectedMisses));
+        }
+
+        {
+            // Do a computeIfAbsent with a different lookup value, assert that it is a cache miss and the search is performed:
+            CountDownLatch queriedDatabaseLatch = new CountDownLatch(1);
+            CountDownLatch notifiedOfResultLatch = new CountDownLatch(1);
+            enrichCache.computeIfAbsent(projectId, "policy1-1", "2", 1, (searchResponseActionListener) -> {
+                SearchResponse searchResponse = convertToSearchResponse(searchResponseMap);
+                searchResponseActionListener.onResponse(searchResponse);
+                searchResponse.decRef();
+                queriedDatabaseLatch.countDown();
+            }, assertNoFailureListener(response -> {
+                assertThat(response, equalTo(searchResponseMap));
+                notifiedOfResultLatch.countDown();
+            }));
+            assertThat(queriedDatabaseLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            assertThat(notifiedOfResultLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            EnrichStatsAction.Response.CacheStats cacheStats = enrichCache.getStats(randomAlphaOfLength(10));
+            assertThat(cacheStats.misses(), equalTo(++expectedMisses));
+        }
+
+        {
+            // Do a computeIfAbsent with a different max matches, assert that it is a cache miss and the search is performed:
+            CountDownLatch queriedDatabaseLatch = new CountDownLatch(1);
+            CountDownLatch notifiedOfResultLatch = new CountDownLatch(1);
+            enrichCache.computeIfAbsent(projectId, "policy1-1", "1", 3, (searchResponseActionListener) -> {
+                SearchResponse searchResponse = convertToSearchResponse(searchResponseMap);
+                searchResponseActionListener.onResponse(searchResponse);
+                searchResponse.decRef();
+                queriedDatabaseLatch.countDown();
+            }, assertNoFailureListener(response -> {
+                assertThat(response, equalTo(searchResponseMap));
+                notifiedOfResultLatch.countDown();
+            }));
+            assertThat(queriedDatabaseLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            assertThat(notifiedOfResultLatch.await(5, TimeUnit.SECONDS), equalTo(true));
+            EnrichStatsAction.Response.CacheStats cacheStats = enrichCache.getStats(randomAlphaOfLength(10));
+            assertThat(cacheStats.misses(), equalTo(++expectedMisses));
+        }
     }
 
     private SearchResponse convertToSearchResponse(List<Map<String, ?>> searchResponseList) {