Merge branch 'main' into agg_values_spatial

Author: craigtaverner

.buildkite/scripts/dra-update-staging.sh

@@ -37,7 +37,7 @@ for BRANCH in "${BRANCHES[@]}"; do
 
   if [[ "$SHOULD_TRIGGER" == "true" ]]; then
     if [[ "$BRANCH" == "9.0" ]]; then
-      export VERSION_QUALIFIER="beta1"
+      export VERSION_QUALIFIER="rc1"
     fi
     echo "Triggering DRA staging workflow for $BRANCH"
     cat << EOF | buildkite-agent pipeline upload

.buildkite/scripts/dra-workflow.trigger.sh

@@ -8,7 +8,7 @@ source .buildkite/scripts/branches.sh
 
 for BRANCH in "${BRANCHES[@]}"; do
   if [[ "$BRANCH" == "9.0" ]]; then
-    export VERSION_QUALIFIER="beta1"
+    export VERSION_QUALIFIER="rc1"
   fi
 
   INTAKE_PIPELINE_SLUG="elasticsearch-intake"

docs/changelog/122409.yaml

@@ -0,0 +1,6 @@
+pr: 122409
+summary: Allow setting the `type` in the reroute processor
+area: Ingest Node
+type: enhancement
+issues:
+ - 121553

docs/changelog/122857.yaml

@@ -0,0 +1,5 @@
+pr: 122857
+summary: Run `TransportGetWatcherSettingsAction` on local node
+area: Watcher
+type: enhancement
+issues: []

docs/internal/DistributedArchitectureGuide.md

@@ -229,19 +229,45 @@ works in parallel with the storage engine.)
 
 # Allocation
 
-(AllocationService runs on the master node)
-
-(Discuss different deciders that limit allocation. Sketch / list the different deciders that we have.)
-
-### APIs for Balancing Operations
-
-(Significant internal APIs for balancing a cluster)
-
-### Heuristics for Allocation
-
-### Cluster Reroute Command
-
-(How does this command behave with the desired auto balancer.)
+### Core Components
+
+The `DesiredBalanceShardsAllocator` is what runs shard allocation decisions. It leverages the `DesiredBalanceComputer` to produce
+`DesiredBalance` instances for the cluster based on the latest cluster changes (add/remove nodes, create/remove indices, load, etc.). Then
+the `DesiredBalanceReconciler` is invoked to choose the next steps to take to move the cluster from the current shard allocation to the
+latest computed `DesiredBalance` shard allocation. The `DesiredBalanceReconciler` will apply changes to a copy of the `RoutingNodes`, which
+is then published in a cluster state update that will reach the data nodes to start the individual shard recovery/deletion/move work.
+
+The `DesiredBalanceReconciler` is throttled by cluster settings, like the max number of concurrent shard moves and recoveries per cluster
+and node: this is why the `DesiredBalanceReconciler` will make, and publish via cluster state updates, incremental changes to the cluster
+shard allocation. The `DesiredBalanceShardsAllocator` is the endpoint for reroute requests, which may trigger immediate requests to the
+`DesiredBalanceReconciler`, but asynchronous requests to the `DesiredBalanceComputer` via the `ContinuousComputation` component. Cluster
+state changes that affect shard balancing (for example index deletion) all call some reroute method interface that reaches the
+`DesiredBalanceShardsAllocator` to run reconciliation and queue a request for the `DesiredBalancerComputer`, leading to desired balance
+computation and reconciliation actions. Asynchronous completion of a new `DesiredBalance` will also invoke a reconciliation action, as will
+cluster state updates completing shard moves/recoveries (unthrottling the next shard move/recovery).
+
+The `ContinuousComputation` saves the latest desired balance computation request, which holds the cluster information at the time of that
+request, and a thread that runs the `DesiredBalanceComputer`. The `ContinuousComputation` thread takes the latest request, with the
+associated cluster information, feeds it into the `DesiredBalanceComputer` and publishes a `DesiredBalance` back to the
+`DesiredBalanceShardsAllocator` to use for reconciliation actions. Sometimes the `ContinuousComputation` thread's desired balance
+computation will be signalled to exit early and publish the initial `DesiredBalance` improvements it has made, when newer rebalancing
+requests (due to cluster state changes) have arrived, or in order to begin recovery of unassigned shards as quickly as possible.
+
+### Rebalancing Process
+
+There are different priorities in shard allocation, reflected in which moves the `DesiredBalancerReconciler` selects to do first given that
+it can only move, recover, or remove a limited number of shards at once. The first priority is assigning unassigned shards, primaries being
+more important than replicas. The second is to move shards that violate any rule (such as node resource limits) as defined by an
+`AllocationDecider`. The `AllocationDeciders` holds a group of `AllocationDecider` implementations that place hard constraints on shard
+allocation. There is a decider, `DiskThresholdDecider`, that manages disk memory usage thresholds, such that further shards may not be
+allowed assignment to a node, or shards may be required to move off because they grew to exceed the disk space; or another,
+`FilterAllocationDecider`, that excludes a configurable list of indices from certain nodes; or `MaxRetryAllocationDecider` that will not
+attempt to recover a shard on a certain node after so many failed retries. The third priority is to rebalance shards to even out the
+relative weight of shards on each node: the intention is to avoid, or ease, future hot-spotting on data nodes due to too many shards being
+placed on the same data node. Node shard weight is based on a sum of factors: disk memory usage, projected shard write load, total number
+of shards, and an incentive to distribute shards within the same index across different nodes. See the `WeightFunction` and
+`NodeAllocationStatsAndWeightsCalculator` classes for more details on the weight calculations that support the `DesiredBalanceComputer`
+decisions.
 
 # Autoscaling
 

