You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@htrace.apache.org by cm...@apache.org on 2015/10/12 22:13:55 UTC
[2/3] incubator-htrace git commit: HTRACE-237. Optimize htraced span
receiver (cmccabe)
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBufferManager.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBufferManager.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBufferManager.java
new file mode 100644
index 0000000..8b59a72
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBufferManager.java
@@ -0,0 +1,340 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.CharBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
+import java.nio.channels.spi.SelectorProvider;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.htrace.core.Span;
+
+class PackedBufferManager implements BufferManager {
+ private static final Log LOG = LogFactory.getLog(PackedBuffer.class);
+ private static final int MAX_PREQUEL_LENGTH = 2048;
+ private static final int METHOD_ID_WRITE_SPANS = 0x1;
+ private final Conf conf;
+ private final ByteBuffer frameBuffer;
+ private final PackedBuffer prequel;
+ private final PackedBuffer spans;
+ private final Selector selector;
+ private int numSpans;
+
+ PackedBufferManager(Conf conf) throws IOException {
+ this.conf = conf;
+ this.frameBuffer = ByteBuffer.allocate(PackedBuffer.HRPC_REQ_FRAME_LENGTH);
+ this.prequel = new PackedBuffer(ByteBuffer.allocate(MAX_PREQUEL_LENGTH));
+ this.spans = new PackedBuffer(ByteBuffer.allocate(conf.bufferSize));
+ this.selector = SelectorProvider.provider().openSelector();
+ clear();
+ }
+
+ @Override
+ public void writeSpan(Span span) throws IOException {
+ spans.writeSpan(span);
+ numSpans++;
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("wrote " + span.toJson() + " to PackedBuffer for " +
+ conf.endpointStr + ". numSpans = " + numSpans +
+ ", buffer position = " + spans.getBuffer().position());
+ }
+ }
+
+ @Override
+ public int contentLength() {
+ return spans.getBuffer().position();
+ }
+
+ @Override
+ public int getNumberOfSpans() {
+ return numSpans;
+ }
+
+ @Override
+ public void prepare() throws IOException {
+ prequel.beginWriteSpansRequest(null, numSpans);
+ long totalLength =
+ prequel.getBuffer().position() + spans.getBuffer().position();
+ if (totalLength > PackedBuffer.MAX_HRPC_BODY_LENGTH) {
+ throw new IOException("Can't send RPC of " + totalLength + " bytes " +
+ "because it is longer than " + PackedBuffer.MAX_HRPC_BODY_LENGTH);
+ }
+ PackedBuffer.writeReqFrame(frameBuffer,
+ METHOD_ID_WRITE_SPANS, 1, (int)totalLength);
+ frameBuffer.flip();
+ prequel.getBuffer().flip();
+ spans.getBuffer().flip();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Preparing to send RPC of length " +
+ (totalLength + PackedBuffer.HRPC_REQ_FRAME_LENGTH) + " to " +
+ conf.endpointStr + ", containing " + numSpans + " spans.");
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ SelectionKey sockKey = null;
+ IOException ioe = null;
+ frameBuffer.position(0);
+ prequel.getBuffer().position(0);
+ spans.getBuffer().position(0);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Preparing to flush " + numSpans + " spans to " +
+ conf.endpointStr);
+ }
+ try {
+ sockKey = doConnect();
+ doSend(sockKey, new ByteBuffer[] {
+ frameBuffer, prequel.getBuffer(), spans.getBuffer() });
+ ByteBuffer response = prequel.getBuffer();
+ readAndValidateResponseFrame(sockKey, response,
+ 1, METHOD_ID_WRITE_SPANS);
+ } catch (IOException e) {
+ // This LOG message is only at debug level because we also log these
+ // exceptions at error level inside HTracedReceiver. The logging in
+ // HTracedReceiver is rate-limited to avoid overwhelming the client log
+ // if htraced goes down. The debug and trace logging is not
+ // rate-limited.
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Got exception during flush", e);
+ }
+ ioe = e;
+ } finally {
+ if (sockKey != null) {
+ sockKey.cancel();
+ try {
+ SocketChannel sock = (SocketChannel)sockKey.attachment();
+ sock.close();
+ } catch (IOException e) {
+ if (ioe != null) {
+ ioe.addSuppressed(e);
+ }
+ }
+ }
+ }
+ if (ioe != null) {
+ throw ioe;
+ }
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Successfully flushed " + numSpans + " spans to " +
+ conf.endpointStr);
+ }
+ }
+
+ private long updateRemainingMs(long startMs, long timeoutMs) {
+ long deltaMs = TimeUtil.deltaMs(startMs, TimeUtil.nowMs());
+ if (deltaMs > timeoutMs) {
+ return 0;
+ }
+ return timeoutMs - deltaMs;
+ }
+
+ private SelectionKey doConnect() throws IOException {
+ SocketChannel sock = SocketChannel.open();
+ SelectionKey sockKey = null;
+ boolean success = false;
+ try {
+ if (sock.isBlocking()) {
+ sock.configureBlocking(false);
+ }
+ InetSocketAddress resolvedEndpoint =
+ new InetSocketAddress(conf.endpoint.getHostString(),
+ conf.endpoint.getPort());
+ resolvedEndpoint.getHostName(); // trigger DNS resolution
+ sock.connect(resolvedEndpoint);
+ sockKey = sock.register(selector, SelectionKey.OP_CONNECT, sock);
+ long startMs = TimeUtil.nowMs();
+ long remainingMs = conf.connectTimeoutMs;
+ while (true) {
+ selector.select(remainingMs);
+ for (SelectionKey key : selector.keys()) {
+ if (key.isConnectable()) {
+ SocketChannel s = (SocketChannel)key.attachment();
+ s.finishConnect();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Successfully connected to " + conf.endpointStr + ".");
+ }
+ success = true;
+ return sockKey;
+ }
+ }
+ remainingMs = updateRemainingMs(startMs, conf.connectTimeoutMs);
+ if (remainingMs == 0) {
+ throw new IOException("Attempt to connect to " + conf.endpointStr +
+ " timed out after " + TimeUtil.deltaMs(startMs, TimeUtil.nowMs()) +
+ " ms.");
+ }
+ }
+ } finally {
+ if (!success) {
+ if (sockKey != null) {
+ sockKey.cancel();
+ }
+ sock.close();
+ }
+ }
+ }
+
+ /**
+ * Send the provided ByteBuffer objects.
+ *
+ * We use non-blocking I/O because Java does not provide write timeouts.
+ * Without a write timeout, the socket could get hung and we'd never recover.
+ * We also use the GatheringByteChannel#write method which calls the pread()
+ * system call under the covers. This ensures that even if TCP_NODELAY is on,
+ * we send the minimal number of packets.
+ */
+ private void doSend(SelectionKey sockKey, ByteBuffer[] bufs)
+ throws IOException {
+ long totalWritten = 0;
+ sockKey.interestOps(SelectionKey.OP_WRITE);
+ SocketChannel sock = (SocketChannel)sockKey.attachment();
+ long startMs = TimeUtil.nowMs();
+ long remainingMs = conf.ioTimeoutMs;
+ while (true) {
+ selector.select(remainingMs);
+ int firstBuf = 0;
+ for (SelectionKey key : selector.selectedKeys()) {
+ if (key.isWritable()) {
+ long written = sock.write(bufs, firstBuf, bufs.length - firstBuf);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Sent " + written + " bytes to " + conf.endpointStr);
+ }
+ totalWritten += written;
+ }
+ }
+ while (true) {
+ if (firstBuf == bufs.length) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Finished sending " + totalWritten + " bytes to " +
+ conf.endpointStr);
+ }
+ return;
+ }
+ if (bufs[firstBuf].remaining() > 0) {
+ break;
+ }
+ firstBuf++;
+ }
+ remainingMs = updateRemainingMs(startMs, conf.ioTimeoutMs);
+ if (remainingMs == 0) {
+ throw new IOException("Attempt to write to " + conf.endpointStr +
+ " timed out after " + TimeUtil.deltaMs(startMs, TimeUtil.nowMs()) +
+ " ms.");
+ }
+ }
+ }
+
+ private void doRecv(SelectionKey sockKey, ByteBuffer response)
+ throws IOException {
+ sockKey.interestOps(SelectionKey.OP_READ);
+ SocketChannel sock = (SocketChannel)sockKey.attachment();
+ int totalRead = response.remaining();
+ long startMs = TimeUtil.nowMs();
+ long remainingMs = conf.ioTimeoutMs;
+ while (remainingMs > 0) {
+ selector.select(remainingMs);
+ for (SelectionKey key : selector.selectedKeys()) {
+ if (key.isReadable()) {
+ sock.read(response);
+ }
+ }
+ if (response.remaining() == 0) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Received all " + totalRead + " bytes from " +
+ conf.endpointStr);
+ }
+ return;
+ }
+ remainingMs = updateRemainingMs(startMs, conf.ioTimeoutMs);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Received " + (totalRead - response.remaining()) +
+ " out of " + totalRead + " bytes from " + conf.endpointStr);
+ }
+ if (remainingMs == 0) {
+ throw new IOException("Attempt to write to " + conf.endpointStr +
+ " timed out after " + TimeUtil.deltaMs(startMs, TimeUtil.nowMs()) +
+ " ms.");
+ }
+ }
+ }
+
+ private void readAndValidateResponseFrame(SelectionKey sockKey,
+ ByteBuffer buf, long expectedSeq, int expectedMethodId)
+ throws IOException {
+ buf.clear();
+ buf.limit(PackedBuffer.HRPC_RESP_FRAME_LENGTH);
+ doRecv(sockKey, buf);
+ buf.flip();
+ buf.order(ByteOrder.LITTLE_ENDIAN);
+ long seq = buf.getLong();
+ if (seq != expectedSeq) {
+ throw new IOException("Expected sequence number " + expectedSeq +
+ ", but got sequence number " + seq);
+ }
+ int methodId = buf.getInt();
+ if (expectedMethodId != methodId) {
+ throw new IOException("Expected method id " + expectedMethodId +
+ ", but got " + methodId);
+ }
+ int errorLength = buf.getInt();
+ buf.getInt();
+ if ((errorLength < 0) ||
+ (errorLength > PackedBuffer.MAX_HRPC_ERROR_LENGTH)) {
+ throw new IOException("Got server error with invalid length " +
+ errorLength);
+ } else if (errorLength > 0) {
+ buf.clear();
+ buf.limit(errorLength);
+ doRecv(sockKey, buf);
+ buf.flip();
+ CharBuffer charBuf = StandardCharsets.UTF_8.decode(buf);
+ String serverErrorStr = charBuf.toString();
+ throw new IOException("Got server error " + serverErrorStr);
+ }
+ }
+
+ @Override
+ public void clear() {
+ frameBuffer.clear();
+ prequel.getBuffer().clear();
+ spans.getBuffer().clear();
+ numSpans = 0;
+ }
+
+ @Override
+ public void close() {
+ clear();
+ prequel.close();
+ spans.close();
+ try {
+ selector.close();
+ } catch (IOException e) {
+ LOG.warn("Error closing selector", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/main/java/org/apache/htrace/impl/RateLimitedLogger.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/RateLimitedLogger.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/RateLimitedLogger.java
new file mode 100644
index 0000000..ac42ee8
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/RateLimitedLogger.java
@@ -0,0 +1,72 @@
+/**
+ * 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.htrace.impl;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+
+/**
+ * A logger which rate-limits its logging to a configurable level.
+ */
+class RateLimitedLogger {
+ private final Log log;
+ private final long timeoutMs;
+ private long lastLogTimeMs;
+
+ public RateLimitedLogger(Log log, long timeoutMs) {
+ this.log = log;
+ this.timeoutMs = timeoutMs;
+ synchronized (this) {
+ this.lastLogTimeMs = 0L;
+ }
+ }
+
+ public void warn(String what) {
+ long now = TimeUnit.MILLISECONDS.convert(System.nanoTime(),
+ TimeUnit.NANOSECONDS);
+ synchronized (this) {
+ if (now >= lastLogTimeMs + timeoutMs) {
+ log.warn(what);
+ lastLogTimeMs = now;
+ }
+ }
+ }
+
+ public void error(String what) {
+ long now = TimeUnit.MILLISECONDS.convert(System.nanoTime(),
+ TimeUnit.NANOSECONDS);
+ synchronized (this) {
+ if (now >= lastLogTimeMs + timeoutMs) {
+ log.error(what);
+ lastLogTimeMs = now;
+ }
+ }
+ }
+
+ public void error(String what, Throwable e) {
+ long now = TimeUnit.MILLISECONDS.convert(System.nanoTime(),
+ TimeUnit.NANOSECONDS);
+ synchronized (this) {
+ if (now >= lastLogTimeMs + timeoutMs) {
+ log.error(what, e);
+ lastLogTimeMs = now;
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/main/java/org/apache/htrace/impl/RestBufferManager.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/RestBufferManager.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/RestBufferManager.java
new file mode 100644
index 0000000..2e1aa70
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/RestBufferManager.java
@@ -0,0 +1,225 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.htrace.core.Span;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.api.ContentProvider;
+import org.eclipse.jetty.client.api.ContentResponse;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.http.HttpMethod;
+import org.eclipse.jetty.http.HttpStatus;
+
+class RestBufferManager implements BufferManager {
+ private static final Log LOG = LogFactory.getLog(RestBufferManager.class);
+ private static ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static ObjectWriter JSON_WRITER = OBJECT_MAPPER.writer();
+ private static final Charset UTF8 = Charset.forName("UTF-8");
+ private static final byte COMMA_BYTE = (byte)0x2c;
+ private static final int MAX_PREQUEL_LENGTH = 512;
+ private static final int MAX_EPILOGUE_LENGTH = 32;
+ private final Conf conf;
+ private final HttpClient httpClient;
+ private final String urlString;
+ private final ByteBuffer prequel;
+ private final ByteBuffer spans;
+ private final ByteBuffer epilogue;
+ private int numSpans;
+
+ private static class RestBufferManagerContentProvider
+ implements ContentProvider {
+ private final ByteBuffer[] bufs;
+
+ private class ByteBufferIterator implements Iterator<ByteBuffer> {
+ private int bufIdx = -1;
+
+ @Override
+ public boolean hasNext() {
+ return (bufIdx + 1) < bufs.length;
+ }
+
+ @Override
+ public ByteBuffer next() {
+ if ((bufIdx + 1) >= bufs.length) {
+ throw new NoSuchElementException();
+ }
+ bufIdx++;
+ return bufs[bufIdx];
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ RestBufferManagerContentProvider(ByteBuffer[] bufs) {
+ this.bufs = bufs;
+ }
+
+ @Override
+ public long getLength() {
+ long total = 0;
+ for (int i = 0; i < bufs.length; i++) {
+ total += bufs[i].remaining();
+ }
+ return total;
+ }
+
+ @Override
+ public Iterator<ByteBuffer> iterator() {
+ return new ByteBufferIterator();
+ }
+ }
+
+ /**
+ * Create an HttpClient instance.
+ *
+ * @param connTimeout The timeout to use for connecting.
+ * @param idleTimeout The idle timeout to use.
+ */
+ static HttpClient createHttpClient(long connTimeout, long idleTimeout) {
+ HttpClient httpClient = new HttpClient();
+ httpClient.setUserAgentField(
+ new HttpField(HttpHeader.USER_AGENT, "HTracedSpanReceiver"));
+ httpClient.setConnectTimeout(connTimeout);
+ httpClient.setIdleTimeout(idleTimeout);
+ return httpClient;
+ }
+
+ RestBufferManager(Conf conf) throws Exception {
+ this.conf = conf;
+ this.httpClient =
+ createHttpClient(conf.connectTimeoutMs, conf.idleTimeoutMs);
+ this.urlString = new URL("http", conf.endpoint.getHostName(),
+ conf.endpoint.getPort(), "/writeSpans").toString();
+ this.prequel = ByteBuffer.allocate(MAX_PREQUEL_LENGTH);
+ this.spans = ByteBuffer.allocate(conf.bufferSize);
+ this.epilogue = ByteBuffer.allocate(MAX_EPILOGUE_LENGTH);
+ clear();
+ this.httpClient.start();
+ }
+
+ @Override
+ public void writeSpan(Span span) throws IOException {
+ byte[] spanJsonBytes = JSON_WRITER.writeValueAsBytes(span);
+ if ((spans.capacity() - spans.position()) < (spanJsonBytes.length + 1)) {
+ // Make sure we have enough space for the span JSON and a comma.
+ throw new IOException("Not enough space remaining in span buffer.");
+ }
+ spans.put(COMMA_BYTE);
+ spans.put(spanJsonBytes);
+ numSpans++;
+ }
+
+ @Override
+ public int contentLength() {
+ return Math.max(spans.position() - 1, 0);
+ }
+
+ @Override
+ public int getNumberOfSpans() {
+ return numSpans;
+ }
+
+ @Override
+ public void prepare() throws IOException {
+ String prequelString = "{\"Spans\":[";
+ prequel.put(prequelString.getBytes(UTF8));
+ prequel.flip();
+
+ spans.flip();
+
+ String epilogueString = "]}";
+ epilogue.put(epilogueString.toString().getBytes(UTF8));
+ epilogue.flip();
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Preparing to send " + contentLength() + " bytes of span " +
+ "data to " + conf.endpointStr + ", containing " + numSpans +
+ " spans.");
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ // Position the buffers at the beginning.
+ prequel.position(0);
+ spans.position(spans.limit() == 0 ? 0 : 1); // Skip the first comma
+ epilogue.position(0);
+
+ RestBufferManagerContentProvider contentProvider =
+ new RestBufferManagerContentProvider(
+ new ByteBuffer[] { prequel, spans, epilogue });
+ long rpcLength = contentProvider.getLength();
+ try {
+ Request request = httpClient.
+ newRequest(urlString).method(HttpMethod.POST);
+ request.header(HttpHeader.CONTENT_TYPE, "application/json");
+ request.content(contentProvider);
+ ContentResponse response = request.send();
+ if (response.getStatus() != HttpStatus.OK_200) {
+ throw new IOException("Got back error response " +
+ response.getStatus() + " from " + conf.endpointStr + "; " +
+ response.getContentAsString());
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Sent WriteSpansReq of length " + rpcLength + " to " + conf.endpointStr);
+ }
+ } catch (InterruptedException e) {
+ throw new IOException("Interrupted while sending spans via REST", e);
+ } catch (TimeoutException e) {
+ throw new IOException("Timed out sending spans via REST", e);
+ } catch (ExecutionException e) {
+ throw new IOException("Execution exception sending spans via REST", e);
+ }
+ }
+
+ @Override
+ public void clear() {
+ prequel.clear();
+ spans.clear();
+ epilogue.clear();
+ numSpans = 0;
+ }
+
+ @Override
+ public void close() {
+ try {
+ httpClient.stop();
+ } catch (Exception e) {
+ LOG.error("Error stopping HTracedReceiver httpClient", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/main/java/org/apache/htrace/impl/TimeUtil.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/TimeUtil.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/TimeUtil.java
new file mode 100644
index 0000000..7361c97
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/TimeUtil.java
@@ -0,0 +1,78 @@
+/**
+ * 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.htrace.impl;
+
+import java.math.BigInteger;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Utilities for dealing with monotonic time.
+ */
+class TimeUtil {
+ /**
+ * Returns the current monotonic time in milliseconds.
+ */
+ static long nowMs() {
+ return TimeUnit.MILLISECONDS.convert(
+ System.nanoTime(), TimeUnit.NANOSECONDS);
+ }
+
+ /**
+ * Get the approximate delta between two monotonic times.
+ *
+ * This function makes the following assumptions:
+ * 1. We read startMs from the monotonic clock prior to endMs.
+ * 2. The two times are not more than 100 years or so apart.
+ *
+ * With these two assumptions in hand, we can smooth over some of the
+ * unpleasant features of the monotonic clock:
+ * 1. It can return either positive or negative values.
+ * 2. When the number of nanoseconds reaches Long.MAX_VALUE it wraps around
+ * to Long.MIN_VALUE.
+ * 3. On some messed up systems it has been known to jump backwards every
+ * now and then. Oops. CPU core synchronization mumble mumble.
+ *
+ * @param startMs The start time.
+ * @param endMs The end time.
+ * @return The delta between the two times.
+ */
+ static long deltaMs(long startMs, long endMs) {
+ BigInteger startNs = BigInteger.valueOf(TimeUnit.NANOSECONDS.
+ convert(startMs, TimeUnit.MILLISECONDS));
+ BigInteger endNs = BigInteger.valueOf(TimeUnit.NANOSECONDS.
+ convert(endMs, TimeUnit.MILLISECONDS));
+ BigInteger deltaNs = endNs.subtract(startNs);
+ if (deltaNs.signum() >= 0) {
+ return TimeUnit.MILLISECONDS.convert(deltaNs.min(
+ BigInteger.valueOf(Long.MAX_VALUE)).longValue(), TimeUnit.NANOSECONDS);
+ }
+ deltaNs = deltaNs.negate();
+ if (deltaNs.compareTo(BigInteger.valueOf(Long.MAX_VALUE / 2)) < 0) {
+ // If the 'startNs' is numerically less than the 'endNs', and the
+ // difference between the two is less than 100 years, it's probably
+ // just clock jitter. Certain old OSes and CPUs had monotonic clocks
+ // that could go backwards under certain conditions (ironic, given
+ // the name).
+ return 0L;
+ }
+ // Handle rollover.
+ BigInteger revDeltaNs = BigInteger.ONE.shiftLeft(64).subtract(deltaNs);
+ return TimeUnit.MILLISECONDS.convert(revDeltaNs.min(
+ BigInteger.valueOf(Long.MAX_VALUE)).longValue(), TimeUnit.NANOSECONDS);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/impl/DataDir.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/DataDir.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/DataDir.java
new file mode 100644
index 0000000..f224f6f
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/DataDir.java
@@ -0,0 +1,58 @@
+/*
+ * 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.htrace.impl;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.UUID;
+
+/**
+ * Small util for making a data directory for tests to use when running tests. We put it up at
+ * target/test-data/UUID. Create an instance of this class per unit test run and it will take
+ * care of setting up the dirs for you. Pass what is returned here as location from which to
+ * have daemons and tests dump data.
+ */
+public class DataDir implements Closeable {
+ private final File dir;
+
+ public DataDir() throws IOException {
+ String baseDir = System.getProperty(
+ "test.data.base.dir", "target");
+ File testData = new File(new File(baseDir), "test-data");
+ this.dir = new File(testData, UUID.randomUUID().toString());
+ Files.createDirectories(this.dir.toPath());
+ }
+
+ public File get() {
+ return dir;
+ }
+
+ @Override
+ public void close() throws IOException {
+ /*for (File file : this.dir.listFiles()) {
+ file.delete();
+ }
+ Files.delete(this.dir.toPath()); */
+ }
+
+ @Override
+ public String toString() {
+ return "DataDir{" + dir.getAbsolutePath() + "}";
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/impl/HTracedProcess.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/HTracedProcess.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/HTracedProcess.java
new file mode 100644
index 0000000..26c1a10
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/HTracedProcess.java
@@ -0,0 +1,277 @@
+/*
+ * 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.htrace.impl;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.URI;
+import java.nio.file.Paths;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.MilliSpan;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.SpanId;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.api.ContentResponse;
+import org.eclipse.jetty.http.HttpStatus;
+import org.junit.Assert;
+
+/**
+ * To get instance of HTraced up and running, create an instance of this class.
+ * Upon successful construction, htraced is running using <code>dataDir</code> as directory to
+ * host data (leveldbs and logs).
+ */
+class HTracedProcess extends Process {
+ private static final Log LOG = LogFactory.getLog(HTracedProcess.class);
+
+ static class Builder {
+ String host = "localhost";
+
+ Builder() {
+ }
+
+ Builder host(String host) {
+ this.host = host;
+ return this;
+ }
+
+ HTracedProcess build() throws Exception {
+ return new HTracedProcess(this);
+ }
+ }
+
+ /**
+ * Path to the htraced binary.
+ */
+ private final File htracedPath;
+
+ /**
+ * Temporary directory for test files.
+ */
+ private final DataDir dataDir;
+
+ /**
+ * The Java Process object for htraced.
+ */
+ private final Process delegate;
+
+ /**
+ * The HTTP host:port returned from htraced.
+ */
+ private final String httpAddr;
+
+ /**
+ * The HRPC host:port returned from htraced.
+ */
+ private final String hrpcAddr;
+
+ /**
+ * REST client to use to talk to htraced.
+ */
+ private final HttpClient httpClient;
+
+ /**
+ * Data send back from the HTraced process on the notification port.
+ */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class StartupNotificationData {
+ /**
+ * The hostname:port pair which the HTraced process uses for HTTP requests.
+ */
+ @JsonProperty("HttpAddr")
+ String httpAddr;
+
+ /**
+ * The hostname:port pair which the HTraced process uses for HRPC requests.
+ */
+ @JsonProperty("HrpcAddr")
+ String hrpcAddr;
+
+ /**
+ * The process ID of the HTraced process.
+ */
+ @JsonProperty("ProcessId")
+ long processId;
+ }
+
+ private HTracedProcess(Builder builder) throws Exception {
+ this.htracedPath = Paths.get(
+ "target", "..", "go", "build", "htraced").toFile();
+ if (!this.htracedPath.exists()) {
+ throw new RuntimeException("No htraced binary exists at " +
+ this.htracedPath);
+ }
+ this.dataDir = new DataDir();
+ // Create a notifier socket bound to a random port.
+ ServerSocket listener = new ServerSocket(0);
+ boolean success = false;
+ Process process = null;
+ HttpClient http = null;
+ try {
+ // Use a random port for the web address. No 'scheme' yet.
+ String random = builder.host + ":0";
+ String logPath = new File(dataDir.get(), "log.txt").getAbsolutePath();
+ // Pass cmdline args to htraced to it uses our test dir for data.
+ ProcessBuilder pb = new ProcessBuilder(htracedPath.getAbsolutePath(),
+ "-Dlog.level=TRACE",
+ "-Dlog.path=" + logPath,
+ "-Dweb.address=" + random,
+ "-Dhrpc.address=" + random,
+ "-Ddata.store.clear=true",
+ "-Dstartup.notification.address=localhost:" + listener.getLocalPort(),
+ "-Ddata.store.directories=" + dataDir.get().getAbsolutePath());
+ pb.redirectErrorStream(true);
+ // Inherit STDERR/STDOUT i/o; dumps on console for now. Can add logs later.
+ pb.inheritIO();
+ pb.directory(dataDir.get());
+ //assert pb.redirectInput() == Redirect.PIPE;
+ //assert pb.redirectOutput().file() == dataDir;
+ process = pb.start();
+ assert process.getInputStream().read() == -1;
+ StartupNotificationData data = readStartupNotification(listener);
+ httpAddr = data.httpAddr;
+ hrpcAddr = data.hrpcAddr;
+ LOG.info("Started htraced process " + data.processId + " with http " +
+ "address " + data.httpAddr + ", logging to " + logPath);
+ http = RestBufferManager.createHttpClient(60000L, 60000L);
+ http.start();
+ success = true;
+ } finally {
+ if (!success) {
+ // Clean up after failure
+ if (process != null) {
+ process.destroy();
+ process = null;
+ }
+ if (http != null) {
+ http.stop();
+ }
+ }
+ delegate = process;
+ listener.close();
+ httpClient = http;
+ }
+ }
+
+ private static StartupNotificationData
+ readStartupNotification(ServerSocket listener) throws IOException {
+ Socket socket = listener.accept();
+ try {
+ InputStream in = socket.getInputStream();
+ ObjectMapper objectMapper = new ObjectMapper();
+ StartupNotificationData data = objectMapper.
+ readValue(in, StartupNotificationData.class);
+ return data;
+ } finally {
+ socket.close();
+ }
+ }
+
+ public int hashCode() {
+ return delegate.hashCode();
+ }
+
+ public OutputStream getOutputStream() {
+ throw new UnsupportedOperationException("Unsupported until complaint; output on STDOUT");
+ }
+
+ public InputStream getInputStream() {
+ throw new UnsupportedOperationException("Unsupported until complaint; output on STDOUT");
+ }
+
+ public boolean equals(Object obj) {
+ return delegate.equals(obj);
+ }
+
+ public InputStream getErrorStream() {
+ throw new UnsupportedOperationException("Unsupported until complaint; output on STDOUT");
+ }
+
+ public int waitFor() throws InterruptedException {
+ return delegate.waitFor();
+ }
+
+ public int exitValue() {
+ return delegate.exitValue();
+ }
+
+ public void destroy() {
+ try {
+ httpClient.stop();
+ } catch (Exception e) {
+ LOG.error("Error stopping httpClient", e);
+ }
+ delegate.destroy();
+ try {
+ dataDir.close();
+ } catch (Exception e) {
+ LOG.error("Error closing " + dataDir, e);
+ }
+ LOG.trace("Destroyed htraced process.");
+ }
+
+ public String toString() {
+ return delegate.toString();
+ }
+
+ public String getHttpAddr() {
+ return httpAddr;
+ }
+
+ public String getHrpcAddr() {
+ return hrpcAddr;
+ }
+
+ /**
+ * Ugly but how else to do file-math?
+ * @param topLevel Presumes top-level of the htrace checkout.
+ * @return Path to the htraced binary.
+ */
+ public static File getPathToHTraceBinaryFromTopLevel(final File topLevel) {
+ return new File(new File(new File(new File(topLevel, "htrace-htraced"), "go"), "build"),
+ "htraced");
+ }
+
+ public String getServerInfoJson() throws Exception {
+ ContentResponse response = httpClient.GET(
+ new URI(String.format("http://%s/server/info", httpAddr)));
+ Assert.assertEquals("application/json", response.getMediaType());
+ Assert.assertEquals(HttpStatus.OK_200, response.getStatus());
+ return response.getContentAsString();
+ }
+
+ public Span getSpan(SpanId spanId) throws Exception {
+ ContentResponse response = httpClient.GET(
+ new URI(String.format("http://%s/span/%s",
+ httpAddr, spanId.toString())));
+ Assert.assertEquals("application/json", response.getMediaType());
+ String responseJson = response.getContentAsString().trim();
+ if (responseJson.isEmpty()) {
+ return null;
+ }
+ return MilliSpan.fromJson(responseJson);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedRESTReceiver.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedRESTReceiver.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedRESTReceiver.java
deleted file mode 100644
index d52f071..0000000
--- a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedRESTReceiver.java
+++ /dev/null
@@ -1,215 +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.htrace.impl;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.net.URL;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.htrace.core.HTraceConfiguration;
-import org.apache.htrace.core.MilliSpan;
-import org.apache.htrace.core.Span;
-import org.apache.htrace.core.SpanId;
-import org.apache.htrace.core.TracerId;
-import org.apache.htrace.util.DataDir;
-import org.apache.htrace.util.HTracedProcess;
-import org.apache.htrace.util.TestUtil;
-import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.client.api.ContentResponse;
-import org.eclipse.jetty.http.HttpStatus;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class TestHTracedRESTReceiver {
- private static final Log LOG =
- LogFactory.getLog(TestHTracedRESTReceiver.class);
- private URL restServerUrl;
- private DataDir dataDir;
- HTracedProcess htraced;
-
- @Before
- public void setUp() throws Exception {
- this.dataDir = new DataDir();
- File tlDir = DataDir.getTopLevelOfCheckout(this.dataDir.getDataDir());
- File pathToHTracedBinary = HTracedProcess.getPathToHTraceBinaryFromTopLevel(tlDir);
- this.htraced = new HTracedProcess(pathToHTracedBinary,
- dataDir.getDataDir(), "localhost");
- this.restServerUrl = new URL("http://" + htraced.getHttpAddr() + "/");
- }
-
- @After
- public void tearDown() throws Exception {
- if (this.htraced != null) this.htraced.destroy();
- }
-
- /**
- * Our simple version of htrace configuration for testing.
- */
- private final class TestHTraceConfiguration extends HTraceConfiguration {
- private final URL restServerUrl;
- final static String TRACER_ID = "TestHTracedRESTReceiver";
-
- public TestHTraceConfiguration(final URL restServerUrl) {
- this.restServerUrl = restServerUrl;
- }
-
- @Override
- public String get(String key) {
- return null;
- }
-
- @Override
- public String get(String key, String defaultValue) {
- if (key.equals(HTracedRESTReceiver.HTRACED_REST_URL_KEY)) {
- return this.restServerUrl.toString();
- } else if (key.equals(TracerId.TRACER_ID_KEY)) {
- return TRACER_ID;
- }
- return defaultValue;
- }
- }
-
- /**
- * Make sure the REST server basically works.
- * @throws Exception
- */
- @Test (timeout = 10000)
- public void testBasicGet() throws Exception {
- HTracedRESTReceiver receiver =
- new HTracedRESTReceiver(new TestHTraceConfiguration(this.restServerUrl));
- HttpClient http = receiver.createHttpClient(60000L, 60000L);
- http.start();
- try {
- // Do basic a GET /server/info against htraced
- ContentResponse response =
- http.GET(restServerUrl + "server/info");
- assertEquals("application/json", response.getMediaType());
- String content = processGET(response);
- assertTrue(content.contains("ReleaseVersion"));
- System.out.println(content);
- } finally {
- http.stop();
- receiver.close();
- }
- }
-
- private String processGET(final ContentResponse response) {
- assertTrue("" + response.getStatus(), HttpStatus.OK_200 <= response.getStatus() &&
- response.getStatus() <= HttpStatus.NO_CONTENT_204);
- return response.getContentAsString();
- }
-
- private void testSendingSpansImpl(boolean testClose) throws Exception {
- final HTracedRESTReceiver receiver =
- new HTracedRESTReceiver(new TestHTraceConfiguration(this.restServerUrl));
- final int NUM_SPANS = 3;
- final HttpClient http = receiver.createHttpClient(60000, 60000);
- http.start();
- Span spans[] = new Span[NUM_SPANS];
- for (int i = 0; i < NUM_SPANS; i++) {
- MilliSpan.Builder builder = new MilliSpan.Builder().
- parents(new SpanId[] { new SpanId(1L, 1L) }).
- spanId(new SpanId(1L, i));
- if (i == NUM_SPANS - 1) {
- builder.tracerId("specialTrid");
- } else {
- builder.tracerId(TestHTraceConfiguration.TRACER_ID);
- }
- spans[i] = builder.build();
- }
- try {
- for (int i = 0; i < NUM_SPANS; i++) {
- LOG.info("receiving " + spans[i].toString());
- receiver.receiveSpan(spans[i]);
- }
- if (testClose) {
- receiver.close();
- } else {
- receiver.startFlushing();
- }
- TestUtil.waitFor(new TestUtil.Supplier<Boolean>() {
- @Override
- public Boolean get() {
- try {
- for (int i = 0; i < NUM_SPANS; i++) {
- // This is what the REST server expects when querying for a
- // span id.
- String findSpan = String.format("span/%s",
- new SpanId(1L, i).toString());
- ContentResponse response =
- http.GET(restServerUrl + findSpan);
- String content = processGET(response);
- if ((content == null) || (content.length() == 0)) {
- LOG.info("Failed to find span " + i);
- return false;
- }
- LOG.info("Got " + content + " for span " + i);
- MilliSpan dspan = MilliSpan.fromJson(content);
- assertEquals(new SpanId(1, i).toString(),
- dspan.getSpanId().toString());
- // Every span should have the tracer ID we set in the
- // configuration... except for the last span, which had
- // a custom value set.
- if (i == NUM_SPANS - 1) {
- assertEquals("specialTrid", dspan.getTracerId());
- } else {
- assertEquals(TestHTraceConfiguration.TRACER_ID,
- dspan.getTracerId());
- }
- }
- return true;
- } catch (Throwable t) {
- LOG.error("Got exception", t);
- return false;
- }
- }
- }, 10, 20000);
- } finally {
- http.stop();
- if (!testClose) {
- receiver.close();
- }
- }
- }
-
- /**
- * Send 100 spans then confirm they made it in.
- * @throws Exception
- */
- @Test (timeout = 60000)
- public void testSendingSpans() throws Exception {
- testSendingSpansImpl(false);
- }
-
- /**
- * Test that the REST receiver blocks during shutdown until all spans are sent
- * (or a long timeout elapses). Otherwise, short-lived client processes will
- * never have a chance to send all their spans and we will have incomplete
- * information.
- */
- @Test (timeout = 60000)
- public void testShutdownBlocksUntilSpanAreSent() throws Exception {
- testSendingSpansImpl(true);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiver.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiver.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiver.java
new file mode 100644
index 0000000..99f00a1
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiver.java
@@ -0,0 +1,572 @@
+/**
+ * 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.htrace.impl;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Random;
+import java.util.concurrent.Semaphore;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.HTraceConfiguration;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.SpanId;
+import org.apache.htrace.core.TracerId;
+import org.apache.htrace.impl.HTracedSpanReceiver.FaultInjector;
+import org.apache.htrace.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestRule;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+
+public class TestHTracedReceiver {
+ private static final Log LOG = LogFactory.getLog(TestHTracedReceiver.class);
+
+ @BeforeClass
+ public static void beforeClass() {
+ // Allow setting really small buffer sizes for testing purposes.
+ // We do not allow setting such small sizes in production.
+ Conf.BUFFER_SIZE_MIN = 0;
+ }
+
+ @Rule
+ public TestRule watcher = new TestWatcher() {
+ protected void starting(Description description) {
+ LOG.info("*** Starting junit test: " + description.getMethodName());
+ }
+
+ protected void finished(Description description) {
+ LOG.info("*** Finished junit test: " + description.getMethodName());
+ }
+ };
+
+ @Test(timeout = 60000)
+ public void testGetServerInfoJson() throws Exception {
+ HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ String response = ht.getServerInfoJson();
+ assertTrue(response.contains("ReleaseVersion"));
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ private void waitForSpans(final HTracedProcess ht, Span[] spans)
+ throws Exception {
+ waitForSpans(ht, spans, spans.length);
+ }
+
+ private void waitForSpans(final HTracedProcess ht, Span[] spans,
+ int numSpans) throws Exception {
+ final LinkedList<SpanId> spanIds = new LinkedList<SpanId>();
+ for (int i = 0; i < numSpans; i++) {
+ spanIds.add(spans[i].getSpanId());
+ }
+ boolean success = false;
+ try {
+ TestUtil.waitFor(new TestUtil.Supplier<Boolean>() {
+ @Override
+ public Boolean get() {
+ for (Iterator<SpanId> iter = spanIds.iterator();
+ iter.hasNext(); ) {
+ SpanId spanId = iter.next();
+ try {
+ if (ht.getSpan(spanId) == null) {
+ return false;
+ }
+ } catch (InterruptedException e) {
+ LOG.error("Got InterruptedException while looking for " +
+ "span ID " + spanId, e);
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ LOG.error("Got error looking for span ID " + spanId, e);
+ return false;
+ }
+ iter.remove();
+ }
+ return true;
+ }
+ }, 10, 30000);
+ success = true;
+ } finally {
+ if (!success) {
+ String prefix = "";
+ StringBuilder idStringBld = new StringBuilder();
+ for (Iterator<SpanId> iter = spanIds.iterator();
+ iter.hasNext(); ) {
+ idStringBld.append(prefix);
+ idStringBld.append(iter.next());
+ prefix = ",";
+ }
+ LOG.error("Unable to find span IDs " + idStringBld.toString());
+ }
+ }
+ }
+
+ /**
+ * Test that we can send spans via the HRPC interface.
+ */
+ @Test(timeout = 10000) //60000)
+ public void testSendSpansViaPacked() throws Exception {
+ final Random rand = new Random(123);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testSendSpansViaPacked");
+ put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+ put(Conf.PACKED_KEY, "true");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "100");
+ put(Conf.ERROR_LOG_PERIOD_MS_KEY, "0");
+ }});
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+ Span[] spans = TestUtil.randomSpans(rand, 10);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ waitForSpans(ht, spans);
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that when the SpanReceiver is closed, we send any spans we have
+ * buffered via the HRPC interface.
+ */
+ @Test(timeout = 60000)
+ public void testSendSpansViaPackedAndClose() throws Exception {
+ final Random rand = new Random(456);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testSendSpansViaPackedAndClose");
+ put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+ put(Conf.PACKED_KEY, "true");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "60000");
+ }});
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+ Span[] spans = TestUtil.randomSpans(rand, 10);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ rcvr.close();
+ waitForSpans(ht, spans);
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that we can send spans via the REST interface.
+ */
+ @Test(timeout = 60000)
+ public void testSendSpansViaRest() throws Exception {
+ final Random rand = new Random(789);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testSendSpansViaRest");
+ put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+ put(Conf.PACKED_KEY, "false");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "100");
+ }});
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+ Span[] spans = TestUtil.randomSpans(rand, 10);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ waitForSpans(ht, spans);
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that when the SpanReceiver is closed, we send any spans we have
+ * buffered via the REST interface.
+ */
+ @Test(timeout = 60000)
+ public void testSendSpansViaRestAndClose() throws Exception {
+ final Random rand = new Random(321);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testSendSpansViaRestAndClose");
+ put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+ put(Conf.PACKED_KEY, "false");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "60000");
+ }});
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+ Span[] spans = TestUtil.randomSpans(rand, 10);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ rcvr.close();
+ waitForSpans(ht, spans);
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ private static class Mutable<T> {
+ private T t;
+
+ Mutable(T t) {
+ this.t = t;
+ }
+
+ void set(T t) {
+ this.t = t;
+ }
+
+ T get() {
+ return this.t;
+ }
+ }
+
+ private static class TestHandleContentLengthTriggerInjector
+ extends HTracedSpanReceiver.FaultInjector {
+ final Semaphore threadStartSem = new Semaphore(0);
+ int contentLengthOnTrigger = 0;
+
+ @Override
+ public synchronized void handleContentLengthTrigger(int len) {
+ contentLengthOnTrigger = len;
+ }
+ @Override
+ public void handleThreadStart() throws Exception {
+ threadStartSem.acquire();
+ }
+
+ public synchronized int getContentLengthOnTrigger() {
+ return contentLengthOnTrigger;
+ }
+ }
+
+ /**
+ * Test that filling up one of the buffers causes us to trigger a flush and
+ * start using the other buffer, when using PackedBufferManager.
+ * This also tests that PackedBufferManager can correctly handle a buffer
+ * getting full.
+ */
+ @Test(timeout = 60000)
+ public void testFullBufferCausesPackedThreadTrigger() throws Exception {
+ final Random rand = new Random(321);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY,
+ "testFullBufferCausesPackedThreadTrigger");
+ put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+ put(Conf.PACKED_KEY, "true");
+ put(Conf.BUFFER_SIZE_KEY, "16384");
+ put(Conf.BUFFER_SEND_TRIGGER_FRACTION_KEY, "0.95");
+ }});
+ TestHandleContentLengthTriggerInjector injector =
+ new TestHandleContentLengthTriggerInjector();
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 47);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ Assert.assertTrue("The wakePostSpansThread should have been " +
+ "triggered by the spans added so far. " +
+ "contentLengthOnTrigger = " + injector.getContentLengthOnTrigger(),
+ injector.getContentLengthOnTrigger() > 16000);
+ injector.threadStartSem.release();
+ rcvr.close();
+ waitForSpans(ht, spans, 45);
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that filling up one of the buffers causes us to trigger a flush and
+ * start using the other buffer, when using RestBufferManager.
+ * This also tests that RestBufferManager can correctly handle a buffer
+ * getting full.
+ */
+ @Test(timeout = 60000)
+ public void testFullBufferCausesRestThreadTrigger() throws Exception {
+ final Random rand = new Random(321);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY,
+ "testFullBufferCausesRestThreadTrigger");
+ put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+ put(Conf.PACKED_KEY, "false");
+ put(Conf.BUFFER_SIZE_KEY, "16384");
+ put(Conf.BUFFER_SEND_TRIGGER_FRACTION_KEY, "0.95");
+ }});
+ TestHandleContentLengthTriggerInjector injector =
+ new TestHandleContentLengthTriggerInjector();
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 34);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ Assert.assertTrue("The wakePostSpansThread should have been " +
+ "triggered by the spans added so far. " +
+ "contentLengthOnTrigger = " + injector.getContentLengthOnTrigger(),
+ injector.getContentLengthOnTrigger() > 16000);
+ injector.threadStartSem.release();
+ rcvr.close();
+ waitForSpans(ht, spans, 33);
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * A FaultInjector that causes all flushes to fail until a specified
+ * number of milliseconds have passed.
+ */
+ private static class TestInjectFlushFaults
+ extends HTracedSpanReceiver.FaultInjector {
+ private long remainingFaults;
+
+ TestInjectFlushFaults(long remainingFaults) {
+ this.remainingFaults = remainingFaults;
+ }
+
+ @Override
+ public synchronized void handleFlush() throws IOException {
+ if (remainingFaults > 0) {
+ remainingFaults--;
+ throw new IOException("Injected IOException into flush " +
+ "code path.");
+ }
+ }
+ }
+
+ /**
+ * Test that even if the flush fails, the system stays stable and we can
+ * still close the span receiver.
+ */
+ @Test(timeout = 60000)
+ public void testPackedThreadHandlesFlushFailure() throws Exception {
+ final Random rand = new Random(321);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testPackedThreadHandlesFlushFailure");
+ put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+ put(Conf.PACKED_KEY, "true");
+ }});
+ TestInjectFlushFaults injector = new TestInjectFlushFaults(Long.MAX_VALUE);
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 15);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that even if the flush fails, the system stays stable and we can
+ * still close the span receiver.
+ */
+ @Test(timeout = 60000)
+ public void testRestThreadHandlesFlushFailure() throws Exception {
+ final Random rand = new Random(321);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testRestThreadHandlesFlushFailure");
+ put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+ put(Conf.PACKED_KEY, "false");
+ }});
+ TestInjectFlushFaults injector = new TestInjectFlushFaults(Long.MAX_VALUE);
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 15);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * A FaultInjector that causes all flushes to fail until a specified
+ * number of milliseconds have passed.
+ */
+ private static class WaitForFlushes
+ extends HTracedSpanReceiver.FaultInjector {
+ final Semaphore flushSem;
+
+ WaitForFlushes(int numFlushes) {
+ this.flushSem = new Semaphore(-numFlushes);
+ }
+
+ @Override
+ public void handleFlush() throws IOException {
+ flushSem.release();
+ }
+ }
+
+ /**
+ * Test that the packed code works when performing multiple flushes.
+ */
+ @Test(timeout = 60000)
+ public void testMultiplePackedFlushes() throws Exception {
+ final Random rand = new Random(123);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testMultiplePackedFlushes");
+ put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+ put(Conf.PACKED_KEY, "true");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1");
+ }});
+ WaitForFlushes injector = new WaitForFlushes(5);
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 3);
+ while (true) {
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ if (injector.flushSem.availablePermits() >= 0) {
+ break;
+ }
+ Thread.sleep(1);
+ }
+ waitForSpans(ht, spans, 3);
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that the REST code works when performing multiple flushes.
+ */
+ @Test(timeout = 60000)
+ public void testMultipleRestFlushes() throws Exception {
+ final Random rand = new Random(123);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testMultipleRestFlushes");
+ put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+ put(Conf.PACKED_KEY, "false");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1");
+ }});
+ WaitForFlushes injector = new WaitForFlushes(5);
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 3);
+ while (true) {
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ if (injector.flushSem.availablePermits() >= 0) {
+ break;
+ }
+ Thread.sleep(1);
+ }
+ waitForSpans(ht, spans, 3);
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that the packed code works when performing multiple flushes.
+ */
+ @Test(timeout = 60000)
+ public void testPackedRetryAfterFlushError() throws Exception {
+ final Random rand = new Random(123);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testPackedRetryAfterFlushError");
+ put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+ put(Conf.PACKED_KEY, "true");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1000");
+ put(Conf.FLUSH_RETRY_DELAYS_KEY, "100,100,100,100,100,100,100");
+ }});
+ TestInjectFlushFaults injector = new TestInjectFlushFaults(5);
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 3);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ waitForSpans(ht, spans);
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+
+ /**
+ * Test that the REST code works when performing multiple flushes.
+ */
+ @Test(timeout = 60000)
+ public void testRestRetryAfterFlushError() throws Exception {
+ final Random rand = new Random(123);
+ final HTracedProcess ht = new HTracedProcess.Builder().build();
+ try {
+ HTraceConfiguration conf = HTraceConfiguration.fromMap(
+ new HashMap<String, String>() {{
+ put(TracerId.TRACER_ID_KEY, "testRestRetryAfterFlushError");
+ put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+ put(Conf.PACKED_KEY, "false");
+ put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1000");
+ put(Conf.FLUSH_RETRY_DELAYS_KEY, "100,100,100,100,100,100,100");
+ }});
+ TestInjectFlushFaults injector = new TestInjectFlushFaults(5);
+ HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+ Span[] spans = TestUtil.randomSpans(rand, 3);
+ for (Span span : spans) {
+ rcvr.receiveSpan(span);
+ }
+ waitForSpans(ht, spans);
+ rcvr.close();
+ } finally {
+ ht.destroy();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiverConf.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiverConf.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiverConf.java
new file mode 100644
index 0000000..bf038f1
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiverConf.java
@@ -0,0 +1,91 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.HTraceConfiguration;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestHTracedReceiverConf {
+ private static final Log LOG =
+ LogFactory.getLog(TestHTracedReceiverConf.class);
+
+ @Test(timeout = 60000)
+ public void testParseHostPort() throws Exception {
+ InetSocketAddress addr = new Conf(
+ HTraceConfiguration.fromKeyValuePairs(
+ Conf.ADDRESS_KEY, "example.com:8080")).endpoint;
+ Assert.assertEquals("example.com", addr.getHostName());
+ Assert.assertEquals(8080, addr.getPort());
+
+ addr = new Conf(
+ HTraceConfiguration.fromKeyValuePairs(
+ Conf.ADDRESS_KEY, "127.0.0.1:8081")).endpoint;
+ Assert.assertEquals("127.0.0.1", addr.getHostName());
+ Assert.assertEquals(8081, addr.getPort());
+
+ addr = new Conf(
+ HTraceConfiguration.fromKeyValuePairs(
+ Conf.ADDRESS_KEY, "[ff02:0:0:0:0:0:0:12]:9095")).endpoint;
+ Assert.assertEquals("ff02:0:0:0:0:0:0:12", addr.getHostName());
+ Assert.assertEquals(9095, addr.getPort());
+ }
+
+ private static void verifyFail(String hostPort) {
+ try {
+ new Conf(HTraceConfiguration.fromKeyValuePairs(
+ Conf.ADDRESS_KEY, hostPort));
+ Assert.fail("Expected bad host:port configuration " + hostPort +
+ " to fail, but it succeeded.");
+ } catch (IOException e) {
+ // expected
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testFailToParseHostPort() throws Exception {
+ verifyFail("localhost"); // no port
+ verifyFail("127.0.0.1"); // no port
+ verifyFail(":8080"); // no hostname
+ verifyFail("bob[ff02:0:0:0:0:0:0:12]:9095"); // bracket at incorrect place
+ }
+
+ @Test(timeout = 60000)
+ public void testGetIntArray() throws Exception {
+ int[] arr = Conf.getIntArray("");
+ Assert.assertEquals(0, arr.length);
+ arr = Conf.getIntArray("123");
+ Assert.assertEquals(1, arr.length);
+ Assert.assertEquals(123, arr[0]);
+ arr = Conf.getIntArray("1,2,3");
+ Assert.assertEquals(3, arr.length);
+ Assert.assertEquals(1, arr[0]);
+ Assert.assertEquals(2, arr[1]);
+ Assert.assertEquals(3, arr[2]);
+ arr = Conf.getIntArray(",-4,5,66,");
+ Assert.assertEquals(3, arr.length);
+ Assert.assertEquals(-4, arr[0]);
+ Assert.assertEquals(5, arr[1]);
+ Assert.assertEquals(66, arr[2]);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/impl/TestPackedBuffer.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestPackedBuffer.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestPackedBuffer.java
new file mode 100644
index 0000000..ed7d904
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestPackedBuffer.java
@@ -0,0 +1,76 @@
+/**
+ * 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.htrace.impl;
+
+import java.nio.ByteBuffer;
+import java.util.Random;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.util.TestUtil;
+import org.junit.Assert;
+import org.junit.Test;
+import org.msgpack.core.MessagePack;
+import org.msgpack.core.MessageUnpacker;
+
+public class TestPackedBuffer {
+ private static final Log LOG = LogFactory.getLog(TestPackedBuffer.class);
+
+ @Test(timeout = 60000)
+ public void testWriteReqFrame() throws Exception {
+ byte[] arr = new byte[PackedBuffer.HRPC_REQ_FRAME_LENGTH];
+ ByteBuffer bb = ByteBuffer.wrap(arr);
+ PackedBuffer buf = new PackedBuffer(bb);
+ PackedBuffer.writeReqFrame(bb, 1, 123, 456);
+ Assert.assertEquals(PackedBuffer.HRPC_REQ_FRAME_LENGTH, bb.position());
+ Assert.assertEquals("48 54 52 43 " +
+ "01 00 00 00 " +
+ "7b 00 00 00 00 00 00 00 " +
+ "c8 01 00 00",
+ buf.toHexString());
+ }
+
+ @Test(timeout = 60000)
+ public void testPackSpans() throws Exception {
+ Random rand = new Random(123);
+ byte[] arr = new byte[16384];
+ ByteBuffer bb = ByteBuffer.wrap(arr);
+ bb.limit(bb.capacity());
+ PackedBuffer buf = new PackedBuffer(bb);
+ final int NUM_TEST_SPANS = 5;
+ Span[] spans = new Span[NUM_TEST_SPANS];
+ for (int i = 0; i < NUM_TEST_SPANS; i++) {
+ spans[i] = TestUtil.randomSpan(rand);
+ }
+ for (int i = 0; i < NUM_TEST_SPANS; i++) {
+ buf.writeSpan(spans[i]);
+ }
+ LOG.info("wrote " + buf.toHexString());
+ MessagePack msgpack = new MessagePack(PackedBuffer.MSGPACK_CONF);
+ MessageUnpacker unpacker = msgpack.newUnpacker(arr, 0, bb.position());
+ Span[] respans = new Span[NUM_TEST_SPANS];
+ for (int i = 0; i < NUM_TEST_SPANS; i++) {
+ respans[i] = PackedBuffer.readSpan(unpacker);
+ }
+ for (int i = 0; i < NUM_TEST_SPANS; i++) {
+ Assert.assertEquals("Failed to read back span " + i,
+ spans[i].toJson(), respans[i].toJson());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/impl/TestTimeUtil.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestTimeUtil.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestTimeUtil.java
new file mode 100644
index 0000000..630a02a
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestTimeUtil.java
@@ -0,0 +1,46 @@
+/**
+ * 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.htrace.impl;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestTimeUtil {
+ /**
+ * Test that our deltaMs function can compute the time difference between any
+ * two monotonic times in milliseconds.
+ */
+ @Test(timeout = 60000)
+ public void testDeltaMs() throws Exception {
+ Assert.assertEquals(0, TimeUtil.deltaMs(0, 0));
+ Assert.assertEquals(1, TimeUtil.deltaMs(0, 1));
+ Assert.assertEquals(0, TimeUtil.deltaMs(1, 0));
+ Assert.assertEquals(10, TimeUtil.deltaMs(1000, 1010));
+ long minMs = TimeUnit.MILLISECONDS.convert(Long.MIN_VALUE,
+ TimeUnit.NANOSECONDS);
+ long maxMs = TimeUnit.MILLISECONDS.convert(Long.MAX_VALUE,
+ TimeUnit.NANOSECONDS);
+ Assert.assertEquals(10, TimeUtil.deltaMs(minMs, minMs + 10));
+ Assert.assertEquals(maxMs, TimeUtil.deltaMs(minMs, maxMs));
+ Assert.assertEquals(11, TimeUtil.deltaMs(maxMs - 10, minMs));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/util/DataDir.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/util/DataDir.java b/htrace-htraced/src/test/java/org/apache/htrace/util/DataDir.java
deleted file mode 100644
index 74731fa..0000000
--- a/htrace-htraced/src/test/java/org/apache/htrace/util/DataDir.java
+++ /dev/null
@@ -1,97 +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.htrace.util;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.UUID;
-
-/**
- * Small util for making a data directory for tests to use when running tests. We put it up at
- * target/test-data/UUID. Create an instance of this class per unit test run and it will take
- * care of setting up the dirs for you. Pass what is returned here as location from which to
- * have daemons and tests dump data.
- * TODO: Add close on exit.
- */
-public class DataDir {
- private File baseTestDir = null;
- private File testDir = null;
-
- /**
- * System property key to get base test directory value
- */
- public static final String TEST_BASE_DIRECTORY_KEY = "test.data.base.dir";
-
- /**
- * Default base directory for test output.
- */
- public static final String TEST_BASE_DIRECTORY_DEFAULT = "target";
-
- public static final String TEST_BASE_DIRECTORY_NAME = "test-data";
-
- /**
- * @return Where to write test data on local filesystem; usually
- * {@link #TEST_BASE_DIRECTORY_DEFAULT}
- * Should not be used directly by the unit tests, hence its's private.
- * Unit test will use a subdirectory of this directory.
- * @see #setupDataTestDir()
- */
- private synchronized File getBaseTestDir() {
- if (this.baseTestDir != null) return this.baseTestDir;
- String testHome = System.getProperty(TEST_BASE_DIRECTORY_KEY, TEST_BASE_DIRECTORY_DEFAULT);
- this.baseTestDir = new File(testHome, TEST_BASE_DIRECTORY_NAME);
- return this.baseTestDir;
- }
-
- /**
- * @return Absolute path to the dir created by this instance.
- * @throws IOException
- */
- public synchronized File getDataDir() throws IOException {
- if (this.testDir != null) return this.testDir;
- this.testDir = new File(getBaseTestDir(), UUID.randomUUID().toString());
- if (!this.testDir.exists()) {
- if (!this.testDir.mkdirs()) throw new IOException("Failed mkdirs for " + this.testDir);
- }
- // Return absolute path. A relative passed to htraced will have it create data dirs relative
- // to its data dir rather than in it.
- return this.testDir.getAbsoluteFile();
- }
-
- /**
- * Fragile. Ugly. Presumes paths. Best we can do for now until htraced comes local to this module
- * and is moved out of src dir.
- * @param dataDir A datadir gotten from {@link #getDataDir()}
- * @return Top-level of the checkout.
- */
- public static File getTopLevelOfCheckout(final File dataDir) {
- // Need absolute else we run out of road when dir is relative to this module.
- File absolute = dataDir.getAbsoluteFile();
- // Check we are where we think we are.
- File testDataDir = absolute.getParentFile();
- if (!testDataDir.getName().equals(TEST_BASE_DIRECTORY_NAME)) {
- throw new IllegalArgumentException(dataDir.toString());
- }
- // Do another check.
- File targetDir = testDataDir.getParentFile();
- if (!targetDir.getName().equals(TEST_BASE_DIRECTORY_DEFAULT)) {
- throw new IllegalArgumentException(dataDir.toString());
- }
- // Back up last two dirs out of the htrace-htraced dir.
- return targetDir.getParentFile().getParentFile();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-htrace/blob/cb2351d2/htrace-htraced/src/test/java/org/apache/htrace/util/HTracedProcess.java
----------------------------------------------------------------------
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/util/HTracedProcess.java b/htrace-htraced/src/test/java/org/apache/htrace/util/HTracedProcess.java
deleted file mode 100644
index 3e800d2..0000000
--- a/htrace-htraced/src/test/java/org/apache/htrace/util/HTracedProcess.java
+++ /dev/null
@@ -1,172 +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.htrace.util;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.lang.ProcessBuilder.Redirect;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.URL;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-/**
- * To get instance of HTraced up and running, create an instance of this class.
- * Upon successful construction, htraced is running using <code>dataDir</code> as directory to
- * host data (leveldbs and logs).
- * TODO: We expect to find the htraced in a very particular place. Fragile. Will break if stuff
- * moves.
- */
-public class HTracedProcess extends Process {
- private static final Log LOG = LogFactory.getLog(HTracedProcess.class);
- private final Process delegate;
-
- private final String httpAddr;
-
- /**
- * Data send back from the HTraced process on the notification port.
- */
- @JsonIgnoreProperties(ignoreUnknown = true)
- public static class StartupNotificationData {
- /**
- * The hostname:port pair which the HTraced process uses for HTTP requests.
- */
- @JsonProperty("HttpAddr")
- String httpAddr;
-
- /**
- * The process ID of the HTraced process.
- */
- @JsonProperty("ProcessId")
- long processId;
- }
-
- public HTracedProcess(final File binPath, final File dataDir,
- final String host) throws IOException {
- // Create a notifier socket bound to a random port.
- ServerSocket listener = new ServerSocket(0);
- boolean success = false;
- Process process = null;
- try {
- // Use a random port for the web address. No 'scheme' yet.
- String webAddress = host + ":0";
- String logPath = new File(dataDir, "log.txt").getAbsolutePath();
- // Pass cmdline args to htraced to it uses our test dir for data.
- ProcessBuilder pb = new ProcessBuilder(binPath.toString(),
- "-Dlog.level=TRACE",
- "-Dlog.path=" + logPath,
- "-Dweb.address=" + webAddress,
- "-Ddata.store.clear=true",
- "-Dstartup.notification.address=localhost:" + listener.getLocalPort(),
- "-Ddata.store.directories=" + dataDir.toString());
- pb.redirectErrorStream(true);
- // Inherit STDERR/STDOUT i/o; dumps on console for now. Can add logs later.
- pb.inheritIO();
- pb.directory(dataDir);
- //assert pb.redirectInput() == Redirect.PIPE;
- //assert pb.redirectOutput().file() == dataDir;
- process = pb.start();
- assert process.getInputStream().read() == -1;
- StartupNotificationData data = readStartupNotification(listener);
- httpAddr = data.httpAddr;
- LOG.info("Started htraced process " + data.processId + " with http " +
- "address " + data.httpAddr + ", logging to " + logPath);
- success = true;
- } finally {
- if (!success) {
- // Clean up after failure
- if (process != null) {
- process.destroy();
- process = null;
- }
- }
- delegate = process;
- listener.close();
- }
- }
-
- private static StartupNotificationData
- readStartupNotification(ServerSocket listener) throws IOException {
- Socket socket = listener.accept();
- try {
- InputStream in = socket.getInputStream();
- ObjectMapper objectMapper = new ObjectMapper();
- StartupNotificationData data = objectMapper.
- readValue(in, StartupNotificationData.class);
- return data;
- } finally {
- socket.close();
- }
- }
-
- public int hashCode() {
- return delegate.hashCode();
- }
-
- public OutputStream getOutputStream() {
- throw new UnsupportedOperationException("Unsupported until complaint; output on STDOUT");
- }
-
- public InputStream getInputStream() {
- throw new UnsupportedOperationException("Unsupported until complaint; output on STDOUT");
- }
-
- public boolean equals(Object obj) {
- return delegate.equals(obj);
- }
-
- public InputStream getErrorStream() {
- throw new UnsupportedOperationException("Unsupported until complaint; output on STDOUT");
- }
-
- public int waitFor() throws InterruptedException {
- return delegate.waitFor();
- }
-
- public int exitValue() {
- return delegate.exitValue();
- }
-
- public void destroy() {
- delegate.destroy();
- }
-
- public String toString() {
- return delegate.toString();
- }
-
- public String getHttpAddr() {
- return httpAddr;
- }
-
- /**
- * Ugly but how else to do file-math?
- * @param topLevel Presumes top-level of the htrace checkout.
- * @return Path to the htraced binary.
- */
- public static File getPathToHTraceBinaryFromTopLevel(final File topLevel) {
- return new File(new File(new File(new File(topLevel, "htrace-htraced"), "go"), "build"),
- "htraced");
- }
-}