Merge branch 'main' into esql_pragma_load_source

Author: nik9000

docs/changelog/120952.yaml

@@ -0,0 +1,5 @@
+pr: 120952
+summary: Add `_metric_names_hash` field to OTel metric mappings
+area: Data streams
+type: bug
+issues: []

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/122637.yaml

@@ -0,0 +1,6 @@
+pr: 122637
+summary: Use `FallbackSyntheticSourceBlockLoader` for `unsigned_long` and `scaled_float`
+  fields
+area: Mapping
+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.

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RerouteProcessor.java

@@ -26,6 +26,7 @@
 import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
 import static org.elasticsearch.ingest.common.RerouteProcessor.DataStreamValueSource.DATASET_VALUE_SOURCE;
 import static org.elasticsearch.ingest.common.RerouteProcessor.DataStreamValueSource.NAMESPACE_VALUE_SOURCE;
+import static org.elasticsearch.ingest.common.RerouteProcessor.DataStreamValueSource.TYPE_VALUE_SOURCE;
 
 public final class RerouteProcessor extends AbstractProcessor {
 
@@ -39,18 +40,25 @@ public final class RerouteProcessor extends AbstractProcessor {
     private static final String DATA_STREAM_DATASET = DATA_STREAM_PREFIX + "dataset";
     private static final String DATA_STREAM_NAMESPACE = DATA_STREAM_PREFIX + "namespace";
     private static final String EVENT_DATASET = "event.dataset";
+    private final List<DataStreamValueSource> type;
     private final List<DataStreamValueSource> dataset;
     private final List<DataStreamValueSource> namespace;
     private final String destination;
 
     RerouteProcessor(
         String tag,
         String description,
+        List<DataStreamValueSource> type,
         List<DataStreamValueSource> dataset,
         List<DataStreamValueSource> namespace,
         String destination
     ) {
         super(tag, description);
+        if (type.isEmpty()) {
+            this.type = List.of(TYPE_VALUE_SOURCE);
+        } else {
+            this.type = type;
+        }
         if (dataset.isEmpty()) {
             this.dataset = List.of(DATASET_VALUE_SOURCE);
         } else {
@@ -71,7 +79,7 @@ public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
             return ingestDocument;
         }
         final String indexName = ingestDocument.getFieldValue(IngestDocument.Metadata.INDEX.getFieldName(), String.class);
-        final String type;
+        final String currentType;
         final String currentDataset;
         final String currentNamespace;
 
@@ -84,10 +92,11 @@ public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
         if (indexOfSecondDash < 0) {
             throw new IllegalArgumentException(format(NAMING_SCHEME_ERROR_MESSAGE, indexName));
         }
-        type = parseDataStreamType(indexName, indexOfFirstDash);
+        currentType = parseDataStreamType(indexName, indexOfFirstDash);
         currentDataset = parseDataStreamDataset(indexName, indexOfFirstDash, indexOfSecondDash);
         currentNamespace = parseDataStreamNamespace(indexName, indexOfSecondDash);
 
+        String type = determineDataStreamField(ingestDocument, this.type, currentType);
         String dataset = determineDataStreamField(ingestDocument, this.dataset, currentDataset);
         String namespace = determineDataStreamField(ingestDocument, this.namespace, currentNamespace);
         String newTarget = type + "-" + dataset + "-" + namespace;
@@ -168,6 +177,15 @@ public RerouteProcessor create(
             String description,
             Map<String, Object> config
         ) throws Exception {
+            List<DataStreamValueSource> type;
+            try {
+                type = ConfigurationUtils.readOptionalListOrString(TYPE, tag, config, "type")
+                    .stream()
+                    .map(DataStreamValueSource::type)
+                    .toList();
+            } catch (IllegalArgumentException e) {
+                throw newConfigurationException(TYPE, tag, "type", e.getMessage());
+            }
             List<DataStreamValueSource> dataset;
             try {
                 dataset = ConfigurationUtils.readOptionalListOrString(TYPE, tag, config, "dataset")
@@ -188,11 +206,11 @@ public RerouteProcessor create(
             }
 
             String destination = ConfigurationUtils.readOptionalStringProperty(TYPE, tag, config, "destination");
-            if (destination != null && (dataset.isEmpty() == false || namespace.isEmpty() == false)) {
-                throw newConfigurationException(TYPE, tag, "destination", "can only be set if dataset and namespace are not set");
+            if (destination != null && (type.isEmpty() == false || dataset.isEmpty() == false || namespace.isEmpty() == false)) {
+                throw newConfigurationException(TYPE, tag, "destination", "can only be set if type, dataset, and namespace are not set");
             }
 
-            return new RerouteProcessor(tag, description, dataset, namespace, destination);
+            return new RerouteProcessor(tag, description, type, dataset, namespace, destination);
         }
     }
 
@@ -203,15 +221,21 @@ static final class DataStreamValueSource {
 
         private static final int MAX_LENGTH = 100;
         private static final String REPLACEMENT = "_";
+        private static final Pattern DISALLOWED_IN_TYPE = Pattern.compile("[\\\\/*?\"<>| ,#:-]");
         private static final Pattern DISALLOWED_IN_DATASET = Pattern.compile("[\\\\/*?\"<>| ,#:-]");
         private static final Pattern DISALLOWED_IN_NAMESPACE = Pattern.compile("[\\\\/*?\"<>| ,#:]");
+        static final DataStreamValueSource TYPE_VALUE_SOURCE = type("{{" + DATA_STREAM_TYPE + "}}");
         static final DataStreamValueSource DATASET_VALUE_SOURCE = dataset("{{" + DATA_STREAM_DATASET + "}}");
         static final DataStreamValueSource NAMESPACE_VALUE_SOURCE = namespace("{{" + DATA_STREAM_NAMESPACE + "}}");
 
         private final String value;
         private final String fieldReference;
         private final Function<String, String> sanitizer;
 
+        public static DataStreamValueSource type(String type) {
+            return new DataStreamValueSource(type, ds -> sanitizeDataStreamField(ds, DISALLOWED_IN_TYPE));
+        }
+
         public static DataStreamValueSource dataset(String dataset) {
             return new DataStreamValueSource(dataset, ds -> sanitizeDataStreamField(ds, DISALLOWED_IN_DATASET));
         }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RerouteProcessorFactoryTests.java

@@ -47,7 +47,7 @@ public void testDestinationAndDataset() {
             ElasticsearchParseException.class,
             () -> create(Map.of("destination", "foo", "dataset", "bar"))
         );
-        assertThat(e.getMessage(), equalTo("[destination] can only be set if dataset and namespace are not set"));
+        assertThat(e.getMessage(), equalTo("[destination] can only be set if type, dataset, and namespace are not set"));
     }
 
     public void testFieldReference() throws Exception {