Merge branch 'main' into bootstrap-entitlements-for-testing

Author: ldematte

muted-tests.yml

@@ -568,6 +568,15 @@ tests:
 - class: org.elasticsearch.search.query.RescoreKnnVectorQueryIT
   method: testKnnSearchRescore
   issue: https://github.com/elastic/elasticsearch/issues/129713
+- class: org.elasticsearch.streams.StreamsYamlTestSuiteIT
+  method: test {yaml=streams/logs/10_basic/Basic toggle of logs state enable to disable and back}
+  issue: https://github.com/elastic/elasticsearch/issues/129733
+- class: org.elasticsearch.streams.StreamsYamlTestSuiteIT
+  method: test {yaml=streams/logs/10_basic/Check for repeated toggle to same state}
+  issue: https://github.com/elastic/elasticsearch/issues/129735
+- class: org.elasticsearch.snapshots.GetSnapshotsIT
+  method: testFilterByState
+  issue: https://github.com/elastic/elasticsearch/issues/129740
 
 # Examples:
 #

qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/TsdbIndexingRollingUpgradeIT.java

@@ -12,19 +12,24 @@
 import com.carrotsearch.randomizedtesting.annotations.Name;
 
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.network.NetworkAddress;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.test.rest.ObjectPath;
+import org.elasticsearch.xcontent.XContentType;
 
+import java.io.IOException;
+import java.io.InputStream;
 import java.time.Instant;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 
-import static org.elasticsearch.upgrades.LogsIndexModeRollingUpgradeIT.getWriteBackingIndex;
-import static org.elasticsearch.upgrades.LogsdbIndexingRollingUpgradeIT.createTemplate;
-import static org.elasticsearch.upgrades.LogsdbIndexingRollingUpgradeIT.getIndexSettingsWithDefaults;
-import static org.elasticsearch.upgrades.LogsdbIndexingRollingUpgradeIT.startTrial;
 import static org.elasticsearch.upgrades.TsdbIT.TEMPLATE;
 import static org.elasticsearch.upgrades.TsdbIT.formatInstant;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -194,4 +199,51 @@ void query(String dataStreamName) throws Exception {
         assertThat(maxTx, notNullValue());
     }
 
