Add multi-project support for health indicator data_stream_lifecycle

modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java

@@ -216,7 +216,7 @@ public Collection<?> createComponents(PluginServices services) {
             )
         );
         dataLifecycleInitialisationService.get().init();
-        dataStreamLifecycleHealthIndicatorService.set(new DataStreamLifecycleHealthIndicatorService());
+        dataStreamLifecycleHealthIndicatorService.set(new DataStreamLifecycleHealthIndicatorService(services.projectResolver()));
 
         components.add(errorStoreInitialisationService.get());
         components.add(dataLifecycleInitialisationService.get());

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleErrorStore.java

@@ -14,8 +14,11 @@
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.health.node.DslErrorInfo;
+import org.elasticsearch.health.node.ProjectIndexName;
 
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -117,23 +120,38 @@ public Set<String> getAllIndices(ProjectId projectId) {
      * retries DSL attempted (descending order) and the number of entries will be limited according to the provided limit parameter.
      * Returns empty list if no entries are present in the error store or none satisfy the predicate.
      */
-    public List<DslErrorInfo> getErrorsInfo(ProjectId projectId, Predicate<ErrorEntry> errorEntryPredicate, int limit) {
-        final var indexNameToError = projectMap.get(projectId);
-        if (indexNameToError == null || indexNameToError.isEmpty()) {
-            return List.of();
-        }
-        return indexNameToError.entrySet()
+    public List<DslErrorInfo> getErrorsInfo(Predicate<ErrorEntry> errorEntryPredicate, int limit) {
+        return projectMap.entrySet()
             .stream()
-            .filter(keyValue -> errorEntryPredicate.test(keyValue.getValue()))
-            .sorted(Map.Entry.comparingByValue())
+            .flatMap(
+                projectToIndexError -> projectToIndexError.getValue()
+                    .entrySet()
+                    .stream()
+                    .map(
+                        indexToError -> new Tuple<>(
+                            new ProjectIndexName(projectToIndexError.getKey(), indexToError.getKey()),
+                            indexToError.getValue()
+                        )
+                    )
+            )
+            .filter(projectIndexAndError -> errorEntryPredicate.test(projectIndexAndError.v2()))
+            .sorted(Comparator.comparing(Tuple::v2))
             .limit(limit)
             .map(
-                keyValue -> new DslErrorInfo(
-                    keyValue.getKey(),
-                    keyValue.getValue().firstOccurrenceTimestamp(),
-                    keyValue.getValue().retryCount()
+                projectIndexAndError -> new DslErrorInfo(
+                    projectIndexAndError.v1().indexName(),
+                    projectIndexAndError.v2().firstOccurrenceTimestamp(),
+                    projectIndexAndError.v2().retryCount(),
+                    projectIndexAndError.v1().projectId()
                 )
             )
             .collect(Collectors.toList());
     }
+
+    /**
+     * Get the total number of error entries in the store
+     */
+    public int getTotalErrorEntries() {
+        return projectMap.values().stream().mapToInt(Map::size).sum();
+    }
 }

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthIndicatorService.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.datastreams.lifecycle.health;
 
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.health.Diagnosis;
 import org.elasticsearch.health.HealthIndicatorDetails;
 import org.elasticsearch.health.HealthIndicatorImpact;
@@ -20,6 +21,7 @@
 import org.elasticsearch.health.node.DataStreamLifecycleHealthInfo;
 import org.elasticsearch.health.node.DslErrorInfo;
 import org.elasticsearch.health.node.HealthInfo;
+import org.elasticsearch.health.node.ProjectIndexName;
 
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -54,6 +56,12 @@ public class DataStreamLifecycleHealthIndicatorService implements HealthIndicato
         DSL_EXPLAIN_HELP_URL
     );
 
+    private final ProjectResolver projectResolver;
+
+    public DataStreamLifecycleHealthIndicatorService(ProjectResolver projectResolver) {
+        this.projectResolver = projectResolver;
+    }
+
     @Override
     public String name() {
         return NAME;
@@ -79,20 +87,20 @@ public HealthIndicatorResult calculate(boolean verbose, int maxAffectedResources
             return createIndicator(
                 HealthStatus.GREEN,
                 "Data streams are executing their lifecycles without issues",
-                createDetails(verbose, dataStreamLifecycleHealthInfo),
+                createDetails(verbose, dataStreamLifecycleHealthInfo, projectResolver.supportsMultipleProjects()),
                 List.of(),
                 List.of()
             );
         } else {
             List<String> affectedIndices = stagnatingBackingIndices.stream()
-                .map(DslErrorInfo::indexName)
+                .map(dslErrorInfo -> indexDisplayName(dslErrorInfo, projectResolver.supportsMultipleProjects()))
                 .limit(Math.min(maxAffectedResourcesCount, stagnatingBackingIndices.size()))
                 .collect(toList());
             return createIndicator(
                 HealthStatus.YELLOW,
                 (stagnatingBackingIndices.size() > 1 ? stagnatingBackingIndices.size() + " backing indices have" : "A backing index has")
                     + " repeatedly encountered errors whilst trying to advance in its lifecycle",
-                createDetails(verbose, dataStreamLifecycleHealthInfo),
+                createDetails(verbose, dataStreamLifecycleHealthInfo, projectResolver.supportsMultipleProjects()),
                 STAGNATING_INDEX_IMPACT,
                 verbose
                     ? List.of(
@@ -106,7 +114,11 @@ public HealthIndicatorResult calculate(boolean verbose, int maxAffectedResources
         }
     }
 
-    private static HealthIndicatorDetails createDetails(boolean verbose, DataStreamLifecycleHealthInfo dataStreamLifecycleHealthInfo) {
+    private static HealthIndicatorDetails createDetails(
+        boolean verbose,
+        DataStreamLifecycleHealthInfo dataStreamLifecycleHealthInfo,
+        boolean supportsMultipleProjects
+    ) {
         if (verbose == false) {
             return HealthIndicatorDetails.EMPTY;
         }
@@ -117,12 +129,16 @@ private static HealthIndicatorDetails createDetails(boolean verbose, DataStreamL
         if (dataStreamLifecycleHealthInfo.dslErrorsInfo().isEmpty() == false) {
             details.put("stagnating_backing_indices", dataStreamLifecycleHealthInfo.dslErrorsInfo().stream().map(dslError -> {
                 LinkedHashMap<String, Object> errorDetails = new LinkedHashMap<>(3, 1L);
-                errorDetails.put("index_name", dslError.indexName());
+                errorDetails.put("index_name", indexDisplayName(dslError, supportsMultipleProjects));
                 errorDetails.put("first_occurrence_timestamp", dslError.firstOccurrence());
                 errorDetails.put("retry_count", dslError.retryCount());
                 return errorDetails;
             }).toList());
         }
         return new SimpleHealthIndicatorDetails(details);
     }
+
+    private static String indexDisplayName(DslErrorInfo dslErrorInfo, boolean supportsMultipleProjects) {
+        return new ProjectIndexName(dslErrorInfo.projectId(), dslErrorInfo.indexName()).toString(supportsMultipleProjects);
+    }
 }

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisher.java

@@ -14,12 +14,10 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
-import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleErrorStore;
 import org.elasticsearch.health.node.DataStreamLifecycleHealthInfo;
 import org.elasticsearch.health.node.DslErrorInfo;
@@ -85,24 +83,22 @@ private void updateNumberOfErrorsToPublish(int newValue) {
      * {@link org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService#DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING}
      */
     public void publishDslErrorEntries(ActionListener<AcknowledgedResponse> actionListener) {
-        @FixForMultiProject(description = "Once the health API becomes project-aware, we shouldn't use the default project ID")
-        final var projectId = Metadata.DEFAULT_PROJECT_ID;
-        // fetching the entries that persist in the error store for more than the signalling retry interval
-        // note that we're reporting this view into the error store on every publishing iteration
-        List<DslErrorInfo> errorEntriesToSignal = errorStore.getErrorsInfo(
-            projectId,
-            entry -> entry.retryCount() >= signallingErrorRetryInterval,
-            maxNumberOfErrorsToPublish
-        );
         DiscoveryNode currentHealthNode = HealthNode.findHealthNode(clusterService.state());
         if (currentHealthNode != null) {
             String healthNodeId = currentHealthNode.getId();
+            // fetching the entries that persist in the error store for more than the signalling retry interval
+            // note that we're reporting this view into the error store on every publishing iteration
+            List<DslErrorInfo> errorEntriesToSignal = errorStore.getErrorsInfo(
+                entry -> entry.retryCount() >= signallingErrorRetryInterval,
+                maxNumberOfErrorsToPublish
+            );
+
             logger.trace("reporting [{}] DSL error entries to to health node [{}]", errorEntriesToSignal.size(), healthNodeId);
             client.execute(
                 UpdateHealthInfoCacheAction.INSTANCE,
                 new UpdateHealthInfoCacheAction.Request(
                     healthNodeId,
-                    new DataStreamLifecycleHealthInfo(errorEntriesToSignal, errorStore.getAllIndices(projectId).size())
+                    new DataStreamLifecycleHealthInfo(errorEntriesToSignal, errorStore.getTotalErrorEntries())
                 ),
                 actionListener
             );

modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleErrorStoreTests.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.IntStream;
@@ -101,26 +102,94 @@ public void testGetFilteredEntries() {
         IntStream.range(0, 5).forEach(i -> errorStore.recordError(projectId, "test5", new NullPointerException("testing")));
 
         {
-            List<DslErrorInfo> entries = errorStore.getErrorsInfo(projectId, entry -> entry.retryCount() > 7, 100);
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 7, 100);
             assertThat(entries.size(), is(1));
+            assertThat(entries.getFirst().indexName(), is("test20"));
+            assertThat(entries.getFirst().projectId(), is(projectId));
+        }
+
+        {
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 7, 0);
+            assertThat(entries.size(), is(0));
+        }
+
+        {
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 50, 100);
+            assertThat(entries.size(), is(0));
+        }
+
+        {
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 2, 100);
+            assertThat(entries.size(), is(2));
             assertThat(entries.get(0).indexName(), is("test20"));
+            assertThat(entries.get(0).projectId(), is(projectId));
+            assertThat(entries.get(1).indexName(), is("test5"));
+            assertThat(entries.get(1).projectId(), is(projectId));
         }
+    }
+
+    public void testGetFilteredEntriesForMultipleProjects() {
+        ProjectId projectId1 = randomProjectIdOrDefault();
+        ProjectId projectId2 = randomUniqueProjectId();
+        IntStream.range(0, 20).forEach(i -> errorStore.recordError(projectId1, "test20", new NullPointerException("testing")));
+        IntStream.range(0, 5).forEach(i -> errorStore.recordError(projectId2, "test5", new NullPointerException("testing")));
 
         {
-            List<DslErrorInfo> entries = errorStore.getErrorsInfo(projectId, entry -> entry.retryCount() > 7, 0);
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 7, 100);
+            assertThat(entries.size(), is(1));
+            assertThat(entries.getFirst().indexName(), is("test20"));
+            assertThat(entries.getFirst().projectId(), is(projectId1));
+        }
+
+        {
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 7, 0);
             assertThat(entries.size(), is(0));
         }
 
         {
-            List<DslErrorInfo> entries = errorStore.getErrorsInfo(projectId, entry -> entry.retryCount() > 50, 100);
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 50, 100);
             assertThat(entries.size(), is(0));
         }
 
         {
-            List<DslErrorInfo> entries = errorStore.getErrorsInfo(projectId, entry -> entry.retryCount() > 2, 100);
+            List<DslErrorInfo> entries = errorStore.getErrorsInfo(entry -> entry.retryCount() > 2, 100);
             assertThat(entries.size(), is(2));
             assertThat(entries.get(0).indexName(), is("test20"));
+            assertThat(entries.get(0).projectId(), is(projectId1));
             assertThat(entries.get(1).indexName(), is("test5"));
+            assertThat(entries.get(1).projectId(), is(projectId2));
+        }
+    }
+
+    public void testTotalErrorCount() {
+        ProjectId projectId1 = randomProjectIdOrDefault();
+        ProjectId projectId2 = randomUniqueProjectId();
+
+        {
+            // empty store
+            assertThat(errorStore.getTotalErrorEntries(), is(0));
+        }
+
+        {
+            // single project multiple indices
+            IntStream.range(1, 20).forEach(i -> errorStore.recordError(projectId1, "index1", new NullPointerException("testing")));
+            IntStream.range(1, 5).forEach(i -> errorStore.recordError(projectId1, "index2", new NullPointerException("testing")));
+            IntStream.range(1, 5).forEach(i -> errorStore.recordError(projectId1, "index2", new IOException("testing")));
+            assertThat(errorStore.getTotalErrorEntries(), is(2));
+        }
+
+        {
+            // clear store
+            errorStore.clearStore();
+            assertThat(errorStore.getTotalErrorEntries(), is(0));
+        }
+
+        {
+            // multiple projects
+            IntStream.range(1, 20).forEach(i -> errorStore.recordError(projectId1, "index1", new NullPointerException("testing")));
+            IntStream.range(1, 5).forEach(i -> errorStore.recordError(projectId1, "index2", new IOException("testing")));
+            IntStream.range(1, 5).forEach(i -> errorStore.recordError(projectId2, "index1", new NullPointerException("testing")));
+            assertThat(errorStore.getTotalErrorEntries(), is(3));
         }
     }
 }

modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthIndicatorServiceTests.java

@@ -10,6 +10,8 @@
 package org.elasticsearch.datastreams.lifecycle.health;
 
 import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.health.Diagnosis;
 import org.elasticsearch.health.HealthIndicatorDetails;
@@ -18,6 +20,7 @@
 import org.elasticsearch.health.node.DataStreamLifecycleHealthInfo;
 import org.elasticsearch.health.node.DslErrorInfo;
 import org.elasticsearch.health.node.HealthInfo;
+import org.elasticsearch.health.node.ProjectIndexName;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
@@ -38,7 +41,7 @@ public class DataStreamLifecycleHealthIndicatorServiceTests extends ESTestCase {
 
     @Before
     public void setupService() {
-        service = new DataStreamLifecycleHealthIndicatorService();
+        service = new DataStreamLifecycleHealthIndicatorService(TestProjectResolvers.singleProjectOnly(randomProjectIdOrDefault()));
     }
 
     public void testGreenWhenNoDSLHealthData() {
@@ -119,6 +122,50 @@ public void testSkippingFieldsWhenVerboseIsFalse() {
         assertThat(result.diagnosisList().isEmpty(), is(true));
     }
 
+    public void testMultiProject() {
+        service = new DataStreamLifecycleHealthIndicatorService(TestProjectResolvers.allProjects());
+        ProjectId projectId1 = randomProjectIdOrDefault();
+        ProjectId projectId2 = randomUniqueProjectId();
+        String index1 = DataStream.getDefaultBackingIndexName("foo", 1L);
+        String index2 = DataStream.getDefaultBackingIndexName("boo", 1L);
+        String index1DisplayName = projectId1 + ProjectIndexName.DELIMITER + index1;
+        String index2DisplayName = projectId2 + ProjectIndexName.DELIMITER + index2;
+
+        HealthIndicatorResult result = service.calculate(
+            true,
+            constructHealthInfo(
+                new DataStreamLifecycleHealthInfo(
+                    List.of(new DslErrorInfo(index1, 1L, 100, projectId1), new DslErrorInfo(index2, 3L, 100, projectId2)),
+                    15
+                )
+            )
+        );
+
+        assertThat(result.status(), is(HealthStatus.YELLOW));
+        assertThat(result.symptom(), is("2 backing indices have repeatedly encountered errors whilst trying to advance in its lifecycle"));
+        assertThat(result.details(), is(not(HealthIndicatorDetails.EMPTY)));
+        String detailsAsString = Strings.toString(result.details());
+        assertThat(detailsAsString, containsString("\"total_backing_indices_in_error\":15"));
+        assertThat(detailsAsString, containsString("\"stagnating_backing_indices_count\":2"));
+        assertThat(
+            detailsAsString,
+            containsString(
+                String.format(
+                    Locale.ROOT,
+                    "\"index_name\":\"%s\","
+                        + "\"first_occurrence_timestamp\":1,\"retry_count\":100},{\"index_name\":\"%s\","
+                        + "\"first_occurrence_timestamp\":3,\"retry_count\":100",
+                    index1DisplayName,
+                    index2DisplayName
+                )
+            )
+        );
+        assertThat(result.impacts(), is(STAGNATING_INDEX_IMPACT));
+        Diagnosis diagnosis = result.diagnosisList().get(0);
+        assertThat(diagnosis.definition(), is(STAGNATING_BACKING_INDICES_DIAGNOSIS_DEF));
+        assertThat(diagnosis.affectedResources().get(0).getValues(), containsInAnyOrder(index1DisplayName, index2DisplayName));
+    }
+
     private HealthInfo constructHealthInfo(DataStreamLifecycleHealthInfo dslHealthInfo) {
         return new HealthInfo(Map.of(), dslHealthInfo, Map.of());
     }