SOLR-16968: The MemoryCircuitBreaker now uses average heap usage

(rebased)

solr/CHANGES.txt

@@ -123,6 +123,8 @@ Improvements
 
 * SOLR-16970: SOLR_OPTS is now able to override options set by the Solr control scripts, "bin/solr" and "bin/solr.cmd". (Houston Putman)
 
+* SOLR-16968: The MemoryCircuitBreaker now uses average heap usage over the last 30 seconds (janhoy, Christine Poerschke)
+
 * SOLR-14886: Suppress stack traces in query response (Isabelle Giguere via Alex Deparvu)
 
 * SOLR-16461: `/solr/coreName/replication?command=backup` now has a v2 equivalent, available at

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

@@ -1088,9 +1088,6 @@ private SolrCore(
       solrMetricsContext = coreMetricManager.getSolrMetricsContext();
       this.coreMetricManager.loadReporters();
 
-      // init pluggable circuit breakers
-      initPlugins(null, CircuitBreaker.class);
-
       if (updateHandler == null) {
         directoryFactory = initDirectoryFactory();
         recoveryStrategyBuilder = initRecoveryStrategyBuilder();
@@ -1115,6 +1112,9 @@ private SolrCore(
       // initialize core metrics
       initializeMetrics(solrMetricsContext, null);
 
+      // init pluggable circuit breakers, after metrics because some circuit breakers use metrics
+      initPlugins(null, CircuitBreaker.class);
+
       SolrFieldCacheBean solrFieldCacheBean = new SolrFieldCacheBean();
       // this is registered at the CONTAINER level because it's not core-specific - for now we
       // also register it here for back-compat
@@ -1764,6 +1764,17 @@ private void doClose() {
 
     ExecutorUtil.shutdownAndAwaitTermination(coreAsyncTaskExecutor);
 
+    // Close circuit breakers that may have background threads, before metrics because some circuit
+    // breakers use metrics
+    try {
+      getCircuitBreakerRegistry().close();
+    } catch (Throwable e) {
+      log.error("Exception closing circuit breakers", e);
+      if (e instanceof Error) {
+        throw (Error) e;
+      }
+    }
+
     // stop reporting metrics
     try {
       coreMetricManager.close();

solr/core/src/java/org/apache/solr/util/circuitbreaker/AveragingMetricProvider.java

@@ -0,0 +1,80 @@
+/*
+ * 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.circuitbreaker;
+
+import com.google.common.util.concurrent.AtomicDouble;
+import java.io.Closeable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.logging.CircularList;
+
+/** Averages the metric value over a period of time */
+public class AveragingMetricProvider implements Closeable {
+  private final CircularList<Double> samplesRingBuffer;
+  private ScheduledExecutorService executor;
+  private final AtomicDouble currentAverageValue = new AtomicDouble(-1);
+
+  /**
+   * Creates an instance with an executor that runs every sampleInterval seconds and averages over
+   * numSamples samples.
+   *
+   * @param metricProvider metric provider that will provide a value
+   * @param numSamples number of samples to calculate average for
+   * @param sampleInterval interval between each sample
+   */
+  public AveragingMetricProvider(
+      MetricProvider metricProvider, int numSamples, long sampleInterval) {
+    this.samplesRingBuffer = new CircularList<>(numSamples);
+    executor =
+        Executors.newSingleThreadScheduledExecutor(
+            new SolrNamedThreadFactory(
+                "AveragingMetricProvider-" + metricProvider.getClass().getSimpleName()));
+    executor.scheduleWithFixedDelay(
+        () -> {
+          samplesRingBuffer.add(metricProvider.getMetricValue());
+          currentAverageValue.set(
+              samplesRingBuffer.toList().stream()
+                  .mapToDouble(Double::doubleValue)
+                  .average()
+                  .orElse(-1));
+        },
+        0,
+        sampleInterval,
+        TimeUnit.SECONDS);
+  }
+
+  /**
+   * Return current average. This is a cached value, so calling this method will not incur any
+   * calculations
+   */
+  public double getMetricValue() {
+    return currentAverageValue.get();
+  }
+
+  @Override
+  public void close() {
+    ExecutorUtil.shutdownAndAwaitTermination(executor);
+  }
+
+  /** Interface to provide the metric value. */
+  public interface MetricProvider {
+    double getMetricValue();
+  }
+}

solr/core/src/java/org/apache/solr/util/circuitbreaker/CircuitBreaker.java

@@ -17,6 +17,8 @@
 
 package org.apache.solr.util.circuitbreaker;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
@@ -41,7 +43,7 @@
  *
  * @lucene.experimental
  */
-public abstract class CircuitBreaker implements NamedListInitializedPlugin {
+public abstract class CircuitBreaker implements NamedListInitializedPlugin, Closeable {
   // Only query requests are checked by default
   private Set<SolrRequestType> requestTypes = Set.of(SolrRequestType.QUERY);
   private final List<SolrRequestType> SUPPORTED_TYPES =
@@ -60,6 +62,11 @@ public CircuitBreaker() {}
   /** Get error message when the circuit breaker triggers */
   public abstract String getErrorMessage();
 
+  @Override
+  public void close() throws IOException {
+    // Nothing to do by default
+  }
+
   /**
    * Set the request types for which this circuit breaker should be checked. If not called, the
    * circuit breaker will be checked for the {@link SolrRequestType#QUERY} request type only.

solr/core/src/java/org/apache/solr/util/circuitbreaker/CircuitBreakerManager.java

@@ -17,6 +17,7 @@
 
 package org.apache.solr.util.circuitbreaker;
 
+import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 import org.apache.solr.common.util.NamedList;
 import org.slf4j.Logger;
@@ -76,6 +77,19 @@ public void init(NamedList<?> args) {
     }
   }
 
+  @Override
+  public void close() throws IOException {
+    try {
+      if (memEnabled) {
+        memCB.close();
+      }
+    } finally {
+      if (cpuEnabled) {
+        cpuCB.close();
+      }
+    }
+  }
+
   // The methods below will be called by super class during init
   public void setMemEnabled(String enabled) {
     this.memEnabled = Boolean.getBoolean(enabled);

solr/core/src/java/org/apache/solr/util/circuitbreaker/CircuitBreakerRegistry.java

@@ -18,12 +18,19 @@
 package org.apache.solr.util.circuitbreaker;
 
 import com.google.common.annotations.VisibleForTesting;
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.solr.client.solrj.SolrRequest.SolrRequestType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Keeps track of all registered circuit breaker instances for various request types. Responsible
@@ -32,26 +39,35 @@
  * @lucene.experimental
  * @since 9.4
  */
-public class CircuitBreakerRegistry {
+public class CircuitBreakerRegistry implements Closeable {
+  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
   private final Map<SolrRequestType, List<CircuitBreaker>> circuitBreakerMap = new HashMap<>();
 
   public CircuitBreakerRegistry() {}
 
   public void register(CircuitBreaker circuitBreaker) {
-    circuitBreaker
-        .getRequestTypes()
-        .forEach(
-            r -> {
-              List<CircuitBreaker> list =
-                  circuitBreakerMap.computeIfAbsent(r, k -> new ArrayList<>());
-              list.add(circuitBreaker);
-            });
+    synchronized (circuitBreakerMap) {
+      circuitBreaker
+          .getRequestTypes()
+          .forEach(
+              r -> {
+                List<CircuitBreaker> list =
+                    circuitBreakerMap.computeIfAbsent(r, k -> new ArrayList<>());
+                list.add(circuitBreaker);
+                if (log.isInfoEnabled()) {
+                  log.info(
+                      "Registered circuit breaker {} for request type(s) {}",
+                      circuitBreaker.getClass().getSimpleName(),
+                      r);
+                }
+              });
+    }
   }
 
   @VisibleForTesting
-  public void deregisterAll() {
-    circuitBreakerMap.clear();
+  public void deregisterAll() throws IOException {
+    this.close();
   }
 
   /**
@@ -97,4 +113,41 @@ public static String toErrorMessage(List<CircuitBreaker> circuitBreakerList) {
   public boolean isEnabled(SolrRequestType requestType) {
     return circuitBreakerMap.containsKey(requestType);
   }
+
+  @Override
+  public void close() throws IOException {
+    synchronized (circuitBreakerMap) {
+      final AtomicInteger closeFailedCounter = new AtomicInteger(0);
+      circuitBreakerMap
+          .values()
+          .forEach(
+              list ->
+                  list.forEach(
+                      it -> {
+                        try {
+                          if (log.isDebugEnabled()) {
+                            log.debug(
+                                "Closed circuit breaker {} for request type(s) {}",
+                                it.getClass().getSimpleName(),
+                                it.getRequestTypes());
+                          }
+                          it.close();
+                        } catch (IOException e) {
+                          if (log.isErrorEnabled()) {
+                            log.error(
+                                String.format(
+                                    Locale.ROOT,
+                                    "Failed to close circuit breaker %s",
+                                    it.getClass().getSimpleName()),
+                                e);
+                          }
+                          closeFailedCounter.incrementAndGet();
+                        }
+                      }));
+      circuitBreakerMap.clear();
+      if (closeFailedCounter.get() > 0) {
+        throw new IOException("Failed to close " + closeFailedCounter.get() + " circuit breakers");
+      }
+    }
+  }
 }