Include project ID in enrich cache key

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,6 +9,7 @@
 
 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;
@@ -90,17 +91,17 @@ private Cache<CacheKey, CacheValue> createCache(long maxWeight, ToLongBiFunction
      * @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
      */
-    @FixForMultiProject(description = "The enrich cache will currently leak data between projects. We need to either disable or fix it.")
     public void computeIfAbsent(
         String enrichIndex,
+        ProjectId projectId,
         Object lookupValue,
         int maxMatches,
         Consumer<ActionListener<SearchResponse>> searchResponseFetcher,
         ActionListener<List<Map<?, ?>>> listener
     ) {
         // 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(enrichIndex, projectId, lookupValue, maxMatches);
         List<Map<?, ?>> response = get(cacheKey);
         long cacheRequestTime = relativeNanoTimeProvider.getAsLong() - cacheStart;
         if (response != null) {
@@ -204,7 +205,7 @@ private static Object innerDeepCopy(Object value, boolean unmodifiable) {
      * should thus be included in the cache key
      */
     // Visibility for testing
-    record CacheKey(String enrichIndex, Object lookupValue, int maxMatches) {
+    record CacheKey(String enrichIndex, ProjectId projectId, 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

@@ -139,6 +139,7 @@ private SearchRunner createSearchRunner(ProjectMetadata project, String indexAli
             // 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(
                 getEnrichIndexKey(project, indexAlias),
+                project.id(),
                 value,
                 maxMatches,
                 (searchResponseActionListener) -> originClient.execute(

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("policy1-1", projectId, "1", 1);
+        var cacheKey2 = new EnrichCache.CacheKey("policy1-1", projectId, "2", 1);
+        var cacheKey3 = new EnrichCache.CacheKey("policy2-1", projectId, "1", 1);
+        var cacheKey4 = new EnrichCache.CacheKey("policy2-1", projectId, "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("policy1-2", projectId, "1", 1);
+        cacheKey2 = new EnrichCache.CacheKey("policy1-2", projectId, "2", 1);
+        cacheKey3 = new EnrichCache.CacheKey("policy2-2", projectId, "1", 1);
+        cacheKey4 = new EnrichCache.CacheKey("policy2-2", projectId, "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("policy1-1", projectId, "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("policy1-1", projectId, "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("policy1-2", projectId, "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("policy1-1", randomUniqueProjectId(), "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("policy1-1", projectId, "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("policy1-1", projectId, "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) {