Merge branch 'main' into consider-aborting-merges-while-enqueued

Author: albertzaharovits

distribution/tools/server-cli/src/test/java/org/elasticsearch/server/cli/MachineDependentHeapTests.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.server.cli;
 
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.test.ESTestCase;
 import org.hamcrest.Matcher;
 
@@ -55,7 +56,8 @@ public void testMasterOnlyOptions() throws Exception {
         assertHeapOptions(64, containsInAnyOrder("-Xmx31744m", "-Xms31744m"), "master");
     }
 
-    public void testMlOnlyOptions() throws Exception {
+    public void testMlOnlyOptions_new() throws Exception {
+        assumeTrue("feature flag must be enabled for new memory computation", new FeatureFlag("new_ml_memory_computation").isEnabled());
         assertHeapOptions(1, containsInAnyOrder("-Xmx272m", "-Xms272m"), "ml");
         assertHeapOptions(4, containsInAnyOrder("-Xmx1092m", "-Xms1092m"), "ml");
         assertHeapOptions(32, containsInAnyOrder("-Xmx5460m", "-Xms5460m"), "ml");
@@ -65,6 +67,20 @@ public void testMlOnlyOptions() throws Exception {
         assertHeapOptions(263, containsInAnyOrder("-Xmx21228m", "-Xms21228m"), "ml");
     }
 
+    public void testMlOnlyOptions_old() throws Exception {
+        assumeTrue(
+            "feature flag must be disabled for old memory computation",
+            new FeatureFlag("new_ml_memory_computation").isEnabled() == false
+        );
+        assertHeapOptions(1, containsInAnyOrder("-Xmx408m", "-Xms408m"), "ml");
+        assertHeapOptions(4, containsInAnyOrder("-Xmx1636m", "-Xms1636m"), "ml");
+        assertHeapOptions(32, containsInAnyOrder("-Xmx8192m", "-Xms8192m"), "ml");
+        assertHeapOptions(64, containsInAnyOrder("-Xmx11468m", "-Xms11468m"), "ml");
+        // We'd never see a node this big in Cloud, but this assertion proves that the 31GB absolute maximum
+        // eventually kicks in (because 0.4 * 16 + 0.1 * (263 - 16) > 31)
+        assertHeapOptions(263, containsInAnyOrder("-Xmx31744m", "-Xms31744m"), "ml");
+    }
+
     public void testDataNodeOptions() throws Exception {
         assertHeapOptions(1, containsInAnyOrder("-Xmx512m", "-Xms512m"), "data");
         assertHeapOptions(8, containsInAnyOrder("-Xmx4096m", "-Xms4096m"), "data");

docs/changelog/129606.yaml

@@ -0,0 +1,35 @@
+pr: 129606
+summary: Release FORK in tech preview
+area: ES|QL
+type: feature
+issues: []
+highlight:
+  title: Release FORK in tech preview
+  body: |-
+    Fork is a foundational building block that allows multiple branches of execution.
+    Conceptually, fork is:
+    - a bifurcation of the stream, with all data going to each fork branch, followed by
+    - a merge of the branches, enhanced with a discriminator column called FORK:
+
+    Example:
+
+    [source,yaml]
+    ----------------------------
+    FROM test
+    | FORK
+    ( WHERE content:"fox" )
+    ( WHERE content:"dog" )
+    | SORT _fork
+    ----------------------------
+
+    The FORK command add a discriminator column called `_fork`:
+
+    [source,yaml]
+    ----------------------------
+    | id  | content   | _fork |
+    |-----|-----------|-------|
+    | 3   | brown fox | fork1 |
+    | 4   | white dog | fork2 |
+    ----------------------------
+
+  notable: true

docs/reference/query-languages/esql/_snippets/functions/layout/categorize.md

@@ -1,2 +1,2 @@
 * [`BUCKET`](../../functions-operators/grouping-functions.md#esql-bucket)
-* [preview] [`CATEGORIZE`](../../functions-operators/grouping-functions.md#esql-categorize)
+* [`CATEGORIZE`](../../functions-operators/grouping-functions.md#esql-categorize)

docs/reference/query-languages/esql/_snippets/lists/grouping-functions.md

@@ -17,6 +17,7 @@
 import org.elasticsearch.ingest.geoip.stats.CacheStats;
 import org.elasticsearch.test.ESTestCase;
 
+import java.nio.file.Path;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
@@ -140,23 +141,25 @@ public void testPurgeCacheEntriesForDatabase() {
         GeoIpCache cache = new GeoIpCache(100);
         ProjectId projectId1 = randomUniqueProjectId();
         ProjectId projectId2 = randomUniqueProjectId();
-        String databasePath1 = "path/to/db1";
-        String databasePath2 = "path/to/db2";
+        // Turn the path strings into Paths to ensure that we always use the canonical string representation (this string literal does not
+        // round-trip when converting to a Path and back again on Windows):
+        Path databasePath1 = PathUtils.get("path/to/db1");
+        Path databasePath2 = PathUtils.get("path/to/db2");
         String ip1 = "127.0.0.1";
         String ip2 = "127.0.0.2";
 
         AbstractResponse response = mock(AbstractResponse.class);
-        cache.putIfAbsent(projectId1, ip1, databasePath1, ip -> response); // cache miss
-        cache.putIfAbsent(projectId1, ip2, databasePath1, ip -> response); // cache miss
-        cache.putIfAbsent(projectId2, ip1, databasePath1, ip -> response); // cache miss
-        cache.putIfAbsent(projectId1, ip1, databasePath2, ip -> response); // cache miss
-        cache.purgeCacheEntriesForDatabase(projectId1, PathUtils.get(databasePath1));
+        cache.putIfAbsent(projectId1, ip1, databasePath1.toString(), ip -> response); // cache miss
+        cache.putIfAbsent(projectId1, ip2, databasePath1.toString(), ip -> response); // cache miss
+        cache.putIfAbsent(projectId2, ip1, databasePath1.toString(), ip -> response); // cache miss
+        cache.putIfAbsent(projectId1, ip1, databasePath2.toString(), ip -> response); // cache miss
+        cache.purgeCacheEntriesForDatabase(projectId1, databasePath1);
         // should have purged entries for projectId1 and databasePath1...
-        assertNull(cache.get(projectId1, ip1, databasePath1));
-        assertNull(cache.get(projectId1, ip2, databasePath1));
+        assertNull(cache.get(projectId1, ip1, databasePath1.toString()));
+        assertNull(cache.get(projectId1, ip2, databasePath1.toString()));
         // ...but left the one for projectId2...
-        assertSame(response, cache.get(projectId2, ip1, databasePath1));
+        assertSame(response, cache.get(projectId2, ip1, databasePath1.toString()));
         // ...and for databasePath2:
-        assertSame(response, cache.get(projectId1, ip1, databasePath2));
+        assertSame(response, cache.get(projectId1, ip1, databasePath2.toString()));
     }
 }

docs/reference/query-languages/esql/kibana/definition/functions/categorize.json

@@ -0,0 +1,41 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+apply plugin: 'elasticsearch.test-with-dependencies'
+apply plugin: 'elasticsearch.internal-cluster-test'
+apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.yaml-rest-compat-test'
+
+esplugin {
+  description = 'The module adds support for the wired streams functionality including logs ingest'
+  classname = 'org.elasticsearch.rest.streams.StreamsPlugin'
+}
+
+restResources {
+  restApi {
+    include '_common', 'streams'
+  }
+}
+
+configurations {
+  basicRestSpecs {
+    attributes {
+      attribute(ArtifactTypeDefinition.ARTIFACT_TYPE_ATTRIBUTE, ArtifactTypeDefinition.DIRECTORY_TYPE)
+    }
+  }
+}
+
+artifacts {
+  basicRestSpecs(new File(projectDir, "src/yamlRestTest/resources/rest-api-spec/test"))
+}
+
+dependencies {
+  testImplementation project(path: ':test:test-clusters')
+}

modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpCacheTests.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.rest.streams;
+
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.cluster.metadata.StreamsMetadata;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.streams.logs.LogsStreamsActivationToggleAction;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.hamcrest.ElasticsearchAssertions;
+import org.elasticsearch.test.transport.MockTransportService;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.Matchers.is;
+
+public class TestToggleIT extends ESIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return List.of(StreamsPlugin.class, MockTransportService.TestPlugin.class);
+    }
+
+    public void testLogStreamToggle() throws IOException, ExecutionException, InterruptedException {
+        boolean[] testParams = new boolean[] { true, false, true };
+        for (boolean enable : testParams) {
+            doLogStreamToggleTest(enable);
+        }
+    }
+
+    private void doLogStreamToggleTest(boolean enable) throws IOException, ExecutionException, InterruptedException {
+        LogsStreamsActivationToggleAction.Request request = new LogsStreamsActivationToggleAction.Request(
+            TEST_REQUEST_TIMEOUT,
+            TEST_REQUEST_TIMEOUT,
+            enable
+        );
+
+        AcknowledgedResponse acknowledgedResponse = client().execute(LogsStreamsActivationToggleAction.INSTANCE, request).get();
+        ElasticsearchAssertions.assertAcked(acknowledgedResponse);
+
+        ClusterStateRequest state = new ClusterStateRequest(TEST_REQUEST_TIMEOUT);
+        ClusterStateResponse clusterStateResponse = client().admin().cluster().state(state).get();
+        ProjectMetadata projectMetadata = clusterStateResponse.getState().metadata().getProject(ProjectId.DEFAULT);
+
+        assertThat(projectMetadata.<StreamsMetadata>custom(StreamsMetadata.TYPE).isLogsEnabled(), is(enable));
+    }
+
+}

modules/streams/build.gradle

@@ -0,0 +1,19 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+module org.elasticsearch.rest.root {
+    requires org.elasticsearch.server;
+    requires org.elasticsearch.xcontent;
+    requires org.apache.lucene.core;
+    requires org.elasticsearch.base;
+    requires org.apache.logging.log4j;
+
+    exports org.elasticsearch.rest.streams;
+    exports org.elasticsearch.rest.streams.logs;
+}

modules/streams/src/internalClusterTest/java/org/elasticsearch/rest/streams/TestToggleIT.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.rest.streams;
+
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.IndexScopedSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.features.NodeFeature;
+import org.elasticsearch.plugins.ActionPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.streams.logs.LogsStreamsActivationToggleAction;
+import org.elasticsearch.rest.streams.logs.RestSetLogStreamsEnabledAction;
+import org.elasticsearch.rest.streams.logs.RestStreamsStatusAction;
+import org.elasticsearch.rest.streams.logs.StreamsStatusAction;
+import org.elasticsearch.rest.streams.logs.TransportLogsStreamsToggleActivation;
+import org.elasticsearch.rest.streams.logs.TransportStreamsStatusAction;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+/**
+ * This plugin provides the Streams feature which builds upon data streams to
+ * provide the user with a more "batteries included" experience for ingesting large
+ * streams of data, such as logs.
+ */
+public class StreamsPlugin extends Plugin implements ActionPlugin {
+
+    @Override
+    public List<RestHandler> getRestHandlers(
+        Settings settings,
+        NamedWriteableRegistry namedWriteableRegistry,
+        RestController restController,
+        ClusterSettings clusterSettings,
+        IndexScopedSettings indexScopedSettings,
+        SettingsFilter settingsFilter,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<DiscoveryNodes> nodesInCluster,
+        Predicate<NodeFeature> clusterSupportsFeature
+    ) {
+        if (DataStream.LOGS_STREAM_FEATURE_FLAG) {
+            return List.of(new RestSetLogStreamsEnabledAction(), new RestStreamsStatusAction());
+        }
+        return Collections.emptyList();
+    }
+
+    @Override
+    public List<ActionHandler> getActions() {
+        return List.of(
+            new ActionHandler(LogsStreamsActivationToggleAction.INSTANCE, TransportLogsStreamsToggleActivation.class),
+            new ActionHandler(StreamsStatusAction.INSTANCE, TransportStreamsStatusAction.class)
+        );
+    }
+}