Misc changes for adaptive operation tracker

ambry-api/src/main/java/com.github.ambry/config/RouterConfig.java

@@ -298,6 +298,41 @@ public class RouterConfig {
   @Default("1000")
   public final long routerOperationTrackerMinDataPointsRequired;
 
+  /**
+   * The maximum number of inflight requests that allowed for adaptive tracker. If current number of inflight requests
+   * is larger than or equal to this threshold, tracker shouldn't send out any request even though the oldest is past due.
+   * {@link RouterConfig#routerGetRequestParallelism} is a suggestive number that operation tracker uses to determine how
+   * many requests can be outstanding in parallel (assuming request gets response in time). Adaptive tracker is allowed
+   * to issue more requests (total inflight requests may exceed #routerGetRequestParallelism) if old request is past due.
+   * {@link RouterConfig#routerOperationTrackerMaxInflightRequests} is the strict upper bound that at any point of time,
+   * number of inflight requests issued by adaptive tracker should not exceed this number. Hence, for adaptive tracker,
+   * inflight requests number should always be within [0, #routerOperationTrackerMaxInflightRequests]
+   */
+  @Config("router.operation.tracker.max.inflight.requests")
+  @Default("2")
+  public final int routerOperationTrackerMaxInflightRequests;
+
+  /**
+   * Indicates whether to enable excluding timed out requests in Histogram reservoir.
+   */
+  @Config("router.operation.tracker.exclude.timeout.enabled")
+  @Default("false")
+  public final boolean routerOperationTrackerExcludeTimeoutEnabled;
+
+  /**
+   * Indicates whether to dump resource-level histogram to log file.
+   */
+  @Config("router.operation.tracker.histogram.dump.enabled")
+  @Default("false")
+  public final boolean routerOperationTrackerHistogramDumpEnabled;
+
+  /**
+   * The period of dumping resource-level histogram (if enabled).
+   */
+  @Config("router.operation.tracker.histogram.dump.period")
+  @Default("600")
+  public final long routerOperationTrackerHistogramDumpPeriod;
+
   /**
    * Create a RouterConfig instance.
    * @param verifiableProperties the properties map to refer to.
@@ -371,5 +406,17 @@ public RouterConfig(VerifiableProperties verifiableProperties) {
         verifiableProperties.getDouble("router.operation.tracker.reservoir.decay.factor", 0.015);
     routerOperationTrackerMinDataPointsRequired =
         verifiableProperties.getLong("router.operation.tracker.min.data.points.required", 1000L);
+    routerOperationTrackerMaxInflightRequests =
+        verifiableProperties.getIntInRange("router.operation.tracker.max.inflight.requests", 2, 1, Integer.MAX_VALUE);
+    routerOperationTrackerExcludeTimeoutEnabled =
+        verifiableProperties.getBoolean("router.operation.tracker.exclude.timeout.enabled", false);
+    routerOperationTrackerHistogramDumpEnabled =
+        verifiableProperties.getBoolean("router.operation.tracker.histogram.dump.enabled", false);
+    routerOperationTrackerHistogramDumpPeriod =
+        verifiableProperties.getLongInRange("router.operation.tracker.histogram.dump.period", 600L, 1L, Long.MAX_VALUE);
+    if (routerGetRequestParallelism > routerOperationTrackerMaxInflightRequests) {
+      throw new IllegalArgumentException(
+          "Operation tracker parallelism is larger than operation tracker max inflight number");
+    }
   }
 }

ambry-router/src/main/java/com.github.ambry.router/AdaptiveOperationTracker.java

@@ -84,6 +84,10 @@ class AdaptiveOperationTracker extends SimpleOperationTracker {
       localDcResourceToHistogram = getResourceToLatencyMap(routerOperation, true);
       crossDcResourceToHistogram = getResourceToLatencyMap(routerOperation, false);
     }
+    if (parallelism > routerConfig.routerOperationTrackerMaxInflightRequests) {
+      throw new IllegalArgumentException(
+          "Operation tracker parallelism is larger than adaptive tracker max inflight number");
+    }
   }
 
   @Override
@@ -95,7 +99,8 @@ public void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequ
     } else {
       elapsedTime = time.milliseconds() - expiredRequestSendTimes.remove(replicaId);
     }
-    if (trackedRequestFinalState != TrackedRequestFinalState.TIMED_OUT) {
+    if (trackedRequestFinalState != TrackedRequestFinalState.TIMED_OUT
+        || !routerConfig.routerOperationTrackerExcludeTimeoutEnabled) {
       getLatencyHistogram(replicaId).update(elapsedTime);
       if (routerConfig.routerOperationTrackerMetricScope != OperationTrackerScope.Datacenter) {
         // This is only used to report whole datacenter histogram for monitoring purpose
@@ -219,7 +224,15 @@ private class OpTrackerIterator implements Iterator<ReplicaId> {
 
     @Override
     public boolean hasNext() {
-      return replicaIterator.hasNext() && (inflightCount < parallelism || isOldestRequestPastDue());
+      if (replicaIterator.hasNext()) {
+        if (inflightCount < parallelism) {
+          return true;
+        }
+        if (inflightCount < routerConfig.routerOperationTrackerMaxInflightRequests && isOldestRequestPastDue()) {
+          return true;
+        }
+      }
+      return false;
     }
 
     @Override

ambry-router/src/main/java/com.github.ambry.router/NonBlockingRouter.java

@@ -410,6 +410,8 @@ public void close() {
     if (cryptoJobHandler != null) {
       cryptoJobHandler.close();
     }
+    // close router metrics
+    routerMetrics.close();
   }
 
   /**

ambry-router/src/main/java/com.github.ambry.router/NonBlockingRouterMetrics.java

@@ -27,11 +27,17 @@
 import com.github.ambry.clustermap.Resource;
 import com.github.ambry.config.RouterConfig;
 import com.github.ambry.utils.SystemTime;
+import com.github.ambry.utils.Utils;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.github.ambry.utils.Utils.*;
 
 
 /**
@@ -41,6 +47,7 @@
  */
 public class NonBlockingRouterMetrics {
   private final MetricRegistry metricRegistry;
+  private static final Logger logger = LoggerFactory.getLogger(NonBlockingRouterMetrics.class);
 
   // Operation rate.
   public final Meter putBlobOperationRate;
@@ -192,17 +199,21 @@ public class NonBlockingRouterMetrics {
   public final CryptoJobMetrics decryptJobMetrics;
 
   // Resource to latency histogram map. Here resource can be DataNode, Partition, Disk, Replica etc.
-  Map<Resource, Histogram> getBlobLocalDcResourceToLatency;
-  Map<Resource, Histogram> getBlobCrossDcResourceToLatency;
+  Map<Resource, Histogram> getBlobLocalDcResourceToLatency = new HashMap<>();
+  Map<Resource, Histogram> getBlobCrossDcResourceToLatency = new HashMap<>();
 
-  Map<Resource, Histogram> getBlobInfoLocalDcResourceToLatency;
-  Map<Resource, Histogram> getBlobInfoCrossDcResourceToLatency;
+  Map<Resource, Histogram> getBlobInfoLocalDcResourceToLatency = new HashMap<>();
+  Map<Resource, Histogram> getBlobInfoCrossDcResourceToLatency = new HashMap<>();
 
   // Map that stores dataNode-level metrics.
   private final Map<DataNodeId, NodeLevelMetrics> dataNodeToMetrics;
+  private final RouterConfig routerConfig;
+  private final HistogramDumper histogramDumper;
+  private ScheduledExecutorService scheduler = null;
 
   public NonBlockingRouterMetrics(ClusterMap clusterMap, RouterConfig routerConfig) {
     metricRegistry = clusterMap.getMetricRegistry();
+    this.routerConfig = routerConfig;
 
     // Operation Rate.
     putBlobOperationRate = metricRegistry.meter(MetricRegistry.name(PutOperation.class, "PutBlobOperationRate"));
@@ -436,8 +447,9 @@ public NonBlockingRouterMetrics(ClusterMap clusterMap, RouterConfig routerConfig
     encryptJobMetrics = new CryptoJobMetrics(PutOperation.class, "Encrypt", metricRegistry);
     decryptJobMetrics = new CryptoJobMetrics(GetOperation.class, "Decrypt", metricRegistry);
 
-    // Custom percentiles
+    // Record type of adaptive tracker and configure custom percentiles
     if (routerConfig != null) {
+      logger.info("The metric scope of adaptive tracker is {}", routerConfig.routerOperationTrackerMetricScope);
       registerCustomPercentiles(GetBlobOperation.class, "LocalDcLatencyMs", getBlobLocalDcLatencyMs,
           routerConfig.routerOperationTrackerCustomPercentiles);
       registerCustomPercentiles(GetBlobOperation.class, "CrossDcLatencyMs", getBlobCrossDcLatencyMs,
@@ -452,6 +464,17 @@ public NonBlockingRouterMetrics(ClusterMap clusterMap, RouterConfig routerConfig
       // pre-populate all resource-to-histogram maps here to allow lock-free hashmap in adaptive operation tracker
       initializeResourceToHistogramMap(clusterMap, routerConfig);
     }
+
+    if (routerConfig != null && routerConfig.routerOperationTrackerHistogramDumpEnabled) {
+      histogramDumper = new HistogramDumper();
+      scheduler = Utils.newScheduler(1, false);
+      logger.info("Scheduling histogram dumper with a period of {} secs",
+          routerConfig.routerOperationTrackerHistogramDumpPeriod);
+      scheduler.scheduleAtFixedRate(histogramDumper, 0, routerConfig.routerOperationTrackerHistogramDumpPeriod,
+          TimeUnit.SECONDS);
+    } else {
+      histogramDumper = null;
+    }
   }
 
   /**
@@ -464,10 +487,6 @@ private void initializeResourceToHistogramMap(ClusterMap clusterMap, RouterConfi
     int reservoirSize = routerConfig.routerOperationTrackerReservoirSize;
     double decayFactor = routerConfig.routerOperationTrackerReservoirDecayFactor;
     String localDatacenterName = clusterMap.getDatacenterName(clusterMap.getLocalDatacenterId());
-    getBlobLocalDcResourceToLatency = new HashMap<>();
-    getBlobInfoLocalDcResourceToLatency = new HashMap<>();
-    getBlobCrossDcResourceToLatency = new HashMap<>();
-    getBlobInfoCrossDcResourceToLatency = new HashMap<>();
     switch (routerConfig.routerOperationTrackerMetricScope) {
       case Partition:
         for (PartitionId partitionId : clusterMap.getAllPartitionIds(null)) {
@@ -892,6 +911,50 @@ void trackAgeAtAccess(long creationTimeMs) {
       }
     }
   }
+
+  /**
+   * Close {@link NonBlockingRouterMetrics} by shutting down scheduler (if present) in this class.
+   */
+  public void close() {
+    if (scheduler != null) {
+      shutDownExecutorService(scheduler, 5, TimeUnit.SECONDS);
+    }
+  }
+
+  /**
+   * A thread that helps periodically dump resource-level histogram (with given percentile) into log file.
+   */
+  private class HistogramDumper implements Runnable {
+
+    @Override
+    public void run() {
+      double quantile = routerConfig.routerLatencyToleranceQuantile;
+      for (Map.Entry<Resource, Histogram> resourceToHistogram : getBlobLocalDcResourceToLatency.entrySet()) {
+        Resource resource = resourceToHistogram.getKey();
+        Histogram histogram = resourceToHistogram.getValue();
+        logger.info("{} GetBlob local DC latency histogram {}th percentile in ms: {}", resource.toString(),
+            quantile * 100, histogram.getSnapshot().getValue(quantile));
+      }
+      for (Map.Entry<Resource, Histogram> resourceToHistogram : getBlobCrossDcResourceToLatency.entrySet()) {
+        Resource resource = resourceToHistogram.getKey();
+        Histogram histogram = resourceToHistogram.getValue();
+        logger.info("{} GetBlob cross DC latency histogram {}th percentile in ms: {}", resource.toString(),
+            quantile * 100, histogram.getSnapshot().getValue(quantile));
+      }
+      for (Map.Entry<Resource, Histogram> resourceToHistogram : getBlobInfoLocalDcResourceToLatency.entrySet()) {
+        Resource resource = resourceToHistogram.getKey();
+        Histogram histogram = resourceToHistogram.getValue();
+        logger.info("{} GetBlobInfo local DC latency histogram {}th percentile in ms: {}", resource.toString(),
+            quantile * 100, histogram.getSnapshot().getValue(quantile));
+      }
+      for (Map.Entry<Resource, Histogram> resourceToHistogram : getBlobInfoCrossDcResourceToLatency.entrySet()) {
+        Resource resource = resourceToHistogram.getKey();
+        Histogram histogram = resourceToHistogram.getValue();
+        logger.info("{} GetBlobInfo cross DC latency histogram {}th percentile in ms: {}", resource.toString(),
+            quantile * 100, histogram.getSnapshot().getValue(quantile));
+      }
+    }
+  }
 }
 
 /**