You are viewing a plain text version of this content. The canonical link for it is here.
Posted to distributedlog-dev@bookkeeper.apache.org by si...@apache.org on 2016/12/13 23:40:34 UTC
incubator-distributedlog git commit: DL-85: Publish both per-stream
stats and aggregation stats
Repository: incubator-distributedlog
Updated Branches:
refs/heads/master 3fceccc6c -> a645c311b
DL-85: Publish both per-stream stats and aggregation stats
merge twittter's changes from Jordan Bull.
Author: Jordan Bull <jb...@twitter.com>
Reviewers: Sijie Guo <si...@apache.org>, Franck Cuny <fc...@apache.org>
Closes #57 from sijie/merge/DL-85
Project: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/commit/a645c311
Tree: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/tree/a645c311
Diff: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/diff/a645c311
Branch: refs/heads/master
Commit: a645c311b3ea724800f0344b71980c788d6a57c8
Parents: 3fceccc
Author: Jordan Bull <jb...@twitter.com>
Authored: Tue Dec 13 15:40:29 2016 -0800
Committer: Sijie Guo <si...@twitter.com>
Committed: Tue Dec 13 15:40:29 2016 -0800
----------------------------------------------------------------------
.../service/DistributedLogServiceImpl.java | 11 ++++++----
.../service/stream/BulkWriteOp.java | 8 ++++++--
.../service/stream/StreamImpl.java | 4 ++--
.../service/stream/StreamOpStats.java | 21 ++++++++++++--------
.../distributedlog/service/stream/WriteOp.java | 8 ++++++--
.../service/streamset/Partition.java | 8 ++++++++
.../service/TestDistributedLogServer.java | 8 ++++----
.../service/TestDistributedLogService.java | 1 +
.../service/stream/TestStreamOp.java | 3 ++-
9 files changed, 49 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/main/java/com/twitter/distributedlog/service/DistributedLogServiceImpl.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/DistributedLogServiceImpl.java b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/DistributedLogServiceImpl.java
index de475ea..751e972 100644
--- a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/DistributedLogServiceImpl.java
+++ b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/DistributedLogServiceImpl.java
@@ -132,6 +132,7 @@ public class DistributedLogServiceImpl implements DistributedLogService.ServiceI
// Stats
private final StatsLogger statsLogger;
private final StatsLogger perStreamStatsLogger;
+ private final StreamPartitionConverter streamPartitionConverter;
private final StreamOpStats streamOpStats;
private final Counter bulkWritePendingStat;
private final Counter writePendingStat;
@@ -158,6 +159,7 @@ public class DistributedLogServiceImpl implements DistributedLogService.ServiceI
this.perStreamStatsLogger = perStreamStatsLogger;
this.dlsnVersion = serverConf.getDlsnVersion();
this.serverRegionId = serverConf.getRegionId();
+ this.streamPartitionConverter = converter;
int serverPort = serverConf.getServerPort();
int shard = serverConf.getServerShardId();
int numThreads = serverConf.getServerThreads();
@@ -396,8 +398,8 @@ public class DistributedLogServiceImpl implements DistributedLogService.ServiceI
public Future<BulkWriteResponse> writeBulkWithContext(final String stream, List<ByteBuffer> data, WriteContext ctx) {
bulkWritePendingStat.inc();
receivedRecordCounter.add(data.size());
- BulkWriteOp op = new BulkWriteOp(stream, data, statsLogger, perStreamStatsLogger, getChecksum(ctx),
- featureChecksumDisabled, accessControlManager);
+ BulkWriteOp op = new BulkWriteOp(stream, data, statsLogger, perStreamStatsLogger, streamPartitionConverter,
+ getChecksum(ctx), featureChecksumDisabled, accessControlManager);
executeStreamOp(op);
return op.result().ensure(new Function0<BoxedUnit>() {
public BoxedUnit apply() {
@@ -675,8 +677,9 @@ public class DistributedLogServiceImpl implements DistributedLogService.ServiceI
ByteBuffer data,
Long checksum,
boolean isRecordSet) {
- return new WriteOp(stream, data, statsLogger, perStreamStatsLogger, serverConfig, dlsnVersion,
- checksum, isRecordSet, featureChecksumDisabled, accessControlManager);
+ return new WriteOp(stream, data, statsLogger, perStreamStatsLogger, streamPartitionConverter,
+ serverConfig, dlsnVersion, checksum, isRecordSet, featureChecksumDisabled,
+ accessControlManager);
}
@VisibleForTesting
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/BulkWriteOp.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/BulkWriteOp.java b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/BulkWriteOp.java
index 96a37cd..c009bb9 100644
--- a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/BulkWriteOp.java
+++ b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/BulkWriteOp.java
@@ -33,6 +33,8 @@ import com.twitter.distributedlog.exceptions.DLException;
import com.twitter.distributedlog.exceptions.OwnershipAcquireFailedException;
import com.twitter.distributedlog.exceptions.RequestDeniedException;
import com.twitter.distributedlog.service.ResponseUtils;
+import com.twitter.distributedlog.service.streamset.Partition;
+import com.twitter.distributedlog.service.streamset.StreamPartitionConverter;
import com.twitter.distributedlog.thrift.service.BulkWriteResponse;
import com.twitter.distributedlog.thrift.service.ResponseHeader;
import com.twitter.distributedlog.thrift.service.StatusCode;
@@ -88,6 +90,7 @@ public class BulkWriteOp extends AbstractStreamOp<BulkWriteResponse> implements
List<ByteBuffer> buffers,
StatsLogger statsLogger,
StatsLogger perStreamStatsLogger,
+ StreamPartitionConverter streamPartitionConverter,
Long checksum,
Feature checksumDisabledFeature,
AccessControlManager accessControlManager) {
@@ -100,6 +103,7 @@ public class BulkWriteOp extends AbstractStreamOp<BulkWriteResponse> implements
}
this.payloadSize = total;
+ final Partition partition = streamPartitionConverter.convert(stream);
// Write record stats
StreamOpStats streamOpStats = new StreamOpStats(statsLogger, perStreamStatsLogger);
this.deniedBulkWriteCounter = streamOpStats.requestDeniedCounter("bulkWrite");
@@ -107,8 +111,8 @@ public class BulkWriteOp extends AbstractStreamOp<BulkWriteResponse> implements
this.failureRecordCounter = streamOpStats.recordsCounter("failure");
this.redirectRecordCounter = streamOpStats.recordsCounter("redirect");
this.bulkWriteBytes = streamOpStats.scopedRequestCounter("bulkWrite", "bytes");
- this.latencyStat = streamOpStats.streamRequestLatencyStat(stream, "bulkWrite");
- this.bytes = streamOpStats.streamRequestCounter(stream, "bulkWrite", "bytes");
+ this.latencyStat = streamOpStats.streamRequestLatencyStat(partition, "bulkWrite");
+ this.bytes = streamOpStats.streamRequestCounter(partition, "bulkWrite", "bytes");
this.accessControlManager = accessControlManager;
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamImpl.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamImpl.java b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamImpl.java
index 04f793f..45630fe 100644
--- a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamImpl.java
+++ b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamImpl.java
@@ -194,12 +194,12 @@ public class StreamImpl implements Stream {
this.dynConf = streamConf;
StatsLogger limiterStatsLogger = BroadCastStatsLogger.two(
streamOpStats.baseScope("stream_limiter"),
- streamOpStats.streamRequestScope(name, "limiter"));
+ streamOpStats.streamRequestScope(partition, "limiter"));
this.limiter = new StreamRequestLimiter(name, dynConf, limiterStatsLogger, featureRateLimitDisabled);
this.requestTimer = requestTimer;
// Stats
- this.streamLogger = streamOpStats.streamRequestStatsLogger(name);
+ this.streamLogger = streamOpStats.streamRequestStatsLogger(partition);
this.limiterStatLogger = streamOpStats.baseScope("request_limiter");
this.streamExceptionStatLogger = streamLogger.scope("exceptions");
this.serviceTimeout = streamOpStats.baseCounter("serviceTimeout");
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamOpStats.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamOpStats.java b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamOpStats.java
index adf89a3..bfbc88c 100644
--- a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamOpStats.java
+++ b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/StreamOpStats.java
@@ -17,6 +17,8 @@
*/
package com.twitter.distributedlog.service.stream;
+import com.twitter.distributedlog.stats.BroadCastStatsLogger;
+import com.twitter.distributedlog.service.streamset.Partition;
import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
@@ -81,19 +83,22 @@ public class StreamOpStats {
return recordsStatsLogger.getCounter(counterName);
}
- public StatsLogger streamRequestStatsLogger(String streamName) {
- return streamStatsLogger.scope(streamName);
+ public StatsLogger streamRequestStatsLogger(Partition partition) {
+ return BroadCastStatsLogger.masterslave(
+ streamStatsLogger.scope(partition.getStream()).scope("partition")
+ .scope(partition.getPaddedId()), streamStatsLogger.scope(partition.getStream())
+ .scope("aggregate"));
}
- public StatsLogger streamRequestScope(String streamName, String scopeName) {
- return streamRequestStatsLogger(streamName).scope(scopeName);
+ public StatsLogger streamRequestScope(Partition partition, String scopeName) {
+ return streamRequestStatsLogger(partition).scope(scopeName);
}
- public OpStatsLogger streamRequestLatencyStat(String streamName, String opName) {
- return streamRequestStatsLogger(streamName).getOpStatsLogger(opName);
+ public OpStatsLogger streamRequestLatencyStat(Partition partition, String opName) {
+ return streamRequestStatsLogger(partition).getOpStatsLogger(opName);
}
- public Counter streamRequestCounter(String streamName, String opName, String counterName) {
- return streamRequestScope(streamName, opName).getCounter(counterName);
+ public Counter streamRequestCounter(Partition partition, String opName, String counterName) {
+ return streamRequestScope(partition, opName).getCounter(counterName);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/WriteOp.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/WriteOp.java b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/WriteOp.java
index c74f2cd..e9f2f4e 100644
--- a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/WriteOp.java
+++ b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/stream/WriteOp.java
@@ -26,6 +26,8 @@ import com.twitter.distributedlog.service.config.ServerConfiguration;
import com.twitter.distributedlog.exceptions.DLException;
import com.twitter.distributedlog.exceptions.RequestDeniedException;
import com.twitter.distributedlog.service.ResponseUtils;
+import com.twitter.distributedlog.service.streamset.Partition;
+import com.twitter.distributedlog.service.streamset.StreamPartitionConverter;
import com.twitter.distributedlog.thrift.service.WriteResponse;
import com.twitter.distributedlog.thrift.service.ResponseHeader;
import com.twitter.distributedlog.thrift.service.StatusCode;
@@ -67,6 +69,7 @@ public class WriteOp extends AbstractWriteOp implements WriteOpWithPayload {
ByteBuffer data,
StatsLogger statsLogger,
StatsLogger perStreamStatsLogger,
+ StreamPartitionConverter streamPartitionConverter,
ServerConfiguration conf,
byte dlsnVersion,
Long checksum,
@@ -78,14 +81,15 @@ public class WriteOp extends AbstractWriteOp implements WriteOpWithPayload {
data.get(payload);
this.isRecordSet = isRecordSet;
+ final Partition partition = streamPartitionConverter.convert(stream);
StreamOpStats streamOpStats = new StreamOpStats(statsLogger, perStreamStatsLogger);
this.successRecordCounter = streamOpStats.recordsCounter("success");
this.failureRecordCounter = streamOpStats.recordsCounter("failure");
this.redirectRecordCounter = streamOpStats.recordsCounter("redirect");
this.deniedWriteCounter = streamOpStats.requestDeniedCounter("write");
this.writeBytes = streamOpStats.scopedRequestCounter("write", "bytes");
- this.latencyStat = streamOpStats.streamRequestLatencyStat(stream, "write");
- this.bytes = streamOpStats.streamRequestCounter(stream, "write", "bytes");
+ this.latencyStat = streamOpStats.streamRequestLatencyStat(partition, "write");
+ this.bytes = streamOpStats.streamRequestCounter(partition, "write", "bytes");
this.dlsnVersion = dlsnVersion;
this.accessControlManager = accessControlManager;
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/main/java/com/twitter/distributedlog/service/streamset/Partition.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/streamset/Partition.java b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/streamset/Partition.java
index f207eee..d199f88 100644
--- a/distributedlog-service/src/main/java/com/twitter/distributedlog/service/streamset/Partition.java
+++ b/distributedlog-service/src/main/java/com/twitter/distributedlog/service/streamset/Partition.java
@@ -57,6 +57,14 @@ public class Partition {
return id;
}
+ /**
+ * Get the 6 digit 0 padded id of this partition as a String.
+ * @return partition id
+ */
+ public String getPaddedId() {
+ return String.format("%06d", getId());
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogServer.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogServer.java b/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogServer.java
index 94e8755..63723ef 100644
--- a/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogServer.java
+++ b/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogServer.java
@@ -106,9 +106,9 @@ public class TestDistributedLogServer extends DistributedLogServerTestCase {
int numRead = 0;
LogRecord r = reader.readNext(false);
while (null != r) {
- int i = Integer.parseInt(new String(r.getPayload()));
- assertEquals(numRead + 1, i);
++numRead;
+ int i = Integer.parseInt(new String(r.getPayload()));
+ assertEquals(numRead, i);
r = reader.readNext(false);
}
assertEquals(10, numRead);
@@ -121,7 +121,7 @@ public class TestDistributedLogServer extends DistributedLogServerTestCase {
*/
@Test(timeout = 60000)
public void testChecksumFlag() throws Exception {
- String name = "dlserver-basic-write";
+ String name = "testChecksumFlag";
LocalRoutingService routingService = LocalRoutingService.newBuilder().build();
routingService.addHost(name, dlServer.getAddress());
DistributedLogClientBuilder dlClientBuilder = DistributedLogClientBuilder.newBuilder()
@@ -134,7 +134,7 @@ public class TestDistributedLogServer extends DistributedLogServerTestCase {
.connectionTimeout(Duration.fromSeconds(1))
.requestTimeout(Duration.fromSeconds(60)))
.checksum(false);
- DistributedLogClient dlClient = (DistributedLogClientImpl) dlClientBuilder.build();
+ DistributedLogClient dlClient = dlClientBuilder.build();
Await.result(dlClient.write(name, ByteBuffer.wrap(("1").getBytes())));
dlClient.close();
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogService.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogService.java b/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogService.java
index ed456b9..17fae4a 100644
--- a/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogService.java
+++ b/distributedlog-service/src/test/java/com/twitter/distributedlog/service/TestDistributedLogService.java
@@ -600,6 +600,7 @@ public class TestDistributedLogService extends TestDistributedLogBase {
ByteBuffer.wrap("test".getBytes()),
new NullStatsLogger(),
new NullStatsLogger(),
+ new IdentityStreamPartitionConverter(),
new ServerConfiguration(),
(byte)0,
checksum,
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/a645c311/distributedlog-service/src/test/java/com/twitter/distributedlog/service/stream/TestStreamOp.java
----------------------------------------------------------------------
diff --git a/distributedlog-service/src/test/java/com/twitter/distributedlog/service/stream/TestStreamOp.java b/distributedlog-service/src/test/java/com/twitter/distributedlog/service/stream/TestStreamOp.java
index 93d900f..41b4c69 100644
--- a/distributedlog-service/src/test/java/com/twitter/distributedlog/service/stream/TestStreamOp.java
+++ b/distributedlog-service/src/test/java/com/twitter/distributedlog/service/stream/TestStreamOp.java
@@ -24,7 +24,7 @@ import com.twitter.distributedlog.acl.DefaultAccessControlManager;
import com.twitter.distributedlog.exceptions.InternalServerException;
import com.twitter.distributedlog.service.ResponseUtils;
import com.twitter.distributedlog.service.config.ServerConfiguration;
-import com.twitter.distributedlog.service.stream.WriteOp;
+import com.twitter.distributedlog.service.streamset.IdentityStreamPartitionConverter;
import com.twitter.distributedlog.thrift.service.StatusCode;
import com.twitter.distributedlog.thrift.service.WriteResponse;
import com.twitter.distributedlog.util.Sequencer;
@@ -67,6 +67,7 @@ public class TestStreamOp {
ByteBuffer.wrap("test".getBytes()),
new NullStatsLogger(),
new NullStatsLogger(),
+ new IdentityStreamPartitionConverter(),
new ServerConfiguration(),
(byte)0,
null,