docs/reference/ingest/processors/reroute.asciidoc

@@ -45,6 +45,9 @@ Otherwise, the document will be rejected with a security exception which looks l
 |======
 | Name          | Required  | Default                      | Description
 | `destination` | no        | -                            | A static value for the target. Can't be set when the `dataset` or `namespace` option is set.
+| `type`        | no        | `{{data_stream.type}}`   a| Field references or a static value for the type part of the data stream name. In addition to the criteria for <<indices-create-api-path-params, index names>>, cannot contain `-` and must be no longer than 100 characters. Example values are `logs` and `metrics`.
+
+Supports field references with a mustache-like syntax (denoted as `{{double}}` or `{{{triple}}}` curly braces). When resolving field references, the processor replaces invalid characters with `_`. Uses the `<type>` part of the index name as a fallback if all field references resolve to a `null`, missing, or non-string value.
 | `dataset`     | no        | `{{data_stream.dataset}}`   a| Field references or a static value for the dataset part of the data stream name. In addition to the criteria for <<indices-create-api-path-params, index names>>, cannot contain `-` and must be no longer than 100 characters. Example values are `nginx.access` and `nginx.error`.
 
 Supports field references with a mustache-like syntax (denoted as `{{double}}` or `{{{triple}}}` curly braces). When resolving field references, the processor replaces invalid characters with `_`. Uses the `<dataset>` part of the index name as a fallback if all field references resolve to a `null`, missing, or non-string value.

libs/entitlement/src/main/java/org/elasticsearch/entitlement/bootstrap/EntitlementBootstrap.java

@@ -28,6 +28,7 @@
 import java.nio.file.Path;
 import java.util.Map;
 import java.util.function.Function;
+import java.util.stream.Stream;
 
 import static java.util.Objects.requireNonNull;
 
