http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/blur-thrift/src/main/java/org/apache/blur/thrift/util/SimpleQueryExample.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/util/SimpleQueryExample.java b/blur-thrift/src/main/java/org/apache/blur/thrift/util/SimpleQueryExample.java index 7bf3bf1..e905ad0 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/util/SimpleQueryExample.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/util/SimpleQueryExample.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; @@ -38,7 +39,7 @@ public class SimpleQueryExample { Iface client = BlurClient.getClient(connectionStr); - String uuid = "123456"; + String uuid = UUID.randomUUID().toString(); Trace.setupTrace(uuid); final BlurQuery blurQuery = new BlurQuery(); Query query = new Query();
http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/blur-util/src/main/java/org/apache/blur/trace/BaseTraceStorage.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/BaseTraceStorage.java b/blur-util/src/main/java/org/apache/blur/trace/BaseTraceStorage.java new file mode 100644 index 0000000..3e17ce4 --- /dev/null +++ b/blur-util/src/main/java/org/apache/blur/trace/BaseTraceStorage.java @@ -0,0 +1,51 @@ +/** + * 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 java.util.List; + +import org.apache.blur.BlurConfiguration; + +public abstract class BaseTraceStorage extends TraceStorage { + + private static final String NOT_SUPPORTED = "Not Supported"; + + public BaseTraceStorage(BlurConfiguration configuration) { + super(configuration); + } + + @Override + public List<String> getTraceIds() { + throw new RuntimeException(NOT_SUPPORTED); + } + + @Override + public List<String> getRequestIds(String traceId) { + throw new RuntimeException(NOT_SUPPORTED); + } + + @Override + public String getRequestContentsJson(String traceId, String requestId) { + throw new RuntimeException(NOT_SUPPORTED); + } + + @Override + public void removeTrace(String traceId) { + throw new RuntimeException(NOT_SUPPORTED); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/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 deleted file mode 100644 index 586f9d7..0000000 --- a/blur-util/src/main/java/org/apache/blur/trace/LogTraceReporter.java +++ /dev/null @@ -1,44 +0,0 @@ -/** - * 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 java.io.IOException; - -import org.apache.blur.BlurConfiguration; -import org.apache.blur.log.Log; -import org.apache.blur.log.LogFactory; - -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(); - LOG.info("Trace Complete [{0}]", json); - } - - @Override - public void close() throws IOException { - - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/blur-util/src/main/java/org/apache/blur/trace/LogTraceStorage.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/LogTraceStorage.java b/blur-util/src/main/java/org/apache/blur/trace/LogTraceStorage.java new file mode 100644 index 0000000..db139bb --- /dev/null +++ b/blur-util/src/main/java/org/apache/blur/trace/LogTraceStorage.java @@ -0,0 +1,44 @@ +/** + * 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 java.io.IOException; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; + +public class LogTraceStorage extends BaseTraceStorage { + + private static final Log LOG = LogFactory.getLog(LogTraceStorage.class); + + public LogTraceStorage(BlurConfiguration configuration) { + super(configuration); + } + + @Override + public void store(TraceCollector collector) { + String json = collector.toJson(); + LOG.info("Trace Complete [{0}]", json); + } + + @Override + public void close() throws IOException { + + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/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 be9140d..7d781c6 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 @@ -72,7 +72,7 @@ public class Trace { } }; private static ThreadLocal<TraceCollector> _tracer = new ThreadLocal<TraceCollector>(); - private static TraceReporter _reporter; + private static TraceStorage _storage; private static String _nodeName; private static ThreadLocal<Random> _random = new ThreadLocal<Random>() { @Override @@ -111,7 +111,7 @@ public class Trace { private static void setupTraceOnNewThread(TraceCollector parentCollector, String requestId, int traceScope) { TraceCollector traceCollector = new TraceCollector(parentCollector, requestId); - TracerImpl tracer = new TracerImpl(traceCollector, parentCollector.getNextId(), traceScope); + TracerImpl tracer = new TracerImpl(traceCollector, parentCollector.getNextId(), traceScope, requestId); parentCollector.add(tracer); _tracer.set(traceCollector); } @@ -123,8 +123,8 @@ public class Trace { public static void tearDownTrace() { TraceCollector collector = _tracer.get(); _tracer.set(null); - if (_reporter != null && collector != null) { - _reporter.report(collector); + if (_storage != null && collector != null) { + _storage.store(collector); } } @@ -138,12 +138,12 @@ public class Trace { return tracer; } - public static TraceReporter getReporter() { - return _reporter; + public static TraceStorage getStorage() { + return _storage; } - public static void setReporter(TraceReporter reporter) { - _reporter = reporter; + public static void setStorage(TraceStorage storage) { + _storage = storage; } public static Runnable getRunnable(final Runnable runnable) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/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 dbc87d3..b6d23fc 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 @@ -60,8 +60,13 @@ public class TraceCollector { public String toJson() { StringBuilder builder = new StringBuilder(); + boolean first = true; for (TracerImpl t : _traces) { - builder.append(" ").append(t.toJson()).append(",\n"); + if (!first) { + builder.append(",\n"); + } + builder.append(" ").append(t.toJson()); + first = false; } return "{\n \"id\":" + _id.toJson() + ",\n \"nodeName\":\"" + (_nodeName == null ? "unknown" : _nodeName) + "\",\n \"pid\":\"" + _pid + "\",\n \"thread\":\"" + _threadName + "\",\n \"created\":" + _now http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/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 deleted file mode 100644 index c7a072c..0000000 --- a/blur-util/src/main/java/org/apache/blur/trace/TraceReporter.java +++ /dev/null @@ -1,33 +0,0 @@ -/** - * 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 java.io.Closeable; - -import org.apache.blur.BlurConfiguration; - -public abstract class TraceReporter implements Closeable { - - protected final 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/e75be0ee/blur-util/src/main/java/org/apache/blur/trace/TraceStorage.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/TraceStorage.java b/blur-util/src/main/java/org/apache/blur/trace/TraceStorage.java new file mode 100644 index 0000000..4b8c6a5 --- /dev/null +++ b/blur-util/src/main/java/org/apache/blur/trace/TraceStorage.java @@ -0,0 +1,43 @@ +/** + * 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 java.io.Closeable; +import java.io.IOException; +import java.util.List; + +import org.apache.blur.BlurConfiguration; + +public abstract class TraceStorage implements Closeable { + + protected final BlurConfiguration _configuration; + + public TraceStorage(BlurConfiguration configuration) { + _configuration = configuration; + } + + public abstract void store(TraceCollector collector); + + public abstract List<String> getTraceIds() throws IOException; + + public abstract List<String> getRequestIds(String traceId) throws IOException; + + public abstract String getRequestContentsJson(String traceId, String requestId) throws IOException; + + public abstract void removeTrace(String traceId) throws IOException; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/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 83d47da..629b241 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 @@ -43,13 +43,13 @@ public class TracerImpl implements Tracer { _traceScope = scope.incrementAndGet(); } - public TracerImpl(TraceCollector traceCollector, long id, int traceScope) { + public TracerImpl(TraceCollector traceCollector, long id, int traceScope, String requestId) { _name = "new thread collector"; _start = System.nanoTime(); _ended = _start; _threadName = Thread.currentThread().getName(); _id = id; - _parameters = new Parameter[] { new Parameter("requestId", Long.toString(id)) }; + _parameters = new Parameter[] { new Parameter("requestId", requestId) }; _traceCollector = traceCollector; _scope = traceCollector.getScope(); _traceScope = traceScope; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/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 deleted file mode 100644 index f39e686..0000000 --- a/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceReporter.java +++ /dev/null @@ -1,139 +0,0 @@ -/** - * 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.trace.Trace.TraceId; -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) { - TraceId id = collector.getId(); - String storeId = id.getRootId(); - 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; - } - - @Override - public void close() throws IOException { - try { - _zooKeeperClient.close(); - } catch (InterruptedException e) { - throw new IOException(e); - } - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceStorage.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceStorage.java b/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceStorage.java new file mode 100644 index 0000000..751b95c --- /dev/null +++ b/blur-util/src/main/java/org/apache/blur/trace/ZooKeeperTraceStorage.java @@ -0,0 +1,234 @@ +/** + * 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.ArrayList; +import java.util.List; +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.trace.Trace.TraceId; +import org.apache.blur.zookeeper.ZkUtils; +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 ZooKeeperTraceStorage extends TraceStorage { + + private static final String UTF_8 = "UTF-8"; + + private static final String TRACE = "trace_"; + + private final static Log LOG = LogFactory.getLog(ZooKeeperTraceStorage.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 ZooKeeperTraceStorage(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.setDaemon(true); + _daemon.setName("ZooKeeper Trace Queue Writer"); + _daemon.start(); + } + + @Override + public void store(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) { + TraceId id = collector.getId(); + String storeId = id.getRootId(); + String requestId = id.getRequestId(); + if (requestId == null) { + requestId = ""; + } + String storePath = getStorePath(storeId); + String json = collector.toJson(); + storeJson(storePath, requestId, json); + } + + public void storeJson(String storePath, String requestId, String json) { + try { + createIfMissing(storePath); + _zooKeeperClient.create(storePath + "/" + TRACE + requestId + "_", 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; + } + + @Override + public void close() throws IOException { + try { + _zooKeeperClient.close(); + } catch (InterruptedException e) { + throw new IOException(e); + } + } + + @Override + public List<String> getTraceIds() throws IOException { + try { + return _zooKeeperClient.getChildren(_storePath, false); + } catch (Exception e) { + throw new IOException(e); + } + } + + @Override + public List<String> getRequestIds(String traceId) throws IOException { + String storePath = getStorePath(traceId); + try { + Stat stats = _zooKeeperClient.exists(storePath, false); + if (stats == null) { + throw new IOException("Trace [" + traceId + "] not found."); + } + List<String> children = _zooKeeperClient.getChildren(storePath, false); + List<String> requestIds = new ArrayList<String>(); + for (String c : children) { + String requestId = getRequestId(c); + if (requestId != null) { + requestIds.add(requestId); + } + } + return requestIds; + } catch (Exception e) { + throw new IOException(e); + } + } + + private String getRequestId(String s) { + if (s.startsWith(TRACE)) { + int lastIndexOf = s.lastIndexOf('_'); + return s.substring(TRACE.length(), lastIndexOf); + } + return null; + } + + @Override + public String getRequestContentsJson(String traceId, String requestId) throws IOException { + String storePath = getStorePath(traceId); + try { + Stat stats = _zooKeeperClient.exists(storePath, false); + if (stats == null) { + throw new IOException("Trace [" + traceId + "] not found."); + } + String requestPath = getRequestStorePath(storePath, requestId); + if (requestPath == null) { + throw new IOException("Request [" + requestId + "] not found."); + } + Stat dataStat = _zooKeeperClient.exists(requestPath, false); + byte[] data = _zooKeeperClient.getData(requestPath, false, dataStat); + return new String(data, UTF_8); + } catch (Exception e) { + throw new IOException(e); + } + } + + private String getRequestStorePath(String storePath, String requestId) throws KeeperException, InterruptedException { + List<String> children = _zooKeeperClient.getChildren(storePath, false); + for (String c : children) { + if (c.startsWith(TRACE + requestId + "_")) { + return storePath + "/" + c; + } + } + return null; + } + + @Override + public void removeTrace(String traceId) throws IOException { + String storePath = getStorePath(traceId); + try { + ZkUtils.rmr(_zooKeeperClient, storePath); + } catch (Exception e) { + throw new IOException(e); + } + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/blur-util/src/main/java/org/apache/blur/zookeeper/ZkUtils.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/zookeeper/ZkUtils.java b/blur-util/src/main/java/org/apache/blur/zookeeper/ZkUtils.java index e82e645..5c96891 100644 --- a/blur-util/src/main/java/org/apache/blur/zookeeper/ZkUtils.java +++ b/blur-util/src/main/java/org/apache/blur/zookeeper/ZkUtils.java @@ -191,4 +191,12 @@ public class ZkUtils { throw new RuntimeException(e); } } + + public static void rmr(ZooKeeper zooKeeper, String path) throws KeeperException, InterruptedException { + List<String> children = zooKeeper.getChildren(path, false); + for (String c : children) { + rmr(zooKeeper, path + "/" + c); + } + zooKeeper.delete(path, -1); + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/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 222a41e..53ef049 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 @@ -29,9 +29,9 @@ public class TraceTest { @Test public void testTrace() throws IOException { - Trace.setReporter(new TraceReporter(new BlurConfiguration()) { + Trace.setStorage(new BaseTraceStorage(new BlurConfiguration()) { @Override - public void report(TraceCollector collector) { + public void store(TraceCollector collector) { assertEquals("test", collector.getId().getRootId()); assertEquals(3, collector.getTraces().size()); } @@ -68,13 +68,13 @@ public class TraceTest { @Test public void testTraceThreadRunnable() throws InterruptedException, IOException { final AtomicLong count = new AtomicLong(); - Trace.setReporter(new TraceReporter(new BlurConfiguration()) { + Trace.setStorage(new BaseTraceStorage(new BlurConfiguration()) { @Override - public void report(TraceCollector collector) { + public void store(TraceCollector collector) { System.out.println(collector.toJson()); TraceId id = collector.getId(); assertEquals("test", id.getRootId()); - assertEquals(8, collector.getTraces().size()); + assertEquals(5, collector.getTraces().size()); count.addAndGet(collector.getTraces().size()); } @@ -112,7 +112,7 @@ public class TraceTest { thread.join(); Trace.tearDownTrace(); - assertEquals(8, count.get()); + assertEquals(5, count.get()); } private static long meth1() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/blur-util/src/test/java/org/apache/blur/trace/ZooKeeperTraceStorageTest.java ---------------------------------------------------------------------- diff --git a/blur-util/src/test/java/org/apache/blur/trace/ZooKeeperTraceStorageTest.java b/blur-util/src/test/java/org/apache/blur/trace/ZooKeeperTraceStorageTest.java new file mode 100644 index 0000000..89cebfb --- /dev/null +++ b/blur-util/src/test/java/org/apache/blur/trace/ZooKeeperTraceStorageTest.java @@ -0,0 +1,122 @@ +/** + * 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 static org.junit.Assert.*; + +import java.io.File; +import java.io.IOException; +import java.util.List; +import java.util.Random; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.zookeeper.ZkMiniCluster; +import org.apache.blur.zookeeper.ZkUtils; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooKeeper; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class ZooKeeperTraceStorageTest { + + private static ZkMiniCluster _zkMiniCluster; + + @BeforeClass + public static void setupZookeeper() { + _zkMiniCluster = new ZkMiniCluster(); + _zkMiniCluster.startZooKeeper(new File("target/ZooKeeperTraceStorageTest").getAbsolutePath(), true); + } + + @AfterClass + public static void tearDownZookeeper() { + _zkMiniCluster.shutdownZooKeeper(); + } + + private ZooKeeper _zooKeeper; + private ZooKeeperTraceStorage _storage; + + @Before + public void setUp() throws IOException, InterruptedException { + final Object lock = new Object(); + synchronized (lock) { + _zooKeeper = new ZooKeeper(_zkMiniCluster.getZkConnectionString(), 10000, new Watcher() { + @Override + public void process(WatchedEvent event) { + synchronized (lock) { + lock.notifyAll(); + } + } + }); + lock.wait(); + } + + BlurConfiguration configuration = new BlurConfiguration(); + configuration.set(BLUR_ZOOKEEPER_CONNECTION, _zkMiniCluster.getZkConnectionString()); + configuration.set(BLUR_ZOOKEEPER_TRACE_PATH, "/test"); + _storage = new ZooKeeperTraceStorage(configuration); + } + + @After + public void tearDown() throws Exception { + _storage.close(); + ZkUtils.rmr(_zooKeeper, "/test"); + _zooKeeper.close(); + } + + @Test + public void testStorage() throws IOException { + Random random = new Random(); + createTraceData(random); + createTraceData(random); + createTraceData(random); + List<String> traceIds = _storage.getTraceIds(); + assertEquals(3, traceIds.size()); + + for (String traceId : traceIds) { + List<String> requestIds = _storage.getRequestIds(traceId); + assertEquals(4, requestIds.size()); + for (String requestId : requestIds) { + String contents = _storage.getRequestContentsJson(traceId, requestId); + assertEquals("{" + requestId + "}", contents); + } + } + + _storage.removeTrace(traceIds.get(0)); + assertEquals(2, _storage.getTraceIds().size()); + } + + private void createTraceData(Random random) { + long traceId = Math.abs(random.nextLong()); + String storePath = "/test/" + traceId; + _storage.storeJson(storePath, Long.toString(traceId), "{" + traceId + "}"); + writeRequest(random, storePath); + writeRequest(random, storePath); + writeRequest(random, storePath); + } + + private void writeRequest(Random random, String storePath) { + String requestId = Long.toString(random.nextLong()); + _storage.storeJson(storePath, requestId, "{" + requestId + "}"); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/distribution/src/main/scripts/interface/Blur.thrift ---------------------------------------------------------------------- diff --git a/distribution/src/main/scripts/interface/Blur.thrift b/distribution/src/main/scripts/interface/Blur.thrift index 0b24392..2dce3e0 100644 --- a/distribution/src/main/scripts/interface/Blur.thrift +++ b/distribution/src/main/scripts/interface/Blur.thrift @@ -1094,6 +1094,40 @@ service Blur { 2:string requestId ) + /** + * Get a list of all the traces. + * @return the list of trace ids. + */ + list<string> traceList() throws (1:BlurException ex) + + /** + * Gets a request list for the given trace. + * @return the list of request ids for the given trace id. + */ + list<string> traceRequestList( + /** the trace id. */ + 1:string traceId + ) throws (1:BlurException ex) + + /** + * Fetches the given trace. + * @return the json for the given trace request. + */ + string traceRequestFetch( + /** the trace id. */ + 1:string traceId, + /** the request id. */ + 2:string requestId + ) throws (1:BlurException ex) + + /** + * Remove the trace for the given trace id. + */ + void traceRemove( + /** the trace id. */ + 1:string traceId + ) throws (1:BlurException ex) + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/distribution/src/main/scripts/interface/gen-html/Blur.html ---------------------------------------------------------------------- diff --git a/distribution/src/main/scripts/interface/gen-html/Blur.html b/distribution/src/main/scripts/interface/gen-html/Blur.html index 80e463b..f5f3fee 100644 --- a/distribution/src/main/scripts/interface/gen-html/Blur.html +++ b/distribution/src/main/scripts/interface/gen-html/Blur.html @@ -46,6 +46,10 @@ <li><a href="Blur.html#Fn_Blur_tableListByCluster">tableListByCluster</a></li> <li><a href="Blur.html#Fn_Blur_tableStats">tableStats</a></li> <li><a href="Blur.html#Fn_Blur_terms">terms</a></li> +<li><a href="Blur.html#Fn_Blur_traceList">traceList</a></li> +<li><a href="Blur.html#Fn_Blur_traceRemove">traceRemove</a></li> +<li><a href="Blur.html#Fn_Blur_traceRequestFetch">traceRequestFetch</a></li> +<li><a href="Blur.html#Fn_Blur_traceRequestList">traceRequestList</a></li> </ul> </td> <td><a href="Blur.html#Struct_BlurException">BlurException</a><br/> @@ -840,4 +844,35 @@ the shard given the same situation. </td></tr> <tr><td>requestId</td><td>the request id, used to connected remote calls together. Client can pass null. </td></tr> +</table></div><div class="definition"><h4 id="Fn_Blur_traceList">Function: Blur.traceList</h4> +<pre><code>list<<code>string</code>></code> traceList() + throws <code><a href="Blur.html#Struct_BlurException">BlurException</a></code> +</pre>Get a list of all the traces. +@return the list of trace ids. +<br/></div><div class="definition"><h4 id="Fn_Blur_traceRequestList">Function: Blur.traceRequestList</h4> +<pre><code>list<<code>string</code>></code> traceRequestList(<code>string</code> traceId) + throws <code><a href="Blur.html#Struct_BlurException">BlurException</a></code> +</pre>Gets a request list for the given trace. +@return the list of request ids for the given trace id. +<br/><br/><h4 id="Parameters_Blur_traceRequestList">Parameters</h4> +<table class="table-bordered table-striped table-condensed"><thead><th>Name</th><th>Description</th></thead><tr><td>traceId</td><td>the trace id. +</td></tr> +</table></div><div class="definition"><h4 id="Fn_Blur_traceRequestFetch">Function: Blur.traceRequestFetch</h4> +<pre><code>string</code> traceRequestFetch(<code>string</code> traceId, + <code>string</code> requestId) + throws <code><a href="Blur.html#Struct_BlurException">BlurException</a></code> +</pre>Fetches the given trace. +@return the json for the given trace request. +<br/><br/><h4 id="Parameters_Blur_traceRequestFetch">Parameters</h4> +<table class="table-bordered table-striped table-condensed"><thead><th>Name</th><th>Description</th></thead><tr><td>traceId</td><td>the trace id. +</td></tr> +<tr><td>requestId</td><td>the request id. +</td></tr> +</table></div><div class="definition"><h4 id="Fn_Blur_traceRemove">Function: Blur.traceRemove</h4> +<pre><code>void</code> traceRemove(<code>string</code> traceId) + throws <code><a href="Blur.html#Struct_BlurException">BlurException</a></code> +</pre>Remove the trace for the given trace id. +<br/><br/><h4 id="Parameters_Blur_traceRemove">Parameters</h4> +<table class="table-bordered table-striped table-condensed"><thead><th>Name</th><th>Description</th></thead><tr><td>traceId</td><td>the trace id. +</td></tr> </table></div></div></body></html> http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e75be0ee/distribution/src/main/scripts/interface/gen-html/index.html ---------------------------------------------------------------------- diff --git a/distribution/src/main/scripts/interface/gen-html/index.html b/distribution/src/main/scripts/interface/gen-html/index.html index b1aa6ac..25b9b96 100644 --- a/distribution/src/main/scripts/interface/gen-html/index.html +++ b/distribution/src/main/scripts/interface/gen-html/index.html @@ -42,6 +42,10 @@ <li><a href="Blur.html#Fn_Blur_tableListByCluster">tableListByCluster</a></li> <li><a href="Blur.html#Fn_Blur_tableStats">tableStats</a></li> <li><a href="Blur.html#Fn_Blur_terms">terms</a></li> +<li><a href="Blur.html#Fn_Blur_traceList">traceList</a></li> +<li><a href="Blur.html#Fn_Blur_traceRemove">traceRemove</a></li> +<li><a href="Blur.html#Fn_Blur_traceRequestFetch">traceRequestFetch</a></li> +<li><a href="Blur.html#Fn_Blur_traceRequestList">traceRequestList</a></li> </ul> </td> <td><a href="Blur.html#Struct_BlurException">BlurException</a><br/>
