Merge branch 'main' into join-limit-pushdown

alex-spies · web-flow · commit d7ca4c5e3efe · 2024-12-11T20:11:03.000+01:00
diff --git a/docs/reference/data-streams/logs.asciidoc b/docs/reference/data-streams/logs.asciidoc
diff --git a/docs/reference/data-streams/tsds.asciidoc b/docs/reference/data-streams/tsds.asciidoc
@@ -17,7 +17,7 @@ metrics data. Only use a TSDS if you typically add metrics data to {es} in near
 real-time and `@timestamp` order.
 
 A TSDS is only intended for metrics data. For other timestamped data, such as
-logs or traces, use a regular data stream.
+logs or traces, use a <<logs-data-stream,logs data stream>> or regular data stream.
 
 [discrete]
 [[differences-from-regular-data-stream]]
diff --git a/docs/reference/images/index-mgmt/management-data-stream-fields.png b/docs/reference/images/index-mgmt/management-data-stream-fields.png
diff --git a/docs/reference/images/index-mgmt/management-data-stream.png b/docs/reference/images/index-mgmt/management-data-stream.png
diff --git a/docs/reference/images/index-mgmt/management-index-templates.png b/docs/reference/images/index-mgmt/management-index-templates.png
diff --git a/docs/reference/index-modules.asciidoc b/docs/reference/index-modules.asciidoc
@@ -113,10 +113,9 @@ Index mode supports the following values:
 
 `standard`::: Standard indexing with default settings.
 
-`time_series`::: Index mode optimized for storage of metrics documented in <<tsds-index-settings,TSDS Settings>>.
+`tsds`::: _(data streams only)_ Index mode optimized for storage of metrics. For more information, see <<tsds-index-settings>>.
 
-`logsdb`::: Index mode optimized for storage of logs. It applies default sort settings on the `hostname` and `timestamp` fields and uses <<synthetic-source,synthetic `_source`>>. <<index-modules-index-sorting,Index sorting>> on different fields is still allowed.
-preview:[]
+`logsdb`::: _(data streams only)_ Index mode optimized for <<logs-data-stream,logs>>.
 
 [[routing-partition-size]] `index.routing_partition_size`::
 
diff --git a/docs/reference/indices/index-mgmt.asciidoc b/docs/reference/indices/index-mgmt.asciidoc
@@ -67,7 +67,7 @@ This value is the time period for which your data is guaranteed to be stored. Da
 Elasticsearch at a later time.
 
 [role="screenshot"]
-image::images/index-mgmt/management-data-stream.png[Data stream details]
+image::images/index-mgmt/management-data-stream-fields.png[Data stream details]
 
 * To view more information about a data stream, such as its generation or its
 current index lifecycle policy, click the stream's name. From this view, you can navigate to *Discover* to
diff --git a/docs/reference/indices/put-index-template.asciidoc b/docs/reference/indices/put-index-template.asciidoc
@@ -115,10 +115,10 @@ See <<create-index-template,create an index template>>.
 
 `index_mode`::
 (Optional, string) Type of data stream to create. Valid values are `null`
-(regular data stream) and `time_series` (<<tsds,time series data stream>>).
+(standard data stream), `time_series` (<<tsds,time series data stream>>) and `logsdb` 
+(<<logs-data-stream,logs data stream>>).
 +
-If `time_series`, each backing index has an `index.mode` index setting of
-`time_series`.
+The template's `index_mode` sets the `index.mode` of the backing index.
 =====
 
 `index_patterns`::
diff --git a/docs/reference/mapping/fields/synthetic-source.asciidoc b/docs/reference/mapping/fields/synthetic-source.asciidoc
@@ -1,17 +1,10 @@
 [[synthetic-source]]
 ==== Synthetic `_source`
 
-IMPORTANT: Synthetic `_source` is Generally Available only for TSDB indices
-(indices that have `index.mode` set to `time_series`). For other indices,
-synthetic `_source` is in technical preview. Features in technical preview may
-be changed or removed in a future release. Elastic will work to fix
-any issues, but features in technical preview are not subject to the support SLA
-of official GA features.
-
 Though very handy to have around, the source field takes up a significant amount
 of space on disk. Instead of storing source documents on disk exactly as you
 send them, Elasticsearch can reconstruct source content on the fly upon retrieval.
