You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by sh...@apache.org on 2020/02/04 10:28:41 UTC
[incubator-ratis] branch master updated: RATIS-803. Add a metric
for count of resource limit of pending requests in ratis leader.
Contributed by Bharat Viswanadham.
This is an automated email from the ASF dual-hosted git repository.
shashikant pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-ratis.git
The following commit(s) were added to refs/heads/master by this push:
new b5528af RATIS-803. Add a metric for count of resource limit of pending requests in ratis leader. Contributed by Bharat Viswanadham.
b5528af is described below
commit b5528afc1fc5f3238b61e3dbc247b91cad0f7865
Author: Shashikant Banerjee <sh...@apache.org>
AuthorDate: Tue Feb 4 15:58:11 2020 +0530
RATIS-803. Add a metric for count of resource limit of pending requests in ratis leader. Contributed by Bharat Viswanadham.
---
pom.xml | 4 ++
.../org/apache/ratis/util/ResourceSemaphore.java | 23 ++++++-
.../apache/ratis/server/impl/PendingRequests.java | 13 ++--
.../ratis/server/impl/RaftServerMetrics.java | 20 +++++++
ratis-test/pom.xml | 6 ++
.../apache/ratis/grpc/TestRaftServerWithGrpc.java | 70 ++++++++++++++++++++++
.../apache/ratis/util/TestResourceSemaphore.java | 21 ++++---
7 files changed, 142 insertions(+), 15 deletions(-)
diff --git a/pom.xml b/pom.xml
index 1d6fec9..482f894 100644
--- a/pom.xml
+++ b/pom.xml
@@ -204,6 +204,10 @@
<!--metrics-->
<dropwizard.version>3.2.5</dropwizard.version>
+
+ <!-- commons lang version -->
+ <commons.lang.version>3.8.1</commons.lang.version>
+
</properties>
<dependencyManagement>
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java b/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java
index 8d89826..3380034 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java
@@ -88,6 +88,13 @@ public class ResourceSemaphore extends Semaphore {
/**
* Track a group of resources with a list of {@link ResourceSemaphore}s.
*/
+
+ public enum ResourceAcquireStatus {
+ SUCCESS,
+ FAILED_IN_ELEMENT_LIMIT,
+ FAILED_IN_BYTE_SIZE_LIMIT
+ }
+
public static class Group {
private final List<ResourceSemaphore> resources;
@@ -108,7 +115,7 @@ public class ResourceSemaphore extends Semaphore {
return resources.get(i);
}
- public boolean tryAcquire(int... permits) {
+ public ResourceAcquireStatus tryAcquire(int... permits) {
Preconditions.assertTrue(permits.length == resources.size(),
() -> "items.length = " + permits.length + " != resources.size() = " + resources.size());
int i = 0;
@@ -118,15 +125,25 @@ public class ResourceSemaphore extends Semaphore {
break;
}
}
+
+
if (i == permits.length) {
- return true; // successfully acquired all resources
+ return ResourceAcquireStatus.SUCCESS; // successfully acquired all resources
+ }
+
+ ResourceAcquireStatus acquireStatus;
+ if (i == 0) {
+ acquireStatus = ResourceAcquireStatus.FAILED_IN_ELEMENT_LIMIT;
+ } else {
+ acquireStatus = ResourceAcquireStatus.FAILED_IN_BYTE_SIZE_LIMIT;
}
// failed at i, releasing all previous resources
for(i--; i >= 0; i--) {
resources.get(i).release(permits[i]);
}
- return false;
+
+ return acquireStatus;
}
public void acquire(int... permits) throws InterruptedException {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java
index f9421f3..42dcf8f 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java
@@ -59,12 +59,11 @@ class PendingRequests {
return get(0).used();
}
- // TODO: add metrics
int getByteSize() {
return get(1).used();
}
- boolean tryAcquire(Message message) {
+ ResourceSemaphore.ResourceAcquireStatus tryAcquire(Message message) {
return tryAcquire(1, Message.getSize(message));
}
@@ -89,13 +88,19 @@ class PendingRequests {
this.raftServerMetrics = raftServerMetrics;
raftServerMetrics.addNumPendingRequestsGauge(resource::getElementCount);
+ raftServerMetrics.addNumPendingRequestsByteSize(resource::getByteSize);
}
Permit tryAcquire(Message message) {
- final boolean acquired = resource.tryAcquire(message);
+ final ResourceSemaphore.ResourceAcquireStatus acquired = resource.tryAcquire(message);
LOG.trace("tryAcquire? {}", acquired);
- if (!acquired) {
+ if (acquired == ResourceSemaphore.ResourceAcquireStatus.FAILED_IN_ELEMENT_LIMIT) {
raftServerMetrics.onRequestQueueLimitHit();
+ raftServerMetrics.onResourceLimitHit();
+ return null;
+ } else if (acquired == ResourceSemaphore.ResourceAcquireStatus.FAILED_IN_BYTE_SIZE_LIMIT) {
+ raftServerMetrics.onRequestByteSizeLimitHit();
+ raftServerMetrics.onResourceLimitHit();
return null;
}
return putPermit();
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerMetrics.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerMetrics.java
index 0689496..02b314b 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerMetrics.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerMetrics.java
@@ -56,7 +56,11 @@ public final class RaftServerMetrics extends RatisMetrics {
public static final String RAFT_CLIENT_WATCH_REQUEST = "clientWatch%sRequest";
public static final String RETRY_REQUEST_CACHE_HIT_COUNTER = "numRetryCacheHits";
public static final String REQUEST_QUEUE_LIMIT_HIT_COUNTER = "numRequestQueueLimitHits";
+ public static final String RESOURCE_LIMIT_HIT_COUNTER = "leaderNumResourceLimitHits";
+ public static final String REQUEST_BYTE_SIZE_LIMIT_HIT_COUNTER = "numRequestsByteSizeLimitHits";
public static final String REQUEST_QUEUE_SIZE = "numPendingRequestInQueue";
+ public static final String REQUEST_BYTE_SIZE = "numPendingRequestByteSize";
+
private Map<String, Long> followerLastHeartbeatElapsedTimeMap = new HashMap<>();
private CommitInfoCache commitInfoCache;
@@ -189,4 +193,20 @@ public final class RaftServerMetrics extends RatisMetrics {
void addNumPendingRequestsGauge(Gauge queueSize) {
registry.gauge(REQUEST_QUEUE_SIZE, () -> queueSize);
}
+
+ void addNumPendingRequestsByteSize(Gauge byteSize) {
+ registry.gauge(REQUEST_BYTE_SIZE, () -> byteSize);
+ }
+
+ void onRequestByteSizeLimitHit() {
+ registry.counter(REQUEST_BYTE_SIZE_LIMIT_HIT_COUNTER).inc();
+ }
+
+ void onResourceLimitHit() {
+ registry.counter(RESOURCE_LIMIT_HIT_COUNTER).inc();
+ }
+
+ public RatisMetricRegistry getRegistry() {
+ return registry;
+ }
}
\ No newline at end of file
diff --git a/ratis-test/pom.xml b/ratis-test/pom.xml
index 31ebc8f..c93ea7b 100644
--- a/ratis-test/pom.xml
+++ b/ratis-test/pom.xml
@@ -103,5 +103,11 @@
<artifactId>metrics-jvm</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>${commons.lang.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java b/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java
index f79149d..512ba1b 100644
--- a/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java
+++ b/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java
@@ -21,7 +21,13 @@ import static org.apache.ratis.server.impl.RaftServerMetrics.RAFT_CLIENT_READ_RE
import static org.apache.ratis.server.impl.RaftServerMetrics.RAFT_CLIENT_STALE_READ_REQUEST;
import static org.apache.ratis.server.impl.RaftServerMetrics.RAFT_CLIENT_WATCH_REQUEST;
import static org.apache.ratis.server.impl.RaftServerMetrics.RAFT_CLIENT_WRITE_REQUEST;
+import static org.apache.ratis.server.impl.RaftServerMetrics.REQUEST_QUEUE_LIMIT_HIT_COUNTER;
+import static org.apache.ratis.server.impl.RaftServerMetrics.REQUEST_BYTE_SIZE;
+import static org.apache.ratis.server.impl.RaftServerMetrics.REQUEST_BYTE_SIZE_LIMIT_HIT_COUNTER;
+import static org.apache.ratis.server.impl.RaftServerMetrics.RESOURCE_LIMIT_HIT_COUNTER;
+import com.codahale.metrics.Gauge;
+import org.apache.commons.lang3.RandomStringUtils;
import org.apache.log4j.Level;
import org.apache.ratis.BaseTest;
import org.apache.ratis.MiniRaftCluster;
@@ -41,6 +47,8 @@ import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.TimeoutIOException;
+import org.apache.ratis.retry.RetryPolicies;
+import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.RaftServerRpc;
import org.apache.ratis.server.impl.RaftServerImpl;
import org.apache.ratis.server.impl.RaftServerMetrics;
@@ -57,6 +65,7 @@ import org.junit.Test;
import java.io.IOException;
import java.nio.channels.OverlappingFileLockException;
+import java.util.SortedMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -179,6 +188,67 @@ public class TestRaftServerWithGrpc extends BaseTest implements MiniRaftClusterW
runWithNewCluster(3, this::testRaftClientRequestMetrics);
}
+ @Test
+ public void testRaftServerMetrics() throws Exception {
+ final RaftProperties p = getProperties();
+ RaftServerConfigKeys.Write.setElementLimit(p, 10);
+ RaftServerConfigKeys.Write.setByteLimit(p, 110);
+ try {
+ runWithNewCluster(3, this::testRequestMetrics);
+ } finally {
+ RaftServerConfigKeys.Write.setElementLimit(p, RaftServerConfigKeys.Write.ELEMENT_LIMIT_DEFAULT);
+ RaftServerConfigKeys.Write.setByteLimit(p, RaftServerConfigKeys.Write.BYTE_LIMIT_DEFAULT.getSizeInt());
+ }
+ }
+
+ void testRequestMetrics(MiniRaftClusterWithGrpc cluster) throws Exception {
+
+ try (RaftClient client = cluster.createClient()) {
+ // send a request to make sure leader is ready
+ final CompletableFuture< RaftClientReply > f = client.sendAsync(new SimpleMessage("testing"));
+ Assert.assertTrue(f.get().isSuccess());
+ }
+
+ SimpleStateMachine4Testing stateMachine = SimpleStateMachine4Testing.get(cluster.getLeader());
+ stateMachine.blockFlushStateMachineData();
+
+ String message = "2nd Message";
+ // Block stateMachine flush data, so that 2nd request will not be
+ // completed, and so it will not be removed from pending request map.
+ cluster.createClient(cluster.getLeader().getId(), RetryPolicies.noRetry()).sendAsync(new SimpleMessage(message));
+
+ final SortedMap< String, Gauge > gaugeMap =
+ cluster.getLeader().getRaftServerMetrics().getRegistry()
+ .getGauges((s, metric) -> s.contains(REQUEST_BYTE_SIZE));
+
+ RaftTestUtil.waitFor(() -> (int) gaugeMap.get(gaugeMap.firstKey()).getValue() == message.length(),
+ 300, 5000);
+
+
+ for (int i = 0; i < 10; i++) {
+ cluster.createClient(cluster.getLeader().getId(), RetryPolicies.noRetry()).sendAsync(new SimpleMessage(message));
+ }
+
+ // Because we have passed 11 requests, and the element queue size is 10.
+ RaftTestUtil.waitFor(() -> cluster.getLeader().getRaftServerMetrics().getCounter(REQUEST_QUEUE_LIMIT_HIT_COUNTER)
+ .getCount() == 1, 300, 5000);
+
+ stateMachine.unblockFlushStateMachineData();
+
+ // Send a message with 120, our byte size limit is 110, so it should fail
+ // and byte size counter limit will be hit.
+
+ cluster.createClient(cluster.getLeader().getId(), RetryPolicies.noRetry())
+ .sendAsync(new SimpleMessage(RandomStringUtils.random(120, true, false)));
+
+ RaftTestUtil.waitFor(() -> cluster.getLeader().getRaftServerMetrics()
+ .getCounter(REQUEST_BYTE_SIZE_LIMIT_HIT_COUNTER).getCount() == 1, 300, 5000);
+
+ Assert.assertEquals(2, cluster.getLeader().getRaftServerMetrics()
+ .getCounter(RESOURCE_LIMIT_HIT_COUNTER).getCount());
+ }
+
+
void testRaftClientRequestMetrics(MiniRaftClusterWithGrpc cluster) throws IOException,
ExecutionException, InterruptedException {
final RaftServerImpl leader = RaftTestUtil.waitForLeader(cluster);
diff --git a/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java b/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java
index ac658d4..dfed847 100644
--- a/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java
+++ b/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java
@@ -24,23 +24,27 @@ import org.junit.Test;
import java.util.concurrent.TimeoutException;
+import static org.apache.ratis.util.ResourceSemaphore.ResourceAcquireStatus.FAILED_IN_BYTE_SIZE_LIMIT;
+import static org.apache.ratis.util.ResourceSemaphore.ResourceAcquireStatus.FAILED_IN_ELEMENT_LIMIT;
+import static org.apache.ratis.util.ResourceSemaphore.ResourceAcquireStatus.SUCCESS;
+
public class TestResourceSemaphore extends BaseTest {
@Test(timeout = 5000)
public void testGroup() throws InterruptedException, TimeoutException {
final ResourceSemaphore.Group g = new ResourceSemaphore.Group(3, 1);
assertUsed(g, 0, 0);
- assertAcquire(g, true, 1, 1);
+ assertAcquire(g, ResourceSemaphore.ResourceAcquireStatus.SUCCESS, 1, 1);
assertUsed(g, 1, 1);
- assertAcquire(g, false, 1, 1);
+ assertAcquire(g, FAILED_IN_BYTE_SIZE_LIMIT, 1, 1);
assertUsed(g, 1, 1);
- assertAcquire(g, false, 0, 1);
+ assertAcquire(g, FAILED_IN_BYTE_SIZE_LIMIT, 0, 1);
assertUsed(g, 1, 1);
- assertAcquire(g, true, 1, 0);
+ assertAcquire(g, SUCCESS, 1, 0);
assertUsed(g, 2, 1);
- assertAcquire(g, true, 1, 0);
+ assertAcquire(g, SUCCESS, 1, 0);
assertUsed(g, 3, 1);
- assertAcquire(g, false, 1, 0);
+ assertAcquire(g, FAILED_IN_ELEMENT_LIMIT, 1, 0);
assertUsed(g, 3, 1);
g.release(1, 1);
@@ -82,8 +86,9 @@ public class TestResourceSemaphore extends BaseTest {
}
}
- static void assertAcquire(ResourceSemaphore.Group g, boolean expected, int... permits) {
- final boolean computed = g.tryAcquire(permits);
+ static void assertAcquire(ResourceSemaphore.Group g, ResourceSemaphore.ResourceAcquireStatus expected,
+ int... permits) {
+ final ResourceSemaphore.ResourceAcquireStatus computed = g.tryAcquire(permits);
Assert.assertEquals(expected, computed);
}