accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From e..@apache.org
Subject [1/2] accumulo git commit: ACCUMULO-3696 limit the size of the queue for outgoing spans
Date Wed, 25 Mar 2015 22:34:10 GMT
Repository: accumulo
Updated Branches:
  refs/heads/master 0228da0a7 -> 483aee470


ACCUMULO-3696 limit the size of the queue for outgoing spans


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/f6ea7e6b
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/f6ea7e6b
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/f6ea7e6b

Branch: refs/heads/master
Commit: f6ea7e6b08033fb64cf65f2c7fc90cc8cec90531
Parents: 85942da
Author: Eric C. Newton <eric.newton@gmail.com>
Authored: Wed Mar 25 17:54:25 2015 -0400
Committer: Eric C. Newton <eric.newton@gmail.com>
Committed: Wed Mar 25 17:54:25 2015 -0400

----------------------------------------------------------------------
 .../apache/accumulo/core/trace/DistributedTrace.java |  2 +-
 .../apache/accumulo/core/trace/ZooTraceClient.java   |  4 ++--
 .../instrument/receivers/AsyncSpanReceiver.java      | 15 +++++++--------
 .../instrument/receivers/SendSpansViaThrift.java     |  9 +++++++--
 .../trace/instrument/receivers/ZooSpanClient.java    |  4 ++--
 5 files changed, 19 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/f6ea7e6b/core/src/main/java/org/apache/accumulo/core/trace/DistributedTrace.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/trace/DistributedTrace.java b/core/src/main/java/org/apache/accumulo/core/trace/DistributedTrace.java
index a5a327d..7b92676 100644
--- a/core/src/main/java/org/apache/accumulo/core/trace/DistributedTrace.java
+++ b/core/src/main/java/org/apache/accumulo/core/trace/DistributedTrace.java
@@ -37,6 +37,6 @@ public class DistributedTrace {
         address = "unknown";
       }
     }
-    Tracer.getInstance().addReceiver(new ZooTraceClient(zoo, path, address, application,
1000));
+    Tracer.getInstance().addReceiver(new ZooTraceClient(zoo, path, address, application,
1000, 5000));
   }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/f6ea7e6b/core/src/main/java/org/apache/accumulo/core/trace/ZooTraceClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/trace/ZooTraceClient.java b/core/src/main/java/org/apache/accumulo/core/trace/ZooTraceClient.java