-Enable this by using the value `synthetic` for the index setting `index.mapping.source.mode`:
+To enable this https://www.elastic.co/subscriptions[subscription] feature, use the value `synthetic` for the index setting `index.mapping.source.mode`:
 
 [source,console,id=enable-synthetic-source-example]
 ----
@@ -30,7 +23,7 @@ PUT idx
 ----
 // TESTSETUP
 
-While this on the fly reconstruction is *generally* slower than saving the source
+While this on-the-fly reconstruction is _generally_ slower than saving the source
 documents verbatim and loading them at query time, it saves a lot of storage
 space. Additional latency can be avoided by not loading `_source` field in queries when it is not needed.
 
diff --git a/muted-tests.yml b/muted-tests.yml
@@ -171,9 +171,6 @@ tests:
 - class: org.elasticsearch.backwards.MixedClusterClientYamlTestSuiteIT
   method: test {p0=synonyms/90_synonyms_reloading_for_synset/Reload analyzers for specific synonym set}
   issue: https://github.com/elastic/elasticsearch/issues/116777
-- class: org.elasticsearch.reservedstate.service.FileSettingsServiceTests
-  method: testStopWorksInMiddleOfProcessing
-  issue: https://github.com/elastic/elasticsearch/issues/117591
 - class: "org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT"
   method: "test {scoring.*}"
   issue: https://github.com/elastic/elasticsearch/issues/117641
diff --git a/server/src/main/java/org/elasticsearch/reservedstate/service/FileSettingsService.java b/server/src/main/java/org/elasticsearch/reservedstate/service/FileSettingsService.java
@@ -37,8 +37,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 
 import static org.elasticsearch.health.HealthStatus.GREEN;
 import static org.elasticsearch.health.HealthStatus.YELLOW;
@@ -122,6 +120,18 @@ public void handleSnapshotRestore(ClusterState clusterState, Metadata.Builder md
         }
     }
 
