Merge branch 'main' of github.com:elastic/elasticsearch into esql_namedqueries

Author: tteofili

docs/changelog/127731.yaml

@@ -0,0 +1,5 @@
+pr: 127731
+summary: ESQL - Enable telemetry for COMPLETION command
+area: Search
+type: feature
+issues: []

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpChannel.java

@@ -12,7 +12,7 @@
 import io.netty.channel.Channel;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.http.HttpChannel;
 import org.elasticsearch.http.HttpResponse;
 
@@ -25,7 +25,7 @@
 public class Netty4HttpChannel implements HttpChannel {
 
     private final Channel channel;
-    private final ListenableFuture<Void> closeContext = new ListenableFuture<>();
+    private final SubscribableListener<Void> closeContext = new SubscribableListener<>();
 
     Netty4HttpChannel(Channel channel) {
         this.channel = channel;

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerChannel.java

@@ -12,7 +12,7 @@
 import io.netty.channel.Channel;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.http.HttpServerChannel;
 
 import java.net.InetSocketAddress;
@@ -22,7 +22,7 @@
 public class Netty4HttpServerChannel implements HttpServerChannel {
 
     private final Channel channel;
-    private final ListenableFuture<Void> closeContext = new ListenableFuture<>();
+    private final SubscribableListener<Void> closeContext = new SubscribableListener<>();
 
     Netty4HttpServerChannel(Channel channel) {
         this.channel = channel;

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpChannel.java

@@ -14,8 +14,8 @@
 import io.netty.channel.ChannelOption;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.transport.TcpChannel;
@@ -30,8 +30,8 @@ public class Netty4TcpChannel implements TcpChannel {
     private final Channel channel;
     private final boolean isServer;
     private final String profile;
-    private final ListenableFuture<Void> connectContext;
-    private final ListenableFuture<Void> closeContext = new ListenableFuture<>();
+    private final SubscribableListener<Void> connectContext = new SubscribableListener<>();
+    private final SubscribableListener<Void> closeContext = new SubscribableListener<>();
     private final ChannelStats stats = new ChannelStats();
     private final boolean rstOnClose;
     /**
@@ -43,7 +43,6 @@ public class Netty4TcpChannel implements TcpChannel {
         this.channel = channel;
         this.isServer = isServer;
         this.profile = profile;
-        this.connectContext = new ListenableFuture<>();
         this.rstOnClose = rstOnClose;
         addListener(connectFuture, connectContext);
         addListener(this.channel.closeFuture(), new ActionListener<>() {

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpServerChannel.java

@@ -12,7 +12,7 @@
 import io.netty.channel.Channel;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.transport.TcpServerChannel;
 
 import java.net.InetSocketAddress;
@@ -22,7 +22,7 @@
 public class Netty4TcpServerChannel implements TcpServerChannel {
 
     private final Channel channel;
-    private final ListenableFuture<Void> closeContext = new ListenableFuture<>();
+    private final SubscribableListener<Void> closeContext = new SubscribableListener<>();
 
     Netty4TcpServerChannel(Channel channel) {
         this.channel = channel;

server/src/main/java/org/elasticsearch/transport/CloseableConnection.java

@@ -10,7 +10,7 @@
 package org.elasticsearch.transport;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.core.AbstractRefCounted;
 
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -20,8 +20,8 @@
  */
 public abstract class CloseableConnection extends AbstractRefCounted implements Transport.Connection {
 
-    private final ListenableFuture<Void> closeContext = new ListenableFuture<>();
-    private final ListenableFuture<Void> removeContext = new ListenableFuture<>();
+    private final SubscribableListener<Void> closeContext = new SubscribableListener<>();
+    private final SubscribableListener<Void> removeContext = new SubscribableListener<>();
 
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final AtomicBoolean removed = new AtomicBoolean(false);

server/src/main/java/org/elasticsearch/transport/ClusterConnectionManager.java

@@ -12,11 +12,11 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.ReferenceDocs;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.AbstractRefCounted;
@@ -43,7 +43,7 @@ public class ClusterConnectionManager implements ConnectionManager {
     private static final Logger logger = LogManager.getLogger(ClusterConnectionManager.class);
 
     private final ConcurrentMap<DiscoveryNode, Transport.Connection> connectedNodes = ConcurrentCollections.newConcurrentMap();
-    private final ConcurrentMap<DiscoveryNode, ListenableFuture<Transport.Connection>> pendingConnections = ConcurrentCollections
+    private final ConcurrentMap<DiscoveryNode, SubscribableListener<Transport.Connection>> pendingConnections = ConcurrentCollections
         .newConcurrentMap();
     private final AbstractRefCounted connectingRefCounter = AbstractRefCounted.of(this::pendingConnectionsComplete);
 
@@ -184,8 +184,8 @@ private void connectToNodeOrRetry(
             return;
         }
 
-        final ListenableFuture<Transport.Connection> currentListener = new ListenableFuture<>();
-        final ListenableFuture<Transport.Connection> existingListener = pendingConnections.putIfAbsent(node, currentListener);
+        final SubscribableListener<Transport.Connection> currentListener = new SubscribableListener<>();
+        final SubscribableListener<Transport.Connection> existingListener = pendingConnections.putIfAbsent(node, currentListener);
         if (existingListener != null) {
             try {
                 // wait on previous entry to complete connection attempt
@@ -203,7 +203,7 @@ private void connectToNodeOrRetry(
         // extra connection to the node. We could _just_ check here, but checking up front skips the work to mark the connection as pending.
         final Transport.Connection existingConnectionRecheck = connectedNodes.get(node);
         if (existingConnectionRecheck != null) {
-            ListenableFuture<Transport.Connection> future = pendingConnections.remove(node);
+            var future = pendingConnections.remove(node);
             assert future == currentListener : "Listener in pending map is different than the expected listener";
             connectingRefCounter.decRef();
             future.onResponse(existingConnectionRecheck);
@@ -257,7 +257,7 @@ private void connectToNodeOrRetry(
                             }
                         }
                     } finally {
-                        ListenableFuture<Transport.Connection> future = pendingConnections.remove(node);
+                        var future = pendingConnections.remove(node);
                         assert future == currentListener : "Listener in pending map is different than the expected listener";
                         managerRefs.decRef();
                         releaseOnce.run();
@@ -387,9 +387,9 @@ private void failConnectionListener(
         DiscoveryNode node,
         RunOnce releaseOnce,
         Exception e,
-        ListenableFuture<Transport.Connection> expectedListener
+        SubscribableListener<Transport.Connection> expectedListener
     ) {
-        ListenableFuture<Transport.Connection> future = pendingConnections.remove(node);
+        final var future = pendingConnections.remove(node);
         releaseOnce.run();
         if (future != null) {
             assert future == expectedListener : "Listener in pending map is different than the expected listener";

test/framework/src/main/java/org/elasticsearch/bootstrap/TestBuildInfo.java

@@ -0,0 +1,14 @@
+/*
+ * 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.bootstrap;
+
+import java.util.List;
+
+record TestBuildInfo(String component, List<TestBuildInfoLocation> locations) {}

test/framework/src/main/java/org/elasticsearch/bootstrap/TestBuildInfoLocation.java

@@ -0,0 +1,12 @@
+/*
+ * 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.bootstrap;
+
+record TestBuildInfoLocation(String representativeClass, String module) {}

test/framework/src/main/java/org/elasticsearch/bootstrap/TestBuildInfoParser.java

@@ -0,0 +1,107 @@
+/*
+ * 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.bootstrap;
+
+import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.xcontent.ObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.XContentFactory;
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
+import org.elasticsearch.xcontent.XContentType;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+
+class TestBuildInfoParser {
+
+    private static final String PLUGIN_TEST_BUILD_INFO_RESOURCES = "META-INF/plugin-test-build-info.json";
+    private static final String SERVER_TEST_BUILD_INFO_RESOURCE = "META-INF/server-test-build-info.json";
+
+    private static final ObjectParser<Builder, Void> PARSER = new ObjectParser<>("test_build_info", Builder::new);
+    private static final ObjectParser<Location, Void> LOCATION_PARSER = new ObjectParser<>("location", Location::new);
+    static {
+        LOCATION_PARSER.declareString(Location::representativeClass, new ParseField("representativeClass"));
+        LOCATION_PARSER.declareString(Location::module, new ParseField("module"));
+
+        PARSER.declareString(Builder::component, new ParseField("component"));
+        PARSER.declareObjectArray(Builder::locations, LOCATION_PARSER, new ParseField("locations"));
+    }
+
+    private static class Location {
+        private String representativeClass;
+        private String module;
+
+        public void module(final String module) {
+            this.module = module;
+        }
+
+        public void representativeClass(final String representativeClass) {
+            this.representativeClass = representativeClass;
+        }
+    }
+
+    private static final class Builder {
+        private String component;
+        private List<Location> locations;
+
+        public void component(final String component) {
+            this.component = component;
+        }
+
+        public void locations(final List<Location> locations) {
+            this.locations = locations;
+        }
+
+        TestBuildInfo build() {
+            return new TestBuildInfo(
+                component,
+                locations.stream().map(l -> new TestBuildInfoLocation(l.representativeClass, l.module)).toList()
+            );
+        }
+    }
+
+    static TestBuildInfo fromXContent(final XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null).build();
+    }
+
+    static List<TestBuildInfo> parseAllPluginTestBuildInfo() throws IOException {
+        var xContent = XContentFactory.xContent(XContentType.JSON);
+        List<TestBuildInfo> pluginsTestBuildInfos = new ArrayList<>();
+        var resources = TestBuildInfoParser.class.getClassLoader().getResources(PLUGIN_TEST_BUILD_INFO_RESOURCES);
+        URL resource;
+        while ((resource = resources.nextElement()) != null) {
+            try (var stream = getStream(resource); var parser = xContent.createParser(XContentParserConfiguration.EMPTY, stream)) {
+                pluginsTestBuildInfos.add(fromXContent(parser));
+            }
+        }
+        return pluginsTestBuildInfos;
+    }
+
+    static TestBuildInfo parseServerTestBuildInfo() throws IOException {
+        var xContent = XContentFactory.xContent(XContentType.JSON);
+        var resource = TestBuildInfoParser.class.getClassLoader().getResource(SERVER_TEST_BUILD_INFO_RESOURCE);
+        // No test-build-info for server: this might be a non-gradle build. Proceed without TestBuildInfo
+        if (resource == null) {
+            return null;
+        }
+        try (var stream = getStream(resource); var parser = xContent.createParser(XContentParserConfiguration.EMPTY, stream)) {
+            return fromXContent(parser);
+        }
+    }
+
+    @SuppressForbidden(reason = "URLs from class loader")
+    private static InputStream getStream(URL resource) throws IOException {
+        return resource.openStream();
+    }
+}