[ML] Transition EIS auth polling to persistent task on a single node

docs/changelog/136713.yaml

@@ -0,0 +1,5 @@
+pr: 136713
+summary: Transition EIS auth polling to persistent task on a single node
+area: Machine Learning
+type: enhancement
+issues: []

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

@@ -9,9 +9,14 @@
 
 package org.elasticsearch.inference;
 
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+
+import java.io.IOException;
 import java.util.Objects;
 
-public class Model {
+public class Model implements Writeable {
     public static String documentId(String modelId) {
         return "model_" + modelId;
     }
@@ -42,6 +47,11 @@ public Model(ModelConfigurations configurations) {
         this(configurations, new ModelSecrets());
     }
 
+    public Model(StreamInput in) throws IOException {
+        this.configurations = new ModelConfigurations(in);
+        this.secrets = new ModelSecrets(in);
+    }
+
     public String getInferenceEntityId() {
         return configurations.getInferenceEntityId();
     }
@@ -111,4 +121,10 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(configurations, secrets);
     }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        configurations.writeTo(out);
+        secrets.writeTo(out);
+    }
 }

server/src/main/resources/transport/definitions/referable/inference_api_eis_authorization_persistent_task.csv

@@ -0,0 +1 @@
+9205000

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-esql_resolve_fields_response_used,9204000
+inference_api_eis_authorization_persistent_task,9205000

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/StoreInferenceEndpointsAction.java

@@ -0,0 +1,105 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.inference.action;
+
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.inference.Model;
+import org.elasticsearch.xpack.core.inference.results.ModelStoreResponse;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Internal action to store inference endpoints and return the results of the store operation. This should only be used internally and not
+ * exposed via a REST API.
+ * For the exposed REST API action see {@link PutInferenceModelAction}.
+ */
+public class StoreInferenceEndpointsAction extends ActionType<StoreInferenceEndpointsAction.Response> {
+
+    public static final StoreInferenceEndpointsAction INSTANCE = new StoreInferenceEndpointsAction();
+    public static final String NAME = "cluster:internal/xpack/inference/create_endpoints";
+
+    public StoreInferenceEndpointsAction() {
+        super(NAME);
+    }
+
+    public static class Request extends AcknowledgedRequest<Request> {
+        private final List<Model> models;
+
+        public Request(List<Model> models, TimeValue timeout) {
+            super(timeout, DEFAULT_ACK_TIMEOUT);
+            this.models = Objects.requireNonNull(models);
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            models = in.readCollectionAsImmutableList(Model::new);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeCollection(models);
+        }
+
+        public List<Model> getModels() {
+            return models;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (o == null || getClass() != o.getClass()) return false;
+            Request request = (Request) o;
+            return Objects.equals(models, request.models);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hashCode(models);
+        }
+    }
+
+    public static class Response extends ActionResponse {
+        private final List<ModelStoreResponse> results;
+
+        public Response(List<ModelStoreResponse> results) {
+            this.results = results;
+        }
+
+        public Response(StreamInput in) throws IOException {
+            results = in.readCollectionAsImmutableList(ModelStoreResponse::new);
+        }
+
+        public List<ModelStoreResponse> getResults() {
+            return results;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeCollection(results);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (o == null || getClass() != o.getClass()) return false;
+            Response response = (Response) o;
+            return Objects.equals(results, response.results);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(results);
+        }
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/results/ModelStoreResponse.java

@@ -0,0 +1,65 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.inference.results;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.rest.RestStatus;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Response for storing a model in the model registry using the bulk API.
+ */
+public record ModelStoreResponse(String inferenceId, RestStatus status, @Nullable Exception failureCause) implements Writeable {
+
+    public ModelStoreResponse(StreamInput in) throws IOException {
+        this(in.readString(), RestStatus.readFrom(in), in.readException());
+    }
+
+    public boolean failed() {
+        return failureCause != null;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(inferenceId);
+        RestStatus.writeTo(out, status);
+        out.writeException(failureCause);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        ModelStoreResponse that = (ModelStoreResponse) o;
+        return status == that.status && Objects.equals(inferenceId, that.inferenceId)
+        // Exception does not have hashCode() or equals() so assume errors are equal iff class and message are equal
+            && Objects.equals(
+                failureCause == null ? null : failureCause.getMessage(),
+                that.failureCause == null ? null : that.failureCause.getMessage()
+            )
+            && Objects.equals(
+                failureCause == null ? null : failureCause.getClass(),
+                that.failureCause == null ? null : that.failureCause.getClass()
+            );
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(
+            inferenceId,
+            status,
+            // Exception does not have hashCode() or equals() so assume errors are equal iff class and message are equal
+            failureCause == null ? null : failureCause.getMessage(),
+            failureCause == null ? null : failureCause.getClass()
+        );
+    }
+}