Merge branch 'main' of https://github.com/Jan-Kazlouski-elastic/elasticsearch into feature/hugging-face-chat-completion-integration

Author: Jan-Kazlouski-elastic

docs/release-notes/breaking-changes.md

@@ -39,6 +39,9 @@ will prevent node from starting [#118366](https://github.com/elastic/elasticsear
 Cluster Coordination:
 * Remove unsupported legacy value for `discovery.type` [#112903](https://github.com/elastic/elasticsearch/pull/112903)
 
+ES|QL
+* Limit the scope of the skip_unavailable setting for remote clusters [#120236](https://github.com/elastic/elasticsearch/pull/120236)
+
 EQL:
 * Set allow_partial_search_results=true by default [#120267](https://github.com/elastic/elasticsearch/pull/120267)
 

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamAutoshardingIT.java

@@ -528,7 +528,7 @@ private static ShardStats getShardStats(IndexMetadata indexMeta, int shardIndex,
         stats.docs = new DocsStats(100, 0, randomByteSizeValue().getBytes());
         stats.store = new StoreStats();
         stats.indexing = new IndexingStats(
-            new IndexingStats.Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, false, 1, targetWriteLoad, targetWriteLoad, 1, 0.123, 0.234)
+            new IndexingStats.Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, false, 1, 234, 234, 1000, 0.123, targetWriteLoad)
         );
         return new ShardStats(shardRouting, new ShardPath(false, path, path, shardId), stats, null, null, null, false, 0);
     }

muted-tests.yml

@@ -447,9 +447,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.mixed.EsqlClientYamlIT
   method: test {p0=esql/120_profile/avg 8.14 or after}
   issue: https://github.com/elastic/elasticsearch/issues/127879
-- class: org.elasticsearch.cluster.routing.allocation.decider.DiskThresholdDeciderIT
-  method: testRestoreSnapshotAllocationDoesNotExceedWatermarkWithMultipleRestores
-  issue: https://github.com/elastic/elasticsearch/issues/127787
 - class: org.elasticsearch.indices.stats.IndexStatsIT
   method: testThrottleStats
   issue: https://github.com/elastic/elasticsearch/issues/126359
@@ -477,6 +474,15 @@ tests:
 - class: org.elasticsearch.compute.aggregation.SampleDoubleAggregatorFunctionTests
   method: testSimpleWithCranky
   issue: https://github.com/elastic/elasticsearch/issues/128024
+- class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
+  method: test {lookup-join.MvJoinKeyOnTheLookupIndex ASYNC}
+  issue: https://github.com/elastic/elasticsearch/issues/128030
+- class: org.elasticsearch.packaging.test.DockerTests
+  method: test072RunEsAsDifferentUserAndGroup
+  issue: https://github.com/elastic/elasticsearch/issues/128031
+- class: org.elasticsearch.packaging.test.DockerTests
+  method: test042KeystorePermissionsAreCorrect
+  issue: https://github.com/elastic/elasticsearch/issues/128019
 
 # Examples:
 #

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDeciderIT.java

@@ -280,7 +280,12 @@ public void testRestoreSnapshotAllocationDoesNotExceedWatermarkWithMultipleResto
             tinyNodeShardIds.size() + tinyNodeShardIdsCopy.size(),
             is(1)
         );
-        assertThat(tinyNodeShardIds.iterator().next(), in(shardSizes.getShardIdsWithSizeSmallerOrEqual(usableSpace)));
+        final var useableSpaceShardSizes = shardSizes.getShardIdsWithSizeSmallerOrEqual(usableSpace);
+        final var tinyNodeShardId = tinyNodeShardIds.isEmpty() == false
+            ? tinyNodeShardIds.iterator().next()
+            // shardSizes only contains the sizes from the original index, not the copy, so we map the copied shard back to the original idx
+            : new ShardId(useableSpaceShardSizes.iterator().next().getIndex(), tinyNodeShardIdsCopy.iterator().next().id());
+        assertThat(tinyNodeShardId, in(useableSpaceShardSizes));
     }
 
     private Set<ShardId> getShardIds(final String nodeId, final String indexName) {

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java

@@ -171,6 +171,7 @@ private void cleanupRepo(String repositoryName, ActionListener<RepositoryCleanup
                     @Override
                     public ClusterState execute(ClusterState currentState) {
                         SnapshotsService.ensureRepositoryExists(repositoryName, currentState);
+                        SnapshotsService.ensureNotReadOnly(currentState, repositoryName);
                         final RepositoryCleanupInProgress repositoryCleanupInProgress = RepositoryCleanupInProgress.get(currentState);
                         if (repositoryCleanupInProgress.hasCleanupInProgress()) {
                             throw new IllegalStateException(

server/src/main/java/org/elasticsearch/action/datastreams/autosharding/DataStreamAutoShardingService.java

@@ -136,7 +136,7 @@ public enum WriteLoadMetric {
     public static final Setting<WriteLoadMetric> DATA_STREAMS_AUTO_SHARDING_INCREASE_SHARDS_LOAD_METRIC = Setting.enumSetting(
         WriteLoadMetric.class,
         "data_streams.auto_sharding.increase_shards.load_metric",
-        WriteLoadMetric.ALL_TIME,
+        WriteLoadMetric.PEAK,
         Setting.Property.Dynamic,
         Setting.Property.NodeScope
     );
@@ -147,7 +147,7 @@ public enum WriteLoadMetric {
     public static final Setting<WriteLoadMetric> DATA_STREAMS_AUTO_SHARDING_DECREASE_SHARDS_LOAD_METRIC = Setting.enumSetting(
         WriteLoadMetric.class,
         "data_streams.auto_sharding.decrease_shards.load_metric",
-        WriteLoadMetric.ALL_TIME,
+        WriteLoadMetric.PEAK,
         Setting.Property.Dynamic,
         Setting.Property.NodeScope
     );

server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java

@@ -50,6 +50,7 @@
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.repositories.VerifyNodeRepositoryAction.Request;
+import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
 import org.elasticsearch.repositories.blobstore.MeteredBlobStoreRepository;
 import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -289,6 +290,7 @@ public ClusterState execute(ClusterState currentState) {
             List<RepositoryMetadata> repositoriesMetadata = new ArrayList<>(repositories.repositories().size() + 1);
             for (RepositoryMetadata repositoryMetadata : repositories.repositories()) {
                 if (repositoryMetadata.name().equals(request.name())) {
+                    rejectInvalidReadonlyFlagChange(repositoryMetadata, request.settings());
                     final RepositoryMetadata newRepositoryMetadata = new RepositoryMetadata(
                         request.name(),
                         // Copy the UUID from the existing instance rather than resetting it back to MISSING_UUID which would force us to
@@ -601,6 +603,7 @@ public static boolean isDedicatedVotingOnlyNode(Set<DiscoveryNodeRole> roles) {
     public void applyClusterState(ClusterChangedEvent event) {
         try {
             final ClusterState state = event.state();
+            assert assertReadonlyRepositoriesNotInUseForWrites(state);
             final RepositoriesMetadata oldMetadata = RepositoriesMetadata.get(event.previousState());
             final RepositoriesMetadata newMetadata = RepositoriesMetadata.get(state);
 
@@ -885,7 +888,7 @@ public static void validateRepositoryName(final String repositoryName) {
         }
     }
 
-    private static void ensureRepositoryNotInUse(ClusterState clusterState, String repository) {
+    private static void ensureRepositoryNotInUseForWrites(ClusterState clusterState, String repository) {
         if (SnapshotsInProgress.get(clusterState).forRepo(repository).isEmpty() == false) {
             throw newRepositoryConflictException(repository, "snapshot is in progress");
         }
@@ -899,13 +902,57 @@ private static void ensureRepositoryNotInUse(ClusterState clusterState, String r
                 throw newRepositoryConflictException(repository, "repository clean up is in progress");
             }
         }
+    }
+
+    private static void ensureRepositoryNotInUse(ClusterState clusterState, String repository) {
+        ensureRepositoryNotInUseForWrites(clusterState, repository);
         for (RestoreInProgress.Entry entry : RestoreInProgress.get(clusterState)) {
             if (repository.equals(entry.snapshot().getRepository())) {
                 throw newRepositoryConflictException(repository, "snapshot restore is in progress");
             }
         }
     }
 
+    public static boolean isReadOnly(Settings repositorySettings) {
+        return Boolean.TRUE.equals(repositorySettings.getAsBoolean(BlobStoreRepository.READONLY_SETTING_KEY, null));
+    }
+
+    /**
+     * Test-only check for the invariant that read-only repositories never have any write activities.
+     */
+    private static boolean assertReadonlyRepositoriesNotInUseForWrites(ClusterState clusterState) {
+        for (final var repositoryMetadata : RepositoriesMetadata.get(clusterState).repositories()) {
+            if (isReadOnly(repositoryMetadata.settings())) {
+                try {
+                    ensureRepositoryNotInUseForWrites(clusterState, repositoryMetadata.name());
+                } catch (Exception e) {
+                    throw new AssertionError("repository [" + repositoryMetadata + "] is readonly but still in use", e);
+                }
+            }
+        }
+        return true;
+    }
+
+    /**
+     * Reject a change to the {@code readonly} setting if there is a pending generation change in progress, i.e. some node somewhere is
+     * updating the root {@link RepositoryData} blob.
+     */
+    private static void rejectInvalidReadonlyFlagChange(RepositoryMetadata existingRepositoryMetadata, Settings newSettings) {
+        if (isReadOnly(newSettings)
+            && isReadOnly(existingRepositoryMetadata.settings()) == false
+            && existingRepositoryMetadata.generation() >= RepositoryData.EMPTY_REPO_GEN
+            && existingRepositoryMetadata.generation() != existingRepositoryMetadata.pendingGeneration()) {
+            throw newRepositoryConflictException(
+                existingRepositoryMetadata.name(),
+                Strings.format(
+                    "currently updating root blob generation from [%d] to [%d], cannot update readonly flag",
+                    existingRepositoryMetadata.generation(),
+                    existingRepositoryMetadata.pendingGeneration()
+                )
+            );
+        }
+    }
+
     private static void ensureNoSearchableSnapshotsIndicesInUse(ClusterState clusterState, RepositoryMetadata repositoryMetadata) {
         long count = 0L;
         List<Index> indices = null;

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -2737,6 +2737,14 @@ protected void writeIndexGen(
             public ClusterState execute(ClusterState currentState) {
                 final RepositoryMetadata meta = getRepoMetadata(currentState);
                 final String repoName = metadata.name();
+
+                if (RepositoriesService.isReadOnly(meta.settings())) {
+                    // Last resort check: we shouldn't have been able to mark the repository as readonly while the operation that led to
+                    // this writeIndexGen() call was in progress, and conversely shouldn't have started any such operation if the repo
+                    // was already readonly, but these invariants are not obviously true and it is disastrous to proceed here.
+                    throw new RepositoryException(meta.name(), "repository is readonly, cannot update root blob");
+                }
+
                 final long genInState = meta.generation();
                 final boolean uninitializedMeta = meta.generation() == RepositoryData.UNKNOWN_REPO_GEN || bestEffortConsistency;
                 if (uninitializedMeta == false && meta.pendingGeneration() != genInState) {

server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

@@ -338,6 +338,7 @@ public ClusterState execute(ClusterState currentState) {
                 ensureRepositoryExists(repositoryName, currentState);
                 ensureSnapshotNameAvailableInRepo(repositoryData, snapshotName, repository);
                 ensureNoCleanupInProgress(currentState, repositoryName, snapshotName, "clone snapshot");
+                ensureNotReadOnly(currentState, repositoryName);
                 final SnapshotsInProgress snapshots = SnapshotsInProgress.get(currentState);
                 ensureSnapshotNameNotRunning(snapshots, repositoryName, snapshotName);
                 validate(repositoryName, snapshotName, currentState);
@@ -433,6 +434,13 @@ private static void ensureNoCleanupInProgress(
         }
     }
 
+    public static void ensureNotReadOnly(final ClusterState currentState, final String repositoryName) {
+        final var repositoryMetadata = RepositoriesMetadata.get(currentState).repository(repositoryName);
+        if (RepositoriesService.isReadOnly(repositoryMetadata.settings())) {
+            throw new RepositoryException(repositoryMetadata.name(), "repository is readonly");
+        }
+    }
+
     private static void ensureSnapshotNameAvailableInRepo(RepositoryData repositoryData, String snapshotName, Repository repository) {
         // check if the snapshot name already exists in the repository
         if (repositoryData.getSnapshotIds().stream().anyMatch(s -> s.getName().equals(snapshotName))) {
@@ -2166,6 +2174,8 @@ public ClusterState execute(ClusterState currentState) {
                     "delete snapshot"
                 );
 
+                ensureNotReadOnly(currentState, repositoryName);
+
                 final SnapshotDeletionsInProgress deletionsInProgress = SnapshotDeletionsInProgress.get(currentState);
 
                 final RestoreInProgress restoreInProgress = RestoreInProgress.get(currentState);
@@ -4075,11 +4085,16 @@ public ClusterState execute(BatchExecutionContext<SnapshotTask> batchExecutionCo
             for (final var taskContext : batchExecutionContext.taskContexts()) {
                 if (taskContext.getTask() instanceof CreateSnapshotTask task) {
                     try {
+                        final var repoMeta = RepositoriesMetadata.get(state).repository(task.snapshot.getRepository());
+                        if (RepositoriesService.isReadOnly(repoMeta.settings())) {
+                            taskContext.onFailure(new RepositoryException(repoMeta.name(), "repository is readonly"));
+                            continue;
+                        }
+
                         registeredPolicySnapshots.addIfSnapshotIsSLMInitiated(
                             task.createSnapshotRequest.userMetadata(),
                             task.snapshot.getSnapshotId()
                         );
-                        final var repoMeta = RepositoriesMetadata.get(state).repository(task.snapshot.getRepository());
                         if (Objects.equals(task.initialRepositoryMetadata, repoMeta)) {
                             snapshotsInProgress = createSnapshot(task, taskContext, state, snapshotsInProgress);
                         } else {