distributedlog-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject incubator-distributedlog git commit: DL-85: Publish both per-stream stats and aggregation stats
Date Tue, 13 Dec 2016 23:40:34 GMT
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 <jbull@twitter.com>

Reviewers: Sijie Guo <sijie@apache.org>, Franck Cuny <fcuny@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 <jbull@twitter.com>
Authored: Tue Dec 13 15:40:29 2016 -0800
Committer: Sijie Guo <sijieg@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,


Mime
View raw message