SOLR-9877: Use instrumented http client and connection pool

solr/CHANGES.txt

@@ -195,6 +195,8 @@ New Features
 
 * SOLR-9725: Substitute properties into JdbcDataSource configuration ( Jamie Jackson, Yuri Sashevsky via Mikhail Khludnev)
 
+* SOLR-9877: Use instrumented http client and connection pool. (shalin)
+
 Optimizations
 ----------------------
 * SOLR-9704: Facet Module / JSON Facet API: Optimize blockChildren facets that have

solr/core/src/java/org/apache/solr/core/CoreContainer.java

@@ -460,10 +460,16 @@ public class CoreContainer {
       }
     }
 
+    metricManager = new SolrMetricManager();
 
     shardHandlerFactory = ShardHandlerFactory.newInstance(cfg.getShardHandlerFactoryPluginInfo(), loader);
+    if (shardHandlerFactory instanceof SolrMetricProducer) {
+      SolrMetricProducer metricProducer = (SolrMetricProducer) shardHandlerFactory;
+      metricProducer.initializeMetrics(metricManager, SolrInfoMBean.Group.http.toString(), "httpShardHandler");
+    }
 
     updateShardHandler = new UpdateShardHandler(cfg.getUpdateShardHandlerConfig());
+    updateShardHandler.initializeMetrics(metricManager, SolrInfoMBean.Group.http.toString(), "updateShardHandler");
 
     solrCores.allocateLazyCores(cfg.getTransientCacheSize(), loader);
 
@@ -476,8 +482,6 @@ public class CoreContainer {
 
     MDCLoggingContext.setNode(this);
 
-    metricManager = new SolrMetricManager();
-
     securityConfHandler = isZooKeeperAware() ? new SecurityConfHandlerZk(this) : new SecurityConfHandlerLocal(this);
     reloadSecurityProperties();
     this.backupRepoFactory = new BackupRepositoryFactory(cfg.getBackupRepositoryPlugins());

solr/core/src/java/org/apache/solr/handler/component/HttpShardHandlerFactory.java

@@ -35,14 +35,21 @@ import org.apache.solr.common.util.StrUtils;
 import org.apache.solr.common.util.URLUtil;
 import org.apache.solr.core.CoreDescriptor;
 import org.apache.solr.core.PluginInfo;
+import org.apache.solr.metrics.SolrMetricManager;
+import org.apache.solr.metrics.SolrMetricProducer;
 import org.apache.solr.update.UpdateShardHandlerConfig;
 import org.apache.solr.request.SolrQueryRequest;
 import org.apache.solr.util.DefaultSolrThreadFactory;
+import org.apache.solr.util.stats.InstrumentedHttpRequestExecutor;
+import org.apache.solr.util.stats.InstrumentedPoolingHttpClientConnectionManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.lang.invoke.MethodHandles;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
@@ -56,7 +63,7 @@ import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 
-public class HttpShardHandlerFactory extends ShardHandlerFactory implements org.apache.solr.util.plugin.PluginInfoInitialized {
+public class HttpShardHandlerFactory extends ShardHandlerFactory implements org.apache.solr.util.plugin.PluginInfoInitialized, SolrMetricProducer {
   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
   private static final String DEFAULT_SCHEME = "http";
   
@@ -74,7 +81,9 @@ public class HttpShardHandlerFactory extends ShardHandlerFactory implements org.
       new DefaultSolrThreadFactory("httpShardExecutor")
   );
 
+  protected InstrumentedPoolingHttpClientConnectionManager clientConnectionManager;
   protected CloseableHttpClient defaultClient;
+  protected InstrumentedHttpRequestExecutor httpRequestExecutor;
   private LBHttpSolrClient loadbalancer;
   //default values:
   int soTimeout = UpdateShardHandlerConfig.DEFAULT_DISTRIBUPDATESOTIMEOUT;
@@ -169,12 +178,12 @@ public class HttpShardHandlerFactory extends ShardHandlerFactory implements org.
     );
 
     ModifiableSolrParams clientParams = getClientParams();
-
-    this.defaultClient = HttpClientUtil.createClient(clientParams);
-    
+    httpRequestExecutor = new InstrumentedHttpRequestExecutor();
+    clientConnectionManager = new InstrumentedPoolingHttpClientConnectionManager(HttpClientUtil.getSchemaRegisteryProvider().getSchemaRegistry());
+    this.defaultClient = HttpClientUtil.createClient(clientParams, clientConnectionManager, false, httpRequestExecutor);
     this.loadbalancer = createLoadbalancer(defaultClient);
   }