+    @Override
+    protected void doStart() {
+        healthIndicatorService.startOccurred();
+        super.doStart();
+    }
+
+    @Override
+    protected void doStop() {
+        super.doStop();
+        healthIndicatorService.stopOccurred();
+    }
+
     /**
      * If the file settings metadata version is set to zero, then we have restored from
      * a snapshot and must reprocess the file.
@@ -211,6 +221,7 @@ protected void processInitialFileMissing() throws ExecutionException, Interrupte
 
     public static class FileSettingsHealthIndicatorService implements HealthIndicatorService {
         static final String NAME = "file_settings";
+        static final String INACTIVE_SYMPTOM = "File-based settings are inactive";
         static final String NO_CHANGES_SYMPTOM = "No file-based setting changes have occurred";
         static final String SUCCESS_SYMPTOM = "The most recent file-based settings were applied successfully";
         static final String FAILURE_SYMPTOM = "The most recent file-based settings encountered an error";
@@ -225,21 +236,33 @@ public static class FileSettingsHealthIndicatorService implements HealthIndicato
             )
         );
 
-        private final AtomicLong changeCount = new AtomicLong(0);
-        private final AtomicLong failureStreak = new AtomicLong(0);
-        private final AtomicReference<String> mostRecentFailure = new AtomicReference<>();
+        private boolean isActive = false;
+        private long changeCount = 0;
+        private long failureStreak = 0;
+        private String mostRecentFailure = null;
 
-        public void changeOccurred() {
-            changeCount.incrementAndGet();
+        public synchronized void startOccurred() {
+            isActive = true;
+            failureStreak = 0;
         }
 
-        public void successOccurred() {
-            failureStreak.set(0);
+        public synchronized void stopOccurred() {
+            isActive = false;
+            mostRecentFailure = null;
         }
 
-        public void failureOccurred(String description) {
-            failureStreak.incrementAndGet();
-            mostRecentFailure.set(description);
+        public synchronized void changeOccurred() {
+            ++changeCount;
+        }
+
+        public synchronized void successOccurred() {
+            failureStreak = 0;
+            mostRecentFailure = null;
+        }
+
+        public synchronized void failureOccurred(String description) {
+            ++failureStreak;
+            mostRecentFailure = description;
         }
 
         @Override
@@ -248,18 +271,20 @@ public String name() {
         }
 
         @Override
-        public HealthIndicatorResult calculate(boolean verbose, int maxAffectedResourcesCount, HealthInfo healthInfo) {
-            if (0 == changeCount.get()) {
+        public synchronized HealthIndicatorResult calculate(boolean verbose, int maxAffectedResourcesCount, HealthInfo healthInfo) {
+            if (isActive == false) {
+                return createIndicator(GREEN, INACTIVE_SYMPTOM, HealthIndicatorDetails.EMPTY, List.of(), List.of());
+            }
+            if (0 == changeCount) {
                 return createIndicator(GREEN, NO_CHANGES_SYMPTOM, HealthIndicatorDetails.EMPTY, List.of(), List.of());
             }
-            long numFailures = failureStreak.get();
-            if (0 == numFailures) {
+            if (0 == failureStreak) {
                 return createIndicator(GREEN, SUCCESS_SYMPTOM, HealthIndicatorDetails.EMPTY, List.of(), List.of());
             } else {
                 return createIndicator(
                     YELLOW,
                     FAILURE_SYMPTOM,
-                    new SimpleHealthIndicatorDetails(Map.of("failure_streak", numFailures, "most_recent_failure", mostRecentFailure.get())),
+                    new SimpleHealthIndicatorDetails(Map.of("failure_streak", failureStreak, "most_recent_failure", mostRecentFailure)),
                     STALE_SETTINGS_IMPACT,
                     List.of()
                 );
diff --git a/server/src/test/java/org/elasticsearch/reservedstate/service/FileSettingsHealthIndicatorServiceTests.java b/server/src/test/java/org/elasticsearch/reservedstate/service/FileSettingsHealthIndicatorServiceTests.java
@@ -22,6 +22,7 @@
 import static org.elasticsearch.health.HealthStatus.GREEN;
 import static org.elasticsearch.health.HealthStatus.YELLOW;
 import static org.elasticsearch.reservedstate.service.FileSettingsService.FileSettingsHealthIndicatorService.FAILURE_SYMPTOM;
+import static org.elasticsearch.reservedstate.service.FileSettingsService.FileSettingsHealthIndicatorService.INACTIVE_SYMPTOM;
 import static org.elasticsearch.reservedstate.service.FileSettingsService.FileSettingsHealthIndicatorService.NO_CHANGES_SYMPTOM;
 import static org.elasticsearch.reservedstate.service.FileSettingsService.FileSettingsHealthIndicatorService.STALE_SETTINGS_IMPACT;
 import static org.elasticsearch.reservedstate.service.FileSettingsService.FileSettingsHealthIndicatorService.SUCCESS_SYMPTOM;
@@ -39,14 +40,27 @@ public void initialize() {
         healthIndicatorService = new FileSettingsHealthIndicatorService();
     }
 
-    public void testInitiallyGreen() {
+    public void testInitiallyGreen() {}
+
+    public void testStartAndStop() {
+        assertEquals(
+            new HealthIndicatorResult("file_settings", GREEN, INACTIVE_SYMPTOM, HealthIndicatorDetails.EMPTY, List.of(), List.of()),
+            healthIndicatorService.calculate(false, null)
+        );
+        healthIndicatorService.startOccurred();
         assertEquals(
             new HealthIndicatorResult("file_settings", GREEN, NO_CHANGES_SYMPTOM, HealthIndicatorDetails.EMPTY, List.of(), List.of()),
             healthIndicatorService.calculate(false, null)
         );
+        healthIndicatorService.stopOccurred();
+        assertEquals(
+            new HealthIndicatorResult("file_settings", GREEN, INACTIVE_SYMPTOM, HealthIndicatorDetails.EMPTY, List.of(), List.of()),
+            healthIndicatorService.calculate(false, null)
+        );
     }
 
     public void testGreenYellowYellowGreen() {
+        healthIndicatorService.startOccurred();
         healthIndicatorService.changeOccurred();
         // This is a strange case: a change occurred, but neither success nor failure have been reported yet.
         // While the change is still in progress, we don't change the status.
diff --git a/server/src/test/java/org/elasticsearch/reservedstate/service/FileSettingsServiceTests.java b/server/src/test/java/org/elasticsearch/reservedstate/service/FileSettingsServiceTests.java
@@ -17,7 +17,6 @@
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.NodeConnectionsService;
-import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.ReservedStateMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -69,6 +68,8 @@
 
 import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
 import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.elasticsearch.health.HealthStatus.GREEN;
+import static org.elasticsearch.health.HealthStatus.YELLOW;
 import static org.elasticsearch.node.Node.NODE_NAME_SETTING;
 import static org.hamcrest.Matchers.anEmptyMap;
 import static org.hamcrest.Matchers.hasEntry;
@@ -82,8 +83,6 @@
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoInteractions;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 public class FileSettingsServiceTests extends ESTestCase {
     private static final Logger logger = LogManager.getLogger(FileSettingsServiceTests.class);
@@ -138,7 +137,7 @@ public void setUp() throws Exception {
                 List.of(new ReservedClusterSettingsAction(clusterSettings))
             )
         );
-        healthIndicatorService = mock(FileSettingsHealthIndicatorService.class);
+        healthIndicatorService = spy(new FileSettingsHealthIndicatorService());
         fileSettingsService = spy(new FileSettingsService(clusterService, controller, env, healthIndicatorService));
     }
 
@@ -170,7 +169,8 @@ public void testStartStop() {
         assertTrue(fileSettingsService.watching());
         fileSettingsService.stop();
         assertFalse(fileSettingsService.watching());
-        verifyNoInteractions(healthIndicatorService);
+        verify(healthIndicatorService, times(1)).startOccurred();
+        verify(healthIndicatorService, times(1)).stopOccurred();
     }
 
     public void testOperatorDirName() {
@@ -218,9 +218,9 @@ public void testInitialFileError() throws Exception {
         // assert we never notified any listeners of successful application of file based settings
         assertFalse(settingsChanged.get());
 
+        assertEquals(YELLOW, healthIndicatorService.calculate(false, null).status());
         verify(healthIndicatorService, times(1)).changeOccurred();
         verify(healthIndicatorService, times(1)).failureOccurred(argThat(s -> s.startsWith(IllegalStateException.class.getName())));
-        verifyNoMoreInteractions(healthIndicatorService);
     }
 
     @SuppressWarnings("unchecked")
@@ -246,9 +246,9 @@ public void testInitialFileWorks() throws Exception {
         verify(fileSettingsService, times(1)).processFileOnServiceStart();
         verify(controller, times(1)).process(any(), any(XContentParser.class), eq(ReservedStateVersionCheck.HIGHER_OR_SAME_VERSION), any());
 
+        assertEquals(GREEN, healthIndicatorService.calculate(false, null).status());
         verify(healthIndicatorService, times(1)).changeOccurred();
         verify(healthIndicatorService, times(1)).successOccurred();
-        verifyNoMoreInteractions(healthIndicatorService);
     }
 
     @SuppressWarnings("unchecked")
@@ -285,9 +285,9 @@ public void testProcessFileChanges() throws Exception {
         verify(fileSettingsService, times(1)).processFileChanges();
         verify(controller, times(1)).process(any(), any(XContentParser.class), eq(ReservedStateVersionCheck.HIGHER_VERSION_ONLY), any());
 
+        assertEquals(GREEN, healthIndicatorService.calculate(false, null).status());
         verify(healthIndicatorService, times(2)).changeOccurred();
         verify(healthIndicatorService, times(2)).successOccurred();
-        verifyNoMoreInteractions(healthIndicatorService);
     }
 
     public void testInvalidJSON() throws Exception {
@@ -323,6 +323,7 @@ public void testInvalidJSON() throws Exception {
         // referring to fileSettingsService.start(). Rather, it is referring to the initialization
         // of the watcher thread itself, which occurs asynchronously when clusterChanged is first called.
 
+        assertEquals(YELLOW, healthIndicatorService.calculate(false, null).status());
         verify(healthIndicatorService).failureOccurred(contains(XContentParseException.class.getName()));
     }
 
@@ -388,14 +389,13 @@ public void testStopWorksInMiddleOfProcessing() throws Exception {
         fileSettingsService.stop();
         assertFalse(fileSettingsService.watching());
         fileSettingsService.close();
+
+        // When the service is stopped, the health indicator should be green
+        assertEquals(GREEN, healthIndicatorService.calculate(false, null).status());
+        verify(healthIndicatorService).stopOccurred();
+
         // let the deadlocked thread end, so we can cleanly exit the test
         deadThreadLatch.countDown();
-
-        verify(healthIndicatorService, times(1)).changeOccurred();
-        verify(healthIndicatorService, times(1)).failureOccurred(
-            argThat(s -> s.startsWith(FailedToCommitClusterStateException.class.getName()))
-        );
-        verifyNoMoreInteractions(healthIndicatorService);
     }
 
     public void testHandleSnapshotRestoreClearsMetadata() throws Exception {
