Skip to content

Commit

Permalink
Misc changes for adaptive operation tracker (#1218)
Browse files Browse the repository at this point in the history 
1. add log info to record which type of adaptive tracker is used
2. enforce max number of inlight requests for adaptive tracker
3. make excluding timedout request configurable
4. support periodically dumping resource-level histogram to log file
  • Loading branch information
@jsjtzyy @cgtz
jsjtzyy authored and cgtz committed Jul 17, 2019
1 parent 79b311b commit 74364c7
Show file tree
Hide file tree
Showing 8 changed files with 288 additions and 51 deletions.
47 changes: 47 additions & 0 deletions ambry-api/src/main/java/com.github.ambry/config/RouterConfig.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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");
}
}
}
17 changes: 15 additions & 2 deletions ambry-router/src/main/java/com.github.ambry.router/AdaptiveOperationTracker.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions ambry-router/src/main/java/com.github.ambry.router/NonBlockingRouter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -410,6 +410,8 @@ public void close() {
if (cryptoJobHandler != null) {
cryptoJobHandler.close();
}
// close router metrics
routerMetrics.close();
}

/**
Expand Down
81 changes: 72 additions & 9 deletions ambry-router/src/main/java/com.github.ambry.router/NonBlockingRouterMetrics.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.*;


/**
Expand All @@ -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;
Expand Down Expand Up @@ -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"));
Expand Down Expand Up @@ -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,
Expand All @@ -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;
}
}

/**
Expand All @@ -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)) {
Expand Down Expand Up @@ -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));
}
}
}
}

/**
Expand Down
Loading

0 comments on commit 74364c7

Please sign in to comment.