+    protected static void startTrial() throws IOException {
+        Request startTrial = new Request("POST", "/_license/start_trial");
+        startTrial.addParameter("acknowledge", "true");
+        try {
+            assertOK(client().performRequest(startTrial));
+        } catch (ResponseException e) {
+            var responseBody = entityAsMap(e.getResponse());
+            String error = ObjectPath.evaluate(responseBody, "error_message");
+            assertThat(error, containsString("Trial was already activated."));
+        }
+    }
+
+    static Map<String, Object> getIndexSettingsWithDefaults(String index) throws IOException {
+        Request request = new Request("GET", "/" + index + "/_settings");
+        request.addParameter("flat_settings", "true");
+        request.addParameter("include_defaults", "true");
+        Response response = client().performRequest(request);
+        try (InputStream is = response.getEntity().getContent()) {
+            return XContentHelper.convertToMap(
+                XContentType.fromMediaType(response.getEntity().getContentType().getValue()).xContent(),
+                is,
+                true
+            );
+        }
+    }
+
+    static void createTemplate(String dataStreamName, String id, String template) throws IOException {
+        final String INDEX_TEMPLATE = """
+            {
+                "index_patterns": ["$DATASTREAM"],
+                "template": $TEMPLATE,
+                "data_stream": {
+                }
+            }""";
+        var putIndexTemplateRequest = new Request("POST", "/_index_template/" + id);
+        putIndexTemplateRequest.setJsonEntity(INDEX_TEMPLATE.replace("$TEMPLATE", template).replace("$DATASTREAM", dataStreamName));
+        assertOK(client().performRequest(putIndexTemplateRequest));
+    }
+
+    @SuppressWarnings("unchecked")
+    static String getWriteBackingIndex(final RestClient client, final String dataStreamName, int backingIndex) throws IOException {
+        final Request request = new Request("GET", "_data_stream/" + dataStreamName);
+        final List<Object> dataStreams = (List<Object>) entityAsMap(client.performRequest(request)).get("data_streams");
+        final Map<String, Object> dataStream = (Map<String, Object>) dataStreams.get(0);
+        final List<Map<String, String>> backingIndices = (List<Map<String, String>>) dataStream.get("indices");
+        return backingIndices.get(backingIndex).get("index_name");
+    }
 }

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -244,7 +244,6 @@ static TransportVersion def(int id) {
     public static final TransportVersion RESCORE_VECTOR_ALLOW_ZERO = def(9_039_0_00);
     public static final TransportVersion PROJECT_ID_IN_SNAPSHOT = def(9_040_0_00);
     public static final TransportVersion INDEX_STATS_AND_METADATA_INCLUDE_PEAK_WRITE_LOAD = def(9_041_0_00);
-    public static final TransportVersion REPOSITORIES_METADATA_AS_PROJECT_CUSTOM = def(9_042_0_00);
     public static final TransportVersion BATCHED_QUERY_PHASE_VERSION = def(9_043_0_00);
     public static final TransportVersion REMOTE_EXCEPTION = def(9_044_0_00);
     public static final TransportVersion ESQL_REMOVE_AGGREGATE_TYPE = def(9_045_0_00);

server/src/main/java/org/elasticsearch/action/bulk/IncrementalBulkService.java

@@ -22,6 +22,8 @@
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexingPressure;
+import org.elasticsearch.telemetry.metric.LongHistogram;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -33,6 +35,7 @@
 import static org.elasticsearch.common.settings.Setting.boolSetting;
 
 public class IncrementalBulkService {
+    public static final String CHUNK_WAIT_TIME_HISTOGRAM_NAME = "es.rest.incremental_bulk.wait_for_next_chunk.duration.histogram";
 
     public static final Setting<Boolean> INCREMENTAL_BULK = boolSetting(
         "rest.incremental_bulk",
@@ -44,9 +47,17 @@ public class IncrementalBulkService {
     private final AtomicBoolean enabledForTests = new AtomicBoolean(true);
     private final IndexingPressure indexingPressure;
 
-    public IncrementalBulkService(Client client, IndexingPressure indexingPressure) {
+    /* Capture in milliseconds because the APM histogram only has a range of 100,000 */
+    private final LongHistogram chunkWaitTimeMillisHistogram;
+
+    public IncrementalBulkService(Client client, IndexingPressure indexingPressure, MeterRegistry meterRegistry) {
         this.client = client;
         this.indexingPressure = indexingPressure;
+        this.chunkWaitTimeMillisHistogram = meterRegistry.registerLongHistogram(
+            CHUNK_WAIT_TIME_HISTOGRAM_NAME,
+            "Total time in millis spent waiting for next chunk of a bulk request",
+            "ms"
+        );
     }
 
     public Handler newBulkRequest() {
@@ -56,7 +67,7 @@ public Handler newBulkRequest() {
 
     public Handler newBulkRequest(@Nullable String waitForActiveShards, @Nullable TimeValue timeout, @Nullable String refresh) {
         ensureEnabled();
-        return new Handler(client, indexingPressure, waitForActiveShards, timeout, refresh);
+        return new Handler(client, indexingPressure, waitForActiveShards, timeout, refresh, chunkWaitTimeMillisHistogram);
     }
 
     private void ensureEnabled() {
@@ -99,6 +110,8 @@ public static class Handler implements Releasable {
         private final ArrayList<Releasable> releasables = new ArrayList<>(4);
         private final ArrayList<BulkResponse> responses = new ArrayList<>(2);
         private final IndexingPressure.Incremental incrementalOperation;
+        // Ideally this should be in RestBulkAction, but it's harder to inject the metric registry there
+        private final LongHistogram chunkWaitTimeMillisHistogram;
         private boolean closed = false;
         private boolean globalFailure = false;
         private boolean incrementalRequestSubmitted = false;
@@ -111,20 +124,26 @@ protected Handler(
             IndexingPressure indexingPressure,
             @Nullable String waitForActiveShards,
             @Nullable TimeValue timeout,
-            @Nullable String refresh
+            @Nullable String refresh,
+            LongHistogram chunkWaitTimeMillisHistogram
         ) {
             this.client = client;
             this.waitForActiveShards = waitForActiveShards != null ? ActiveShardCount.parseString(waitForActiveShards) : null;
             this.timeout = timeout;
             this.refresh = refresh;
             this.incrementalOperation = indexingPressure.startIncrementalCoordinating(0, 0, false);
+            this.chunkWaitTimeMillisHistogram = chunkWaitTimeMillisHistogram;
             createNewBulkRequest(EMPTY_STATE);
         }
 
         public IndexingPressure.Incremental getIncrementalOperation() {
             return incrementalOperation;
         }
 
+        public void updateWaitForChunkMetrics(long chunkWaitTimeInMillis) {
+            chunkWaitTimeMillisHistogram.record(chunkWaitTimeInMillis);
+        }
+
         public void addItems(List<DocWriteRequest<?>> items, Releasable releasable, Runnable nextItems) {
             assert closed == false;
             assert bulkInProgress == false;