[ML] Inference API add configurable connection pool TTL (elastic#127585)

* Adding ttl

* Using 60 seconds as default

Author: jonathan-buttner

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

@@ -278,13 +278,17 @@ public Collection<?> createComponents(PluginServices services) {
         var inferenceServices = new ArrayList<>(inferenceServiceExtensions);
         inferenceServices.add(this::getInferenceServiceFactories);
 
+        var inferenceServiceSettings = new ElasticInferenceServiceSettings(settings);
+        inferenceServiceSettings.init(services.clusterService());
+
         // Create a separate instance of HTTPClientManager with its own SSL configuration (`xpack.inference.elastic.http.ssl.*`).
         var elasticInferenceServiceHttpClientManager = HttpClientManager.create(
             settings,
             services.threadPool(),
             services.clusterService(),
             throttlerManager,
-            getSslService()
+            getSslService(),
+            inferenceServiceSettings.getConnectionTtl()
         );
 
         var elasticInferenceServiceRequestSenderFactory = new HttpRequestSender.Factory(
@@ -294,9 +298,6 @@ public Collection<?> createComponents(PluginServices services) {
         );
         elasicInferenceServiceFactory.set(elasticInferenceServiceRequestSenderFactory);
 
-        var inferenceServiceSettings = new ElasticInferenceServiceSettings(settings);
-        inferenceServiceSettings.init(services.clusterService());
-
         var authorizationHandler = new ElasticInferenceServiceAuthorizationRequestHandler(
             inferenceServiceSettings.getElasticInferenceServiceUrl(),
             services.threadPool()

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

@@ -32,6 +32,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.xpack.inference.services.elastic.ElasticInferenceServiceSettings.ELASTIC_INFERENCE_SERVICE_SSL_CONFIGURATION_PREFIX;
@@ -112,14 +113,15 @@ public static HttpClientManager create(
         ThreadPool threadPool,
         ClusterService clusterService,
         ThrottlerManager throttlerManager,
-        SSLService sslService
+        SSLService sslService,
+        TimeValue connectionTtl
     ) {
         // Set the sslStrategy to ensure an encrypted connection, as Elastic Inference Service requires it.
         SSLIOSessionStrategy sslioSessionStrategy = sslService.sslIOSessionStrategy(
             sslService.getSSLConfiguration(ELASTIC_INFERENCE_SERVICE_SSL_CONFIGURATION_PREFIX)
         );
 
-        PoolingNHttpClientConnectionManager connectionManager = createConnectionManager(sslioSessionStrategy);
+        PoolingNHttpClientConnectionManager connectionManager = createConnectionManager(sslioSessionStrategy, connectionTtl);
         return new HttpClientManager(settings, connectionManager, threadPool, clusterService, throttlerManager);
     }
 
@@ -146,7 +148,7 @@ public static HttpClientManager create(
         this.addSettingsUpdateConsumers(clusterService);
     }
 
-    private static PoolingNHttpClientConnectionManager createConnectionManager(SSLIOSessionStrategy sslStrategy) {
+    private static PoolingNHttpClientConnectionManager createConnectionManager(SSLIOSessionStrategy sslStrategy, TimeValue connectionTtl) {
         ConnectingIOReactor ioReactor;
         try {
             var configBuilder = IOReactorConfig.custom().setSoKeepAlive(true);
@@ -162,7 +164,15 @@ private static PoolingNHttpClientConnectionManager createConnectionManager(SSLIO
             .register("https", sslStrategy)
             .build();
 
-        return new PoolingNHttpClientConnectionManager(ioReactor, registry);
+        return new PoolingNHttpClientConnectionManager(
+            ioReactor,
+            null,
+            registry,
+            null,
+            null,
+            Math.toIntExact(connectionTtl.getMillis()),
+            TimeUnit.MILLISECONDS
+        );
     }
 
     private static PoolingNHttpClientConnectionManager createConnectionManager() {

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/elastic/ElasticInferenceServiceSettings.java

@@ -70,20 +70,33 @@ public class ElasticInferenceServiceSettings {
         Setting.Property.NodeScope
     );
 
+    /**
+     * Total time to live (TTL)  defines maximum life span of persistent connections regardless of their
+     * expiration setting. No persistent connection will be re-used past its TTL value.
+     * Using a TTL of -1 will disable the expiration of persistent connections (the idle connection evictor will still apply).
+     */
+    public static final Setting<TimeValue> CONNECTION_TTL_SETTING = Setting.timeSetting(
+        "xpack.inference.elastic.http.connection_ttl",
+        TimeValue.timeValueSeconds(60),
+        Setting.Property.NodeScope
+    );
+
     @Deprecated
     private final String eisGatewayUrl;
 
     private final String elasticInferenceServiceUrl;
     private final boolean periodicAuthorizationEnabled;
     private volatile TimeValue authRequestInterval;
     private volatile TimeValue maxAuthorizationRequestJitter;
+    private final TimeValue connectionTtl;
 
     public ElasticInferenceServiceSettings(Settings settings) {
         eisGatewayUrl = EIS_GATEWAY_URL.get(settings);
         elasticInferenceServiceUrl = ELASTIC_INFERENCE_SERVICE_URL.get(settings);
         periodicAuthorizationEnabled = PERIODIC_AUTHORIZATION_ENABLED.get(settings);
         authRequestInterval = AUTHORIZATION_REQUEST_INTERVAL.get(settings);
         maxAuthorizationRequestJitter = MAX_AUTHORIZATION_REQUEST_JITTER.get(settings);
+        connectionTtl = CONNECTION_TTL_SETTING.get(settings);
     }
 
     /**
@@ -115,6 +128,10 @@ public TimeValue getMaxAuthorizationRequestJitter() {
         return maxAuthorizationRequestJitter;
     }
 
+    public TimeValue getConnectionTtl() {
+        return connectionTtl;
+    }
+
     public static List<Setting<?>> getSettingsDefinitions() {
         ArrayList<Setting<?>> settings = new ArrayList<>();
         settings.add(EIS_GATEWAY_URL);
@@ -124,6 +141,7 @@ public static List<Setting<?>> getSettingsDefinitions() {
         settings.add(PERIODIC_AUTHORIZATION_ENABLED);
         settings.add(AUTHORIZATION_REQUEST_INTERVAL);
         settings.add(MAX_AUTHORIZATION_REQUEST_JITTER);
+        settings.add(CONNECTION_TTL_SETTING);
         return settings;
     }