Merge branch 'main' into reindex-data-stream-make-source-read-only

Author: parkertimmins

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/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
 

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/extras/ScaledFloatFieldMapper.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.index.mapper.BlockLoader;
 import org.elasticsearch.index.mapper.BlockSourceReader;
 import org.elasticsearch.index.mapper.DocumentParserContext;
+import org.elasticsearch.index.mapper.FallbackSyntheticSourceBlockLoader;
 import org.elasticsearch.index.mapper.FieldMapper;
 import org.elasticsearch.index.mapper.IgnoreMalformedStoredValues;
 import org.elasticsearch.index.mapper.MapperBuilderContext;
@@ -195,7 +196,9 @@ public ScaledFloatFieldMapper build(MapperBuilderContext context) {
                 scalingFactor.getValue(),
                 nullValue.getValue(),
                 metric.getValue(),
-                indexMode
+                indexMode,
+                coerce.getValue().value(),
+                context.isSourceSynthetic()
             );
             return new ScaledFloatFieldMapper(leafName(), type, builderParams(this, context), context.isSourceSynthetic(), this);
         }
@@ -209,6 +212,8 @@ public static final class ScaledFloatFieldType extends SimpleMappedFieldType {
         private final Double nullValue;
         private final TimeSeriesParams.MetricType metricType;
         private final IndexMode indexMode;
+        private final boolean coerce;
+        private final boolean isSyntheticSource;
 
         public ScaledFloatFieldType(
             String name,
@@ -219,21 +224,25 @@ public ScaledFloatFieldType(
             double scalingFactor,
             Double nullValue,
             TimeSeriesParams.MetricType metricType,
-            IndexMode indexMode
+            IndexMode indexMode,
+            boolean coerce,
+            boolean isSyntheticSource
         ) {
             super(name, indexed, stored, hasDocValues, TextSearchInfo.SIMPLE_MATCH_WITHOUT_TERMS, meta);
             this.scalingFactor = scalingFactor;
             this.nullValue = nullValue;
             this.metricType = metricType;
             this.indexMode = indexMode;
+            this.coerce = coerce;
+            this.isSyntheticSource = isSyntheticSource;
         }
 
         public ScaledFloatFieldType(String name, double scalingFactor) {
             this(name, scalingFactor, true);
         }
 
         public ScaledFloatFieldType(String name, double scalingFactor, boolean indexed) {
-            this(name, indexed, false, true, Collections.emptyMap(), scalingFactor, null, null, null);
+            this(name, indexed, false, true, Collections.emptyMap(), scalingFactor, null, null, null, false, false);
         }
 
         public double getScalingFactor() {
@@ -315,13 +324,73 @@ public BlockLoader blockLoader(BlockLoaderContext blContext) {
                 double scalingFactorInverse = 1d / scalingFactor;
                 return new BlockDocValuesReader.DoublesBlockLoader(name(), l -> l * scalingFactorInverse);
             }
+            if (isSyntheticSource) {
+                return new FallbackSyntheticSourceBlockLoader(fallbackSyntheticSourceBlockLoaderReader(), name()) {
+                    @Override
+                    public Builder builder(BlockFactory factory, int expectedCount) {
+                        return factory.doubles(expectedCount);
+                    }
+                };
+            }
+
             ValueFetcher valueFetcher = sourceValueFetcher(blContext.sourcePaths(name()));
             BlockSourceReader.LeafIteratorLookup lookup = isStored() || isIndexed()
                 ? BlockSourceReader.lookupFromFieldNames(blContext.fieldNames(), name())
                 : BlockSourceReader.lookupMatchingAll();
             return new BlockSourceReader.DoublesBlockLoader(valueFetcher, lookup);
         }
 
+        private FallbackSyntheticSourceBlockLoader.Reader<?> fallbackSyntheticSourceBlockLoaderReader() {
+            var nullValueAdjusted = nullValue != null ? adjustSourceValue(nullValue, scalingFactor) : null;
+
+            return new FallbackSyntheticSourceBlockLoader.ReaderWithNullValueSupport<>(nullValue) {
+                @Override
+                public void convertValue(Object value, List<Double> accumulator) {
+                    if (coerce && value.equals("")) {
+                        if (nullValueAdjusted != null) {
+                            accumulator.add(nullValueAdjusted);
+                        }
+                    }
+
+                    try {
+                        // Convert to doc_values format
+                        var converted = adjustSourceValue(NumberFieldMapper.NumberType.objectToDouble(value), scalingFactor);
+                        accumulator.add(converted);
+                    } catch (Exception e) {
+                        // Malformed value, skip it
+                    }
+                }
+
+                @Override
+                protected void parseNonNullValue(XContentParser parser, List<Double> accumulator) throws IOException {
+                    // Aligned with implementation of `parseCreateField(XContentParser)`
+                    if (coerce && parser.currentToken() == XContentParser.Token.VALUE_STRING && parser.textLength() == 0) {
+                        if (nullValueAdjusted != null) {
+                            accumulator.add(nullValueAdjusted);
+                        }
+                    }
+
+                    try {
+                        double value = parser.doubleValue(coerce);
+                        // Convert to doc_values format
+                        var converted = adjustSourceValue(value, scalingFactor);
+                        accumulator.add(converted);
+                    } catch (Exception e) {
+                        // Malformed value, skip it
+                    }
+                }
+
+                @Override
+                public void writeToBlock(List<Double> values, BlockLoader.Builder blockBuilder) {
+                    var longBuilder = (BlockLoader.DoubleBuilder) blockBuilder;
+
+                    for (var value : values) {
+                        longBuilder.appendDouble(value);
+                    }
+                }
+            };
+        }
+
         @Override
         public IndexFieldData.Builder fielddataBuilder(FieldDataContext fieldDataContext) {
             FielddataOperation operation = fieldDataContext.fielddataOperation();
@@ -386,12 +455,16 @@ protected Double parseSourceValue(Object value) {
                         doubleValue = NumberFieldMapper.NumberType.objectToDouble(value);
                     }
 
-                    double factor = getScalingFactor();
-                    return Math.round(doubleValue * factor) / factor;
+                    return adjustSourceValue(doubleValue, getScalingFactor());
                 }
             };
         }
 