index fd8db85..58108fb 100644
--- a/core/src/main/java/org/apache/accumulo/core/trace/ZooTraceClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/trace/ZooTraceClient.java
@@ -44,8 +44,8 @@ public class ZooTraceClient extends SendSpansViaThrift implements Watcher
{
   final Random random = new Random();
   final List<String> hosts = new ArrayList<String>();
 
-  public ZooTraceClient(ZooReader zoo, String path, String host, String service, long millis)
throws IOException, KeeperException, InterruptedException {
-    super(host, service, millis);
+  public ZooTraceClient(ZooReader zoo, String path, String host, String service, long millis,
int maxQueueSize) throws IOException, KeeperException, InterruptedException {
+    super(host, service, millis, maxQueueSize);
     this.path = path;
     this.zoo = zoo;
     updateHosts(path, zoo.getChildren(path, this));

http://git-wip-us.apache.org/repos/asf/accumulo/blob/f6ea7e6b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/AsyncSpanReceiver.java
----------------------------------------------------------------------
diff --git a/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/AsyncSpanReceiver.java
b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/AsyncSpanReceiver.java
index 6ebdee1..1226e03 100644
--- a/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/AsyncSpanReceiver.java
+++ b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/AsyncSpanReceiver.java
@@ -51,10 +51,12 @@ public abstract class AsyncSpanReceiver<SpanKey,Destination> implements
SpanRece
 
   Timer timer = new Timer("SpanSender", true);
   final AbstractQueue<RemoteSpan> sendQueue = new ConcurrentLinkedQueue<RemoteSpan>();
+  final int maxQueueSize;
 
-  public AsyncSpanReceiver(String host, String service, long millis) {
+  public AsyncSpanReceiver(String host, String service, long millis, int maxQueueSize) {
     this.host = host;
     this.service = service;
+    this.maxQueueSize = maxQueueSize;
     timer.schedule(new TimerTask() {
       @Override
       public void run() {
@@ -72,13 +74,6 @@ public abstract class AsyncSpanReceiver<SpanKey,Destination> implements
SpanRece
     while (!sendQueue.isEmpty()) {
       boolean sent = false;
       RemoteSpan s = sendQueue.peek();
-      if (s.stop - s.start < 1) {
-        synchronized (sendQueue) {
-          sendQueue.remove();
-          sendQueue.notifyAll();
-        }
-        continue;
-      }
       SpanKey dest = getSpanKey(s.data);
       Destination client = clients.get(dest);
       if (client == null) {
@@ -108,6 +103,10 @@ public abstract class AsyncSpanReceiver<SpanKey,Destination> implements
SpanRece
   @Override
   public void span(long traceId, long spanId, long parentId, long start, long stop, String
description, Map<String,String> data) {
 
+    if (sendQueue.size() > maxQueueSize) {
+      return;
+    }
+
     SpanKey dest = getSpanKey(data);
     if (dest != null) {
       sendQueue.add(new RemoteSpan(host, service, traceId, spanId, parentId, start, stop,
description, data));

http://git-wip-us.apache.org/repos/asf/accumulo/blob/f6ea7e6b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/SendSpansViaThrift.java
----------------------------------------------------------------------
diff --git a/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/SendSpansViaThrift.java
b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/SendSpansViaThrift.java
index 2fff6ca..23d943a 100644
--- a/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/SendSpansViaThrift.java
+++ b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/SendSpansViaThrift.java
@@ -16,6 +16,7 @@
  */
 package org.apache.accumulo.trace.instrument.receivers;
 
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.Map;
@@ -36,8 +37,8 @@ public class SendSpansViaThrift extends AsyncSpanReceiver<String,Client>
{
 
   private static final String THRIFT = "thrift://";
 
-  public SendSpansViaThrift(String host, String service, long millis) {
-    super(host, service, millis);
+  public SendSpansViaThrift(String host, String service, long millis, int maxQueueSize) {
+    super(host, service, millis, maxQueueSize);
   }
 
   @Override
@@ -55,6 +56,9 @@ public class SendSpansViaThrift extends AsyncSpanReceiver<String,Client>
{
       TTransport transport = new TSocket(sock);
       TProtocol prot = new TBinaryProtocol(transport);
       return new Client(prot);
+    } catch (IOException ex) {
+      log.trace(ex, ex);
+      return null;
     } catch (Exception ex) {
       log.error(ex, ex);
       return null;
@@ -73,6 +77,7 @@ public class SendSpansViaThrift extends AsyncSpanReceiver<String,Client>
{
     }
   }
 
+  @Override
   protected String getSpanKey(Map<String,String> data) {
     String dest = data.get("dest");
     if (dest != null && dest.startsWith(THRIFT)) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/f6ea7e6b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/ZooSpanClient.java
----------------------------------------------------------------------
diff --git a/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/ZooSpanClient.java
b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/ZooSpanClient.java
index 0466426..7a6e44f 100644
--- a/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/ZooSpanClient.java
+++ b/trace/src/main/java/org/apache/accumulo/trace/instrument/receivers/ZooSpanClient.java
@@ -46,8 +46,8 @@ public class ZooSpanClient extends SendSpansViaThrift {
   final Random random = new Random();
   final List<String> hosts = new ArrayList<String>();
 
-  public ZooSpanClient(String keepers, final String path, String host, String service, long
millis) throws IOException, KeeperException, InterruptedException {
-    super(host, service, millis);
+  public ZooSpanClient(String keepers, final String path, String host, String service, long
millis, int maxQueueSize) throws IOException, KeeperException, InterruptedException {
+    super(host, service, millis, maxQueueSize);
     this.path = path;
     zoo = new ZooKeeper(keepers, 30 * 1000, new Watcher() {
       @Override


Mime
View raw message