-  
+
   protected ModifiableSolrParams getClientParams() {
     ModifiableSolrParams clientParams = new ModifiableSolrParams();
     clientParams.set(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, maxConnectionsPerHost);
@@ -219,6 +228,9 @@ public class HttpShardHandlerFactory extends ShardHandlerFactory implements org.
         if (defaultClient != null) {
           HttpClientUtil.close(defaultClient);
         }
+        if (clientConnectionManager != null)  {
+          clientConnectionManager.close();
+        }
       }
     }
   }
@@ -350,4 +362,47 @@ public class HttpShardHandlerFactory extends ShardHandlerFactory implements org.
     
     return url;
   }
+
+  @Override
+  public String getName() {
+    return this.getClass().getName();
+  }
+
+  @Override
+  public String getVersion() {
+    return getClass().getPackage().getSpecificationVersion();
+  }
+
+  @Override
+  public Collection<String> initializeMetrics(SolrMetricManager manager, String registry, String scope) {
+    List<String> metricNames = new ArrayList<>(4);
+    metricNames.addAll(clientConnectionManager.initializeMetrics(manager, registry, scope));
+    metricNames.addAll(httpRequestExecutor.initializeMetrics(manager, registry, scope));
+    return metricNames;
+  }
+
+  @Override
+  public String getDescription() {
+    return "Metrics tracked by HttpShardHandlerFactory for distributed query requests";
+  }
+
+  @Override
+  public Category getCategory() {
+    return Category.OTHER;
+  }
+
+  @Override
+  public String getSource() {
+    return null;
+  }
+
+  @Override
+  public URL[] getDocs() {
+    return new URL[0];
+  }
+
+  @Override
+  public NamedList getStatistics() {
+    return null;
+  }
 }

solr/core/src/java/org/apache/solr/update/UpdateShardHandler.java

@@ -17,6 +17,10 @@
 package org.apache.solr.update;
 
 import java.lang.invoke.MethodHandles;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 
 import org.apache.http.client.HttpClient;
@@ -27,11 +31,16 @@ import org.apache.solr.cloud.RecoveryStrategy;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
 import org.apache.solr.common.util.SolrjNamedThreadFactory;