+        // Adjusts precision of a double value so that it looks like it came from doc_values.
+        private static Double adjustSourceValue(double value, double scalingFactor) {
+            return Math.round(value * scalingFactor) / scalingFactor;
+        }
+
         @Override
         public Object valueForDisplay(Object value) {
             if (value == null) {

modules/mapper-extras/src/test/java/org/elasticsearch/index/mapper/extras/ScaledFloatFieldBlockLoaderTests.java

@@ -0,0 +1,48 @@
+/*
+ * 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.index.mapper.extras;
+
+import org.elasticsearch.index.mapper.NumberFieldBlockLoaderTestCase;
+import org.elasticsearch.logsdb.datageneration.FieldType;
+import org.elasticsearch.plugins.Plugin;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+public class ScaledFloatFieldBlockLoaderTests extends NumberFieldBlockLoaderTestCase<Double> {
+    public ScaledFloatFieldBlockLoaderTests() {
+        super(FieldType.SCALED_FLOAT);
+    }
+
+    @Override
+    protected Double convert(Number value, Map<String, Object> fieldMapping) {
+        var scalingFactor = ((Number) fieldMapping.get("scaling_factor")).doubleValue();
+
+        var docValues = (boolean) fieldMapping.getOrDefault("doc_values", false);
+
+        // There is a slight inconsistency between values that are read from doc_values and from source.
+        // Due to how precision reduction is applied to source values so that they are consistent with doc_values.
+        // See #122547.
+        if (docValues) {
+            var reverseScalingFactor = 1d / scalingFactor;
+            return Math.round(value.doubleValue() * scalingFactor) * reverseScalingFactor;
+        }
+
+        // Adjust values coming from source to the way they are stored in doc_values.
+        // See mapper implementation.
+        return Math.round(value.doubleValue() * scalingFactor) / scalingFactor;
+    }
+
+    @Override
+    protected Collection<? extends Plugin> getPlugins() {
+        return List.of(new MapperExtrasPlugin());
+    }
+}

modules/mapper-extras/src/test/java/org/elasticsearch/index/mapper/extras/ScaledFloatFieldTypeTests.java

@@ -95,7 +95,9 @@ public void testRangeQuery() throws IOException {
             0.1 + randomDouble() * 100,
             null,
             null,
-            null
+            null,
+            false,
+            false
         );
         Directory dir = newDirectory();
         IndexWriter w = new IndexWriter(dir, new IndexWriterConfig(null));

muted-tests.yml

@@ -315,15 +315,9 @@ tests:
 - class: org.elasticsearch.xpack.autoscaling.storage.ReactiveStorageIT
   method: testScaleWhileShrinking
   issue: https://github.com/elastic/elasticsearch/issues/122119
-- class: org.elasticsearch.xpack.searchablesnapshots.FrozenSearchableSnapshotsIntegTests
-  method: testCreateAndRestorePartialSearchableSnapshot
-  issue: https://github.com/elastic/elasticsearch/issues/122693
 - class: org.elasticsearch.search.basic.SearchWithRandomDisconnectsIT
   method: testSearchWithRandomDisconnects
   issue: https://github.com/elastic/elasticsearch/issues/122707
-- class: org.elasticsearch.indices.recovery.IndexRecoveryIT
-  method: testSourceThrottling
-  issue: https://github.com/elastic/elasticsearch/issues/122712
 - class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
   issue: https://github.com/elastic/elasticsearch/issues/122810
 - class: org.elasticsearch.snapshots.DedicatedClusterSnapshotRestoreIT

server/src/internalClusterTest/java/org/elasticsearch/indices/recovery/IndexRecoveryIT.java

@@ -342,9 +342,17 @@ public void assertNodeHasThrottleTimeAndNoRecoveries(String nodeName, Boolean is
         assertThat(recoveryStats.currentAsSource(), equalTo(0));
         assertThat(recoveryStats.currentAsTarget(), equalTo(0));
         if (isRecoveryThrottlingNode) {
-            assertThat("Throttling should be >0 for '" + nodeName + "'", recoveryStats.throttleTime().millis(), greaterThan(0L));
+            assertThat(
+                "Throttling should be >0 for '" + nodeName + "'. Node stats: " + nodesStatsResponse,
+                recoveryStats.throttleTime().millis(),
+                greaterThan(0L)
+            );
         } else {
-            assertThat("Throttling should be =0 for '" + nodeName + "'", recoveryStats.throttleTime().millis(), equalTo(0L));
+            assertThat(
+                "Throttling should be =0 for '" + nodeName + "'. Node stats: " + nodesStatsResponse,
+                recoveryStats.throttleTime().millis(),
+                equalTo(0L)
+            );
         }
     }
 
@@ -1967,7 +1975,14 @@ public void testPostRecoveryMerge() throws Exception {
         internalCluster().startMasterOnlyNode();
         final var dataNode = internalCluster().startDataOnlyNode();
         final var indexName = randomIdentifier();
-        createIndex(indexName, indexSettings(1, 0).put(INDEX_MERGE_ENABLED, false).build());
+        final var indexSettingsBuilder = indexSettings(1, 0).put(INDEX_MERGE_ENABLED, false);
+        if (randomBoolean()) {
+            indexSettingsBuilder.put(
+                IndexMetadata.SETTING_VERSION_CREATED,
+                IndexVersionUtils.randomVersionBetween(random(), IndexVersions.UPGRADE_TO_LUCENE_10_0_0, IndexVersion.current())
+            );
+        }
+        createIndex(indexName, indexSettingsBuilder.build());
 
         final var initialSegmentCount = 20;
         for (int i = 0; i < initialSegmentCount; i++) {
@@ -2051,7 +2066,7 @@ public void testPostRecoveryMergeDisabledOnOlderIndices() throws Exception {
                     IndexVersionUtils.randomVersionBetween(
                         random(),
                         IndexVersionUtils.getLowestWriteCompatibleVersion(),
-                        IndexVersionUtils.getPreviousVersion(IndexVersions.MERGE_ON_RECOVERY_VERSION)
+                        IndexVersionUtils.getPreviousVersion(IndexVersions.UPGRADE_TO_LUCENE_10_0_0)
                     )
                 )
                 .build()

server/src/internalClusterTest/java/org/elasticsearch/search/retriever/RetrieverRewriteIT.java

@@ -127,10 +127,7 @@ public void testRewriteCompoundRetrieverShouldThrowForPartialResults() throws Ex
                 SearchPhaseExecutionException.class,
                 client().prepareSearch(testIndex).setSource(source)::get
             );
-            assertThat(
-                ex.getDetailedMessage(),
-                containsString("[open_point_in_time] action requires all shards to be available. Missing shards")
-            );
+            assertThat(ex.getDetailedMessage(), containsString("Search rejected due to missing shards"));
         } finally {
             internalCluster().restartNode(randomDataNode);
         }

server/src/main/java/org/elasticsearch/action/search/CanMatchPreFilterSearchPhase.java

@@ -201,7 +201,7 @@ private synchronized void consumeResult(int shardIndex, boolean canMatch, MinAnd
 
     private void checkNoMissingShards(List<SearchShardIterator> shards) {
         assert assertSearchCoordinationThread();
-        SearchPhase.doCheckNoMissingShards("can_match", request, shards, SearchPhase::makeMissingShardsError);
+        SearchPhase.doCheckNoMissingShards("can_match", request, shards);
     }
 
     private Map<SendingTarget, List<SearchShardIterator>> groupByNode(List<SearchShardIterator> shards) {