@@ -36,19 +37,24 @@ public class EntitlementBootstrap {
     public record BootstrapArgs(
         Map<String, Policy> pluginPolicies,
         Function<Class<?>, String> pluginResolver,
+        Function<String, String> settingResolver,
+        Function<String, Stream<String>> settingGlobResolver,
         Path[] dataDirs,
         Path configDir,
-        Path tempDir,
-        Path logsDir
+        Path logsDir,
+        Path tempDir
     ) {
         public BootstrapArgs {
             requireNonNull(pluginPolicies);
             requireNonNull(pluginResolver);
+            requireNonNull(settingResolver);
+            requireNonNull(settingGlobResolver);
             requireNonNull(dataDirs);
             if (dataDirs.length == 0) {
                 throw new IllegalArgumentException("must provide at least one data directory");
             }
             requireNonNull(configDir);
+            requireNonNull(logsDir);
             requireNonNull(tempDir);
         }
     }
@@ -73,16 +79,27 @@ public static BootstrapArgs bootstrapArgs() {
     public static void bootstrap(
         Map<String, Policy> pluginPolicies,
         Function<Class<?>, String> pluginResolver,
+        Function<String, String> settingResolver,
+        Function<String, Stream<String>> settingGlobResolver,
         Path[] dataDirs,
         Path configDir,
-        Path tempDir,
-        Path logsDir
+        Path logsDir,
+        Path tempDir
     ) {
         logger.debug("Loading entitlement agent");
         if (EntitlementBootstrap.bootstrapArgs != null) {
             throw new IllegalStateException("plugin data is already set");
         }
-        EntitlementBootstrap.bootstrapArgs = new BootstrapArgs(pluginPolicies, pluginResolver, dataDirs, configDir, tempDir, logsDir);
+        EntitlementBootstrap.bootstrapArgs = new BootstrapArgs(
+            pluginPolicies,
+            pluginResolver,
+            settingResolver,
+            settingGlobResolver,
+            dataDirs,
+            configDir,
+            logsDir,
+            tempDir
+        );
         exportInitializationToAgent();
         loadAgent(findAgentJar());
         selfTest();

libs/entitlement/src/main/java/org/elasticsearch/entitlement/initialization/EntitlementInitialization.java

@@ -135,8 +135,14 @@ private static Class<?>[] findClassesToRetransform(Class<?>[] loadedClasses, Set
     private static PolicyManager createPolicyManager() {
         EntitlementBootstrap.BootstrapArgs bootstrapArgs = EntitlementBootstrap.bootstrapArgs();
         Map<String, Policy> pluginPolicies = bootstrapArgs.pluginPolicies();
-        var pathLookup = new PathLookup(getUserHome(), bootstrapArgs.configDir(), bootstrapArgs.dataDirs(), bootstrapArgs.tempDir());
-        Path logsDir = EntitlementBootstrap.bootstrapArgs().logsDir();
+        var pathLookup = new PathLookup(
+            getUserHome(),
+            bootstrapArgs.configDir(),
+            bootstrapArgs.dataDirs(),
+            bootstrapArgs.tempDir(),
+            bootstrapArgs.settingResolver(),
+            bootstrapArgs.settingGlobResolver()
+        );
 
         List<Scope> serverScopes = new ArrayList<>();
         Collections.addAll(

libs/entitlement/src/main/java/org/elasticsearch/entitlement/runtime/policy/FileAccessTree.java

@@ -42,8 +42,9 @@ private FileAccessTree(FilesEntitlement filesEntitlement, PathLookup pathLookup)
         }
 
         // everything has access to the temp dir
-        readPaths.add(pathLookup.tempDir().toString());
-        writePaths.add(pathLookup.tempDir().toString());
+        String tempDir = normalizePath(pathLookup.tempDir());
+        readPaths.add(tempDir);
+        writePaths.add(tempDir);
 
         readPaths.sort(String::compareTo);
         writePaths.sort(String::compareTo);

libs/entitlement/src/main/java/org/elasticsearch/entitlement/runtime/policy/PathLookup.java

@@ -10,5 +10,14 @@
 package org.elasticsearch.entitlement.runtime.policy;
 
 import java.nio.file.Path;
+import java.util.function.Function;
+import java.util.stream.Stream;
 
-public record PathLookup(Path homeDir, Path configDir, Path[] dataDirs, Path tempDir) {}
+public record PathLookup(
+    Path homeDir,
+    Path configDir,
+    Path[] dataDirs,
+    Path tempDir,
+    Function<String, String> settingResolver,
+    Function<String, Stream<String>> settingGlobResolver
+) {}