Improve scalability of get-license action (#134457)

Today this action runs on the transport worker thread and forwards the
request on to the master by default. It turns out that Elastic Agent
uses this API as a readiness check whenever opening a new connection, so
a thundering herd of 1000s of agents can prevent the transport worker
threads from doing more useful work for far too long, leading to high
latency and timeouts.

This commit changes the default behaviour to run the action on the local
node rather than forwarding to the master (although the option remains
to specify `?local=false`) and dispatches the work off of the transport
worker early.

Author: DaveCTurner

docs/changelog/134457.yaml

@@ -0,0 +1,5 @@
+pr: 134457
+summary: Improve scalability of get-license action
+area: License
+type: enhancement
+issues: []

x-pack/plugin/core/src/main/java/org/elasticsearch/license/RestGetLicenseAction.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.license;
 
+import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.logging.DeprecationLogger;
@@ -20,6 +21,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestBuilderListener;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 
@@ -77,9 +79,10 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         final ToXContent.Params params = new ToXContent.DelegatingMapParams(overrideParams, request);
         GetLicenseRequest getLicenseRequest = new GetLicenseRequest(RestUtils.getMasterNodeTimeout(request));
         getLicenseRequest.local(request.paramAsBoolean("local", getLicenseRequest.local()));
-        return channel -> client.admin()
-            .cluster()
-            .execute(GetLicenseAction.INSTANCE, getLicenseRequest, new RestBuilderListener<>(channel) {
+        return channel -> client.threadPool()
+            .executor(ThreadPool.Names.MANAGEMENT)
+            // dispatching to MANAGEMENT here as a workaround for https://github.com/elastic/elasticsearch/issues/97916
+            .execute(ActionRunnable.wrap(new RestBuilderListener<GetLicenseResponse>(channel) {
                 @Override
                 public RestResponse buildResponse(GetLicenseResponse response, XContentBuilder builder) throws Exception {
                     // Default to pretty printing, but allow ?pretty=false to disable
@@ -96,7 +99,7 @@ public RestResponse buildResponse(GetLicenseResponse response, XContentBuilder b
                     builder.endObject();
                     return new RestResponse(hasLicense ? OK : NOT_FOUND, builder);
                 }
-            });
+            }, responseListener -> client.admin().cluster().execute(GetLicenseAction.INSTANCE, getLicenseRequest, responseListener)));
     }
 
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/license/TransportGetLicenseAction.java

@@ -14,7 +14,6 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.protocol.xpack.license.GetLicenseRequest;
 import org.elasticsearch.tasks.Task;
@@ -41,7 +40,7 @@ public TransportGetLicenseAction(
             actionFilters,
             GetLicenseRequest::new,
             GetLicenseResponse::new,
-            EsExecutors.DIRECT_EXECUTOR_SERVICE
+            threadPool.executor(ThreadPool.Names.MANAGEMENT)
         );
         this.licenseService = licenseService;
     }
@@ -58,6 +57,7 @@ protected void masterOperation(
         ClusterState state,
         final ActionListener<GetLicenseResponse> listener
     ) throws ElasticsearchException {
+        assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.MANAGEMENT);
         if (licenseService instanceof ClusterStateLicenseService clusterStateLicenseService) {
             listener.onResponse(new GetLicenseResponse(clusterStateLicenseService.getLicense(state.metadata())));
         } else {

x-pack/plugin/core/src/main/java/org/elasticsearch/protocol/xpack/license/GetLicenseRequest.java

@@ -17,6 +17,7 @@ public class GetLicenseRequest extends MasterNodeReadRequest<GetLicenseRequest>
 
     public GetLicenseRequest(TimeValue masterNodeTimeout) {
         super(masterNodeTimeout);
+        local = true; // default to a local response, no need to forward to the master in most cases
     }
 
     public GetLicenseRequest(StreamInput in) throws IOException {