Updated Branches: refs/heads/apache-blur-0.2 89f5a5b08 -> a00ec88e2
Adding more tracing logic Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/a00ec88e Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/a00ec88e Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/a00ec88e Branch: refs/heads/apache-blur-0.2 Commit: a00ec88e20f22c16dcaec8911b50ae8e37082272 Parents: 89f5a5b Author: Aaron McCurry <[email protected]> Authored: Fri Nov 22 14:50:40 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Fri Nov 22 14:50:40 2013 -0500 ---------------------------------------------------------------------- .../org/apache/blur/manager/IndexManager.java | 15 ++- .../blur/manager/writer/BlurNRTIndex.java | 3 +- .../blur/thrift/BlurControllerServer.java | 9 +- .../blur/thrift/ThriftBlurControllerServer.java | 6 +- .../blur/thrift/ThriftBlurShardServer.java | 3 +- .../org/apache/blur/thrift/ThriftServer.java | 15 ++- .../java/org/apache/blur/utils/BlurUtil.java | 3 + .../main/java/org/apache/blur/shell/Main.java | 44 ++++++ .../java/org/apache/blur/thrift/BlurClient.java | 5 - .../apache/blur/thrift/BlurClientManager.java | 5 + .../org/apache/blur/thrift/util/StatsTable.java | 6 +- .../org/apache/blur/concurrent/Executors.java | 8 +- .../org/apache/blur/trace/LogTraceReporter.java | 7 +- .../main/java/org/apache/blur/trace/Trace.java | 111 +++++++++++++-- .../org/apache/blur/trace/TraceCollector.java | 20 +-- .../org/apache/blur/trace/TraceReporter.java | 12 +- .../java/org/apache/blur/trace/TracerImpl.java | 8 +- .../blur/trace/ZooKeeperTraceReporter.java | 135 +++++++++++++++++++ .../org/apache/blur/utils/BlurConstants.java | 1 + .../apache/blur/zookeeper/ZooKeeperClient.java | 41 +++--- .../src/main/resources/blur-default.properties | 3 + .../java/org/apache/blur/trace/TraceTest.java | 26 +++- 22 files changed, 413 insertions(+), 73 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java index 0b0be0a..3a0ddcc 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java +++ b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java @@ -84,6 +84,8 @@ import org.apache.blur.thrift.generated.RowMutationType; import org.apache.blur.thrift.generated.Schema; import org.apache.blur.thrift.generated.ScoreType; import org.apache.blur.thrift.generated.Selector; +import org.apache.blur.trace.Trace; +import org.apache.blur.trace.Tracer; import org.apache.blur.utils.BlurExecutorCompletionService; import org.apache.blur.utils.BlurExecutorCompletionService.Cancel; import org.apache.blur.utils.BlurUtil; @@ -223,7 +225,8 @@ public class IndexManager { } catch (InterruptedException e) { throw new BException("Unkown error while fetching batch table [{0}] selectors [{1}].", e, table, selectors); } catch (ExecutionException e) { - throw new BException("Unkown error while fetching batch table [{0}] selectors [{1}].", e.getCause(), table, selectors); + throw new BException("Unkown error while fetching batch table [{0}] selectors [{1}].", e.getCause(), table, + selectors); } } return results; @@ -429,6 +432,7 @@ public class IndexManager { LOG.error("Unknown error while trying to fetch index readers.", e); throw new BException(e.getMessage(), e); } + Tracer trace = Trace.trace("query setup"); ShardServerContext shardServerContext = ShardServerContext.getShardServerContext(); ParallelCall<Entry<String, BlurIndex>, BlurResultIterable> call; TableContext context = getTableContext(table); @@ -444,6 +448,7 @@ public class IndexManager { preFilter); call = new SimpleQueryParallelCall(running, table, status, facetedQuery, blurQuery.selector, _queriesInternalMeter, shardServerContext, runSlow, _fetchCount, _maxHeapPerRowFetch, context.getSimilarity()); + trace.done(); MergerBlurResultIterable merger = new MergerBlurResultIterable(blurQuery); return ForkJoin.execute(_executor, blurIndexes.entrySet(), call, new Cancel() { @Override @@ -1120,6 +1125,7 @@ public class IndexManager { _status.attachThread(shard); BlurIndex index = entry.getValue(); IndexSearcherClosable searcher = index.getIndexSearcher(); + Tracer trace2 = null; try { IndexReader indexReader = searcher.getIndexReader(); if (indexReader instanceof ExitableReader) { @@ -1132,16 +1138,20 @@ public class IndexManager { _shardServerContext.setIndexSearcherClosable(_table, shard, searcher); } searcher.setSimilarity(_similarity); + Tracer trace1 = Trace.trace("query rewrite"); Query rewrite; try { rewrite = searcher.rewrite((Query) _query.clone()); } catch (ExitingReaderException e) { LOG.info("Query [{0}] has been cancelled during the rewrite phase.", _query); throw e; + } finally { + trace1.done(); } // BlurResultIterableSearcher will close searcher, if shard server // context is null. + trace2 = Trace.trace("query initial search"); return new BlurResultIterableSearcher(_running, rewrite, _table, shard, searcher, _selector, _shardServerContext == null, _runSlow, _fetchCount, _maxHeapPerRowFetch); } catch (BlurException e) { @@ -1157,6 +1167,9 @@ public class IndexManager { throw e; } } finally { + if (trace2 != null) { + trace2.done(); + } _queriesInternalMeter.mark(); _status.deattachThread(shard); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java index 7e77db2..b56b53b 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java @@ -137,7 +137,8 @@ public class BlurNRTIndex extends BlurIndex { _searcherFactory = new SearcherFactory() { @Override public IndexSearcher newSearcher(IndexReader reader) throws IOException { - return new IndexSearcherClosableNRT(reader, searchExecutor, _nrtManagerRef, _directory); +// return new IndexSearcherClosableNRT(reader, searchExecutor, _nrtManagerRef, _directory); + return new IndexSearcherClosableNRT(reader, null, _nrtManagerRef, _directory); } }; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java index ed15d34..299abdc 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java @@ -79,6 +79,8 @@ import org.apache.blur.thrift.generated.ShardState; import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.thrift.generated.TableStats; import org.apache.blur.thrift.generated.User; +import org.apache.blur.trace.Trace; +import org.apache.blur.trace.Tracer; import org.apache.blur.utils.BlurConstants; import org.apache.blur.utils.BlurExecutorCompletionService; import org.apache.blur.utils.BlurIterator; @@ -111,7 +113,12 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public <T> T execute(String node, BlurCommand<T> command, int maxRetries, long backOffTime, long maxBackOffTime) throws BlurException, TException, IOException { - return BlurClientManager.execute(node + "#" + _timeout, command, maxRetries, backOffTime, maxBackOffTime); + Tracer trace = Trace.trace("remote call - " + node); + try { + return BlurClientManager.execute(node + "#" + _timeout, command, maxRetries, backOffTime, maxBackOffTime); + } finally { + trace.done(); + } } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java index 91e5d04..3d44c4e 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java @@ -58,6 +58,7 @@ import org.apache.blur.metrics.ReporterSetup; import org.apache.blur.server.ControllerServerEventHandler; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TNonblockingServerSocket; import org.apache.blur.thrift.generated.Blur.Iface; +import org.apache.blur.trace.Trace; import org.apache.blur.utils.BlurUtil; import org.apache.blur.utils.MemoryReporter; import org.apache.blur.zookeeper.ZkUtils; @@ -134,10 +135,11 @@ public class ThriftBlurControllerServer extends ThriftServer { controllerServer.init(); + Trace.setReporter(setupTraceReporter(configuration)); + Iface iface= BlurUtil.wrapFilteredBlurServer(configuration, controllerServer, false); - iface = BlurUtil.recordMethodCallsAndAverageTimes(iface, Iface.class, true); - + iface = BlurUtil.runTrace(iface, true); int threadCount = configuration.getInt(BLUR_CONTROLLER_SERVER_THRIFT_THREAD_COUNT, 32); ControllerServerEventHandler eventHandler = new ControllerServerEventHandler(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java index 506a8b3..06305f3 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java @@ -86,7 +86,6 @@ import org.apache.blur.thirdparty.thrift_0_9_0.server.TServlet; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TNonblockingServerSocket; import org.apache.blur.thrift.generated.Blur; import org.apache.blur.thrift.generated.Blur.Iface; -import org.apache.blur.trace.LogTraceReporter; import org.apache.blur.trace.Trace; import org.apache.blur.utils.BlurUtil; import org.apache.blur.utils.GCWatcher; @@ -232,7 +231,7 @@ public class ThriftBlurShardServer extends ThriftServer { shardServer.setConfiguration(configuration); shardServer.init(); - Trace.setReporter(new LogTraceReporter()); + Trace.setReporter(setupTraceReporter(configuration)); Iface iface = BlurUtil.wrapFilteredBlurServer(configuration, shardServer, true); iface = BlurUtil.recordMethodCallsAndAverageTimes(iface, Iface.class, false); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java index 81423cd..a7e240c 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java @@ -22,6 +22,7 @@ import static org.apache.blur.metrics.MetricsConstants.JVM; import static org.apache.blur.metrics.MetricsConstants.LOAD_AVERAGE; import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; import static org.apache.blur.metrics.MetricsConstants.SYSTEM; +import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TRACE_PATH; import java.io.BufferedReader; import java.io.IOException; @@ -53,13 +54,16 @@ import org.apache.blur.thrift.generated.Blur; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.server.TThreadedSelectorServer; import org.apache.blur.thrift.server.TThreadedSelectorServer.Args.AcceptPolicy; +import org.apache.blur.trace.LogTraceReporter; +import org.apache.blur.trace.TraceReporter; +import org.apache.blur.trace.ZooKeeperTraceReporter; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Gauge; import com.yammer.metrics.core.MetricName; public class ThriftServer { - + private static final Log LOG = LogFactory.getLog(ThriftServer.class); private String _nodeName; @@ -85,6 +89,14 @@ public class ThriftServer { _serverTransport = serverTransport; } + public static TraceReporter setupTraceReporter(BlurConfiguration configuration) throws IOException { + String path = configuration.get(BLUR_ZOOKEEPER_TRACE_PATH); + if (path == null || path.isEmpty()) { + return new LogTraceReporter(configuration); + } + return new ZooKeeperTraceReporter(configuration); + } + public static void printUlimits() throws IOException { ProcessBuilder processBuilder = new ProcessBuilder("bash", "-c", "ulimit -a"); Process process; @@ -293,7 +305,6 @@ public class ThriftServer { _eventHandler = eventHandler; } - public int getAcceptQueueSizePerThread() { return _acceptQueueSizePerThread; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java index 9a6c004..d11c291 100644 --- a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java +++ b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java @@ -89,6 +89,7 @@ import org.apache.blur.thrift.generated.RowMutationType; import org.apache.blur.thrift.generated.Selector; import org.apache.blur.thrift.util.ResetableTMemoryBuffer; import org.apache.blur.trace.Trace; +import org.apache.blur.trace.Tracer; import org.apache.hadoop.conf.Configurable; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; @@ -187,6 +188,7 @@ public class BlurUtil { String name = method.getName(); boolean error = false; LoggerArgsState loggerArgsState = null; + Tracer trace = Trace.trace("thrift - "+method.getName()); try { if (REQUEST_LOG.isInfoEnabled()) { if (argsStr == null) { @@ -200,6 +202,7 @@ public class BlurUtil { error = true; throw e.getTargetException(); } finally { + trace.done(); long end = System.nanoTime(); double ms = (end - start) / 1000000.0; if (RESPONSE_LOG.isInfoEnabled()) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-shell/src/main/java/org/apache/blur/shell/Main.java ---------------------------------------------------------------------- diff --git a/blur-shell/src/main/java/org/apache/blur/shell/Main.java b/blur-shell/src/main/java/org/apache/blur/shell/Main.java index aacc9dd..c202f8a 100644 --- a/blur-shell/src/main/java/org/apache/blur/shell/Main.java +++ b/blur-shell/src/main/java/org/apache/blur/shell/Main.java @@ -27,6 +27,7 @@ import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.Map.Entry; import java.util.Set; import java.util.TreeMap; @@ -49,6 +50,7 @@ import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.Selector; import org.apache.blur.thrift.generated.User; +import org.apache.blur.trace.Trace; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableMap.Builder; @@ -59,6 +61,8 @@ public class Main { static boolean debug = false; /** is timing enabled - off by default */ static boolean timed = false; + /** is tracing enabled - off by default */ + static boolean trace = false; /** is highlight enabled - off by default */ static boolean highlight = false; /** default selector */ @@ -333,6 +337,36 @@ public class Main { } + private static class TraceCommand extends Command { + + @Override + public void doit(PrintWriter out, Blur.Iface client, String[] args) throws CommandException, TException, + BlurException { + if (trace == true) { + trace = false; + } else { + trace = true; + } + out.println("tracing of commands is now " + (trace ? "on" : "off")); + } + + @Override + public String description() { + return "Toggle tracing of commands on/off."; + } + + @Override + public String usage() { + return ""; + } + + @Override + public String name() { + return "trace"; + } + + } + private static class HighlightCommand extends Command { @Override @@ -534,7 +568,16 @@ public class Main { long start = System.nanoTime(); try { client.setUser(user); + String traceId = null; + if (trace) { + traceId = UUID.randomUUID().toString(); + Trace.setupTrace(traceId); + out.println("Running trace with id: " + traceId); + } command.doit(out, client, commandArgs); + if (trace) { + Trace.tearDownTrace(); + } } catch (QuitCommandException e) { // exit gracefully System.exit(0); @@ -626,6 +669,7 @@ public class Main { register(builder, new DiscoverFileBufferSizeUtil()); register(builder, new WhoAmICommand()); register(builder, new UserCommand()); + register(builder, new TraceCommand()); commands = builder.build(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java index 9bf98dd..086a1ad 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java @@ -31,7 +31,6 @@ import org.apache.blur.thrift.generated.Blur.Client; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.User; -import org.apache.blur.trace.Trace; public class BlurClient { @@ -80,10 +79,6 @@ public class BlurClient { if (_setUser) { client.setUser(user); } - String traceId = Trace.getTraceId(); - if (traceId != null) { - client.startTrace(traceId); - } return method.invoke(client, args); } } catch (IllegalArgumentException e) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClientManager.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClientManager.java b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClientManager.java index 40604ff..6c389fa 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClientManager.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClientManager.java @@ -40,6 +40,7 @@ import org.apache.blur.thrift.generated.Blur; import org.apache.blur.thrift.generated.Blur.Client; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.ErrorType; +import org.apache.blur.trace.Trace; public class BlurClientManager { @@ -164,6 +165,10 @@ public class BlurClientManager { } } try { + String traceId = Trace.getTraceId(); + if (traceId != null) { + client.get().startTrace(traceId); + } T result = command.call((CLIENT) client.get(), connection); allBad = false; if (command.isDetachClient()) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-thrift/src/main/java/org/apache/blur/thrift/util/StatsTable.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/util/StatsTable.java b/blur-thrift/src/main/java/org/apache/blur/thrift/util/StatsTable.java index a37a558..9d9c4e4 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/util/StatsTable.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/util/StatsTable.java @@ -17,6 +17,7 @@ package org.apache.blur.thrift.util; * limitations under the License. */ import java.io.IOException; +import java.util.UUID; import org.apache.blur.thirdparty.thrift_0_9_0.TException; import org.apache.blur.thrift.BlurClient; @@ -30,10 +31,11 @@ public class StatsTable { String connectionStr = args[0]; final String tableName = args[1]; - Trace.setupTrace("1234"); + String uuid = UUID.randomUUID().toString(); + Trace.setupTrace(uuid); + System.out.println("Tracing on [" + uuid + "]"); Iface client = BlurClient.getClient(connectionStr); System.out.println(client.tableStats(tableName)); - System.out.println(client.tableStats(tableName)); Trace.tearDownTrace(); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/concurrent/Executors.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/concurrent/Executors.java b/blur-util/src/main/java/org/apache/blur/concurrent/Executors.java index f6b8fb2..e9fce56 100644 --- a/blur-util/src/main/java/org/apache/blur/concurrent/Executors.java +++ b/blur-util/src/main/java/org/apache/blur/concurrent/Executors.java @@ -23,6 +23,8 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.apache.blur.trace.Trace; + public class Executors { public static ExecutorService newThreadPool(String prefix, int threadCount) { @@ -33,13 +35,13 @@ public class Executors { ThreadPoolExecutor executorService = new ThreadPoolExecutor(threadCount, threadCount, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new BlurThreadFactory(prefix)); executorService.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); if (watch) { - return ThreadWatcher.instance().watch(executorService); + return Trace.getExecutorService(ThreadWatcher.instance().watch(executorService)); } - return executorService; + return Trace.getExecutorService(executorService); } public static ExecutorService newSingleThreadExecutor(String prefix) { - return java.util.concurrent.Executors.newSingleThreadExecutor(new BlurThreadFactory(prefix)); + return Trace.getExecutorService(java.util.concurrent.Executors.newSingleThreadExecutor(new BlurThreadFactory(prefix))); } public static class BlurThreadFactory implements ThreadFactory { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/trace/LogTraceReporter.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/LogTraceReporter.java b/blur-util/src/main/java/org/apache/blur/trace/LogTraceReporter.java index 83e9b25..4c0f280 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/LogTraceReporter.java +++ b/blur-util/src/main/java/org/apache/blur/trace/LogTraceReporter.java @@ -16,13 +16,18 @@ */ package org.apache.blur.trace; +import org.apache.blur.BlurConfiguration; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; -public class LogTraceReporter implements TraceReporter { +public class LogTraceReporter extends TraceReporter { private static final Log LOG = LogFactory.getLog(LogTraceReporter.class); + public LogTraceReporter(BlurConfiguration configuration) { + super(configuration); + } + @Override public void report(TraceCollector collector) { String json = collector.toJson(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/trace/Trace.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/Trace.java b/blur-util/src/main/java/org/apache/blur/trace/Trace.java index 7446950..a8b0b3f 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/Trace.java +++ b/blur-util/src/main/java/org/apache/blur/trace/Trace.java @@ -16,7 +16,16 @@ */ package org.apache.blur.trace; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.UUID; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; public class Trace { @@ -33,6 +42,15 @@ public class Trace { TraceCollector collector = new TraceCollector(id); _tracer.set(collector); } + + private static void setupTraceOnNewThread(TraceCollector parentCollector) { + TraceCollector collector = new TraceCollector(getNewThreadId(parentCollector)); + _tracer.set(collector); + } + + private static String getNewThreadId(TraceCollector parentCollector) { + return parentCollector._id + ":" + Thread.currentThread().getName() + "/" + UUID.randomUUID().toString(); + } public static void tearDownTrace() { TraceCollector collector = _tracer.get(); @@ -47,7 +65,7 @@ public class Trace { if (collector == null) { return DO_NOTHING; } - TracerImpl tracer = new TracerImpl(name, collector.getNextId(), collector.getParentThreadId()); + TracerImpl tracer = new TracerImpl(name, collector.getNextId()); collector.add(tracer); return tracer; } @@ -61,43 +79,53 @@ public class Trace { } public static Runnable getRunnable(final Runnable runnable) { - TraceCollector tc = _tracer.get(); + final TraceCollector tc = _tracer.get(); if (tc == null) { return runnable; } - final TraceCollector traceCollector = new TraceCollector(tc); return new Runnable() { @Override public void run() { - _tracer.set(traceCollector); + setupTraceOnNewThread(tc); try { runnable.run(); } finally { - _tracer.set(null); + tearDownTrace(); } } }; } - public static <V> Callable<V> getRunnable(final Callable<V> callable) { - TraceCollector tc = _tracer.get(); + public static <V> Callable<V> getCallable(final Callable<V> callable) { + final TraceCollector tc = _tracer.get(); if (tc == null) { return callable; } - final TraceCollector traceCollector = new TraceCollector(tc); return new Callable<V>() { @Override public V call() throws Exception { - _tracer.set(traceCollector); + setupTraceOnNewThread(tc); try { return callable.call(); } finally { - _tracer.set(null); + tearDownTrace(); } } }; } + public static <V> Collection<Callable<V>> getCallables(final Collection<Callable<V>> callables) { + TraceCollector tc = _tracer.get(); + if (tc == null) { + return callables; + } + Collection<Callable<V>> list = new ArrayList<Callable<V>>(callables.size()); + for (Callable<V> c : callables) { + list.add(getCallable(c)); + } + return list; + } + public static String getTraceId() { TraceCollector collector = _tracer.get(); if (collector == null) { @@ -114,4 +142,67 @@ public class Trace { return true; } + @SuppressWarnings("unchecked") + public static ExecutorService getExecutorService(final ExecutorService service) { + return new ExecutorService() { + ExecutorService _service = service; + + public void execute(Runnable command) { + _service.execute(getRunnable(command)); + } + + public <T> Future<T> submit(Callable<T> task) { + return _service.submit(getCallable(task)); + } + + public <T> Future<T> submit(Runnable task, T result) { + return _service.submit(getRunnable(task), result); + } + + public Future<?> submit(Runnable task) { + return _service.submit(getRunnable(task)); + } + + public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException { + return _service.invokeAll((Collection<? extends Callable<T>>) getCallable((Callable<T>) tasks)); + } + + public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) + throws InterruptedException { + return _service.invokeAll((Collection<? extends Callable<T>>) getCallable((Callable<T>) tasks), timeout, unit); + } + + public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException { + return _service.invokeAny((Collection<? extends Callable<T>>) getCallable((Callable<T>) tasks)); + } + + public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + return _service.invokeAny((Collection<? extends Callable<T>>) getCallable((Callable<T>) tasks), timeout, unit); + } + + // pass-through + + public void shutdown() { + _service.shutdown(); + } + + public List<Runnable> shutdownNow() { + return _service.shutdownNow(); + } + + public boolean isShutdown() { + return _service.isShutdown(); + } + + public boolean isTerminated() { + return _service.isTerminated(); + } + + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + return _service.awaitTermination(timeout, unit); + } + + }; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java b/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java index d06ac1a..e3ab598 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java +++ b/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java @@ -16,6 +16,7 @@ */ package org.apache.blur.trace; +import java.lang.management.ManagementFactory; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicLong; @@ -25,20 +26,14 @@ public class TraceCollector { protected final String _id; protected final List<TracerImpl> _traces; protected final AtomicLong _traceCounter; - protected final long _parentThreadId; + protected final long _now = System.nanoTime(); + protected final String _pid; public TraceCollector(String id) { _id = id; _traces = new CopyOnWriteArrayList<TracerImpl>(); _traceCounter = new AtomicLong(); - _parentThreadId = Long.MAX_VALUE; - } - - public TraceCollector(TraceCollector traceCollector) { - _id = traceCollector._id; - _traces = traceCollector._traces; - _traceCounter = traceCollector._traceCounter; - _parentThreadId = Thread.currentThread().getId(); + _pid = ManagementFactory.getRuntimeMXBean().getName(); } public void add(TracerImpl tracer) { @@ -57,7 +52,7 @@ public class TraceCollector { } - return "{\n \"id\"=\"" + _id + "\",\n \"traces\"=[\n" + builder.toString() + " ]\n}"; + return "{\n \"id\"=\"" + _id + "\"\n \"pid\"=\"" + _pid + "\"\n \"created\"=" + _now + ",\n \"traces\"=[\n" + builder.toString() + " ]\n}"; } public String getId() { @@ -71,9 +66,4 @@ public class TraceCollector { public long getNextId() { return _traceCounter.incrementAndGet(); } - - public long getParentThreadId() { - return _parentThreadId; - } - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/trace/TraceReporter.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/TraceReporter.java b/blur-util/src/main/java/org/apache/blur/trace/TraceReporter.java index fa73ac1..96cf053 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/TraceReporter.java +++ b/blur-util/src/main/java/org/apache/blur/trace/TraceReporter.java @@ -16,8 +16,16 @@ */ package org.apache.blur.trace; -public interface TraceReporter { +import org.apache.blur.BlurConfiguration; - void report(TraceCollector collector); +public abstract class TraceReporter { + + protected BlurConfiguration _configuration; + + public TraceReporter(BlurConfiguration configuration) { + _configuration = configuration; + } + + public abstract void report(TraceCollector collector); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java b/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java index 3056a73..0f4eb87 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java +++ b/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java @@ -23,14 +23,12 @@ public class TracerImpl implements Tracer { protected long _ended; protected final String _threadName; protected final long _id; - protected final long _parentThreadId; - public TracerImpl(String name, long id, long parentThreadId) { + public TracerImpl(String name, long id) { _name = name; _start = System.nanoTime(); _threadName = Thread.currentThread().getName(); _id = id; - _parentThreadId = parentThreadId; } @Override @@ -61,8 +59,8 @@ public class TracerImpl implements Tracer { } public String toJson() { - return "{\"id\"=" + _id + ", \"parentThread\"=" + _parentThreadId + ", \"name\"=\"" + _name + "\", \"thread\"=\"" - + _threadName + "\", \"took\"=" + (_ended - _start) + ", \"started\"=" + _start + ", \"ended\"=" + _ended + "}"; + return "{\"id\"=" + _id + ", \"name\"=\"" + _name + "\", \"thread\"=\"" + _threadName + "\", \"took\"=" + + (_ended - _start) + ", \"started\"=" + _start + ", \"ended\"=" + _ended + "}"; } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceReporter.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceReporter.java b/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceReporter.java new file mode 100644 index 0000000..32bb6b9 --- /dev/null +++ b/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceReporter.java @@ -0,0 +1,135 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.blur.trace; + +import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; +import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TRACE_PATH; + +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; +import org.apache.blur.zookeeper.ZooKeeperClient; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.KeeperException.Code; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooDefs.Ids; +import org.apache.zookeeper.data.Stat; + +public class ZooKeeperTraceReporter extends TraceReporter { + + private final static Log LOG = LogFactory.getLog(ZooKeeperTraceReporter.class); + + private final String _zkConnectionStr; + private final ZooKeeperClient _zooKeeperClient; + private final String _storePath; + private final BlockingQueue<TraceCollector> _queue = new LinkedBlockingQueue<TraceCollector>(); + private final Thread _daemon; + + public ZooKeeperTraceReporter(BlurConfiguration configuration) throws IOException { + super(configuration); + _zkConnectionStr = configuration.get(BLUR_ZOOKEEPER_CONNECTION); + _storePath = configuration.get(BLUR_ZOOKEEPER_TRACE_PATH); + _zooKeeperClient = new ZooKeeperClient(_zkConnectionStr, 30000, new Watcher() { + @Override + public void process(WatchedEvent event) { + + } + }); + createIfMissing(_storePath); + _daemon = new Thread(new Runnable() { + @Override + public void run() { + while (true) { + TraceCollector collector; + try { + collector = _queue.take(); + } catch (InterruptedException e) { + return; + } + try { + writeCollector(collector); + } catch (Throwable t) { + LOG.error("Unknown error while trying to write collector.", t); + } + } + } + }); + _daemon.setName("ZooKeeper Trace Queue Writer"); + _daemon.start(); + } + + @Override + public void report(TraceCollector collector) { + try { + _queue.put(collector); + } catch (InterruptedException e) { + LOG.error("Unknown error while trying to add collector on queue", e); + } + } + + private void writeCollector(TraceCollector collector) { + String id = collector.getId(); + int indexOf = id.indexOf(":"); + String storeId; + if (indexOf < 0) { + storeId = id; + } else { + storeId = id.substring(0, indexOf); + } + String storePath = getStorePath(storeId); + createIfMissing(storePath); + String json = collector.toJson(); + try { + _zooKeeperClient.create(storePath + "/trace-", json.getBytes("UTF-8"), Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT_SEQUENTIAL); + } catch (UnsupportedEncodingException e) { + throw new RuntimeException(e); + } catch (KeeperException e) { + throw new RuntimeException(e); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + private void createIfMissing(String storePath) { + try { + Stat stat = _zooKeeperClient.exists(storePath, false); + if (stat == null) { + _zooKeeperClient.create(storePath, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + } + } catch (KeeperException e) { + if (e.code() == Code.NODEEXISTS) { + return; + } + throw new RuntimeException(e); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + private String getStorePath(String storeId) { + return _storePath + "/" + storeId; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java index 324fa20..525b4bd 100644 --- a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java +++ b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java @@ -38,6 +38,7 @@ public class BlurConstants { public static final String BLUR_TABLE_PATH = "blur.table.path"; public static final String BLUR_ZOOKEEPER_CONNECTION = "blur.zookeeper.connection"; + public static final String BLUR_ZOOKEEPER_TRACE_PATH = "blur.zookeeper.trace.path"; public static final String BLUR_ZOOKEEPER_TIMEOUT = "blur.zookeeper.timeout"; public static final int BLUR_ZOOKEEPER_TIMEOUT_DEFAULT = 30000; public static final String BLUR_SHARD_HOSTNAME = "blur.shard.hostname"; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java b/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java index f67c1d0..9ea5a30 100644 --- a/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java +++ b/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java @@ -21,13 +21,15 @@ import java.util.List; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; +import org.apache.blur.trace.Trace; +import org.apache.blur.trace.Tracer; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.Op; import org.apache.zookeeper.OpResult; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; -import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.data.Stat; @@ -60,6 +62,12 @@ public class ZooKeeperClient extends ZooKeeper { } static abstract class ZKExecutor<T> { + String _name; + + ZKExecutor(String name) { + _name = name; + } + abstract T execute() throws KeeperException, InterruptedException; } @@ -70,6 +78,7 @@ public class ZooKeeperClient extends ZooKeeper { sessionTimeout = internalSessionTimeout; } while (true) { + Tracer trace = Trace.trace("remote call - zookeeper - " + executor._name); try { return executor.execute(); } catch (KeeperException e) { @@ -79,6 +88,8 @@ public class ZooKeeperClient extends ZooKeeper { continue; } throw e; + } finally { + trace.done(); } } } @@ -86,7 +97,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public String create(final String path, final byte[] data, final List<ACL> acl, final CreateMode createMode) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<String>() { + return execute(new ZKExecutor<String>("create") { @Override String execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - create [{0}] [{1}] [{2}] [{3}]", path, data, acl, createMode); @@ -97,7 +108,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public void delete(final String path, final int version) throws InterruptedException, KeeperException { - execute(new ZKExecutor<Void>() { + execute(new ZKExecutor<Void>("delete") { @Override Void execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - delete [{0}] [{1}]", path, version); @@ -109,7 +120,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public List<OpResult> multi(final Iterable<Op> ops) throws InterruptedException, KeeperException { - return execute(new ZKExecutor<List<OpResult>>() { + return execute(new ZKExecutor<List<OpResult>>("multi") { @Override List<OpResult> execute() throws KeeperException, InterruptedException { return ZooKeeperClient.super.multi(ops); @@ -119,7 +130,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public Stat exists(final String path, final Watcher watcher) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<Stat>() { + return execute(new ZKExecutor<Stat>("exists") { @Override Stat execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - exists [{0}] [{1}]", path, watcher); @@ -130,7 +141,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public Stat exists(final String path, final boolean watch) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<Stat>() { + return execute(new ZKExecutor<Stat>("exists") { @Override Stat execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - exists [{0}] [{1}]", path, watch); @@ -142,7 +153,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public byte[] getData(final String path, final Watcher watcher, final Stat stat) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<byte[]>() { + return execute(new ZKExecutor<byte[]>("getData") { @Override byte[] execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - getData [{0}] [{1}] [{2}]", path, watcher, stat); @@ -154,7 +165,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public byte[] getData(final String path, final boolean watch, final Stat stat) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<byte[]>() { + return execute(new ZKExecutor<byte[]>("getData") { @Override byte[] execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - getData [{0}] [{1}] [{2}]", path, watch, stat); @@ -166,7 +177,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public Stat setData(final String path, final byte[] data, final int version) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<Stat>() { + return execute(new ZKExecutor<Stat>("setData") { @Override Stat execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - setData [{0}] [{1}] [{2}]", path, data, version); @@ -177,7 +188,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public List<ACL> getACL(final String path, final Stat stat) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<List<ACL>>() { + return execute(new ZKExecutor<List<ACL>>("getACL") { @Override List<ACL> execute() throws KeeperException, InterruptedException { return ZooKeeperClient.super.getACL(path, stat); @@ -188,7 +199,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public Stat setACL(final String path, final List<ACL> acl, final int version) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<Stat>() { + return execute(new ZKExecutor<Stat>("setACL") { @Override Stat execute() throws KeeperException, InterruptedException { return ZooKeeperClient.super.setACL(path, acl, version); @@ -199,7 +210,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public List<String> getChildren(final String path, final Watcher watcher) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<List<String>>() { + return execute(new ZKExecutor<List<String>>("getChildren") { @Override List<String> execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - getChildren [{0}] [{1}]", path, watcher); @@ -210,7 +221,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public List<String> getChildren(final String path, final boolean watch) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<List<String>>() { + return execute(new ZKExecutor<List<String>>("getChildren") { @Override List<String> execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - getChildren [{0}] [{1}]", path, watch); @@ -222,7 +233,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public List<String> getChildren(final String path, final Watcher watcher, final Stat stat) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<List<String>>() { + return execute(new ZKExecutor<List<String>>("getChildren") { @Override List<String> execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - getChildren [{0}] [{1}] [{2}]", path, watcher, stat); @@ -234,7 +245,7 @@ public class ZooKeeperClient extends ZooKeeper { @Override public List<String> getChildren(final String path, final boolean watch, final Stat stat) throws KeeperException, InterruptedException { - return execute(new ZKExecutor<List<String>>() { + return execute(new ZKExecutor<List<String>>("getChildren") { @Override List<String> execute() throws KeeperException, InterruptedException { LOG.debug("ZK Call - getChildren [{0}] [{1}] [{2}]", path, watch, stat); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/main/resources/blur-default.properties ---------------------------------------------------------------------- diff --git a/blur-util/src/main/resources/blur-default.properties b/blur-util/src/main/resources/blur-default.properties index 4e4c997..d575daf 100644 --- a/blur-util/src/main/resources/blur-default.properties +++ b/blur-util/src/main/resources/blur-default.properties @@ -16,6 +16,9 @@ # The zookeeper session timeout blur.zookeeper.timeout=90000 +# The path in ZooKeeper where the distributed traces will be stored, if blank trace output will be written to the log +blur.zookeeper.trace.path= + # The maximum number of results that can be fetched in a single request blur.query.max.results.fetch=1000 http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a00ec88e/blur-util/src/test/java/org/apache/blur/trace/TraceTest.java ---------------------------------------------------------------------- diff --git a/blur-util/src/test/java/org/apache/blur/trace/TraceTest.java b/blur-util/src/test/java/org/apache/blur/trace/TraceTest.java index f1fbdfa..dd5faaf 100644 --- a/blur-util/src/test/java/org/apache/blur/trace/TraceTest.java +++ b/blur-util/src/test/java/org/apache/blur/trace/TraceTest.java @@ -18,13 +18,17 @@ package org.apache.blur.trace; import static org.junit.Assert.assertEquals; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicLong; + +import org.apache.blur.BlurConfiguration; import org.junit.Test; public class TraceTest { @Test - public void testTrace() { - Trace.setReporter(new TraceReporter() { + public void testTrace() throws IOException { + Trace.setReporter(new TraceReporter(new BlurConfiguration()) { @Override public void report(TraceCollector collector) { assertEquals("test", collector.getId()); @@ -56,13 +60,21 @@ public class TraceTest { } @Test - public void testTraceThreadRunnable() throws InterruptedException { - Trace.setReporter(new TraceReporter() { + public void testTraceThreadRunnable() throws InterruptedException, IOException { + final AtomicLong count = new AtomicLong(); + Trace.setReporter(new TraceReporter(new BlurConfiguration()) { @Override public void report(TraceCollector collector) { System.out.println(collector.toJson()); - assertEquals("test", collector.getId()); - assertEquals(6, collector.getTraces().size()); + String id = collector.getId(); + int indexOf = id.indexOf(":"); + if (indexOf < 0) { + assertEquals("test", id); + } else { + assertEquals("test", id.substring(0,indexOf)); + } + assertEquals(3, collector.getTraces().size()); + count.addAndGet(collector.getTraces().size()); } }); @@ -93,6 +105,8 @@ public class TraceTest { System.out.println(meth1); thread.join(); Trace.tearDownTrace(); + + assertEquals(6, count.get()); } private static long meth1() {
