[ML] Move code specific to the Elasticsearch in cluster services to those sevices (elastic#113749)

Remove the platform arch argument from parseRequest and move code
used by internal services out of the transport action into the service.

Author: davidkyle

muted-tests.yml

@@ -278,12 +278,9 @@ tests:
 - class: org.elasticsearch.xpack.ml.integration.MlJobIT
   method: testCreateJobsWithIndexNameOption
   issue: https://github.com/elastic/elasticsearch/issues/113528
-- class: org.elasticsearch.xpack.inference.TextEmbeddingCrudIT
-  method: testPutE5WithTrainedModelAndInference
-  issue: https://github.com/elastic/elasticsearch/issues/113565
-- class: org.elasticsearch.xpack.inference.TextEmbeddingCrudIT
-  method: testPutE5Small_withPlatformAgnosticVariant
-  issue: https://github.com/elastic/elasticsearch/issues/113577
+- class: org.elasticsearch.validation.DotPrefixClientYamlTestSuiteIT
+  method: test {p0=dot_prefix/10_basic/Deprecated index template with a dot prefix index pattern}
+  issue: https://github.com/elastic/elasticsearch/issues/113529
 - class: org.elasticsearch.xpack.ml.integration.MlJobIT
   method: testCantCreateJobWithSameID
   issue: https://github.com/elastic/elasticsearch/issues/113581

server/src/main/java/org/elasticsearch/inference/InferenceService.java

@@ -39,17 +39,9 @@ default void init(Client client) {}
      * @param modelId               Model Id
      * @param taskType              The model task type
      * @param config                Configuration options including the secrets
-     * @param platformArchitectures The Set of platform architectures (OS name and hardware architecture)
-     *                              the cluster nodes and models are running on.
      * @param parsedModelListener   A listener which will handle the resulting model or failure
      */
-    void parseRequestConfig(
-        String modelId,
-        TaskType taskType,
-        Map<String, Object> config,
-        Set<String> platformArchitectures,
-        ActionListener<Model> parsedModelListener
-    );
+    void parseRequestConfig(String modelId, TaskType taskType, Map<String, Object> config, ActionListener<Model> parsedModelListener);
 
     /**
      * Parse model configuration from {@code config map} from persisted storage and return the parsed {@link Model}. This requires that
@@ -155,17 +147,6 @@ default void putModel(Model modelVariant, ActionListener<Boolean> listener) {
         listener.onResponse(true);
     }
 
-    /**
-     * Checks if the modelId has been downloaded to the local Elasticsearch cluster using the trained models API
-     * The default action does nothing except acknowledge the request (false).
-     * Any internal services should Override this method.
-     * @param model
-     * @param listener The listener
-     */
-    default void isModelDownloaded(Model model, ActionListener<Boolean> listener) {
-        listener.onResponse(false);
-    };
-
     /**
      * Optionally test the new model configuration in the inference service.
      * This function should be called when the model is first created, the
@@ -188,14 +169,6 @@ default Model updateModelWithEmbeddingDetails(Model model, int embeddingSize) {
         return model;
     }
 
-    /**
-     * Return true if this model is hosted in the local Elasticsearch cluster
-     * @return True if in cluster
-     */
-    default boolean isInClusterService() {
-        return false;
-    }
-
     /**
      * Defines the version required across all clusters to use this service
      * @return {@link TransportVersion} specifying the version

server/src/main/java/org/elasticsearch/inference/InferenceServiceExtension.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.inference;
 
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.List;
 
@@ -20,7 +21,7 @@ public interface InferenceServiceExtension {
 
     List<Factory> getInferenceServiceFactories();
 
-    record InferenceServiceFactoryContext(Client client) {}
+    record InferenceServiceFactoryContext(Client client, ThreadPool threadPool) {}
 
     interface Factory {
         /**

x-pack/plugin/inference/qa/inference-service-tests/src/javaRestTest/java/org/elasticsearch/xpack/inference/InferenceBaseRestTest.java

@@ -53,7 +53,10 @@ protected String getTestRestCluster() {
     @Override
     protected Settings restClientSettings() {
         String token = basicAuthHeaderValue("x_pack_rest_user", new SecureString("x-pack-test-password".toCharArray()));
-        return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", token).build();
+        return Settings.builder()
+            .put(ThreadContext.PREFIX + ".Authorization", token)
+            .put(CLIENT_SOCKET_TIMEOUT, "120s")  // Long timeout for model download
+            .build();
     }
 
     static String mockSparseServiceModelConfig() {

x-pack/plugin/inference/qa/test-service-plugin/src/main/java/org/elasticsearch/xpack/inference/mock/TestDenseInferenceServiceExtension.java

@@ -38,7 +38,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 public class TestDenseInferenceServiceExtension implements InferenceServiceExtension {
     @Override
@@ -76,7 +75,6 @@ public void parseRequestConfig(
             String modelId,
             TaskType taskType,
             Map<String, Object> config,
-            Set<String> platformArchitectures,
             ActionListener<Model> parsedModelListener
         ) {
             var serviceSettingsMap = (Map<String, Object>) config.remove(ModelConfigurations.SERVICE_SETTINGS);

x-pack/plugin/inference/qa/test-service-plugin/src/main/java/org/elasticsearch/xpack/inference/mock/TestRerankingServiceExtension.java

@@ -34,7 +34,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 public class TestRerankingServiceExtension implements InferenceServiceExtension {
     @Override
@@ -67,7 +66,6 @@ public void parseRequestConfig(
             String modelId,
             TaskType taskType,
             Map<String, Object> config,
-            Set<String> platformArchitectures,
             ActionListener<Model> parsedModelListener
         ) {
             var serviceSettingsMap = (Map<String, Object>) config.remove(ModelConfigurations.SERVICE_SETTINGS);

x-pack/plugin/inference/qa/test-service-plugin/src/main/java/org/elasticsearch/xpack/inference/mock/TestSparseInferenceServiceExtension.java

@@ -37,7 +37,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 public class TestSparseInferenceServiceExtension implements InferenceServiceExtension {
     @Override
@@ -70,7 +69,6 @@ public void parseRequestConfig(
             String modelId,
             TaskType taskType,
             Map<String, Object> config,
-            Set<String> platformArchitectures,
             ActionListener<Model> parsedModelListener
         ) {
             var serviceSettingsMap = (Map<String, Object>) config.remove(ModelConfigurations.SERVICE_SETTINGS);

x-pack/plugin/inference/qa/test-service-plugin/src/main/java/org/elasticsearch/xpack/inference/mock/TestStreamingCompletionServiceExtension.java

@@ -67,7 +67,6 @@ public void parseRequestConfig(
             String modelId,
             TaskType taskType,
             Map<String, Object> config,
-            Set<String> platformArchitectures,
             ActionListener<Model> parsedModelListener
         ) {
             var serviceSettingsMap = (Map<String, Object>) config.remove(ModelConfigurations.SERVICE_SETTINGS);

x-pack/plugin/inference/src/internalClusterTest/java/org/elasticsearch/xpack/inference/integration/ModelRegistryIT.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.reindex.ReindexPlugin;
 import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xpack.inference.InferencePlugin;
@@ -117,7 +118,9 @@ public void testGetModel() throws Exception {
 
         assertEquals(model.getConfigurations().getService(), modelHolder.get().service());
 
-        var elserService = new ElserInternalService(new InferenceServiceExtension.InferenceServiceFactoryContext(mock(Client.class)));
+        var elserService = new ElserInternalService(
+            new InferenceServiceExtension.InferenceServiceFactoryContext(mock(Client.class), mock(ThreadPool.class))
+        );
         ElserInternalModel roundTripModel = elserService.parsePersistedConfigWithSecrets(
             modelHolder.get().inferenceEntityId(),
             modelHolder.get().taskType(),

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/InferencePlugin.java

@@ -206,7 +206,7 @@ public Collection<?> createComponents(PluginServices services) {
             );
         }
 
-        var factoryContext = new InferenceServiceExtension.InferenceServiceFactoryContext(services.client());
+        var factoryContext = new InferenceServiceExtension.InferenceServiceFactoryContext(services.client(), services.threadPool());
         // This must be done after the HttpRequestSenderFactory is created so that the services can get the
         // reference correctly
         var registry = new InferenceServiceRegistry(inferenceServices, factoryContext);
@@ -299,15 +299,17 @@ public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings sett
 
     @Override
     public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settingsToUse) {
-        return List.of(
-            new ScalingExecutorBuilder(
-                UTILITY_THREAD_POOL_NAME,
-                0,
-                10,
-                TimeValue.timeValueMinutes(10),
-                false,
-                "xpack.inference.utility_thread_pool"
-            )
+        return List.of(inferenceUtilityExecutor(settings));
+    }
+
+    public static ExecutorBuilder<?> inferenceUtilityExecutor(Settings settings) {
+        return new ScalingExecutorBuilder(
+            UTILITY_THREAD_POOL_NAME,
+            0,
+            10,
+            TimeValue.timeValueMinutes(10),
+            false,
+            "xpack.inference.utility_thread_pool"
         );
     }