Merge branch 'main' into esql/fix_telemetry_fork

Author: luigidellaquila

docs/changelog/129738.yaml

@@ -0,0 +1,5 @@
+pr: 129738
+summary: Watch SSL files instead of directories
+area: TLS
+type: bug
+issues: []

muted-tests.yml

@@ -551,23 +551,18 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeIT
   method: test
   issue: https://github.com/elastic/elasticsearch/issues/129819
-- class: org.elasticsearch.index.engine.ThreadPoolMergeExecutorServiceDiskSpaceTests
-  method: testAbortingOrRunningMergeTaskHoldsUpBudget
-  issue: https://github.com/elastic/elasticsearch/issues/129823
 - class: org.elasticsearch.index.store.FsDirectoryFactoryTests
   method: testPreload
   issue: https://github.com/elastic/elasticsearch/issues/129852
 - class: org.elasticsearch.xpack.rank.rrf.RRFRankClientYamlTestSuiteIT
   method: test {yaml=rrf/950_pinned_interaction/rrf with pinned retriever as a sub-retriever}
   issue: https://github.com/elastic/elasticsearch/issues/129845
-- class: org.elasticsearch.index.mapper.vectors.DenseVectorFieldMapperTests
-  method: testExistsQueryMinimalMapping
-  issue: https://github.com/elastic/elasticsearch/issues/129911
 - class: org.elasticsearch.gradle.internal.InternalDistributionBwcSetupPluginFuncTest
   method: "builds distribution from branches via archives extractedAssemble [bwcDistVersion: 8.2.1, bwcProject: bugfix, expectedAssembleTaskName:
     extractedAssemble, #2]"
   issue: https://github.com/elastic/elasticsearch/issues/119871
 