+import org.apache.solr.metrics.SolrMetricManager;
+import org.apache.solr.metrics.SolrMetricProducer;
+import org.apache.solr.util.stats.InstrumentedHttpRequestExecutor;
+import org.apache.solr.util.stats.InstrumentedPoolingHttpClientConnectionManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class UpdateShardHandler {
+public class UpdateShardHandler implements SolrMetricProducer {
   
   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
@@ -50,10 +59,12 @@ public class UpdateShardHandler {
   
   private final CloseableHttpClient client;
 
-  private final PoolingHttpClientConnectionManager clientConnectionManager;
+  private final InstrumentedPoolingHttpClientConnectionManager clientConnectionManager;
+
+  private final InstrumentedHttpRequestExecutor httpRequestExecutor;
 
   public UpdateShardHandler(UpdateShardHandlerConfig cfg) {
-    clientConnectionManager = new PoolingHttpClientConnectionManager(HttpClientUtil.getSchemaRegisteryProvider().getSchemaRegistry());
+    clientConnectionManager = new InstrumentedPoolingHttpClientConnectionManager(HttpClientUtil.getSchemaRegisteryProvider().getSchemaRegistry());
     if (cfg != null ) {
       clientConnectionManager.setMaxTotal(cfg.getMaxUpdateConnections());
       clientConnectionManager.setDefaultMaxPerRoute(cfg.getMaxUpdateConnectionsPerHost());
@@ -64,7 +75,8 @@ public class UpdateShardHandler {
       clientParams.set(HttpClientUtil.PROP_SO_TIMEOUT, cfg.getDistributedSocketTimeout());
       clientParams.set(HttpClientUtil.PROP_CONNECTION_TIMEOUT, cfg.getDistributedConnectionTimeout());
     }
-    client = HttpClientUtil.createClient(clientParams, clientConnectionManager);
+    httpRequestExecutor = new InstrumentedHttpRequestExecutor();
+    client = HttpClientUtil.createClient(clientParams, clientConnectionManager, false, httpRequestExecutor);
 
     // following is done only for logging complete configuration.
     // The maxConnections and maxConnectionsPerHost have already been specified on the connection manager
@@ -74,7 +86,50 @@ public class UpdateShardHandler {
     }
     log.debug("Created UpdateShardHandler HTTP client with params: {}", clientParams);
   }
-  
+
+  @Override
+  public String getName() {
+    return this.getClass().getName();
+  }
+
+  @Override
+  public String getVersion() {
+    return getClass().getPackage().getSpecificationVersion();
+  }
+
+  @Override
+  public Collection<String> initializeMetrics(SolrMetricManager manager, String registry, String scope) {
+    List<String> metricNames = new ArrayList<>(4);
+    metricNames.addAll(clientConnectionManager.initializeMetrics(manager, registry, scope));
+    metricNames.addAll(httpRequestExecutor.initializeMetrics(manager, registry, scope));
+    return metricNames;
+  }
+
+  @Override
+  public String getDescription() {
+    return "Metrics tracked by UpdateShardHandler for ";
+  }
+
+  @Override
+  public Category getCategory() {
+    return null;
+  }
+
+  @Override
+  public String getSource() {
+    return null;
+  }
+
+  @Override
+  public URL[] getDocs() {
+    return new URL[0];
+  }
+
+  @Override
+  public NamedList getStatistics() {
+    return null;
+  }
+
   public HttpClient getHttpClient() {
     return client;
   }

solr/core/src/java/org/apache/solr/util/stats/InstrumentedHttpRequestExecutor.java

@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.util.stats;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Locale;
+
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Timer;
+import org.apache.http.HttpClientConnection;
+import org.apache.http.HttpException;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.RequestLine;
+import org.apache.http.client.methods.HttpRequestWrapper;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.protocol.HttpContext;
+import org.apache.http.protocol.HttpRequestExecutor;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.metrics.SolrMetricManager;
+import org.apache.solr.metrics.SolrMetricProducer;
+
+/**
+ * Sub-class of HttpRequestExecutor which tracks metrics interesting to solr
+ * Inspired and partially copied from dropwizard httpclient library
+ */
+public class InstrumentedHttpRequestExecutor extends HttpRequestExecutor implements SolrMetricProducer {
+  protected MetricRegistry metricsRegistry;
+  protected String scope;
+
+  private static String methodNameString(HttpRequest request) {
+    return request.getRequestLine().getMethod().toLowerCase(Locale.ROOT) + "-requests";
+  }
+
+  @Override
+  public HttpResponse execute(HttpRequest request, HttpClientConnection conn, HttpContext context) throws IOException, HttpException {
+    assert metricsRegistry != null;
+    final Timer.Context timerContext = timer(request).time();
+    try {
+      return super.execute(request, conn, context);
+    } finally {
+      timerContext.stop();
+    }
+  }
+
+  private Timer timer(HttpRequest request) {
+    return metricsRegistry.timer(getNameFor(request));
+  }
+
+  @Override
+  public String getName() {
+    return this.getClass().getName();
+  }
+
+  @Override
+  public String getVersion() {
+    return getClass().getPackage().getSpecificationVersion();
+  }
+
+  @Override
+  public Collection<String> initializeMetrics(SolrMetricManager manager, String registry, String scope) {
+    this.metricsRegistry = manager.registry(registry);
+    this.scope = scope;
+    return Collections.emptyList(); // we do not know the names of the metrics yet
+  }
+
+  @Override
+  public String getDescription() {
+    return null;
+  }
+
+  @Override
+  public Category getCategory() {
+    return Category.OTHER;
+  }
+
+  @Override
+  public String getSource() {
+    return null;
+  }
+
+  @Override
+  public URL[] getDocs() {
+    return null;
+  }
+
+  @Override
+  public NamedList getStatistics() {
+    return null;
+  }
+
+  private String getNameFor(HttpRequest request) {
+    try {
+      final RequestLine requestLine = request.getRequestLine();
+      String schemeHostPort = null;
+      if (request instanceof HttpRequestWrapper) {
+        HttpRequestWrapper wrapper = (HttpRequestWrapper) request;
+        schemeHostPort = wrapper.getTarget().getSchemeName() + "://" + wrapper.getTarget().getHostName() + ":" +  wrapper.getTarget().getPort();
+      }
+      final URIBuilder url = new URIBuilder(requestLine.getUri());
+      return SolrMetricManager.mkName((schemeHostPort != null ? schemeHostPort : "") + url.removeQuery().build().toString() + "." + methodNameString(request), scope);
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
+}

solr/core/src/java/org/apache/solr/util/stats/InstrumentedPoolingHttpClientConnectionManager.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.util.stats;
+
+import java.net.URL;
+import java.util.Arrays;
+import java.util.Collection;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.MetricRegistry;
+import org.apache.http.config.Registry;
+import org.apache.http.conn.socket.ConnectionSocketFactory;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.metrics.SolrMetricManager;
+import org.apache.solr.metrics.SolrMetricProducer;
+
+/**
+ * Sub-class of PoolingHttpClientConnectionManager which tracks metrics interesting to Solr.
+ * Inspired by dropwizard metrics-httpclient library implementation.
+ */
+public class InstrumentedPoolingHttpClientConnectionManager extends PoolingHttpClientConnectionManager implements SolrMetricProducer {
+
+  protected MetricRegistry metricsRegistry;
+
+  public InstrumentedPoolingHttpClientConnectionManager(Registry<ConnectionSocketFactory> socketFactoryRegistry) {
+    super(socketFactoryRegistry);
+  }
+
+  public MetricRegistry getMetricsRegistry() {
+    return metricsRegistry;
+  }
+
+  public void setMetricsRegistry(MetricRegistry metricRegistry) {
+    this.metricsRegistry = metricRegistry;
+  }
+
+  @Override
+  public String getName() {
+    return this.getClass().getName();
+  }
+
+  @Override
+  public String getVersion() {
+    return getClass().getPackage().getSpecificationVersion();
+  }
+
+  @Override
+  public Collection<String> initializeMetrics(SolrMetricManager manager, String registry, String scope) {
+    this.metricsRegistry = manager.registry(registry);
+    metricsRegistry.register(SolrMetricManager.mkName("availableConnections", scope),
+        (Gauge<Integer>) () -> {
+          // this acquires a lock on the connection pool; remove if contention sucks
+          return getTotalStats().getAvailable();
+        });
+    metricsRegistry.register(SolrMetricManager.mkName("leasedConnections", scope),
+        (Gauge<Integer>) () -> {
+          // this acquires a lock on the connection pool; remove if contention sucks
+          return getTotalStats().getLeased();
+        });
+    metricsRegistry.register(SolrMetricManager.mkName("maxConnections", scope),
+        (Gauge<Integer>) () -> {
+          // this acquires a lock on the connection pool; remove if contention sucks
+          return getTotalStats().getMax();
+        });
+    metricsRegistry.register(SolrMetricManager.mkName("pendingConnections", scope),
+        (Gauge<Integer>) () -> {
+          // this acquires a lock on the connection pool; remove if contention sucks
+          return getTotalStats().getPending();
+        });
+    return Arrays.asList("availableConnections", "leasedConnections", "maxConnections", "pendingConnections");
+  }
+
+  @Override
+  public String getDescription() {
+    return "";
+  }
+
+  @Override
+  public Category getCategory() {
+    return Category.OTHER;
+  }
+
+  @Override
+  public String getSource() {
+    return null;
+  }
+
+  @Override
+  public URL[] getDocs() {
+    return null;
+  }
+
+  @Override
+  public NamedList getStatistics() {
+    return null;
+  }
+}

solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpClientUtil.java

@@ -55,6 +55,7 @@ import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
 import org.apache.http.impl.client.HttpClientBuilder;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.apache.http.protocol.HttpContext;
+import org.apache.http.protocol.HttpRequestExecutor;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.params.SolrParams;
 import org.apache.solr.common.util.ObjectReleaseTracker;
@@ -213,22 +214,18 @@ public class HttpClientUtil {
     
     return createClient(params, cm, false);
   }
-  
-  /**
-   * Creates new http client by using the provided configuration.
-   * 
-   */
-  public static CloseableHttpClient createClient(final SolrParams params, PoolingHttpClientConnectionManager cm, boolean sharedConnectionManager) {
+
+  public static CloseableHttpClient createClient(final SolrParams params, PoolingHttpClientConnectionManager cm, boolean sharedConnectionManager, HttpRequestExecutor httpRequestExecutor)  {
     final ModifiableSolrParams config = new ModifiableSolrParams(params);
     if (logger.isDebugEnabled()) {
       logger.debug("Creating new http client, config:" + config);
     }
- 
+
     cm.setMaxTotal(params.getInt(HttpClientUtil.PROP_MAX_CONNECTIONS, 10000));
     cm.setDefaultMaxPerRoute(params.getInt(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, 10000));
     cm.setValidateAfterInactivity(Integer.getInteger(VALIDATE_AFTER_INACTIVITY, VALIDATE_AFTER_INACTIVITY_DEFAULT));
 
-    
+
     HttpClientBuilder newHttpClientBuilder = HttpClientBuilder.create();
 
     if (sharedConnectionManager) {
@@ -236,7 +233,7 @@ public class HttpClientUtil {
     } else {
       newHttpClientBuilder.setConnectionManagerShared(false);
     }
-    
+
     ConnectionKeepAliveStrategy keepAliveStrat = new ConnectionKeepAliveStrategy() {
       @Override
       public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
@@ -256,18 +253,30 @@ public class HttpClientUtil {
     }
 
     newHttpClientBuilder.addInterceptorLast(new DynamicInterceptor());
-    
+
     newHttpClientBuilder = newHttpClientBuilder.setKeepAliveStrategy(keepAliveStrat)
         .evictIdleConnections((long) Integer.getInteger(EVICT_IDLE_CONNECTIONS, EVICT_IDLE_CONNECTIONS_DEFAULT), TimeUnit.MILLISECONDS);
-    
+
+    if (httpRequestExecutor != null)  {
+      newHttpClientBuilder.setRequestExecutor(httpRequestExecutor);
+    }
+
     HttpClientBuilder builder = setupBuilder(newHttpClientBuilder, params);
-    
+
     HttpClient httpClient = builder.setConnectionManager(cm).build();
-    
+
     assert ObjectReleaseTracker.track(httpClient);
     return (CloseableHttpClient) httpClient;
   }
   
+  /**
+   * Creates new http client by using the provided configuration.
+   * 
+   */
+  public static CloseableHttpClient createClient(final SolrParams params, PoolingHttpClientConnectionManager cm, boolean sharedConnectionManager) {
+    return createClient(params, cm, sharedConnectionManager, null);
+  }
+  
   private static HttpClientBuilder setupBuilder(HttpClientBuilder builder, SolrParams config) {
    
     Builder requestConfigBuilder = RequestConfig.custom()