[Inference API] Remove worst-case additional 50ms latency for non-rate limited requests (#136167)

Author: timgrein

docs/changelog/136167.yaml

@@ -0,0 +1,6 @@
+pr: 136167
+summary: "[Inference API] Remove worst-case additional 50ms latency for non-rate limited\
+  \ requests"
+area: Machine Learning
+type: bug
+issues: []

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

@@ -61,6 +61,10 @@ public static boolean isInternalTypeOrUnspecified(InputType inputType) {
         return inputType == InputType.INTERNAL_INGEST || inputType == InputType.INTERNAL_SEARCH || inputType == InputType.UNSPECIFIED;
     }
 
+    public static boolean isIngest(InputType inputType) {
+        return inputType == InputType.INGEST || inputType == InputType.INTERNAL_INGEST;
+    }
+
     public static boolean isSpecified(InputType inputType) {
         return inputType != null && inputType != InputType.UNSPECIFIED;
     }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/http/sender/RequestExecutorService.java

@@ -125,7 +125,7 @@ public RateLimitSettings(long requestsPerTimeUnit, TimeUnit timeUnit) {
     }
 
     // This should only be used for testing.
-    RateLimitSettings(long requestsPerTimeUnit, TimeUnit timeUnit, boolean enabled) {
+    public RateLimitSettings(long requestsPerTimeUnit, TimeUnit timeUnit, boolean enabled) {
         if (requestsPerTimeUnit <= 0) {
             throw new IllegalArgumentException("requests per minute must be positive");
         }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/settings/RateLimitSettings.java

@@ -64,6 +64,10 @@ public static InputType randomWithInternalAndUnspecified() {
         return randomFrom(InputType.INTERNAL_SEARCH, InputType.INTERNAL_INGEST, InputType.UNSPECIFIED);
     }
 
+    public static InputType randomIngest() {
+        return randomFrom(InputType.INGEST, InputType.INTERNAL_INGEST);
+    }
+
     public void testFromRestString_ValidInputType() {
         for (String internal : List.of("search", "ingest", "classification", "clustering", "unspecified")) {
             assertEquals(InputType.fromRestString(internal), InputType.fromString(internal));
@@ -211,4 +215,8 @@ public void testValidateInputTypeTranslationValues_ThrowsAnException_WhenValueIs
             )
         );
     }
+
+    public void testIsIngest() {
+        assertTrue(InputType.isIngest(randomFrom(InputType.INGEST, InputType.INTERNAL_INGEST)));
+    }
 }

x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/InputTypeTests.java

@@ -354,7 +354,12 @@ public void testHttpRequestSender_Throws_WhenCallingSendBeforeStart() throws Exc
             PlainActionFuture<InferenceServiceResults> listener = new PlainActionFuture<>();
             var thrownException = expectThrows(
                 AssertionError.class,
-                () -> sender.send(RequestManagerTests.createMock(), new EmbeddingsInput(List.of(), null), null, listener)
+                () -> sender.send(
+                    RequestManagerTests.createMockWithRateLimitingEnabled(),
+                    new EmbeddingsInput(List.of(), null),
+                    null,
+                    listener
+                )
             );
             assertThat(thrownException.getMessage(), is("call start() before sending a request"));
         }
@@ -375,7 +380,12 @@ public void testHttpRequestSender_Throws_WhenATimeoutOccurs() throws Exception {
             sender.startSynchronously();
 
             PlainActionFuture<InferenceServiceResults> listener = new PlainActionFuture<>();
-            sender.send(RequestManagerTests.createMock(), new EmbeddingsInput(List.of(), null), TimeValue.timeValueNanos(1), listener);
+            sender.send(
+                RequestManagerTests.createMockWithRateLimitingEnabled(),
+                new EmbeddingsInput(List.of(), null),
+                TimeValue.timeValueNanos(1),
+                listener
+            );
 
             var thrownException = expectThrows(ElasticsearchStatusException.class, () -> listener.actionGet(TIMEOUT));
 
@@ -397,7 +407,12 @@ public void testHttpRequestSenderWithTimeout_Throws_WhenATimeoutOccurs() throws
             sender.startSynchronously();
 
             PlainActionFuture<InferenceServiceResults> listener = new PlainActionFuture<>();
-            sender.send(RequestManagerTests.createMock(), new EmbeddingsInput(List.of(), null), TimeValue.timeValueNanos(1), listener);
+            sender.send(
+                RequestManagerTests.createMockWithRateLimitingEnabled(),
+                new EmbeddingsInput(List.of(), null),
+                TimeValue.timeValueNanos(1),
+                listener
+            );
 
             var thrownException = expectThrows(ElasticsearchStatusException.class, () -> listener.actionGet(TIMEOUT));
 

x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/external/http/sender/HttpRequestSenderTests.java

@@ -15,26 +15,36 @@
 import org.elasticsearch.xpack.inference.external.request.RequestTests;
 import org.elasticsearch.xpack.inference.services.settings.RateLimitSettings;
 
+import java.util.concurrent.TimeUnit;
+
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class RequestManagerTests {
-    public static RequestManager createMock() {
-        return createMock(mock(RequestSender.class));
+    public static RequestManager createMockWithRateLimitingDisabled(RequestSender requestSender, String inferenceEntityId) {
+        return createMock(requestSender, inferenceEntityId, new RateLimitSettings(1, TimeUnit.MINUTES, false));
+    }
+
+    public static RequestManager createMockWithRateLimitingDisabled(String inferenceEntityId) {
+        return createMock(mock(RequestSender.class), inferenceEntityId, new RateLimitSettings(1, TimeUnit.MINUTES, false));
+    }
+
+    public static RequestManager createMockWithRateLimitingEnabled() {
+        return createMockWithRateLimitingEnabled(mock(RequestSender.class));
     }
 
-    public static RequestManager createMock(String inferenceEntityId) {
-        return createMock(mock(RequestSender.class), inferenceEntityId);
+    public static RequestManager createMockWithRateLimitingEnabled(String inferenceEntityId) {
+        return createMockWithRateLimitingEnabled(mock(RequestSender.class), inferenceEntityId);
     }
 
-    public static RequestManager createMock(RequestSender requestSender) {
-        return createMock(requestSender, "id", new RateLimitSettings(1));
+    public static RequestManager createMockWithRateLimitingEnabled(RequestSender requestSender) {
+        return createMock(requestSender, "id", new RateLimitSettings(1, TimeUnit.MINUTES, true));
     }
 
-    public static RequestManager createMock(RequestSender requestSender, String inferenceEntityId) {
-        return createMock(requestSender, inferenceEntityId, new RateLimitSettings(1));
+    public static RequestManager createMockWithRateLimitingEnabled(RequestSender requestSender, String inferenceEntityId) {
+        return createMock(requestSender, inferenceEntityId, new RateLimitSettings(1, TimeUnit.MINUTES, true));
     }
 
     public static RequestManager createMock(RequestSender requestSender, String inferenceEntityId, RateLimitSettings settings) {