+
 # Examples:
 #
 #  Mute a single test case in a YAML test suite:

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -322,6 +322,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion CLUSTER_STATE_PROJECTS_SETTINGS = def(9_108_0_00);
     public static final TransportVersion ML_INFERENCE_ELASTIC_DENSE_TEXT_EMBEDDINGS_ADDED = def(9_109_00_0);
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION = def(9_110_0_00);
+    public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN = def(9_111_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/watcher/FileWatcher.java

@@ -57,6 +57,11 @@ public FileWatcher(Path path, boolean checkFileContents) {
         rootFileObserver = new FileObserver(path);
     }
 
+    // For testing
+    public Path getPath() {
+        return path;
+    }
+
     /**
      * Clears any state with the FileWatcher, making all files show up as new
      */

server/src/test/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorServiceDiskSpaceTests.java

@@ -534,7 +534,7 @@ public void testAbortingOrRunningMergeTaskHoldsUpBudget() throws Exception {
         aFileStore.totalSpace = randomLongBetween(1_000L, 10_000L);
         bFileStore.totalSpace = randomLongBetween(1_000L, 10_000L);
         aFileStore.usableSpace = randomLongBetween(900L, aFileStore.totalSpace);
-        bFileStore.usableSpace = randomLongBetween(900L, bFileStore.totalSpace);
+        bFileStore.usableSpace = randomValueOtherThan(aFileStore.usableSpace, () -> randomLongBetween(900L, bFileStore.totalSpace));
         boolean aHasMoreSpace = aFileStore.usableSpace > bFileStore.usableSpace;
         try (
             ThreadPoolMergeExecutorService threadPoolMergeExecutorService = ThreadPoolMergeExecutorService
@@ -613,7 +613,7 @@ public void testBackloggedMergeTasksDoNotHoldUpBudget() throws Exception {
         aFileStore.totalSpace = randomLongBetween(1_000L, 10_000L);
         bFileStore.totalSpace = randomLongBetween(1_000L, 10_000L);
         aFileStore.usableSpace = randomLongBetween(900L, aFileStore.totalSpace);
-        bFileStore.usableSpace = randomLongBetween(900L, bFileStore.totalSpace);
+        bFileStore.usableSpace = randomValueOtherThan(aFileStore.usableSpace, () -> randomLongBetween(900L, bFileStore.totalSpace));
         boolean aHasMoreSpace = aFileStore.usableSpace > bFileStore.usableSpace;
         try (
             ThreadPoolMergeExecutorService threadPoolMergeExecutorService = ThreadPoolMergeExecutorService
@@ -900,7 +900,7 @@ public void testEnqueuedMergeTasksAreUnblockedWhenEstimatedMergeSizeChanges() th
         long diskSpaceLimitBytes = randomLongBetween(10L, 100L);
         aFileStore.usableSpace = diskSpaceLimitBytes + randomLongBetween(1L, 100L);
         aFileStore.totalSpace = aFileStore.usableSpace + randomLongBetween(1L, 10L);
-        bFileStore.usableSpace = diskSpaceLimitBytes + randomLongBetween(1L, 100L);
+        bFileStore.usableSpace = randomValueOtherThan(aFileStore.usableSpace, () -> diskSpaceLimitBytes + randomLongBetween(1L, 100L));
         bFileStore.totalSpace = bFileStore.usableSpace + randomLongBetween(1L, 10L);
         boolean aHasMoreSpace = aFileStore.usableSpace > bFileStore.usableSpace;
         Settings.Builder settingsBuilder = Settings.builder().put(settings);
@@ -1001,7 +1001,10 @@ public void testMergeTasksAreUnblockedWhenMoreDiskSpaceBecomesAvailable() throws
         bFileStore.totalSpace = randomLongBetween(300L, 1_000L);
         long grantedUsableSpaceBuffer = randomLongBetween(10L, 50L);
         aFileStore.usableSpace = randomLongBetween(200L, aFileStore.totalSpace - grantedUsableSpaceBuffer);
-        bFileStore.usableSpace = randomLongBetween(200L, bFileStore.totalSpace - grantedUsableSpaceBuffer);
+        bFileStore.usableSpace = randomValueOtherThan(
+            aFileStore.usableSpace,
+            () -> randomLongBetween(200L, bFileStore.totalSpace - grantedUsableSpaceBuffer)
+        );
         boolean aHasMoreSpace = aFileStore.usableSpace > bFileStore.usableSpace;
         Settings.Builder settingsBuilder = Settings.builder().put(settings);
         // change the watermark level, just for coverage and it's easier with the calculations

server/src/test/java/org/elasticsearch/index/mapper/vectors/DenseVectorFieldMapperTests.java

@@ -65,6 +65,7 @@
 
 import static org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat.DEFAULT_BEAM_WIDTH;
 import static org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat.DEFAULT_MAX_CONN;
+import static org.apache.lucene.tests.index.BaseKnnVectorsFormatTestCase.randomNormalizedVector;
 import static org.elasticsearch.index.codec.vectors.IVFVectorsFormat.DYNAMIC_NPROBE;
 import static org.elasticsearch.index.mapper.vectors.DenseVectorFieldMapper.DEFAULT_OVERSAMPLE;
 import static org.elasticsearch.index.mapper.vectors.DenseVectorFieldMapper.IVF_FORMAT;
@@ -147,7 +148,17 @@ private void indexMapping(XContentBuilder b, IndexVersion indexVersion) throws I
 
     @Override
     protected Object getSampleValueForDocument() {
-        return elementType == ElementType.FLOAT ? List.of(0.5, 0.5, 0.5, 0.5) : List.of((byte) 1, (byte) 1, (byte) 1, (byte) 1);
+        return elementType == ElementType.FLOAT
+            ? convertToList(randomNormalizedVector(this.dims))
+            : List.of((byte) 1, (byte) 1, (byte) 1, (byte) 1);
+    }
+
+    private static List<Float> convertToList(float[] vector) {
+        List<Float> list = new ArrayList<>(vector.length);
+        for (float v : vector) {
+            list.add(v);
+        }
+        return list;
     }
 
     @Override

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -2713,9 +2713,11 @@ public static Request newXContentRequest(HttpMethod method, String endpoint, ToX
     }
 
     protected static MapMatcher getProfileMatcher() {
-        return matchesMap().entry("query", instanceOf(Map.class))
+        return matchesMap() //
+            .entry("query", instanceOf(Map.class))
             .entry("planning", instanceOf(Map.class))
-            .entry("drivers", instanceOf(List.class));
+            .entry("drivers", instanceOf(List.class))
+            .entry("plans", instanceOf(List.class));
     }
 
     protected static MapMatcher getResultMatcher(boolean includeMetadata, boolean includePartial, boolean includeDocumentsFound) {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ssl/SSLConfigurationReloader.java

@@ -22,10 +22,8 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.function.Consumer;
@@ -80,7 +78,7 @@ private static Consumer<SslConfiguration> reloadConsumer(Future<SSLService> futu
     }
 
     /**
-     * Collects all of the directories that need to be monitored for the provided {@link SslConfiguration} instances and ensures that
+     * Collects all of the files that need to be monitored for the provided {@link SslConfiguration} instances and ensures that
      * they are being watched for changes
      */
     private static void startWatching(
@@ -91,8 +89,8 @@ private static void startWatching(
         Map<Path, List<SslConfiguration>> pathToConfigurationsMap = new HashMap<>();
         for (SslConfiguration sslConfiguration : sslConfigurations) {
             final Collection<Path> filesToMonitor = sslConfiguration.getDependentFiles();
-            for (Path directory : directoriesToMonitor(filesToMonitor)) {
-                pathToConfigurationsMap.compute(directory, (path, list) -> {
+            for (Path file : filesToMonitor) {
+                pathToConfigurationsMap.compute(file, (path, list) -> {
                     if (list == null) {
                         list = new ArrayList<>();
                     }
@@ -109,22 +107,11 @@ private static void startWatching(
             try {
                 resourceWatcherService.add(fileWatcher, Frequency.HIGH);
             } catch (IOException | SecurityException e) {
-                logger.error("failed to start watching directory [{}] for ssl configurations [{}] - {}", path, configurations, e);
+                logger.error("failed to start watching file [{}] for ssl configurations [{}] - {}", path, configurations, e);
             }
         });
     }
 
-    /**
-     * Returns a unique set of directories that need to be monitored based on the provided file paths
-     */
-    private static Set<Path> directoriesToMonitor(Iterable<Path> filePaths) {
-        Set<Path> paths = new HashSet<>();
-        for (Path path : filePaths) {
-            paths.add(path.getParent());
-        }
-        return paths;
-    }
-
     private static class ChangeListener implements FileChangesListener {
 
         private final List<SslConfiguration> sslConfigurations;

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ssl/SSLConfigurationReloaderTests.java

@@ -37,6 +37,8 @@
 import org.elasticsearch.test.http.MockWebServer;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.watcher.FileWatcher;
+import org.elasticsearch.watcher.ResourceWatcher;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.junit.After;
 import org.junit.Before;
@@ -66,7 +68,9 @@
 import java.security.cert.CertificateException;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
@@ -79,6 +83,7 @@
 import javax.net.ssl.SSLSocket;
 
 import static org.elasticsearch.test.TestMatchers.throwableWithMessage;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.sameInstance;
 
@@ -559,6 +564,38 @@ public void testFailureToReadFileDoesntFail() throws Exception {
         }
     }
 
+    /**
+     * Due to exclusive access entitlements
+     * (see {@link org.elasticsearch.entitlement.runtime.policy.entitlements.FilesEntitlement.FileData#exclusive}),
+     * it is not safe to monitor a directory or any files that are not an explicit part of this SSL configuration.
+     */
+    public void testReloaderOnlyWatchesSpecifiedFiles() throws Exception {
+        final Set<Path> watchedPaths = new HashSet<>();
+        final ResourceWatcherService mockResourceWatcher = Mockito.mock(ResourceWatcherService.class);
+        Mockito.when(mockResourceWatcher.add(Mockito.any(ResourceWatcher.class), Mockito.any(ResourceWatcherService.Frequency.class)))
+            .then(inv -> {
+                final FileWatcher fileWatcher = asInstanceOf(FileWatcher.class, inv.getArguments()[0]);
+                watchedPaths.add(fileWatcher.getPath());
+                return null;
+            });
+
+        final Path tempDir = createTempDir();
+        final Path clientCertPath = tempDir.resolve("testclient.crt");
+        Settings settings = baseKeystoreSettings(tempDir, null).putList(
+            "xpack.security.transport.ssl.certificate_authorities",
+            clientCertPath.toString()
+        ).put("path.home", createTempDir()).build();
+
+        final Environment env = newEnvironment(settings);
+        final Collection<SslConfiguration> configurations = SSLService.getSSLConfigurations(env).values();
+        new SSLConfigurationReloader(ignore -> {}, mockResourceWatcher, configurations);
+
+        assertThat(
+            watchedPaths,
+            containsInAnyOrder(tempDir.resolve("testclient.pem"), tempDir.resolve("testclient.crt"), tempDir.resolve("testclientcert.crt"))
+        );
+    }
+
     private Settings.Builder baseKeystoreSettings(Path tempDir, MockSecureSettings secureSettings) throws IOException {
         final Path keyPath = tempDir.resolve("testclient.pem");
         final Path certPath = tempDir.resolve("testclientcert.crt"); // testclient.crt filename already used in #testPEMTrustReloadException

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/DriverCompletionInfo.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.compute.operator;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -24,23 +25,34 @@
  *                     <strong>roughly</strong> the number of documents times the number of
  *                     fields per document. Except {@code null} values don't count.
  *                     And multivalued fields count as many times as there are values.
- * @param collectedProfiles {@link DriverProfile}s from each driver. These are fairly cheap to build but
+ * @param driverProfiles {@link DriverProfile}s from each driver. These are fairly cheap to build but
  *                          not free so this will be empty if the {@code profile} option was not set in
  *                          the request.
  */
-public record DriverCompletionInfo(long documentsFound, long valuesLoaded, List<DriverProfile> collectedProfiles) implements Writeable {
+public record DriverCompletionInfo(
+    long documentsFound,
+    long valuesLoaded,
+    List<DriverProfile> driverProfiles,
+    List<PlanProfile> planProfiles
+) implements Writeable {
 
     /**
      * Completion info we use when we didn't properly complete any drivers.
      * Usually this is returned with an error, but it's also used when receiving
      * responses from very old nodes.
      */
-    public static final DriverCompletionInfo EMPTY = new DriverCompletionInfo(0, 0, List.of());
+    public static final DriverCompletionInfo EMPTY = new DriverCompletionInfo(0, 0, List.of(), List.of());
 
     /**
      * Build a {@link DriverCompletionInfo} for many drivers including their profile output.
      */
-    public static DriverCompletionInfo includingProfiles(List<Driver> drivers) {
+    public static DriverCompletionInfo includingProfiles(
+        List<Driver> drivers,
+        String description,
+        String clusterName,
+        String nodeName,
+        String planTree
+    ) {
         long documentsFound = 0;
         long valuesLoaded = 0;
         List<DriverProfile> collectedProfiles = new ArrayList<>(drivers.size());
@@ -52,7 +64,12 @@ public static DriverCompletionInfo includingProfiles(List<Driver> drivers) {
             }
             collectedProfiles.add(p);
         }
-        return new DriverCompletionInfo(documentsFound, valuesLoaded, collectedProfiles);
+        return new DriverCompletionInfo(
+            documentsFound,
+            valuesLoaded,
+            collectedProfiles,
+            List.of(new PlanProfile(description, clusterName, nodeName, planTree))
+        );
     }
 
     /**
@@ -69,49 +86,63 @@ public static DriverCompletionInfo excludingProfiles(List<Driver> drivers) {
                 valuesLoaded += o.valuesLoaded();
             }
         }
-        return new DriverCompletionInfo(documentsFound, valuesLoaded, List.of());
+        return new DriverCompletionInfo(documentsFound, valuesLoaded, List.of(), List.of());
     }
 
-    public DriverCompletionInfo(StreamInput in) throws IOException {
-        this(in.readVLong(), in.readVLong(), in.readCollectionAsImmutableList(DriverProfile::readFrom));
+    public static DriverCompletionInfo readFrom(StreamInput in) throws IOException {
+        return new DriverCompletionInfo(
+            in.readVLong(),
+            in.readVLong(),
+            in.readCollectionAsImmutableList(DriverProfile::readFrom),
+            in.getTransportVersion().onOrAfter(TransportVersions.ESQL_PROFILE_INCLUDE_PLAN)
+                ? in.readCollectionAsImmutableList(PlanProfile::readFrom)
+                : List.of()
+        );
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeVLong(documentsFound);
         out.writeVLong(valuesLoaded);
-        out.writeCollection(collectedProfiles, (o, v) -> v.writeTo(o));
+        out.writeCollection(driverProfiles);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_PROFILE_INCLUDE_PLAN)) {
+            out.writeCollection(planProfiles);
+        }
     }
 
     public static class Accumulator {
         private long documentsFound;
         private long valuesLoaded;
-        private final List<DriverProfile> collectedProfiles = new ArrayList<>();
+        private final List<DriverProfile> driverProfiles = new ArrayList<>();
+        private final List<PlanProfile> planProfiles = new ArrayList<>();
 
         public void accumulate(DriverCompletionInfo info) {
             this.documentsFound += info.documentsFound;
             this.valuesLoaded += info.valuesLoaded;
-            this.collectedProfiles.addAll(info.collectedProfiles);
+            this.driverProfiles.addAll(info.driverProfiles);
+            this.planProfiles.addAll(info.planProfiles);
         }
 
         public DriverCompletionInfo finish() {
-            return new DriverCompletionInfo(documentsFound, valuesLoaded, collectedProfiles);
+            return new DriverCompletionInfo(documentsFound, valuesLoaded, driverProfiles, planProfiles);
         }
     }
 
     public static class AtomicAccumulator {
         private final AtomicLong documentsFound = new AtomicLong();
         private final AtomicLong valuesLoaded = new AtomicLong();
         private final List<DriverProfile> collectedProfiles = Collections.synchronizedList(new ArrayList<>());
+        private final List<PlanProfile> planProfiles = Collections.synchronizedList(new ArrayList<>());
 
         public void accumulate(DriverCompletionInfo info) {
             this.documentsFound.addAndGet(info.documentsFound);
             this.valuesLoaded.addAndGet(info.valuesLoaded);
-            this.collectedProfiles.addAll(info.collectedProfiles);
+            this.collectedProfiles.addAll(info.driverProfiles);
+            this.planProfiles.addAll(info.planProfiles);
         }
 
         public DriverCompletionInfo finish() {
-            return new DriverCompletionInfo(documentsFound.get(), valuesLoaded.get(), collectedProfiles);
+            return new DriverCompletionInfo(documentsFound.get(), valuesLoaded.get(), collectedProfiles, planProfiles);
         }
     }
 }