You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by jy...@apache.org on 2014/07/28 15:40:52 UTC
[1/3] PHOENIX-177: Collect usage and performance metrics
Repository: phoenix
Updated Branches:
refs/heads/master 9185f760b -> b7f46c105
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TestableMetricsWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TestableMetricsWriter.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TestableMetricsWriter.java
new file mode 100644
index 0000000..b6bc75d
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TestableMetricsWriter.java
@@ -0,0 +1,30 @@
+/**
+ * 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.phoenix.trace;
+
+import org.apache.phoenix.metrics.MetricsWriter;
+
+/**
+ * Marker interface for a MetricsWriter that can be registered to the current metrics system. The
+ * writer should convert from the metrics information it receives from the metrics system to Phoenix
+ * records that the MetricsWriter can read (and subsequently write).
+ */
+public interface TestableMetricsWriter {
+
+ public void setWriterForTesting(MetricsWriter writer);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TracingCompat.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TracingCompat.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TracingCompat.java
new file mode 100644
index 0000000..6ec12de
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/TracingCompat.java
@@ -0,0 +1,94 @@
+/**
+ * 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.phoenix.trace;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.CompatibilityFactory;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.metrics.MetricsWriter;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.SpanReceiver;
+
+/**
+ * Utilities for tracing that are common among the compatibility and core classes.
+ */
+public class TracingCompat {
+
+ private static final Log LOG = LogFactory.getLog(TracingCompat.class);
+
+ /**
+ * @return a new SpanReceiver that will write to the correct metrics system
+ */
+ public static SpanReceiver newTraceMetricSource() {
+ return CompatibilityFactory.getInstance(PhoenixSpanReceiver.class);
+ }
+
+ public static final String DEFAULT_STATS_TABLE_NAME = "PHOENIX.TRACING_STATS";
+
+ /**
+ * Configuration key to overwrite the tablename that should be used as the target table
+ */
+ public static final String TARGET_TABLE_CONF_KEY =
+ "org.apache.phoenix._internal.trace.tablename";
+
+ public static final String METRIC_SOURCE_KEY = "phoenix.";
+
+ /** Set context to enable filtering */
+ public static final String METRICS_CONTEXT = "tracing";
+
+ public static void addAnnotation(Span span, String message, int value) {
+ span.addKVAnnotation(message.getBytes(), Bytes.toBytes(value));
+ }
+
+ public static Pair<String, String> readAnnotation(byte[] key, byte[] value) {
+ return new Pair<String, String>(new String(key), Integer.toString(Bytes.toInt(value)));
+ }
+
+ public static MetricsWriter initializeWriter(String clazz) {
+ try {
+ MetricsWriter writer =
+ Class.forName(clazz).asSubclass(MetricsWriter.class).newInstance();
+ writer.initialize();
+ return writer;
+ } catch (InstantiationException e) {
+ LOG.error("Failed to create metrics writer: " + clazz, e);
+ } catch (IllegalAccessException e) {
+ LOG.error("Failed to create metrics writer: " + clazz, e);
+ } catch (ClassNotFoundException e) {
+ LOG.error("Failed to create metrics writer: " + clazz, e);
+ }
+ return null;
+ }
+
+ /**
+ * @see #getTraceMetricName(String)
+ */
+ public static final String getTraceMetricName(long traceId) {
+ return getTraceMetricName(Long.toString(traceId));
+ }
+
+ /**
+ * @param traceId unique id of the trace
+ * @return the name of the metric record that should be generated for a given trace
+ */
+ public static final String getTraceMetricName(String traceId) {
+ return METRIC_SOURCE_KEY + traceId;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/LoggingSink.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/LoggingSink.java b/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/LoggingSink.java
new file mode 100644
index 0000000..97682b3
--- /dev/null
+++ b/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/LoggingSink.java
@@ -0,0 +1,56 @@
+/**
+ * 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.phoenix.metrics;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.phoenix.trace.TracingCompat;
+
+/**
+ * Simple sink that just logs the output of all the metrics that start with
+ * {@link TracingCompat#METRIC_SOURCE_KEY}
+ */
+public class LoggingSink implements MetricsWriter {
+
+ private static final Log LOG = LogFactory.getLog(LoggingSink.class);
+
+ @Override
+ public void initialize() {
+ }
+
+ @Override
+ public void addMetrics(PhoenixMetricsRecord record) {
+ // we could wait until flush, but this is a really lightweight process, so we just write
+ // them
+ // as soon as we get them
+ if (!LOG.isDebugEnabled()) {
+ return;
+ }
+ LOG.debug("Found record:" + record.name());
+ for (PhoenixAbstractMetric metric : record.metrics()) {
+ // just print the metric we care about
+ if (metric.getName().startsWith(TracingCompat.METRIC_SOURCE_KEY)) {
+ LOG.debug("\t metric:" + metric);
+ }
+ }
+ }
+
+ @Override
+ public void flush() {
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/TracingTestCompat.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/TracingTestCompat.java b/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/TracingTestCompat.java
new file mode 100644
index 0000000..8dd8a41
--- /dev/null
+++ b/phoenix-hadoop-compat/src/test/java/org/apache/phoenix/metrics/TracingTestCompat.java
@@ -0,0 +1,45 @@
+/**
+ * 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.phoenix.metrics;
+
+import org.apache.hadoop.hbase.CompatibilityFactory;
+import org.apache.phoenix.trace.TestableMetricsWriter;
+
+/**
+ * Utility class for testing tracing
+ */
+public class TracingTestCompat {
+
+ private TracingTestCompat() {
+ assert false;
+ }
+
+ public static TestableMetricsWriter newTraceMetricSink() {
+ return CompatibilityFactory.getInstance(TestableMetricsWriter.class);
+ }
+
+ /**
+ * Register the sink with the metrics system, so we don't need to specify it in the conf
+ * @param sink
+ */
+ public static void registerSink(MetricsWriter sink) {
+ TestableMetricsWriter writer = newTraceMetricSink();
+ writer.setWriterForTesting(sink);
+ Metrics.getManager().register("phoenix", "test sink gets logged", writer);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/pom.xml b/phoenix-hadoop2-compat/pom.xml
index 2781bcf..7944a51 100644
--- a/phoenix-hadoop2-compat/pom.xml
+++ b/phoenix-hadoop2-compat/pom.xml
@@ -29,4 +29,49 @@
</parent>
<artifactId>phoenix-hadoop2-compat</artifactId>
<name>Phoenix Hadoop2 Compatibility</name>
-</project>
+
+ <dependencies>
+ <!-- Intra-project dependencies -->
+ <dependency>
+ <groupId>org.apache.phoenix</groupId>
+ <artifactId>phoenix-hadoop-compat</artifactId>
+ </dependency>
+ <!-- HBase -->
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <exclusions>
+ <exclusion>
+ <artifactId>hadoop-core</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <!-- Hadoop -->
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <version>${hadoop-two.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-annotations</artifactId>
+ <version>${hadoop-two.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop-two.version}</version>
+ </dependency>
+ <!-- Other -->
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <!-- Test -->
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ </dependency>
+ </dependencies>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/metrics/MetricsManagerImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/metrics/MetricsManagerImpl.java b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/metrics/MetricsManagerImpl.java
new file mode 100644
index 0000000..03e06a5
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/metrics/MetricsManagerImpl.java
@@ -0,0 +1,71 @@
+/**
+ * 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.phoenix.metrics;
+
+import java.util.Arrays;
+
+import org.apache.hadoop.metrics2.MetricsSink;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.MetricsSystem;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+
+import com.google.common.base.Preconditions;
+
+/**
+ *
+ */
+public class MetricsManagerImpl implements MetricsManager {
+
+ private MetricsSystem system;
+
+ @Override
+ /**
+ * Register a metrics sink
+ * @param <T> the type of the sink.
+ * @param sink to register
+ * @param name of the sink. Must be unique.
+ * @param desc the description of the sink
+ * @return the sink
+ * @throws IllegalArgumentException if sink is not a MetricsSink
+ */
+ public <T> T register(String name, String desc, T sink) {
+ isA(sink, MetricsSink.class);
+ return (T) system.register(name, desc, (MetricsSink) sink);
+ }
+
+ public <T> T registerSource(String name, String desc, T source) {
+ isA(source, MetricsSource.class);
+ return (T) system.register(name, desc, (MetricsSource) source);
+ }
+
+ @Override
+ public void initialize(String prefix) {
+ this.system = DefaultMetricsSystem.initialize(prefix);
+ }
+
+ private <T> void isA(T object, Class<?>... classes) {
+ boolean match = false;
+ for (Class<?> clazz : classes) {
+ if (clazz.isAssignableFrom(object.getClass())) {
+ match = true;
+ break;
+ }
+ }
+ Preconditions.checkArgument(match, object + " is not one of " + Arrays.toString(classes));
+ }
+
+ @Override
+ public void shutdown() {
+ if (this.system != null) {
+ this.system.shutdown();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/MetricsInfoImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/MetricsInfoImpl.java b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/MetricsInfoImpl.java
new file mode 100644
index 0000000..47c1dda
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/MetricsInfoImpl.java
@@ -0,0 +1,63 @@
+/**
+ * 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.phoenix.trace;
+
+import com.google.common.base.Objects;
+import static com.google.common.base.Preconditions.*;
+import org.apache.hadoop.metrics2.MetricsInfo;
+
+/**
+ * Making implementing metric info a little easier
+ * <p>
+ * Just a copy of the same from Hadoop, but exposed for usage.
+ */
+public class MetricsInfoImpl implements MetricsInfo {
+ private final String name, description;
+
+ MetricsInfoImpl(String name, String description) {
+ this.name = checkNotNull(name, "name");
+ this.description = checkNotNull(description, "description");
+ }
+
+ @Override public String name() {
+ return name;
+ }
+
+ @Override public String description() {
+ return description;
+ }
+
+ @Override public boolean equals(Object obj) {
+ if (obj instanceof MetricsInfo) {
+ MetricsInfo other = (MetricsInfo) obj;
+ return Objects.equal(name, other.name()) &&
+ Objects.equal(description, other.description());
+ }
+ return false;
+ }
+
+ @Override public int hashCode() {
+ return Objects.hashCode(name, description);
+ }
+
+ @Override public String toString() {
+ return Objects.toStringHelper(this)
+ .add("name", name).add("description", description)
+ .toString();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/PhoenixMetricsWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/PhoenixMetricsWriter.java b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/PhoenixMetricsWriter.java
new file mode 100644
index 0000000..03230ee
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/PhoenixMetricsWriter.java
@@ -0,0 +1,176 @@
+/**
+ * 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.phoenix.trace;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+
+import javax.annotation.Nullable;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.SubsetConfiguration;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsSink;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.phoenix.metrics.MetricsWriter;
+import org.apache.phoenix.metrics.PhoenixAbstractMetric;
+import org.apache.phoenix.metrics.PhoenixMetricTag;
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterators;
+
+/**
+ * Translate metrics from a Hadoop2 metrics2 metric to a generic PhoenixMetric that a
+ * {@link MetricsWriter} can then write out.
+ * <p>
+ * This class becomes unnecessary once we drop Hadoop1 support.
+ */
+public class PhoenixMetricsWriter implements MetricsSink, TestableMetricsWriter {
+
+ /**
+ * Metrics configuration key for the class that should be used for writing the output
+ */
+ public static final String PHOENIX_METRICS_WRITER_CLASS = "phoenix.sink.writer-class";
+
+ public static void setWriterClass(MetricsWriter writer, Configuration conf) {
+ conf.setProperty(PHOENIX_METRICS_WRITER_CLASS, writer.getClass().getName());
+ }
+
+ private MetricsWriter writer;
+
+ @Override
+ public void init(SubsetConfiguration config) {
+ // instantiate the configured writer class
+ String clazz = config.getString(PHOENIX_METRICS_WRITER_CLASS);
+ this.writer = TracingCompat.initializeWriter(clazz);
+ Preconditions.checkNotNull(writer, "Could not correctly initialize metrics writer!");
+ }
+
+ @Override
+ @VisibleForTesting
+ public void setWriterForTesting(MetricsWriter writer) {
+ this.writer = writer;
+ }
+
+ @Override
+ public void putMetrics(MetricsRecord record) {
+ writer.addMetrics(wrap(record));
+ }
+
+ @Override
+ public void flush() {
+ writer.flush();
+ }
+
+ /**
+ * Convert the passed record to a {@link PhoenixMetricsRecord}
+ * @param record to convert
+ * @return a generic {@link PhoenixMetricsRecord} that delegates to the record in all things
+ */
+ private PhoenixMetricsRecord wrap(final MetricsRecord record) {
+ return new PhoenixMetricsRecord() {
+
+ @Override
+ public String name() {
+ return record.name();
+ }
+
+ @Override
+ public String description() {
+ return record.description();
+ }
+
+ @Override
+ public Iterable<PhoenixAbstractMetric> metrics() {
+ final Iterable<AbstractMetric> iterable = record.metrics();
+ return new Iterable<PhoenixAbstractMetric>(){
+
+ @Override
+ public Iterator<PhoenixAbstractMetric> iterator() {
+ final Iterator<AbstractMetric> iter = iterable.iterator();
+ return Iterators.transform(iter, new Function<AbstractMetric, PhoenixAbstractMetric>() {
+
+ @Override
+ @Nullable
+ public PhoenixAbstractMetric apply(@Nullable final AbstractMetric input) {
+ if (input == null) {
+ return null;
+ }
+ return new PhoenixAbstractMetric() {
+
+ @Override
+ public Number value() {
+ return input.value();
+ }
+
+ @Override
+ public String getName() {
+ return input.name();
+ }
+
+ @Override
+ public String toString() {
+ return input.toString();
+ }
+ };
+ }
+ });
+ }
+ };
+ }
+
+ @Override
+ public Collection<PhoenixMetricTag> tags() {
+ Collection<PhoenixMetricTag> tags = new ArrayList<PhoenixMetricTag>();
+ Collection<MetricsTag> origTags = record.tags();
+ for (final MetricsTag tag : origTags) {
+ tags.add(new PhoenixMetricTag() {
+
+ @Override
+ public String name() {
+ return tag.name();
+ }
+
+ @Override
+ public String description() {
+ return tag.description();
+ }
+
+ @Override
+ public String value() {
+ return tag.value();
+ }
+
+ @Override
+ public String toString() {
+ return tag.toString();
+ }
+
+ });
+ }
+ return tags;
+ }
+
+ };
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/TraceMetricSource.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/TraceMetricSource.java b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/TraceMetricSource.java
new file mode 100644
index 0000000..5876771
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/main/java/org/apache/phoenix/trace/TraceMetricSource.java
@@ -0,0 +1,192 @@
+/**
+ * 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.phoenix.trace;
+
+import static org.apache.phoenix.metrics.MetricInfo.ANNOTATION;
+import static org.apache.phoenix.metrics.MetricInfo.TAG;
+import static org.apache.phoenix.metrics.MetricInfo.END;
+import static org.apache.phoenix.metrics.MetricInfo.PARENT;
+import static org.apache.phoenix.metrics.MetricInfo.SPAN;
+import static org.apache.phoenix.metrics.MetricInfo.START;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.metrics2.MetricsCollector;
+import org.apache.hadoop.metrics2.MetricsInfo;
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.hadoop.metrics2.lib.Interns;
+import org.apache.phoenix.metrics.MetricInfo;
+import org.apache.phoenix.metrics.Metrics;
+import org.apache.phoenix.metrics.MetricsManager;
+import org.cloudera.htrace.HTraceConfiguration;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.SpanReceiver;
+import org.cloudera.htrace.TimelineAnnotation;
+import org.cloudera.htrace.impl.MilliSpan;
+
+/**
+ * Sink for request traces ({@link SpanReceiver}) that pushes writes to {@link MetricsSource} in a
+ * format that we can more easily consume.
+ * <p>
+ * <p>
+ * Rather than write directly to a phoenix table, we drop it into the metrics queue so we can more
+ * cleanly handle it asyncrhonously.Currently, {@link MilliSpan} submits the span in a synchronized
+ * block to all the receivers, which could have a lot of overhead if we are submitting to multiple
+ * receivers.
+ * <p>
+ * The format of the generated metrics is this:
+ * <ol>
+ * <li>All Metrics from the same span have the same name (allowing correlation in the sink)</li>
+ * <li>The description of the metric describes what it contains. For instance,
+ * <ul>
+ * <li>{@link MetricInfo#PARENT} is the id of the parent of this span. (Root span is
+ * {@link Span#ROOT_SPAN_ID}).</li>
+ * <li>{@value MetricInfo#START} is the start time of the span</li>
+ * <li>{@value MetricInfo#END} is the end time of the span</li>
+ * </ul></li>
+ * <li>Each span's messages are contained in a {@link MetricsTag} with the same name as above and a
+ * generic counter for the number of messages (to differentiate messages and provide timeline
+ * ordering).</li>
+ * </ol>
+ * <p>
+ * <i>So why even submit to metrics2 framework if we only have a single source?</i>
+ * <p>
+ * This allows us to make the updates in batches. We might have spans that finish before other spans
+ * (for instance in the same parent). By batching the updates we can lessen the overhead on the
+ * client, which is also busy doing 'real' work. <br>
+ * We could make our own queue and manage batching and filtering and dropping extra metrics, but
+ * that starts to get complicated fast (its not as easy as it sounds) so we use metrics2 to abstract
+ * out that pipeline and also provides us flexibility to dump metrics to other sources.
+ * <p>
+ * This is a somewhat rough implementation - we do excessive locking for correctness,
+ * rather than trying to make it fast, for the moment.
+ */
+public class TraceMetricSource implements PhoenixSpanReceiver, MetricsSource {
+
+ private static final String EMPTY_STRING = "";
+
+ /** This must match the prefix that we are using in the hadoop-metrics2 config */
+ private static final String METRICS_SYSTEM_NAME = "phoenix";
+ private static final String CONTEXT = "tracing";
+
+ private List<Metric> spans = new ArrayList<Metric>();
+
+ public TraceMetricSource() {
+ MetricsManager manager = Metrics.getManager();
+ manager.initialize(METRICS_SYSTEM_NAME);
+
+ // Register this instance.
+ // For right now, we ignore the MBean registration issues that show up in DEBUG logs. Basically,
+ // we need a Jmx MBean compliant name. We'll get to a better name when we want that later
+ manager.registerSource(CONTEXT, "Phoenix call tracing", this);
+ }
+
+ @Override
+ public void receiveSpan(Span span) {
+ Metric builder = new Metric(span);
+ // add all the metrics for the span
+ builder.addCounter(Interns.info(SPAN.traceName, EMPTY_STRING), span.getSpanId());
+ builder.addCounter(Interns.info(PARENT.traceName, EMPTY_STRING), span.getParentId());
+ builder.addCounter(Interns.info(START.traceName, EMPTY_STRING), span.getStartTimeMillis());
+ builder.addCounter(Interns.info(END.traceName, EMPTY_STRING), span.getStopTimeMillis());
+ // add the tags to the span. They were written in order received so we mark them as such
+ for (TimelineAnnotation ta : span.getTimelineAnnotations()) {
+ builder.add(new MetricsTag(Interns.info(TAG.traceName, Long.toString(ta.getTime())), ta
+ .getMessage()));
+ }
+
+ // add the annotations. We assume they are serialized as strings and integers, but that can
+ // change in the future
+ Map<byte[], byte[]> annotations = span.getKVAnnotations();
+ for (Entry<byte[], byte[]> annotation : annotations.entrySet()) {
+ Pair<String, String> val =
+ TracingCompat.readAnnotation(annotation.getKey(), annotation.getValue());
+ builder.add(new MetricsTag(Interns.info(ANNOTATION.traceName, val.getFirst()), val
+ .getSecond()));
+ }
+
+ // add the span to the list we care about
+ synchronized (this) {
+ spans.add(builder);
+ }
+ }
+
+ @Override
+ public void getMetrics(MetricsCollector collector, boolean all) {
+ synchronized (this) {
+ for (Metric span : spans) {
+ MetricsRecordBuilder builder = collector.addRecord(new MetricsInfoImpl(TracingCompat
+ .getTraceMetricName(span.id), span.desc));
+ builder.setContext(TracingCompat.METRICS_CONTEXT);
+ for (Pair<MetricsInfo, Long> metric : span.counters) {
+ builder.addCounter(metric.getFirst(), metric.getSecond());
+ }
+ for (MetricsTag tag : span.tags) {
+ builder.add(tag);
+ }
+ }
+ // reset the spans so we don't keep a big chunk of memory around
+ spans = new ArrayList<Metric>();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ // noop
+ }
+
+ @Override
+ public void configure(HTraceConfiguration conf) {
+ // noop
+ }
+
+ private class Metric {
+
+ List<Pair<MetricsInfo, Long>> counters = new ArrayList<Pair<MetricsInfo, Long>>();
+ List<MetricsTag> tags = new ArrayList<MetricsTag>();
+ private String id;
+ private String desc;
+
+ public Metric(Span span) {
+ this.id = Long.toString(span.getTraceId());
+ this.desc = span.getDescription();
+ }
+
+ /**
+ * @param metricsInfoImpl
+ * @param startTimeMillis
+ */
+ public void addCounter(MetricsInfo metricsInfoImpl, long startTimeMillis) {
+ counters.add(new Pair<MetricsInfo, Long>(metricsInfoImpl, startTimeMillis));
+ }
+
+ /**
+ * @param metricsTag
+ */
+ public void add(MetricsTag metricsTag) {
+ tags.add(metricsTag);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.metrics.MetricsManager
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.metrics.MetricsManager b/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.metrics.MetricsManager
new file mode 100644
index 0000000..8430a48
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.metrics.MetricsManager
@@ -0,0 +1 @@
+org.apache.phoenix.metrics.MetricsManagerImpl
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.PhoenixSpanReceiver
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.PhoenixSpanReceiver b/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.PhoenixSpanReceiver
new file mode 100644
index 0000000..3694093
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.PhoenixSpanReceiver
@@ -0,0 +1 @@
+org.apache.phoenix.trace.TraceMetricSource
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.TestableMetricsWriter
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.TestableMetricsWriter b/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.TestableMetricsWriter
new file mode 100644
index 0000000..b10af39
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/main/resources/META-INF/services/org.apache.phoenix.trace.TestableMetricsWriter
@@ -0,0 +1 @@
+org.apache.phoenix.trace.PhoenixMetricsWriter
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricCounterLong.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricCounterLong.java b/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricCounterLong.java
new file mode 100644
index 0000000..33ca738
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricCounterLong.java
@@ -0,0 +1,35 @@
+/*
+ * 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.hadoop.metrics2.impl;
+
+import org.apache.hadoop.metrics2.MetricsInfo;
+import org.apache.hadoop.metrics2.impl.MetricCounterLong;
+
+/**
+ *
+ */
+public class ExposedMetricCounterLong extends MetricCounterLong {
+
+ /**
+ * @param info
+ * @param value
+ */
+ public ExposedMetricCounterLong(MetricsInfo info, long value) {
+ super(info, value);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricsRecordImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricsRecordImpl.java b/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricsRecordImpl.java
new file mode 100644
index 0000000..bcb8b43
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/impl/ExposedMetricsRecordImpl.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.hadoop.metrics2.impl;
+
+import java.util.List;
+
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsInfo;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.hadoop.metrics2.impl.MetricsRecordImpl;
+
+/**
+ * Helper class to access the package-private {@link MetricsRecordImpl}
+ */
+@SuppressWarnings("javadoc")
+public class ExposedMetricsRecordImpl extends MetricsRecordImpl {
+
+ /**
+ * @param info
+ * @param timestamp
+ * @param tags
+ * @param metrics
+ */
+ public ExposedMetricsRecordImpl(MetricsInfo info, long timestamp, List<MetricsTag> tags,
+ Iterable<AbstractMetric> metrics) {
+ super(info, timestamp, tags, metrics);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/lib/ExposedMetricsInfoImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/lib/ExposedMetricsInfoImpl.java b/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/lib/ExposedMetricsInfoImpl.java
new file mode 100644
index 0000000..6daf604
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/test/java/org/apache/hadoop/metrics2/lib/ExposedMetricsInfoImpl.java
@@ -0,0 +1,32 @@
+/**
+ * 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.hadoop.metrics2.lib;
+
+/**
+ * Helper class to expose access to the {@link MetricsInfoImpl}
+ */
+public class ExposedMetricsInfoImpl extends MetricsInfoImpl {
+
+ /**
+ * @param name
+ * @param description
+ */
+ public ExposedMetricsInfoImpl(String name, String description) {
+ super(name, description);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/PhoenixMetricsWriterTest.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/PhoenixMetricsWriterTest.java b/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/PhoenixMetricsWriterTest.java
new file mode 100644
index 0000000..0c8bbab
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/PhoenixMetricsWriterTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.phoenix.trace;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsInfo;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.hadoop.metrics2.impl.ExposedMetricCounterLong;
+import org.apache.hadoop.metrics2.impl.ExposedMetricsRecordImpl;
+import org.apache.hadoop.metrics2.lib.ExposedMetricsInfoImpl;
+import org.apache.phoenix.metrics.MetricInfo;
+import org.apache.phoenix.metrics.MetricsWriter;
+import org.apache.phoenix.metrics.PhoenixAbstractMetric;
+import org.apache.phoenix.metrics.PhoenixMetricTag;
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Test that we correctly convert between hadoop2 metrics2 and generic phoenix metrics.
+ */
+public class PhoenixMetricsWriterTest {
+
+ @Test
+ public void testTranslation() throws Exception {
+ // hook up a sink we can test
+ MetricsWriter mockSink = Mockito.mock(MetricsWriter.class);
+
+ // writer that will translate to the sink (specific to hadoop version used)
+ PhoenixMetricsWriter writer = new PhoenixMetricsWriter();
+ writer.setWriterForTesting(mockSink);
+
+ // create a simple metrics record
+ final long traceid = 987654;
+ MetricsInfo info = new ExposedMetricsInfoImpl(TracingCompat.getTraceMetricName(traceid),
+ "Some generic trace");
+ // setup some metrics for the span
+ long spanid = 10;
+ AbstractMetric span = new ExposedMetricCounterLong(new ExposedMetricsInfoImpl(
+ MetricInfo.SPAN.traceName, ""), spanid);
+ long parentid = 11;
+ AbstractMetric parent = new ExposedMetricCounterLong(new ExposedMetricsInfoImpl(
+ MetricInfo.PARENT.traceName, ""), parentid);
+ long startTime = 12;
+ AbstractMetric start = new ExposedMetricCounterLong(new ExposedMetricsInfoImpl(
+ MetricInfo.START.traceName, ""), startTime);
+ long endTime = 13;
+ AbstractMetric end = new ExposedMetricCounterLong(new ExposedMetricsInfoImpl(
+ MetricInfo.END.traceName, ""), endTime);
+ final List<AbstractMetric> metrics = Lists.newArrayList(span, parent, start, end);
+
+ // create an annotation as well
+ String annotation = "test annotation for a span";
+ MetricsTag tag = new MetricsTag(
+ new ExposedMetricsInfoImpl(MetricInfo.ANNOTATION.traceName, "0"), annotation);
+ String hostnameValue = "host-name.value";
+ MetricsTag hostname = new MetricsTag(new ExposedMetricsInfoImpl(MetricInfo.HOSTNAME.traceName,
+ ""), hostnameValue);
+ final List<MetricsTag> tags = Lists.newArrayList(hostname, tag);
+
+ MetricsRecord record = new ExposedMetricsRecordImpl(info, System.currentTimeMillis(), tags,
+ metrics);
+
+ // setup the mocking/validation for the sink
+ Mockito.doAnswer(new Answer<Void>() {
+
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ PhoenixMetricsRecord record = (PhoenixMetricsRecord) invocation.getArguments()[0];
+ //validate that we got the right fields in the record
+ assertEquals("phoenix.987654", record.name());
+ assertEquals("Some generic trace", record.description());
+ int count = 0;
+ for (PhoenixAbstractMetric metric : record.metrics()) {
+ count++;
+ //find the matching metric in the list
+ boolean found = false;
+ for(AbstractMetric expected : metrics){
+ if(expected.name().equals(metric.getName())){
+ found = true;
+ // make sure the rest of the info matches
+ assertEquals("Metric value mismatch", expected.value(), metric.value());
+ }
+ }
+ assertTrue("Didn't find an expected metric to match "+metric, found);
+ }
+ assertEquals("Number of metrics is received is wrong", metrics.size(), count);
+
+ count = 0;
+ for (PhoenixMetricTag tag : record.tags()) {
+ count++;
+ // find the matching metric in the list
+ boolean found = false;
+ for (MetricsTag expected : tags) {
+ if (expected.name().equals(tag.name())) {
+ found = true;
+ // make sure the rest of the info matches
+ assertEquals("Tag value mismatch", expected.value(), tag.value());
+ assertEquals("Tag description mismatch", expected.description(), tag.description());
+ }
+ }
+ assertTrue("Didn't find an expected metric to match " + tag, found);
+ }
+ assertEquals("Number of tags is received is wrong", tags.size(), count);
+ return null;
+ }
+
+ }).when(mockSink).addMetrics(Mockito.any(PhoenixMetricsRecord.class));
+
+ // actually do the update
+ writer.putMetrics(record);
+ writer.flush();
+
+ Mockito.verify(mockSink).addMetrics(Mockito.any(PhoenixMetricsRecord.class));
+ Mockito.verify(mockSink).flush();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/TracingTest.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/TracingTest.java b/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/TracingTest.java
new file mode 100644
index 0000000..ffe6c82
--- /dev/null
+++ b/phoenix-hadoop2-compat/src/test/java/org/apache/phoenix/trace/TracingTest.java
@@ -0,0 +1,34 @@
+/**
+ * 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.phoenix.trace;
+
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.Test;
+
+public class TracingTest {
+
+ /**
+ * Test that we can correctly load a class that will convert the tracing output to metrics
+ * @throws Exception on failure
+ */
+ @Test
+ public void testLoadTracingToMetrics() throws Exception{
+ assertNotNull("Didn't find a trace receiver", TracingCompat.newTraceMetricSource());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index b51d115..180d9da 100644
--- a/pom.xml
+++ b/pom.xml
@@ -559,7 +559,8 @@
<id>hadoop-1</id>
<activation>
<property>
- <name>!hadoop.profile</name>
+ <name>hadoop.profile</name>
+ <value>1</value>
</property>
</activation>
<modules>
@@ -633,6 +634,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase-hadoop1.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-test</artifactId>
@@ -649,8 +655,7 @@
<id>hadoop-2</id>
<activation>
<property>
- <name>hadoop.profile</name>
- <value>2</value>
+ <name>!hadoop.profile</name>
</property>
</activation>
<modules>
@@ -670,7 +675,7 @@
<version>${project.version}</version>
</dependency>
- <!-- Hadoop dependencies -->
+ <!-- HBase dependencies -->
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-testing-util</artifactId>
@@ -689,6 +694,11 @@
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
<artifactId>hbase-protocol</artifactId>
<version>${hbase-hadoop2.version}</version>
</dependency>
@@ -698,24 +708,38 @@
<version>${hbase-hadoop2.version}</version>
</dependency>
<dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Hadoop Dependencies -->
+ <dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
- <version>${hadoop.version}</version>
+ <version>${hadoop-two.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-annotations</artifactId>
- <version>${hadoop.version}</version>
+ <version>${hadoop-two.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
- <version>${hadoop.version}</version>
+ <version>${hadoop-two.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minicluster</artifactId>
- <version>${hadoop.version}</version>
+ <version>${hadoop-two.version}</version>
<optional>true</optional>
<scope>test</scope>
</dependency>
[2/3] PHOENIX-177: Collect usage and performance metrics
Posted by jy...@apache.org.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index e24591d..b3bec6e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -51,6 +51,7 @@ import org.apache.hadoop.hbase.regionserver.ScanType;
import org.apache.hadoop.hbase.regionserver.Store;
import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.hbase.index.builder.IndexBuildManager;
import org.apache.phoenix.hbase.index.builder.IndexBuilder;
@@ -64,6 +65,11 @@ import org.apache.phoenix.hbase.index.write.IndexWriter;
import org.apache.phoenix.hbase.index.write.recovery.PerRegionIndexWriteCache;
import org.apache.phoenix.hbase.index.write.recovery.StoreFailuresInCachePolicy;
import org.apache.phoenix.hbase.index.write.recovery.TrackingParallelWriterIndexCommitter;
+import org.apache.phoenix.trace.TracingCompat;
+import org.apache.phoenix.trace.util.NullSpan;
+import org.apache.phoenix.trace.util.Tracing;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.Trace;
import com.google.common.collect.Multimap;
@@ -134,10 +140,18 @@ public class Indexer extends BaseRegionObserver {
private static final int INDEX_WAL_COMPRESSION_MINIMUM_SUPPORTED_VERSION = VersionUtil
.encodeVersion("0.94.9");
+ /**
+ * Raw configuration, for tracing. Coprocessors generally will get a subset configuration (if
+ * they are on a per-table basis), so we need the raw one from the server, so we can get the
+ * actual configuration keys
+ */
+ private Configuration rawConf;
+
@Override
public void start(CoprocessorEnvironment e) throws IOException {
try {
final RegionCoprocessorEnvironment env = (RegionCoprocessorEnvironment) e;
+ this.rawConf = env.getRegionServerServices().getConfiguration();
String serverName = env.getRegionServerServices().getServerName().getServerName();
if (env.getConfiguration().getBoolean(CHECK_VERSION_CONF_KEY, true)) {
// make sure the right version <-> combinations are allowed.
@@ -312,12 +326,24 @@ public class Indexer extends BaseRegionObserver {
// don't worry which one we get
WALEdit edit = miniBatchOp.getWalEdit(0);
+ // get the current span, or just use a null-span to avoid a bunch of if statements
+ Span current = Trace.startSpan("Starting to build index updates").getSpan();
+ if (current == null) {
+ current = NullSpan.INSTANCE;
+ }
+
// get the index updates for all elements in this batch
Collection<Pair<Mutation, byte[]>> indexUpdates =
this.builder.getIndexUpdate(miniBatchOp, mutations.values());
+ current.addTimelineAnnotation("Built index updates, doing preStep");
+ TracingCompat.addAnnotation(current, "index update count", indexUpdates.size());
+
// write them, either to WAL or the index tables
doPre(indexUpdates, edit, durability);
+
+ // close the span
+ current.stop();
}
private class MultiMutation extends Mutation {
@@ -458,16 +484,24 @@ public class Indexer extends BaseRegionObserver {
return;
}
+ // get the current span, or just use a null-span to avoid a bunch of if statements
+ Span current = Trace.startSpan("Completing index writes").getSpan();
+ if (current == null) {
+ current = NullSpan.INSTANCE;
+ }
+
// there is a little bit of excess here- we iterate all the non-indexed kvs for this check first
// and then do it again later when getting out the index updates. This should be pretty minor
// though, compared to the rest of the runtime
IndexedKeyValue ikv = getFirstIndexedKeyValue(edit);
+
/*
* early exit - we have nothing to write, so we don't need to do anything else. NOTE: we don't
* release the WAL Rolling lock (INDEX_UPDATE_LOCK) since we never take it in doPre if there are
* no index updates.
*/
if (ikv == null) {
+ current.stop();
return;
}
@@ -483,6 +517,7 @@ public class Indexer extends BaseRegionObserver {
// references originally - therefore, we just pass in a null factory here and use the ones
// already specified on each reference
try {
+ current.addTimelineAnnotation("Actually doing index update for first time");
writer.writeAndKillYourselfOnFailure(indexUpdates);
} finally {
// With a custom kill policy, we may throw instead of kill the server.
@@ -492,6 +527,10 @@ public class Indexer extends BaseRegionObserver {
// mark the batch as having been written. In the single-update case, this never gets check
// again, but in the batch case, we will check it again (see above).
ikv.markBatchFinished();
+
+ // finish the span
+
+ current.stop();
}
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
index 92fc54a..2af5896 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
@@ -40,6 +40,7 @@ import org.apache.phoenix.query.*;
import org.apache.phoenix.schema.*;
import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTable.ViewType;
+import org.apache.phoenix.trace.util.Tracing;
import org.apache.phoenix.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -250,7 +251,7 @@ public class ParallelIterators extends ExplainTable implements ResultIterators {
// Delay the swapping of start/stop row until row so we don't muck with the intersect logic
ScanUtil.swapStartStopRowIfReversed(splitScan);
Future<PeekingResultIterator> future =
- executor.submit(new JobCallable<PeekingResultIterator>() {
+ executor.submit(Tracing.wrap(new JobCallable<PeekingResultIterator>() {
@Override
public PeekingResultIterator call() throws Exception {
@@ -274,7 +275,7 @@ public class ParallelIterators extends ExplainTable implements ResultIterators {
public Object getJobId() {
return ParallelIterators.this;
}
- });
+ }, "Parallel scanner for table: " + tableRef.getTable().getName().getString()));
futures.add(new Pair<byte[],Future<PeekingResultIterator>>(split.getLowerRange(),future));
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index 609c2ba..d6dd5f7 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -51,7 +51,10 @@ import java.util.concurrent.Executor;
import javax.annotation.Nullable;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HConstants;
+import org.apache.phoenix.call.CallRunner;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.execute.MutationState;
@@ -72,6 +75,8 @@ import org.apache.phoenix.schema.PMetaData.Pruner;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.trace.TracingCompat;
+import org.apache.phoenix.trace.util.Tracing;
import org.apache.phoenix.util.DateUtil;
import org.apache.phoenix.util.JDBCUtil;
import org.apache.phoenix.util.NumberUtil;
@@ -80,6 +85,8 @@ import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SQLCloseable;
import org.apache.phoenix.util.SQLCloseables;
+import org.cloudera.htrace.Sampler;
+import org.cloudera.htrace.Trace;
import com.google.common.base.Objects;
import com.google.common.base.Strings;
@@ -101,6 +108,8 @@ import com.google.common.collect.Maps;
* @since 0.1
*/
public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jdbc7Shim.Connection, MetaDataMutated {
+ private static final Log LOG = LogFactory.getLog(PhoenixConnection.class);
+
private final String url;
private final ConnectionQueryServices services;
private final Properties info;
@@ -115,7 +124,19 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
private final String datePattern;
private boolean isClosed = false;
+ private Sampler<?> sampler;
+ static {
+ // add the phoenix span receiver so we can log the traces. We have a single trace
+ // source for the whole JVM
+ try {
+ Trace.addReceiver(TracingCompat.newTraceMetricSource());
+ } catch (RuntimeException e) {
+ LOG.error("Tracing will outputs will not be written to any metrics sink! No "
+ + "TraceMetricsSink found on the classpath", e);
+ }
+ }
+
private static Properties newPropsWithSCN(long scn, Properties props) {
props = new Properties(props);
props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(scn));
@@ -125,15 +146,18 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
public PhoenixConnection(PhoenixConnection connection) throws SQLException {
this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.getMetaDataCache());
this.isAutoCommit = connection.isAutoCommit;
+ this.sampler = connection.sampler;
}
public PhoenixConnection(PhoenixConnection connection, long scn) throws SQLException {
this(connection.getQueryServices(), connection, scn);
+ this.sampler = connection.sampler;
}
public PhoenixConnection(ConnectionQueryServices services, PhoenixConnection connection, long scn) throws SQLException {
this(services, connection.getURL(), newPropsWithSCN(scn,connection.getClientInfo()), connection.getMetaDataCache());
this.isAutoCommit = connection.isAutoCommit;
+ this.sampler = connection.sampler;
}
public PhoenixConnection(ConnectionQueryServices services, String url, Properties info, PMetaData metaData) throws SQLException {
@@ -199,6 +223,13 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
});
this.mutationState = new MutationState(maxSize, this);
this.services.addConnection(this);
+
+ // setup tracing, if its enabled
+ this.sampler = Tracing.getConfiguredSampler(this);
+ }
+
+ public Sampler<?> getSampler() {
+ return this.sampler;
}
public int executeStatements(Reader reader, List<Object> binds, PrintStream out) throws IOException, SQLException {
@@ -359,7 +390,12 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
@Override
public void commit() throws SQLException {
- mutationState.commit();
+ CallRunner.run(new CallRunner.CallableThrowable<Void, SQLException>() {
+ public Void call() throws SQLException {
+ mutationState.commit();
+ return null;
+ }
+ }, Tracing.withTracing(this, "committing mutations"));
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index f5b8367..42acc60 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -31,9 +31,9 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.call.CallRunner;
import org.apache.phoenix.compile.ColumnProjector;
import org.apache.phoenix.compile.ColumnResolver;
import org.apache.phoenix.compile.CreateIndexCompiler;
@@ -110,6 +110,7 @@ import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.tuple.SingleKeyValueTuple;
import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.trace.util.Tracing;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.PhoenixContextExecutor;
@@ -186,7 +187,7 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
return new PhoenixResultSet(iterator, projector, this);
}
- protected boolean execute(CompilableStatement stmt) throws SQLException {
+ protected boolean execute(final CompilableStatement stmt) throws SQLException {
if (stmt.getOperation().isMutation()) {
executeMutation(stmt);
return false;
@@ -202,13 +203,16 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
protected PhoenixResultSet executeQuery(final CompilableStatement stmt) throws SQLException {
try {
- return PhoenixContextExecutor.call(new Callable<PhoenixResultSet>() {
+ return CallRunner.run(
+ new CallRunner.CallableThrowable<PhoenixResultSet, SQLException>() {
@Override
- public PhoenixResultSet call() throws Exception {
+ public PhoenixResultSet call() throws SQLException {
try {
QueryPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.RESERVE_SEQUENCE);
plan = connection.getQueryServices().getOptimizer().optimize(
PhoenixStatement.this, plan);
+ // this will create its own trace internally, so we don't wrap this
+ // whole thing in tracing
PhoenixResultSet rs = newResultSet(plan.iterator(), plan.getProjector());
resultSets.add(rs);
setLastQueryPlan(plan);
@@ -225,7 +229,7 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
throw e;
}
}
- });
+ }, PhoenixContextExecutor.inContext());
} catch (Exception e) {
Throwables.propagateIfInstanceOf(e, SQLException.class);
throw Throwables.propagate(e);
@@ -234,11 +238,11 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
protected int executeMutation(final CompilableStatement stmt) throws SQLException {
try {
- return PhoenixContextExecutor.call(
- new Callable<Integer>() {
+ return CallRunner
+ .run(
+ new CallRunner.CallableThrowable<Integer, SQLException>() {
@Override
- public Integer call() throws Exception {
-
+ public Integer call() throws SQLException {
// Note that the upsert select statements will need to commit any open transaction here,
// since they'd update data directly from coprocessors, and should thus operate on
// the latest state
@@ -266,7 +270,8 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
throw e;
}
}
- });
+ }, PhoenixContextExecutor.inContext(),
+ Tracing.withTracing(connection, this.toString()));
} catch (Exception e) {
Throwables.propagateIfInstanceOf(e, SQLException.class);
throw Throwables.propagate(e);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixTableMetricsWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixTableMetricsWriter.java b/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixTableMetricsWriter.java
new file mode 100644
index 0000000..b6d3c67
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixTableMetricsWriter.java
@@ -0,0 +1,255 @@
+/**
+ * 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.phoenix.trace;
+
+import static org.apache.phoenix.metrics.MetricInfo.ANNOTATION;
+import static org.apache.phoenix.metrics.MetricInfo.TAG;
+import static org.apache.phoenix.metrics.MetricInfo.DESCRIPTION;
+import static org.apache.phoenix.metrics.MetricInfo.END;
+import static org.apache.phoenix.metrics.MetricInfo.HOSTNAME;
+import static org.apache.phoenix.metrics.MetricInfo.PARENT;
+import static org.apache.phoenix.metrics.MetricInfo.SPAN;
+import static org.apache.phoenix.metrics.MetricInfo.START;
+import static org.apache.phoenix.metrics.MetricInfo.TRACE;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.phoenix.metrics.MetricInfo;
+import org.apache.phoenix.metrics.MetricsWriter;
+import org.apache.phoenix.metrics.PhoenixAbstractMetric;
+import org.apache.phoenix.metrics.PhoenixMetricTag;
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+import org.apache.phoenix.util.QueryUtil;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Joiner;
+
+/**
+ * Sink that writes phoenix metrics to a phoenix table
+ * <p>
+ * Each metric record should only correspond to a single completed span. Each span is only updated
+ * in the phoenix table <i>once</i>
+ */
+public class PhoenixTableMetricsWriter implements MetricsWriter {
+
+ private static final String VARIABLE_VALUE = "?";
+
+ public static final Log LOG = LogFactory.getLog(PhoenixTableMetricsWriter.class);
+
+ private static final Joiner COLUMN_JOIN = Joiner.on(".");
+ static final String TAG_FAMILY = "tags";
+ /** Count of the number of tags we are storing for this row */
+ static final String TAG_COUNT = COLUMN_JOIN.join(TAG_FAMILY, "count");
+
+ static final String ANNOTATION_FAMILY = "annotations";
+ static final String ANNOTATION_COUNT = COLUMN_JOIN.join(ANNOTATION_FAMILY, "count");
+
+ /** Join strings on a comma */
+ private static final Joiner COMMAS = Joiner.on(',');
+
+ private Connection conn;
+
+ private String table;
+
+ @Override
+ public void initialize() {
+ try {
+ // create the phoenix connection
+ Configuration conf = HBaseConfiguration.create();
+ Connection conn = QueryUtil.getConnection(conf);
+ // enable bulk loading when we have enough data
+ conn.setAutoCommit(true);
+
+ String tableName =
+ conf.get(TracingCompat.TARGET_TABLE_CONF_KEY,
+ TracingCompat.DEFAULT_STATS_TABLE_NAME);
+
+ initializeInternal(conn, tableName);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void initializeInternal(Connection conn, String tableName) throws SQLException {
+ this.conn = conn;
+
+ // ensure that the target table already exists
+ createTable(conn, tableName);
+ }
+
+ /**
+ * Used for <b>TESTING ONLY</b>
+ * <p>
+ * Initialize the connection and setup the table to use the
+ * {@link TracingCompat#DEFAULT_STATS_TABLE_NAME}
+ * @param conn to store for upserts and to create the table (if necessary)
+ * @throws SQLException if any phoenix operation fails
+ */
+ @VisibleForTesting
+ public void initForTesting(Connection conn) throws SQLException {
+ initializeInternal(conn, TracingCompat.DEFAULT_STATS_TABLE_NAME);
+ }
+
+ /**
+ * Create a stats table with the given name. Stores the name for use later when creating upsert
+ * statements
+ * @param conn connection to use when creating the table
+ * @param table name of the table to create
+ * @throws SQLException if any phoenix operations fails
+ */
+ private void createTable(Connection conn, String table) throws SQLException {
+ // only primary-key columns can be marked non-null
+ String ddl =
+ "create table if not exists " + table + "( " +
+ TRACE.columnName + " bigint not null, " +
+ PARENT.columnName + " bigint not null, " +
+ SPAN.columnName + " bigint not null, " +
+ DESCRIPTION.columnName + " varchar, " +
+ START.columnName + " bigint, " +
+ END.columnName + " bigint, " +
+ HOSTNAME.columnName + " varchar, " +
+ TAG_COUNT + " smallint, " +
+ ANNOTATION_COUNT + " smallint" +
+ " CONSTRAINT pk PRIMARY KEY (" + TRACE.columnName + ", "
+ + PARENT.columnName + ", " + SPAN.columnName + "))\n";
+ PreparedStatement stmt = conn.prepareStatement(ddl);
+ stmt.execute();
+ this.table = table;
+ }
+
+ @Override
+ public void flush() {
+ try {
+ this.conn.commit();
+ this.conn.rollback();
+ } catch (SQLException e) {
+ LOG.error("Failed to commit changes to table", e);
+ }
+ }
+
+ /**
+ * Add a new metric record to be written.
+ * @param record
+ */
+ @Override
+ public void addMetrics(PhoenixMetricsRecord record) {
+ // its not a tracing record, we are done. This could also be handled by filters, but safer
+ // to do it here, in case it gets misconfigured
+ if (!record.name().startsWith(TracingCompat.METRIC_SOURCE_KEY)) {
+ return;
+ }
+ String stmt = "UPSERT INTO " + table + " (";
+ // drop it into the queue of things that should be written
+ List<String> keys = new ArrayList<String>();
+ List<Object> values = new ArrayList<Object>();
+ // we need to keep variable values in a separate set since they may have spaces, which
+ // causes the parser to barf. Instead, we need to add them after the statement is prepared
+ List<String> variableValues = new ArrayList<String>(record.tags().size());
+ keys.add(TRACE.columnName);
+ values.add(Long.parseLong(record.name().substring(TracingCompat.METRIC_SOURCE_KEY.length())));
+
+ keys.add(DESCRIPTION.columnName);
+ values.add(VARIABLE_VALUE);
+ variableValues.add(record.description());
+
+ // add each of the metrics
+ for (PhoenixAbstractMetric metric : record.metrics()) {
+ // name of the metric is also the column name to which we write
+ keys.add(MetricInfo.getColumnName(metric.getName()));
+ values.add((Long) metric.value());
+ }
+
+ // get the tags out so we can set them later (otherwise, need to be a single value)
+ int annotationCount = 0;
+ int tagCount = 0;
+ for (PhoenixMetricTag tag : record.tags()) {
+ if (tag.name().equals(ANNOTATION.traceName)) {
+ addDynamicEntry(keys, values, variableValues, ANNOTATION_FAMILY, tag, ANNOTATION,
+ annotationCount);
+ annotationCount++;
+ } else if (tag.name().equals(TAG.traceName)) {
+ addDynamicEntry(keys, values, variableValues, TAG_FAMILY, tag, TAG, tagCount);
+ tagCount++;
+ } else if (tag.name().equals(HOSTNAME.traceName)) {
+ keys.add(HOSTNAME.columnName);
+ values.add(VARIABLE_VALUE);
+ variableValues.add(tag.value());
+ } else if (tag.name().equals("Context")) {
+ // ignored
+ } else {
+ LOG.error("Got an unexpected tag: " + tag);
+ }
+ }
+
+ // add the tag count, now that we know it
+ keys.add(TAG_COUNT);
+ // ignore the hostname in the tags, if we know it
+ values.add(tagCount);
+
+ keys.add(ANNOTATION_COUNT);
+ values.add(annotationCount);
+
+ // compile the statement together
+ stmt += COMMAS.join(keys);
+ stmt += ") VALUES (" + COMMAS.join(values) + ")";
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Logging metrics to phoenix table via: " + stmt);
+ LOG.trace("With tags: " + variableValues);
+ }
+ try {
+ PreparedStatement ps = conn.prepareStatement(stmt);
+ // add everything that wouldn't/may not parse
+ int index = 1;
+ for (String tag : variableValues) {
+ ps.setString(index++, tag);
+ }
+ ps.execute();
+ } catch (SQLException e) {
+ LOG.error("Could not write metric: \n" + record + " to prepared statement:\n" + stmt, e);
+ }
+ }
+
+ public static String getDynamicColumnName(String family, String column, int count) {
+ return COLUMN_JOIN.join(family, column) + count;
+ }
+
+ private void addDynamicEntry(List<String> keys, List<Object> values,
+ List<String> variableValues, String family, PhoenixMetricTag tag,
+ MetricInfo metric, int count) {
+ // <family><.dynColumn><count> <VARCHAR>
+ keys.add(getDynamicColumnName(family, metric.columnName, count) + " VARCHAR");
+
+ // build the annotation value
+ String val = tag.description() + " - " + tag.value();
+ values.add(VARIABLE_VALUE);
+ variableValues.add(val);
+ }
+
+ public void clearForTesting() throws SQLException {
+ this.conn.rollback();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/trace/TraceReader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/trace/TraceReader.java b/phoenix-core/src/main/java/org/apache/phoenix/trace/TraceReader.java
new file mode 100644
index 0000000..4760329
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/trace/TraceReader.java
@@ -0,0 +1,375 @@
+/**
+ * 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.phoenix.trace;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.phoenix.metrics.MetricInfo;
+import org.apache.phoenix.trace.PhoenixTableMetricsWriter;
+import org.cloudera.htrace.Span;
+
+import com.google.common.base.Joiner;
+import com.google.common.primitives.Longs;
+
+/**
+ * Read the traces written to phoenix tables by the {@link PhoenixTableMetricsWriter}.
+ */
+public class TraceReader {
+
+ private static final Log LOG = LogFactory.getLog(TraceReader.class);
+ private static final int DEFAULT_PAGE_SIZE = 100;
+ private static final String PAGE_SIZE_CONF_KEY = "phoenix.trace.read.pagesize";
+ private final Joiner comma = Joiner.on(',');
+ private String knownColumns;
+ {
+ // the order here dictates the order we pull out the values below. For now, just keep them
+ // in sync - so we can be efficient pulling them off the results.
+ knownColumns =
+ comma.join(MetricInfo.TRACE.columnName, MetricInfo.PARENT.columnName,
+ MetricInfo.SPAN.columnName, MetricInfo.DESCRIPTION.columnName,
+ MetricInfo.START.columnName, MetricInfo.END.columnName,
+ MetricInfo.HOSTNAME.columnName, PhoenixTableMetricsWriter.TAG_COUNT,
+ PhoenixTableMetricsWriter.ANNOTATION_COUNT);
+ }
+
+ private Connection conn;
+ private String table;
+ private int pageSize;
+
+ public TraceReader(Connection conn, String statsTableName) throws SQLException {
+ this.conn = conn;
+ this.table = statsTableName;
+ String ps = conn.getClientInfo(PAGE_SIZE_CONF_KEY);
+ this.pageSize = ps == null ? DEFAULT_PAGE_SIZE : Integer.parseInt(ps);
+ }
+
+ public TraceReader(Connection conn) throws SQLException {
+ this(conn, TracingCompat.DEFAULT_STATS_TABLE_NAME);
+ }
+
+ /**
+ * Read all the currently stored traces.
+ * <p>
+ * <b>Be Careful!</b> This could cause an OOME if there are a lot of traces.
+ * @param limit max number of traces to return. If -1, returns all known traces.
+ * @return the found traces
+ * @throws SQLException
+ */
+ public Collection<TraceHolder> readAll(int limit) throws SQLException {
+ Set<TraceHolder> traces = new HashSet<TraceHolder>();
+ // read all the known columns from the table, sorting first by trace column (so the same
+ // trace
+ // goes together), and then by start time (so parent spans always appear before child spans)
+ String query =
+ "SELECT " + knownColumns + " FROM " + TracingCompat.DEFAULT_STATS_TABLE_NAME
+ + " ORDER BY " + MetricInfo.TRACE.columnName + " DESC, "
+ + MetricInfo.START.columnName + " ASC" + " LIMIT " + pageSize;
+ int resultCount = 0;
+ ResultSet results = conn.prepareStatement(query).executeQuery();
+ TraceHolder trace = null;
+ // the spans that are not the root span, but haven't seen their parent yet
+ List<SpanInfo> orphans = null;
+ while (results.next()) {
+ int index = 1;
+ long traceid = results.getLong(index++);
+ long parent = results.getLong(index++);
+ long span = results.getLong(index++);
+ String desc = results.getString(index++);
+ long start = results.getLong(index++);
+ long end = results.getLong(index++);
+ String host = results.getString(index++);
+ int tagCount = results.getInt(index++);
+ int annotationCount = results.getInt(index++);
+ // we have a new trace
+ if (trace == null || traceid != trace.traceid) {
+ // only increment if we are on a new trace, to ensure we get at least one
+ if (trace != null) {
+ resultCount++;
+ }
+ // we beyond the limit, so we stop
+ if (resultCount >= limit) {
+ break;
+ }
+ trace = new TraceHolder();
+ // add the orphans, so we can track them later
+ orphans = new ArrayList<SpanInfo>();
+ trace.orphans = orphans;
+ trace.traceid = traceid;
+ traces.add(trace);
+ }
+
+ // search the spans to determine the if we have a known parent
+ SpanInfo parentSpan = null;
+ if (parent != Span.ROOT_SPAN_ID) {
+ // find the parent
+ for (SpanInfo p : trace.spans) {
+ if (p.id == parent) {
+ parentSpan = p;
+ break;
+ }
+ }
+ }
+ SpanInfo spanInfo =
+ new SpanInfo(parentSpan, parent, span, desc, start, end, host, tagCount,
+ annotationCount);
+ // search the orphans to see if this is the parent id
+
+ for (int i = 0; i < orphans.size(); i++) {
+ SpanInfo orphan = orphans.get(i);
+ // we found the parent for the orphan
+ if (orphan.parentId == span) {
+ // update the bi-directional relationship
+ orphan.parent = spanInfo;
+ spanInfo.children.add(orphan);
+ // / its no longer an orphan
+ LOG.trace("Found parent for span: " + span);
+ orphans.remove(i--);
+ }
+ }
+
+ if (parentSpan != null) {
+ // add this as a child to the parent span
+ parentSpan.children.add(spanInfo);
+ } else if (parent != Span.ROOT_SPAN_ID) {
+ // add the span to the orphan pile to check for the remaining spans we see
+ LOG.info("No parent span found for span: " + span + " (root span id: "
+ + Span.ROOT_SPAN_ID + ")");
+ orphans.add(spanInfo);
+ }
+
+ // add the span to the full known list
+ trace.spans.add(spanInfo);
+
+ // go back and find the tags for the row
+ spanInfo.tags.addAll(getTags(traceid, parent, span, tagCount));
+
+ spanInfo.annotations.addAll(getAnnotations(traceid, parent, span, annotationCount));
+ }
+
+ // make sure we clean up after ourselves
+ results.close();
+
+ return traces;
+ }
+
+ private Collection<? extends String> getTags(long traceid, long parent, long span, int count)
+ throws SQLException {
+ return getDynamicCountColumns(traceid, parent, span, count,
+ PhoenixTableMetricsWriter.TAG_FAMILY, MetricInfo.TAG.columnName);
+ }
+
+ private Collection<? extends String> getAnnotations(long traceid, long parent, long span,
+ int count) throws SQLException {
+ return getDynamicCountColumns(traceid, parent, span, count,
+ PhoenixTableMetricsWriter.ANNOTATION_FAMILY, MetricInfo.ANNOTATION.columnName);
+ }
+
+ private Collection<? extends String> getDynamicCountColumns(long traceid, long parent,
+ long span, int count, String family, String columnName) throws SQLException {
+ if (count == 0) {
+ return Collections.emptyList();
+ }
+
+ // build the column strings, family.column<index>
+ String[] parts = new String[count];
+ for (int i = 0; i < count; i++) {
+ parts[i] = PhoenixTableMetricsWriter.getDynamicColumnName(family, columnName, i);
+ }
+ // join the columns together
+ String columns = comma.join(parts);
+
+ // redo them and add "VARCHAR to the end, so we can specify the columns
+ for (int i = 0; i < count; i++) {
+ parts[i] = parts[i] + " VARCHAR";
+ }
+
+ String dynamicColumns = comma.join(parts);
+ String request =
+ "SELECT " + columns + " from " + table + "(" + dynamicColumns + ") WHERE "
+ + MetricInfo.TRACE.columnName + "=" + traceid + " AND "
+ + MetricInfo.PARENT.columnName + "=" + parent + " AND "
+ + MetricInfo.SPAN.columnName + "=" + span;
+ LOG.trace("Requesting columns with: " + request);
+ ResultSet results = conn.createStatement().executeQuery(request);
+ List<String> cols = new ArrayList<String>();
+ while (results.next()) {
+ for (int index = 1; index <= count; index++) {
+ cols.add(results.getString(index));
+ }
+ }
+ if (cols.size() < count) {
+ LOG.error("Missing tags! Expected " + count + ", but only got " + cols.size()
+ + " tags from rquest " + request);
+ }
+ return cols;
+ }
+
+ /**
+ * Holds information about a trace
+ */
+ public static class TraceHolder {
+ public List<SpanInfo> orphans;
+ public long traceid;
+ public TreeSet<SpanInfo> spans = new TreeSet<SpanInfo>();
+
+ @Override
+ public int hashCode() {
+ return new Long(traceid).hashCode();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof TraceHolder) {
+ return traceid == ((TraceHolder) o).traceid;
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("Trace: " + traceid + "\n");
+ // get the first span, which is always going to be the root span
+ SpanInfo root = spans.iterator().next();
+ if (root.parent != null) {
+ sb.append("Root span not present! Just printing found spans\n");
+ for (SpanInfo span : spans) {
+ sb.append(span.toString() + "\n");
+ }
+ } else {
+ // print the tree of spans
+ List<SpanInfo> toPrint = new ArrayList<SpanInfo>();
+ toPrint.add(root);
+ while (!toPrint.isEmpty()) {
+ SpanInfo span = toPrint.remove(0);
+ sb.append(span.toString() + "\n");
+ toPrint.addAll(span.children);
+ }
+ }
+ if (orphans.size() > 0) {
+ sb.append("Found orphan spans:\n" + orphans);
+ }
+ return sb.toString();
+ }
+ }
+
+ public static class SpanInfo implements Comparable<SpanInfo> {
+ public SpanInfo parent;
+ public List<SpanInfo> children = new ArrayList<SpanInfo>();
+ public String description;
+ public long id;
+ public long start;
+ public long end;
+ public String hostname;
+ public int tagCount;
+ public List<String> tags = new ArrayList<String>();
+ public int annotationCount;
+ public List<String> annotations = new ArrayList<String>();
+ private long parentId;
+
+ public SpanInfo(SpanInfo parent, long parentid, long span, String desc, long start,
+ long end, String host, int tagCount, int annotationCount) {
+ this.parent = parent;
+ this.parentId = parentid;
+ this.id = span;
+ this.description = desc;
+ this.start = start;
+ this.end = end;
+ this.hostname = host;
+ this.tagCount = tagCount;
+ this.annotationCount = annotationCount;
+ }
+
+ @Override
+ public int hashCode() {
+ return new Long(id).hashCode();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof SpanInfo) {
+ return id == ((SpanInfo) o).id;
+ }
+ return false;
+ }
+
+ /**
+ * Do the same sorting that we would get from reading the table with a {@link TraceReader},
+ * specifically, by trace and then by start/end. However, these are only every stored in a
+ * single trace, so we can just sort on start/end times.
+ */
+ @Override
+ public int compareTo(SpanInfo o) {
+ // root span always comes first
+ if (this.parentId == Span.ROOT_SPAN_ID) {
+ return -1;
+ } else if (o.parentId == Span.ROOT_SPAN_ID) {
+ return 1;
+ }
+
+ int compare = Longs.compare(start, o.start);
+ if (compare == 0) {
+ compare = Longs.compare(end, o.end);
+ if (compare == 0) {
+ return Longs.compare(id, o.id);
+ }
+ }
+ return compare;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("Span: " + id + "\n");
+ sb.append("\tdescription=" + description);
+ sb.append("\n");
+ sb.append("\tparent="
+ + (parent == null ? (parentId == Span.ROOT_SPAN_ID ? "ROOT" : "[orphan - id: "
+ + parentId + "]") : parent.id));
+ sb.append("\n");
+ sb.append("\tstart,end=" + start + "," + end);
+ sb.append("\n");
+ sb.append("\telapsed=" + (end - start));
+ sb.append("\n");
+ sb.append("\thostname=" + hostname);
+ sb.append("\n");
+ sb.append("\ttags=(" + tagCount + ") " + tags);
+ sb.append("\n");
+ sb.append("\tannotations=(" + annotationCount + ") " + annotations);
+ sb.append("\n");
+ sb.append("\tchildren=");
+ for (SpanInfo child : children) {
+ sb.append(child.id + ", ");
+ }
+ sb.append("\n");
+ return sb.toString();
+ }
+
+ public long getParentIdForTesting() {
+ return parentId;
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/trace/TracingIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/trace/TracingIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/trace/TracingIterator.java
new file mode 100644
index 0000000..0b12c43
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/trace/TracingIterator.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.phoenix.trace;
+
+import java.sql.SQLException;
+
+import org.apache.phoenix.iterate.DelegateResultIterator;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.schema.tuple.Tuple;
+import org.cloudera.htrace.TraceScope;
+
+/**
+ * A simple iterator that closes the trace scope when the iterator is closed.
+ */
+public class TracingIterator extends DelegateResultIterator {
+
+ private TraceScope scope;
+ private boolean started;
+
+ /**
+ * @param scope a scope with a non-null span
+ * @param iterator delegate
+ */
+ public TracingIterator(TraceScope scope, ResultIterator iterator) {
+ super(iterator);
+ this.scope = scope;
+ }
+
+ @Override
+ public void close() throws SQLException {
+ scope.close();
+ super.close();
+ }
+
+ @Override
+ public Tuple next() throws SQLException {
+ if (!started) {
+ scope.getSpan().addTimelineAnnotation("First request completed");
+ started = true;
+ }
+ return super.next();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/trace/util/ConfigurationAdapter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/trace/util/ConfigurationAdapter.java b/phoenix-core/src/main/java/org/apache/phoenix/trace/util/ConfigurationAdapter.java
new file mode 100644
index 0000000..a96be7d
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/trace/util/ConfigurationAdapter.java
@@ -0,0 +1,56 @@
+/**
+ * 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.phoenix.trace.util;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.phoenix.jdbc.PhoenixConnection;
+
+/**
+ * Helper class to wrap access to configured properties.
+ */
+abstract class ConfigurationAdapter {
+
+ public abstract String get(String key);
+
+ public static class ConnectionConfigurationAdapter extends ConfigurationAdapter {
+ private PhoenixConnection conn;
+
+ public ConnectionConfigurationAdapter(PhoenixConnection connection) {
+ this.conn = connection;
+ }
+
+ @Override
+ public String get(String key) {
+ return conn.getClientInfo(key);
+ }
+ }
+
+ public static class HadoopConfigConfigurationAdapter extends ConfigurationAdapter {
+ private Configuration conf;
+
+ public HadoopConfigConfigurationAdapter(Configuration conf) {
+ this.conf = conf;
+ }
+
+ @Override
+ public String get(String key) {
+ return conf.get(key);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/trace/util/NullSpan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/trace/util/NullSpan.java b/phoenix-core/src/main/java/org/apache/phoenix/trace/util/NullSpan.java
new file mode 100644
index 0000000..3799fdb
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/trace/util/NullSpan.java
@@ -0,0 +1,112 @@
+/**
+ * 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.phoenix.trace.util;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.TimelineAnnotation;
+
+/**
+ * Fake {@link Span} that doesn't save any state, in place of <tt>null</tt> return values, to avoid
+ * <tt>null</tt> check.
+ */
+public class NullSpan implements Span {
+
+ public static Span INSTANCE = new NullSpan();
+
+ /**
+ * Private constructor to limit garbage
+ */
+ private NullSpan() {
+ }
+
+ @Override
+ public void stop() {
+ }
+
+ @Override
+ public long getStartTimeMillis() {
+ return 0;
+ }
+
+ @Override
+ public long getStopTimeMillis() {
+ return 0;
+ }
+
+ @Override
+ public long getAccumulatedMillis() {
+ return 0;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return false;
+ }
+
+ @Override
+ public String getDescription() {
+ return "NullSpan";
+ }
+
+ @Override
+ public long getSpanId() {
+ return 0;
+ }
+
+ @Override
+ public long getTraceId() {
+ return 0;
+ }
+
+ @Override
+ public Span child(String description) {
+ return INSTANCE;
+ }
+
+ @Override
+ public long getParentId() {
+ return 0;
+ }
+
+ @Override
+ public void addKVAnnotation(byte[] key, byte[] value) {
+ }
+
+ @Override
+ public void addTimelineAnnotation(String msg) {
+ }
+
+ @Override
+ public Map<byte[], byte[]> getKVAnnotations() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public List<TimelineAnnotation> getTimelineAnnotations() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public String getProcessId() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/trace/util/Tracing.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/trace/util/Tracing.java b/phoenix-core/src/main/java/org/apache/phoenix/trace/util/Tracing.java
new file mode 100644
index 0000000..5913cfb
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/trace/util/Tracing.java
@@ -0,0 +1,282 @@
+/**
+ * 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.phoenix.trace.util;
+
+import java.util.Properties;
+import java.util.concurrent.Callable;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.OperationWithAttributes;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.call.CallRunner;
+import org.apache.phoenix.call.CallWrapper;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.cloudera.htrace.Sampler;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.Trace;
+import org.cloudera.htrace.TraceInfo;
+import org.cloudera.htrace.TraceScope;
+import org.cloudera.htrace.Tracer;
+import org.cloudera.htrace.impl.ProbabilitySampler;
+import org.cloudera.htrace.wrappers.TraceCallable;
+import org.cloudera.htrace.wrappers.TraceRunnable;
+
+import com.google.common.base.Function;
+
+/**
+ * Helper class to manage using the {@link Tracer} within Phoenix
+ */
+public class Tracing {
+
+ private static final Log LOG = LogFactory.getLog(Tracing.class);
+
+ private static final String SEPARATOR = ".";
+ // Constants for tracing across the wire
+ public static final String TRACE_ID_ATTRIBUTE_KEY = "phoenix.trace.traceid";
+ public static final String SPAN_ID_ATTRIBUTE_KEY = "phoenix.trace.spanid";
+ private static final String START_SPAN_MESSAGE = "Span received on server. Starting child";
+
+ // Constants for passing into the metrics system
+ public static final String TRACE_METRIC_PREFIX = "phoenix.trace.instance";
+ // Constants for for configuring tracing
+ public static final String TRACING_LEVEL_KEY = "org.apache.phoenix.trace.frequency";
+ protected static final String PROBABILITY_THRESHOLD_KEY =
+ "com.salesforce.phoenix.trace.probability.threshold";
+
+ /**
+ * We always trace on the server, assuming the client has requested tracing on the request
+ */
+ public static Sampler<?> SERVER_TRACE_LEVEL = Sampler.ALWAYS;
+
+ /**
+ * Manage the types of frequencies that we support. By default, we never turn on tracing.
+ */
+ public static enum Frequency {
+ NEVER("never", CREATE_NEVER), // default
+ ALWAYS("always", CREATE_ALWAYS), PROBABILITY("probability", CREATE_PROBABILITY);
+
+ String key;
+ Function<ConfigurationAdapter, Sampler<?>> builder;
+
+ private Frequency(String key, Function<ConfigurationAdapter, Sampler<?>> builder) {
+ this.key = key;
+ this.builder = builder;
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ static Frequency getSampler(String key) {
+ for (Frequency type : Frequency.values()) {
+ if (type.key.equals(key)) {
+ return type;
+ }
+ }
+ return NEVER;
+ }
+ }
+
+ private static Function<ConfigurationAdapter, Sampler<?>> CREATE_ALWAYS =
+ new Function<ConfigurationAdapter, Sampler<?>>() {
+ @Override
+ public Sampler<?> apply(ConfigurationAdapter arg0) {
+ return Sampler.ALWAYS;
+ }
+ };
+
+ private static Function<ConfigurationAdapter, Sampler<?>> CREATE_NEVER =
+ new Function<ConfigurationAdapter, Sampler<?>>() {
+ @Override
+ public Sampler<?> apply(ConfigurationAdapter arg0) {
+ return Sampler.NEVER;
+ }
+ };
+
+ private static Function<ConfigurationAdapter, Sampler<?>> CREATE_PROBABILITY =
+ new Function<ConfigurationAdapter, Sampler<?>>() {
+ @Override
+ public Sampler<?> apply(ConfigurationAdapter conn) {
+ // get the connection properties for the probability information
+ double threshold = Double.parseDouble(conn.get(PROBABILITY_THRESHOLD_KEY));
+ return new ProbabilitySampler(threshold);
+ }
+ };
+
+ public static Sampler<?> getConfiguredSampler(PhoenixConnection connection) {
+ String tracelevel = connection.getClientInfo(TRACING_LEVEL_KEY);
+ return getSampler(tracelevel, new ConfigurationAdapter.ConnectionConfigurationAdapter(
+ connection));
+ }
+
+ public static Sampler<?> getConfiguredSampler(Configuration conf) {
+ String tracelevel = conf.get(TRACING_LEVEL_KEY);
+ return getSampler(tracelevel, new ConfigurationAdapter.HadoopConfigConfigurationAdapter(
+ conf));
+ }
+
+ private static Sampler<?> getSampler(String traceLevel, ConfigurationAdapter conf) {
+ return Frequency.getSampler(traceLevel).builder.apply(conf);
+ }
+
+ public static void setSampling(Properties props, Frequency freq) {
+ props.setProperty(TRACING_LEVEL_KEY, freq.key);
+ }
+
+ /**
+ * Start a span with the currently configured sampling frequency. Creates a new 'current' span
+ * on this thread - the previous 'current' span will be replaced with this newly created span.
+ * <p>
+ * Hands back the direct span as you shouldn't be detaching the span - use {@link TraceRunnable}
+ * instead to detach a span from this operation.
+ * @param connection from which to read parameters
+ * @param string description of the span to start
+ * @return the underlying span.
+ */
+ public static TraceScope startNewSpan(PhoenixConnection connection, String string) {
+ Sampler<?> sampler = connection.getSampler();
+ TraceScope scope = Trace.startSpan(string, sampler);
+ return scope;
+ }
+
+ public static String getSpanName(Span span) {
+ return Tracing.TRACE_METRIC_PREFIX + span.getTraceId() + SEPARATOR + span.getParentId()
+ + SEPARATOR + span.getSpanId();
+ }
+
+ /**
+ * Check to see if tracing is current enabled. The trace for this thread is returned, if we are
+ * already tracing. Otherwise, checks to see if mutation has tracing enabled, and if so, starts
+ * a new span with the {@link Mutation}'s specified span as its parent.
+ * <p>
+ * This should only be run on the server-side as we base tracing on if we are currently tracing
+ * (started higher in the call-stack) or if the {@link Mutation} has the tracing attributes
+ * defined. As such, we would expect to continue the trace on the server-side based on the
+ * original sampling parameters.
+ * @param scan {@link Mutation} to check
+ * @param conf {@link Configuration} to read for the current sampler
+ * @param description description of the child span to start
+ * @return <tt>null</tt> if tracing is not enabled, or the parent {@link Span}
+ */
+ public static Span childOnServer(OperationWithAttributes scan, Configuration conf,
+ String description) {
+ // check to see if we are currently tracing. Generally, this will only work when we go to
+ // 0.96. CPs should always be setting up and tearing down their own tracing
+ Span current = Trace.currentSpan();
+ if (current == null) {
+ // its not tracing yet, but maybe it should be.
+ current = enable(scan, conf, description);
+ } else {
+ current = Trace.startSpan(description, current).getSpan();
+ }
+ return current;
+ }
+
+ /**
+ * Check to see if this mutation has tracing enabled, and if so, get a new span with the
+ * {@link Mutation}'s specified span as its parent.
+ * @param map mutation to check
+ * @param conf {@link Configuration} to check for the {@link Sampler} configuration, if we are
+ * tracing
+ * @param description on the child to start
+ * @return a child span of the mutation, or <tt>null</tt> if tracing is not enabled.
+ */
+ @SuppressWarnings("unchecked")
+ private static Span enable(OperationWithAttributes map, Configuration conf, String description) {
+ byte[] traceid = map.getAttribute(TRACE_ID_ATTRIBUTE_KEY);
+ if (traceid == null) {
+ return NullSpan.INSTANCE;
+ }
+ byte[] spanid = map.getAttribute(SPAN_ID_ATTRIBUTE_KEY);
+ if (spanid == null) {
+ LOG.error("TraceID set to " + Bytes.toLong(traceid) + ", but span id was not set!");
+ return NullSpan.INSTANCE;
+ }
+ Sampler<?> sampler = SERVER_TRACE_LEVEL;
+ TraceInfo parent = new TraceInfo(Bytes.toLong(traceid), Bytes.toLong(spanid));
+ return Trace.startSpan(START_SPAN_MESSAGE + ": " + description,
+ (Sampler<TraceInfo>) sampler, parent).getSpan();
+ }
+
+ public static Span child(Span s, String d) {
+ if (s == null) {
+ return NullSpan.INSTANCE;
+ }
+ return s.child(d);
+ }
+
+ /**
+ * Wrap the callable in a TraceCallable, if tracing.
+ * @param callable to call
+ * @param description description of the operation being run. If <tt>null</tt> uses the current
+ * thread name
+ * @return The callable provided, wrapped if tracing, 'callable' if not.
+ */
+ public static <V> Callable<V> wrap(Callable<V> callable, String description) {
+ if (Trace.isTracing()) {
+ return new TraceCallable<V>(Trace.currentSpan(), callable, description);
+ }
+ return callable;
+ }
+
+ /**
+ * Helper to automatically start and complete tracing on the given method, used in conjuction
+ * with {@link CallRunner#run}.
+ * <p>
+ * This will always attempt start a new span (which will always start, unless the
+ * {@link Sampler} says it shouldn't be traced). If you are just looking for flexible tracing
+ * that only turns on if the current thread/query is already tracing, use
+ * {@link #wrap(Callable, String)} or {@link Trace#wrap(Callable)}.
+ * <p>
+ * Ensures that the trace is closed, even if there is an exception from the
+ * {@link org.apache.phoenix.call.CallRunner.CallableThrowable}.
+ * <p>
+ * Generally, this should wrap a long-running operation.
+ * @param conn connection from which to determine if we are tracing, ala
+ * {@link #startNewSpan(PhoenixConnection, String)}
+ * @param desc description of the operation being run
+ * @return the value returned from the call
+ */
+ public static CallWrapper withTracing(PhoenixConnection conn, String desc) {
+ return new TracingWrapper(conn, desc);
+ }
+
+ private static class TracingWrapper implements CallWrapper {
+ private TraceScope scope;
+ private final PhoenixConnection conn;
+ private final String desc;
+
+ public TracingWrapper(PhoenixConnection conn, String desc){
+ this.conn = conn;
+ this.desc = desc;
+ }
+
+ @Override
+ public void before() {
+ scope = Tracing.startNewSpan(conn, "Executing " + desc);
+ }
+
+ @Override
+ public void after() {
+ scope.close();
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixContextExecutor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixContextExecutor.java b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixContextExecutor.java
index a725c2c..9106a5d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixContextExecutor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixContextExecutor.java
@@ -19,6 +19,8 @@ package org.apache.phoenix.util;
import java.util.concurrent.Callable;
+import org.apache.phoenix.call.CallWrapper;
+
import com.google.common.base.Throwables;
/**
@@ -34,6 +36,27 @@ import com.google.common.base.Throwables;
*/
public class PhoenixContextExecutor {
+ private static class CurrentContextWrapper implements CallWrapper {
+ private ClassLoader saveCcl;
+
+ @Override
+ public void before() {
+ saveCcl = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(
+ PhoenixContextExecutor.class.getClassLoader());
+ }
+
+ @Override
+ public void after() {
+ Thread.currentThread().setContextClassLoader(saveCcl);
+
+ };
+ }
+
+ public static CallWrapper inContext() {
+ return new CurrentContextWrapper();
+ }
+
/**
* Execute an operation (synchronously) using the context classloader used to load this class,
* instead of the currently-set context classloader of the current thread. This allows loading
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
index ebd8465..055c2d8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
@@ -20,13 +20,22 @@ package org.apache.phoenix.util;
import static org.apache.phoenix.util.SchemaUtil.getEscapedFullColumnName;
+import java.sql.Connection;
import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import javax.annotation.Nullable;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.zookeeper.ZKConfig;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
@@ -35,6 +44,8 @@ import com.google.common.collect.Lists;
public final class QueryUtil {
+ private static final Log LOG = LogFactory.getLog(QueryUtil.class);
+
/**
* Column family name index within ResultSet resulting from {@link DatabaseMetaData#getColumns(String, String, String, String)}
*/
@@ -152,6 +163,12 @@ public final class QueryUtil {
return PhoenixRuntime.JDBC_PROTOCOL + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + server;
}
+ public static String getUrl(String server, long port) {
+ String serverUrl = getUrl(server);
+ return serverUrl + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + port
+ + PhoenixRuntime.JDBC_PROTOCOL_TERMINATOR;
+ }
+
public static String getExplainPlan(ResultSet rs) throws SQLException {
StringBuilder buf = new StringBuilder();
while (rs.next()) {
@@ -163,4 +180,16 @@ public final class QueryUtil {
}
return buf.toString();
}
+ public static Connection getConnection(Configuration conf) throws ClassNotFoundException,
+ SQLException {
+ // read the hbase properties from the configuration
+ String server = ZKConfig.getZKQuorumServersString(conf);
+ // make sure we load the phoenix driver
+ Class.forName(PhoenixDriver.class.getName());
+ int port =
+ conf.getInt(HConstants.ZOOKEEPER_CLIENT_PORT, HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT);
+ String jdbcUrl = getUrl(server, port);
+ LOG.info("Creating connection with the jdbc url:" + jdbcUrl);
+ return DriverManager.getConnection(jdbcUrl);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/test/resources/hadoop-metrics2.properties
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/resources/hadoop-metrics2.properties b/phoenix-core/src/test/resources/hadoop-metrics2.properties
new file mode 100644
index 0000000..ebbbe84
--- /dev/null
+++ b/phoenix-core/src/test/resources/hadoop-metrics2.properties
@@ -0,0 +1,25 @@
+#There are two options with file names:
+# 1. hadoop-metrics2-[prefix].properties
+# 2. hadoop-metrics2.properties
+# Either will be loaded by the metrics system (but not both).
+#
+# NOTE: The metrics system is only initialized once per JVM (but does ref-counting, so we can't
+#shutdown and restart), so we only load the first prefix that we find. Generally, this will be
+#phoenix (unless someone else registers first, but for many clients, there should only be one).
+#
+# Usually, you would use hadoop-metrics2-phoenix.properties, but we use the generic
+# hadoop-metrics2.properties to ensure it these are loaded regardless of where we are running,
+# assuming there isn't another config on the classpath.
+
+#
+# When specifying sinks, the syntax to use is:
+# [prefix].[source|sink].[instance].[options]
+# See javadoc of package-info.java for org.apache.hadoop.metrics2 for detail
+
+
+# Don't attempt to start jmx mbeans for all sources.
+# For right now, all metrics are exported to a phoenix table
+*.source.start_mbeans=false
+
+# Frequency, in seconds, of sampling from the sources
+*.period=10
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/resources/log4j.properties b/phoenix-core/src/test/resources/log4j.properties
index 6b1ce50..8e54793 100644
--- a/phoenix-core/src/test/resources/log4j.properties
+++ b/phoenix-core/src/test/resources/log4j.properties
@@ -52,12 +52,12 @@ log4j.appender.DRFA.layout.ConversionPattern=%d %-5p [%t] %C{2}(%L): %m%n
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d %-5p [%t] %C{2}(%L): %m%n
+log4j.appender.console.layout.ConversionPattern=%d %-5p [%t] %C(%L): %m%n
# Custom Logging levels
#log4j.logger.org.apache.hadoop.fs.FSNamesystem=DEBUG
-
+log4j.logger.org.mortbay.log=WARN
log4j.logger.org.apache.hadoop=WARN
log4j.logger.org.apache.zookeeper=ERROR
log4j.logger.org.apache.hadoop.hbase=DEBUG
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/pom.xml b/phoenix-hadoop-compat/pom.xml
index a51f571..c2f810a 100644
--- a/phoenix-hadoop-compat/pom.xml
+++ b/phoenix-hadoop-compat/pom.xml
@@ -39,7 +39,7 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
- <version>2.4</version>
+ <version>2.4</version><!--$NO-MVN-MAN-VER$-->
<executions>
<execution>
<phase>prepare-package
@@ -59,4 +59,31 @@
</plugin>
</plugins>
</build>
-</project>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.cloudera.htrace</groupId>
+ <artifactId>htrace-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ </dependency>
+ </dependencies>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricInfo.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricInfo.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricInfo.java
new file mode 100644
index 0000000..e6ad976
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricInfo.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.phoenix.metrics;
+
+/**
+ * Metrics and their conversion from the trace name to the name we store in the stats table
+ */
+public enum MetricInfo {
+
+ TRACE("", "trace_id"),
+ SPAN("span_id", "span_id"),
+ PARENT("parent_id", "parent_id"),
+ START("start_time", "start_time"),
+ END("end_time", "end_time"),
+ TAG("phoenix.tag", "t"),
+ ANNOTATION("phoenix.annotation", "a"),
+ HOSTNAME("Hostname", "hostname"),
+ DESCRIPTION("", "description");
+
+ public final String traceName;
+ public final String columnName;
+
+ private MetricInfo(String traceName, String columnName) {
+ this.traceName = traceName;
+ this.columnName = columnName;
+ }
+
+ public static String getColumnName(String traceName) {
+ for (MetricInfo info : MetricInfo.values()) {
+ if (info.traceName.equals(traceName)) {
+ return info.columnName;
+ }
+ }
+ throw new IllegalArgumentException("Unknown tracename: " + traceName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/Metrics.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/Metrics.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/Metrics.java
new file mode 100644
index 0000000..e0667ab
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/Metrics.java
@@ -0,0 +1,39 @@
+/**
+ * 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.phoenix.metrics;
+
+import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
+
+public class Metrics {
+
+ private static volatile MetricsManager manager;
+
+ /**
+ * @return get the first {@link MetricsManager} on the classpath. Always returns the same object
+ */
+ public static MetricsManager getManager(){
+ if(manager == null){
+ synchronized(Metrics.class){
+ if(manager == null){
+ manager = CompatibilitySingletonFactory.getInstance(MetricsManager.class);
+ }
+ }
+ }
+ return manager;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsManager.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsManager.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsManager.java
new file mode 100644
index 0000000..13c9435
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsManager.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.phoenix.metrics;
+
+/**
+ * Metrics management system. Backed by the underlying hadoop metrics, depending on the project on
+ * the classpath.
+ * <p>
+ * The underlying types passed to method must match the expected metrics type - this will vary for
+ * the underlying metrics systems (hadoop1 vs hadoop2), but can't be specified at this layer because
+ * we must be compatible with both systems.
+ */
+public interface MetricsManager {
+
+ /**
+ * @param metricsSystemName the metrics prefix to initialize, if it hasn't already been
+ * initialized. Not assumed to be thread-safe, unless otherwise noted in the
+ * implementation.
+ */
+ public abstract void initialize(String metricsSystemName);
+
+ /**
+ * Register a metrics sink
+ * @param <T> the type of the sink
+ * @param sink to register
+ * @param name of the sink. Must be unique.
+ * @param desc the description of the sink
+ * @return the sink
+ */
+ public abstract <T> T register(String name, String desc, T sink);
+
+ /**
+ * Register a metrics source.
+ * @param name name of the source - must be unique
+ * @param description description of the source
+ * @param source to register.
+ * @param <T> the type of the source
+ * @return the source
+ */
+ public abstract <T> T registerSource(String name, String description, T source);
+
+ public void shutdown();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsWriter.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsWriter.java
new file mode 100644
index 0000000..0e8b9fe
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/MetricsWriter.java
@@ -0,0 +1,31 @@
+/**
+ * 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.phoenix.metrics;
+
+
+/**
+ * Generic writer for a phoenix metric
+ */
+public interface MetricsWriter {
+
+ public void initialize();
+
+ public void flush();
+
+ public void addMetrics(PhoenixMetricsRecord record);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixAbstractMetric.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixAbstractMetric.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixAbstractMetric.java
new file mode 100644
index 0000000..27ae6b8
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixAbstractMetric.java
@@ -0,0 +1,30 @@
+/**
+ * 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.phoenix.metrics;
+
+
+public interface PhoenixAbstractMetric {
+
+ public String getName();
+
+ /**
+ * Get the value of the metric
+ * @return the value of the metric
+ */
+ public Number value();
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricTag.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricTag.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricTag.java
new file mode 100644
index 0000000..123cc1c
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricTag.java
@@ -0,0 +1,27 @@
+/**
+ * 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.phoenix.metrics;
+
+public interface PhoenixMetricTag {
+
+ public String name();
+
+ public String description();
+
+ public String value();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricsRecord.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricsRecord.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricsRecord.java
new file mode 100644
index 0000000..68f7c46
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/metrics/PhoenixMetricsRecord.java
@@ -0,0 +1,35 @@
+/**
+ * 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.phoenix.metrics;
+
+import java.util.Collection;
+
+
+/**
+ *
+ */
+public interface PhoenixMetricsRecord {
+
+ public String name();
+
+ public String description();
+
+ public Iterable<PhoenixAbstractMetric> metrics();
+
+ public Collection<PhoenixMetricTag> tags();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/PhoenixSpanReceiver.java
----------------------------------------------------------------------
diff --git a/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/PhoenixSpanReceiver.java b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/PhoenixSpanReceiver.java
new file mode 100644
index 0000000..7e4e09c
--- /dev/null
+++ b/phoenix-hadoop-compat/src/main/java/org/apache/phoenix/trace/PhoenixSpanReceiver.java
@@ -0,0 +1,26 @@
+/**
+ * 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.phoenix.trace;
+
+import org.cloudera.htrace.SpanReceiver;
+
+/**
+ * Marker interface for phoenix specific receivers.
+ */
+public interface PhoenixSpanReceiver extends SpanReceiver {
+}
\ No newline at end of file
[3/3] git commit: PHOENIX-177: Collect usage and performance metrics
Posted by jy...@apache.org.
PHOENIX-177: Collect usage and performance metrics
Add basic Dapper-like tracing (using Cloudera's HTrace library)
to phoenix requests. This is the basic infrastructure to
support more holistic, non-profiler based analysis.
This patch includes, among other things, the
infrastructure to use HTrace, async-tracing handling
via the Hadoop metrics2 framework, and trace read/write
to a phoenix table.
Currently, do NOT support Hadoop1 (though does work
against Hadoop1).
Default builds to hadoop2, rather than hadoop1 (particularly as hadoop1 is
now a second-class citizen).
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/b7f46c10
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/b7f46c10
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/b7f46c10
Branch: refs/heads/master
Commit: b7f46c1051de3e23630dccb82677a0a16985f27c
Parents: 9185f76
Author: Jesse Yates <jy...@apache.org>
Authored: Fri Jun 6 16:11:32 2014 -0700
Committer: Jesse Yates <jy...@apache.org>
Committed: Mon Jul 28 06:37:49 2014 -0700
----------------------------------------------------------------------
phoenix-core/pom.xml | 47 +--
.../apache/phoenix/trace/BaseTracingTestIT.java | 117 ++++++
.../phoenix/trace/DelegatingConnection.java | 328 +++++++++++++++
.../phoenix/trace/DisableableMetricsWriter.java | 83 ++++
.../trace/Hadoop1TracingTestEnabler.java | 86 ++++
.../apache/phoenix/trace/PhoenixMetricImpl.java | 44 ++
.../phoenix/trace/PhoenixMetricRecordImpl.java | 71 ++++
.../trace/PhoenixTableMetricsWriterIT.java | 119 ++++++
.../apache/phoenix/trace/PhoenixTagImpl.java | 52 +++
.../phoenix/trace/PhoenixTracingEndToEndIT.java | 401 +++++++++++++++++++
.../apache/phoenix/trace/TraceReaderTest.java | 181 +++++++++
.../org/apache/phoenix/call/CallRunner.java | 66 +++
.../org/apache/phoenix/call/CallWrapper.java | 29 ++
.../coprocessor/BaseScannerRegionObserver.java | 36 +-
.../coprocessor/DelegateRegionScanner.java | 78 ++++
.../apache/phoenix/execute/BasicQueryPlan.java | 17 +-
.../apache/phoenix/execute/MutationState.java | 24 +-
.../org/apache/phoenix/hbase/index/Indexer.java | 39 ++
.../phoenix/iterate/ParallelIterators.java | 5 +-
.../apache/phoenix/jdbc/PhoenixConnection.java | 38 +-
.../apache/phoenix/jdbc/PhoenixStatement.java | 25 +-
.../trace/PhoenixTableMetricsWriter.java | 255 ++++++++++++
.../org/apache/phoenix/trace/TraceReader.java | 375 +++++++++++++++++
.../apache/phoenix/trace/TracingIterator.java | 58 +++
.../trace/util/ConfigurationAdapter.java | 56 +++
.../org/apache/phoenix/trace/util/NullSpan.java | 112 ++++++
.../org/apache/phoenix/trace/util/Tracing.java | 282 +++++++++++++
.../phoenix/util/PhoenixContextExecutor.java | 23 ++
.../java/org/apache/phoenix/util/QueryUtil.java | 29 ++
.../test/resources/hadoop-metrics2.properties | 25 ++
.../src/test/resources/log4j.properties | 4 +-
phoenix-hadoop-compat/pom.xml | 31 +-
.../org/apache/phoenix/metrics/MetricInfo.java | 51 +++
.../org/apache/phoenix/metrics/Metrics.java | 39 ++
.../apache/phoenix/metrics/MetricsManager.java | 58 +++
.../apache/phoenix/metrics/MetricsWriter.java | 31 ++
.../phoenix/metrics/PhoenixAbstractMetric.java | 30 ++
.../phoenix/metrics/PhoenixMetricTag.java | 27 ++
.../phoenix/metrics/PhoenixMetricsRecord.java | 35 ++
.../phoenix/trace/PhoenixSpanReceiver.java | 26 ++
.../phoenix/trace/TestableMetricsWriter.java | 30 ++
.../org/apache/phoenix/trace/TracingCompat.java | 94 +++++
.../org/apache/phoenix/metrics/LoggingSink.java | 56 +++
.../phoenix/metrics/TracingTestCompat.java | 45 +++
phoenix-hadoop2-compat/pom.xml | 47 ++-
.../phoenix/metrics/MetricsManagerImpl.java | 71 ++++
.../apache/phoenix/trace/MetricsInfoImpl.java | 63 +++
.../phoenix/trace/PhoenixMetricsWriter.java | 176 ++++++++
.../apache/phoenix/trace/TraceMetricSource.java | 192 +++++++++
.../org.apache.phoenix.metrics.MetricsManager | 1 +
...org.apache.phoenix.trace.PhoenixSpanReceiver | 1 +
...g.apache.phoenix.trace.TestableMetricsWriter | 1 +
.../metrics2/impl/ExposedMetricCounterLong.java | 35 ++
.../metrics2/impl/ExposedMetricsRecordImpl.java | 43 ++
.../metrics2/lib/ExposedMetricsInfoImpl.java | 32 ++
.../phoenix/trace/PhoenixMetricsWriterTest.java | 142 +++++++
.../org/apache/phoenix/trace/TracingTest.java | 34 ++
pom.xml | 40 +-
58 files changed, 4479 insertions(+), 57 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index f5c3ace..cfdee95 100644
--- a/phoenix-core/pom.xml
+++ b/phoenix-core/pom.xml
@@ -299,25 +299,23 @@
<dependency>
<groupId>commons-collections</groupId>
<artifactId>commons-collections</artifactId>
- <version>${collections.version}</version>
- </dependency>
+ </dependency>
</dependencies>
<profiles>
- <!-- Profile for building against Hadoop 1. Active by default. Not used if another
- Hadoop profile is specified with mvn -Dhadoop.profile=foo -->
+ <!-- Profile for building against Hadoop 1. Activate using: mvn -Dhadoop.profile=1-->
<profile>
<id>hadoop-1</id>
<activation>
<property>
- <name>!hadoop.profile</name>
+ <name>hadoop.profile</name>
+ <value>1</value>
</property>
</activation>
<dependencies>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-testing-util</artifactId>
- <version>${hbase-hadoop1.version}</version>
<exclusions>
<exclusion>
<groupId>org.jruby</groupId>
@@ -341,22 +339,18 @@
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
- <version>${hbase-hadoop1.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-protocol</artifactId>
- <version>${hbase-hadoop1.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
- <version>${hbase-hadoop1.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-core</artifactId>
- <version>${hadoop-one.version}</version>
<exclusions>
<exclusion>
<groupId>hsqldb</groupId>
@@ -383,27 +377,29 @@
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-test</artifactId>
- <version>${hadoop-one.version}</version>
<optional>true</optional>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.phoenix</groupId>
+ <artifactId>phoenix-hadoop1-compat</artifactId>
+ </dependency>
</dependencies>
</profile>
- <!-- Profile for building against Hadoop 2. Activate using: mvn -Dhadoop.profile=2 -->
+ <!-- Profile for building against Hadoop 2. Active by default. Not used if another
+ Hadoop profile is specified with mvn -Dhadoop.profile=foo -->
<profile>
<id>hadoop-2</id>
<activation>
<property>
- <name>hadoop.profile</name>
- <value>2</value>
+ <name>!hadoop.profile</name>
</property>
</activation>
<dependencies>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-testing-util</artifactId>
- <version>${hbase-hadoop2.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
@@ -415,23 +411,14 @@
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
- <version>${hbase-hadoop2.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase-common</artifactId>
- <version>${hbase-hadoop2.version}</version>
- <type>test-jar</type>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-protocol</artifactId>
- <version>${hbase-hadoop2.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
- <version>${hbase-hadoop2.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
@@ -447,13 +434,11 @@
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-hadoop-compat</artifactId>
- <version>${hbase-hadoop2.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-hadoop-compat</artifactId>
- <version>${hbase-hadoop2.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
@@ -486,6 +471,16 @@
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minicluster</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.phoenix</groupId>
+ <artifactId>phoenix-hadoop2-compat</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.phoenix</groupId>
+ <artifactId>phoenix-hadoop2-compat</artifactId>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/BaseTracingTestIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/BaseTracingTestIT.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/BaseTracingTestIT.java
new file mode 100644
index 0000000..59cd871
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/BaseTracingTestIT.java
@@ -0,0 +1,117 @@
+/**
+ * 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.phoenix.trace;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.phoenix.end2end.BaseHBaseManagedTimeIT;
+import org.apache.phoenix.metrics.MetricInfo;
+import org.apache.phoenix.metrics.Metrics;
+import org.apache.phoenix.metrics.PhoenixAbstractMetric;
+import org.apache.phoenix.metrics.PhoenixMetricTag;
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+import org.apache.phoenix.trace.util.Tracing;
+import org.apache.phoenix.trace.util.Tracing.Frequency;
+
+/**
+ * Base test for tracing tests - helps manage getting tracing/non-tracing
+ * connections, as well as any supporting utils.
+ */
+public class BaseTracingTestIT extends BaseHBaseManagedTimeIT {
+ private static final Log LOG = LogFactory.getLog(BaseTracingTestIT.class);
+
+ /**
+ * Hadoop1 doesn't yet support tracing (need metrics library support) so we just skip those
+ * tests for the moment
+ * @return <tt>true</tt> if the test should exit because some necessary classes are missing, or
+ * <tt>false</tt> if the tests can continue normally
+ */
+ static boolean shouldEarlyExitForHadoop1Test() {
+ try {
+ // get a receiver for the spans
+ TracingCompat.newTraceMetricSource();
+ // which also needs to a source for the metrics system
+ Metrics.getManager();
+ return false;
+ } catch (RuntimeException e) {
+ LOG.error("Shouldn't run test because can't instantiate necessary metrics/tracing classes!");
+ }
+
+ return true;
+ }
+
+ public static Connection getConnectionWithoutTracing() throws SQLException {
+ Properties props = new Properties(TEST_PROPERTIES);
+ return getConnectionWithoutTracing(props);
+ }
+
+ public static Connection getConnectionWithoutTracing(Properties props) throws SQLException {
+ Connection conn = getConnectionWithTracingFrequency(props, Frequency.NEVER);
+ conn.setAutoCommit(false);
+ return conn;
+ }
+
+ public static Connection getTracingConnection() throws Exception {
+ Properties props = new Properties(TEST_PROPERTIES);
+ return getConnectionWithTracingFrequency(props, Tracing.Frequency.ALWAYS);
+ }
+
+ public static Connection getConnectionWithTracingFrequency(Properties props,
+ Tracing.Frequency frequency) throws SQLException {
+ Tracing.setSampling(props, frequency);
+ return DriverManager.getConnection(getUrl(), props);
+ }
+
+ public static PhoenixMetricsRecord createRecord(long traceid, long parentid, long spanid,
+ String desc, long startTime, long endTime, String hostname, String... tags) {
+ PhoenixMetricRecordImpl record =
+ new PhoenixMetricRecordImpl(TracingCompat.getTraceMetricName(traceid), desc);
+ PhoenixAbstractMetric span = new PhoenixMetricImpl(MetricInfo.SPAN.traceName, spanid);
+ record.addMetric(span);
+
+ PhoenixAbstractMetric parent = new PhoenixMetricImpl(MetricInfo.PARENT.traceName, parentid);
+ record.addMetric(parent);
+
+ PhoenixAbstractMetric start = new PhoenixMetricImpl(MetricInfo.START.traceName, startTime);
+ record.addMetric(start);
+
+ PhoenixAbstractMetric end = new PhoenixMetricImpl(MetricInfo.END.traceName, endTime);
+ record.addMetric(end);
+
+ int tagCount = 0;
+ for (String annotation : tags) {
+ PhoenixMetricTag tag =
+ new PhoenixTagImpl(MetricInfo.ANNOTATION.traceName,
+ Integer.toString(tagCount++), annotation);
+ record.addTag(tag);
+ }
+ String hostnameValue = "host-name.value";
+ PhoenixMetricTag hostnameTag =
+ new PhoenixTagImpl(MetricInfo.HOSTNAME.traceName, "", hostnameValue);
+ record.addTag(hostnameTag);
+
+ return record;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/DelegatingConnection.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/DelegatingConnection.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/DelegatingConnection.java
new file mode 100644
index 0000000..261522d
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/DelegatingConnection.java
@@ -0,0 +1,328 @@
+/*
+ * 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.phoenix.trace;
+
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.CallableStatement;
+import java.sql.Clob;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.NClob;
+import java.sql.PreparedStatement;
+import java.sql.SQLClientInfoException;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.SQLXML;
+import java.sql.Savepoint;
+import java.sql.Statement;
+import java.sql.Struct;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+import org.apache.phoenix.jdbc.Jdbc7Shim;
+
+/**
+ * Simple {@link Connection} that just delegates to an underlying {@link Connection}.
+ * @param <D> delegate type that is both a {@link Connection} and a {@link Jdbc7Shim#Connection}
+ */
+public class DelegatingConnection<D extends Connection & Jdbc7Shim.Connection> implements
+ Connection, Jdbc7Shim.Connection {
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ return conn.unwrap(iface);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ return conn.isWrapperFor(iface);
+ }
+
+ @Override
+ public Statement createStatement() throws SQLException {
+ return conn.createStatement();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql) throws SQLException {
+ return conn.prepareStatement(sql);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql) throws SQLException {
+ return conn.prepareCall(sql);
+ }
+
+ @Override
+ public String nativeSQL(String sql) throws SQLException {
+ return conn.nativeSQL(sql);
+ }
+
+ @Override
+ public void setAutoCommit(boolean autoCommit) throws SQLException {
+ conn.setAutoCommit(autoCommit);
+ }
+
+ @Override
+ public boolean getAutoCommit() throws SQLException {
+ return conn.getAutoCommit();
+ }
+
+ @Override
+ public void commit() throws SQLException {
+ conn.commit();
+ }
+
+ @Override
+ public void rollback() throws SQLException {
+ conn.rollback();
+ }
+
+ @Override
+ public void close() throws SQLException {
+ conn.close();
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return conn.isClosed();
+ }
+
+ @Override
+ public DatabaseMetaData getMetaData() throws SQLException {
+ return conn.getMetaData();
+ }
+
+ @Override
+ public void setReadOnly(boolean readOnly) throws SQLException {
+ conn.setReadOnly(readOnly);
+ }
+
+ @Override
+ public boolean isReadOnly() throws SQLException {
+ return conn.isReadOnly();
+ }
+
+ @Override
+ public void setCatalog(String catalog) throws SQLException {
+ conn.setCatalog(catalog);
+ }
+
+ @Override
+ public String getCatalog() throws SQLException {
+ return conn.getCatalog();
+ }
+
+ @Override
+ public void setTransactionIsolation(int level) throws SQLException {
+ conn.setTransactionIsolation(level);
+ }
+
+ @Override
+ public int getTransactionIsolation() throws SQLException {
+ return conn.getTransactionIsolation();
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ return conn.getWarnings();
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ conn.clearWarnings();
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
+ return conn.createStatement(resultSetType, resultSetConcurrency);
+ }
+
+ @Override
+ public PreparedStatement
+ prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
+ return conn.prepareStatement(sql, resultSetType, resultSetConcurrency);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency)
+ throws SQLException {
+ return conn.prepareCall(sql, resultSetType, resultSetConcurrency);
+ }
+
+ @Override
+ public Map<String, Class<?>> getTypeMap() throws SQLException {
+ return conn.getTypeMap();
+ }
+
+ @Override
+ public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
+ conn.setTypeMap(map);
+ }
+
+ @Override
+ public void setHoldability(int holdability) throws SQLException {
+ conn.setHoldability(holdability);
+ }
+
+ @Override
+ public int getHoldability() throws SQLException {
+ return conn.getHoldability();
+ }
+
+ @Override
+ public Savepoint setSavepoint() throws SQLException {
+ return conn.setSavepoint();
+ }
+
+ @Override
+ public Savepoint setSavepoint(String name) throws SQLException {
+ return conn.setSavepoint(name);
+ }
+
+ @Override
+ public void rollback(Savepoint savepoint) throws SQLException {
+ conn.rollback(savepoint);
+ }
+
+ @Override
+ public void releaseSavepoint(Savepoint savepoint) throws SQLException {
+ conn.releaseSavepoint(savepoint);
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
+ return conn.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType,
+ int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ return conn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
+ return conn.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
+ return conn.prepareStatement(sql, autoGeneratedKeys);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
+ return conn.prepareStatement(sql, columnIndexes);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
+ return conn.prepareStatement(sql, columnNames);
+ }
+
+ @Override
+ public Clob createClob() throws SQLException {
+ return conn.createClob();
+ }
+
+ @Override
+ public Blob createBlob() throws SQLException {
+ return conn.createBlob();
+ }
+
+ @Override
+ public NClob createNClob() throws SQLException {
+ return conn.createNClob();
+ }
+
+ @Override
+ public SQLXML createSQLXML() throws SQLException {
+ return conn.createSQLXML();
+ }
+
+ @Override
+ public boolean isValid(int timeout) throws SQLException {
+ return conn.isValid(timeout);
+ }
+
+ @Override
+ public void setClientInfo(String name, String value) throws SQLClientInfoException {
+ conn.setClientInfo(name, value);
+ }
+
+ @Override
+ public void setClientInfo(Properties properties) throws SQLClientInfoException {
+ conn.setClientInfo(properties);
+ }
+
+ @Override
+ public String getClientInfo(String name) throws SQLException {
+ return conn.getClientInfo(name);
+ }
+
+ @Override
+ public Properties getClientInfo() throws SQLException {
+ return conn.getClientInfo();
+ }
+
+ @Override
+ public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
+ return conn.createArrayOf(typeName, elements);
+ }
+
+ @Override
+ public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
+ return conn.createStruct(typeName, attributes);
+ }
+
+ private D conn;
+
+ public DelegatingConnection(D conn) {
+ this.conn = conn;
+ }
+
+ @Override
+ public void setSchema(String schema) throws SQLException {
+ conn.setSchema(schema);
+ }
+
+ @Override
+ public String getSchema() throws SQLException {
+ return conn.getSchema();
+ }
+
+ @Override
+ public void abort(Executor executor) throws SQLException {
+ conn.abort(executor);
+ }
+
+ @Override
+ public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
+ conn.setNetworkTimeout(executor, milliseconds);
+ }
+
+ @Override
+ public int getNetworkTimeout() throws SQLException {
+ return conn.getNetworkTimeout();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/DisableableMetricsWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/DisableableMetricsWriter.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/DisableableMetricsWriter.java
new file mode 100644
index 0000000..a054bf2
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/DisableableMetricsWriter.java
@@ -0,0 +1,83 @@
+/**
+ * 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.phoenix.trace;
+
+import java.sql.SQLException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.phoenix.metrics.MetricsWriter;
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+
+/**
+ *
+ */
+public class DisableableMetricsWriter implements MetricsWriter {
+
+ private static final Log LOG = LogFactory.getLog(DisableableMetricsWriter.class);
+ private PhoenixTableMetricsWriter writer;
+ private AtomicBoolean disabled = new AtomicBoolean(false);
+
+ public DisableableMetricsWriter(PhoenixTableMetricsWriter writer) {
+ this.writer = writer;
+ }
+
+ @Override
+ public void initialize() {
+ if (this.disabled.get()) return;
+ writer.initialize();
+ }
+
+ @Override
+ public void flush() {
+ if (this.disabled.get()) {
+ clear();
+ return;
+ }
+ writer.flush();
+
+ }
+
+ @Override
+ public void addMetrics(PhoenixMetricsRecord record) {
+ if (this.disabled.get()) return;
+ writer.addMetrics(record);
+ }
+
+ public void disable() {
+ this.disabled.set(true);
+ }
+
+ public void enable() {
+ this.disabled.set(false);
+ }
+
+ public void clear() {
+ // clear any pending writes
+ try {
+ writer.clearForTesting();
+ } catch (SQLException e) {
+ LOG.error("Couldn't clear the delgate writer when flush called and disabled", e);
+ }
+ }
+
+ public PhoenixTableMetricsWriter getDelegate() {
+ return this.writer;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/Hadoop1TracingTestEnabler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/Hadoop1TracingTestEnabler.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/Hadoop1TracingTestEnabler.java
new file mode 100644
index 0000000..07e2305
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/Hadoop1TracingTestEnabler.java
@@ -0,0 +1,86 @@
+/**
+ * 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.phoenix.trace;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.runner.notification.RunNotifier;
+import org.junit.runners.BlockJUnit4ClassRunner;
+import org.junit.runners.model.FrameworkMethod;
+import org.junit.runners.model.InitializationError;
+
+/**
+ * Test runner to run classes that depend on Hadoop1 compatibility that may not be present for the
+ * feature
+ */
+public class Hadoop1TracingTestEnabler extends BlockJUnit4ClassRunner {
+
+ public Hadoop1TracingTestEnabler(Class<?> klass) throws InitializationError {
+ super(klass);
+ }
+
+ @Override
+ public void runChild(FrameworkMethod method, RunNotifier notifier) {
+ // if the class is already disabled, then we can disable on the class level, otherwise we
+ // just check the per-method
+ Hadoop1Disabled condition =
+ getTestClass().getJavaClass().getAnnotation(Hadoop1Disabled.class);
+ if (condition == null) {
+ condition = method
+ .getAnnotation(Hadoop1Disabled.class);
+ }
+
+ // if this has the flag, then we want to disable it if hadoop1 is not enabled for that
+ // feature
+ if (condition != null && getEnabled(condition.value())) {
+ super.runChild(method, notifier);
+ } else {
+ notifier.fireTestIgnored(describeChild(method));
+ }
+ }
+
+ /**
+ * Simple check that just uses if-else logic. We can move to something more complex, policy
+ * based later when this gets more complex.
+ * @param feature name of the feature to check
+ * @return <tt>true</tt> if the test method is enabled for the given feature, <tt>false</tt>
+ * otherwise
+ */
+ private boolean getEnabled(String feature) {
+ if (feature.equals("tracing")) {
+ return !BaseTracingTestIT.shouldEarlyExitForHadoop1Test();
+ }
+ return true;
+ }
+
+ /**
+ * Marker that a class/method should be disabled if hadoop1 features are not enabled. It takes a
+ * value for the Hadoop1 feature on which this class/method depends, for instance "tracing" is
+ * not supported in Hadoop1 (yet).
+ */
+ @Target({ ElementType.TYPE, ElementType.METHOD })
+ @Retention(RetentionPolicy.RUNTIME)
+ public static @interface Hadoop1Disabled {
+ String value();
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricImpl.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricImpl.java
new file mode 100644
index 0000000..985504f
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricImpl.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.phoenix.trace;
+
+import org.apache.phoenix.metrics.PhoenixAbstractMetric;
+
+/**
+ * Simple metric implementation for testing
+ */
+public class PhoenixMetricImpl implements PhoenixAbstractMetric {
+
+ private String name;
+ private Number value;
+
+ public PhoenixMetricImpl(String name, Number value) {
+ this.name = name;
+ this.value = value;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public Number value() {
+ return value;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricRecordImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricRecordImpl.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricRecordImpl.java
new file mode 100644
index 0000000..45cabf0
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixMetricRecordImpl.java
@@ -0,0 +1,71 @@
+/**
+ * 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.phoenix.trace;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.phoenix.metrics.PhoenixAbstractMetric;
+import org.apache.phoenix.metrics.PhoenixMetricTag;
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+
+import com.google.common.collect.Lists;
+
+/**
+ *
+ */
+public class PhoenixMetricRecordImpl implements PhoenixMetricsRecord {
+
+ private String name;
+ private String description;
+ private final List<PhoenixAbstractMetric> metrics = Lists.newArrayList();
+ private final List<PhoenixMetricTag> tags = Lists.newArrayList();
+
+ public PhoenixMetricRecordImpl(String name, String description) {
+ this.name = name;
+ this.description = description;
+ }
+
+ public void addMetric(PhoenixAbstractMetric metric) {
+ this.metrics.add(metric);
+ }
+
+ public void addTag(PhoenixMetricTag tag) {
+ this.tags.add(tag);
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ @Override
+ public String description() {
+ return this.description;
+ }
+
+ @Override
+ public Iterable<PhoenixAbstractMetric> metrics() {
+ return metrics;
+ }
+
+ @Override
+ public Collection<PhoenixMetricTag> tags() {
+ return tags;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTableMetricsWriterIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTableMetricsWriterIT.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTableMetricsWriterIT.java
new file mode 100644
index 0000000..b1544e8
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTableMetricsWriterIT.java
@@ -0,0 +1,119 @@
+/**
+ * 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.phoenix.trace;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.sql.Connection;
+import java.util.Collection;
+
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+import org.apache.phoenix.trace.Hadoop1TracingTestEnabler.Hadoop1Disabled;
+import org.apache.phoenix.trace.TraceReader.SpanInfo;
+import org.apache.phoenix.trace.TraceReader.TraceHolder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+/**
+ * Test that the logging sink stores the expected metrics/stats
+ */
+@RunWith(Hadoop1TracingTestEnabler.class)
+@Hadoop1Disabled("tracing")
+public class PhoenixTableMetricsWriterIT extends BaseTracingTestIT {
+
+ /**
+ * IT should create the target table if it hasn't been created yet, but not fail if the table
+ * has already been created
+ * @throws Exception on failure
+ */
+ @Test
+ public void testCreatesTable() throws Exception {
+ PhoenixTableMetricsWriter sink = new PhoenixTableMetricsWriter();
+ Connection conn = getConnectionWithoutTracing();
+ sink.initForTesting(conn);
+
+ // check for existence of the tracing table
+ try {
+ String ddl = "CREATE TABLE " + TracingCompat.DEFAULT_STATS_TABLE_NAME;
+ conn.createStatement().execute(ddl);
+ fail("Table " + TracingCompat.DEFAULT_STATS_TABLE_NAME
+ + " was not created by the metrics sink");
+ } catch (Exception e) {
+ // expected
+ }
+
+ // initialize sink again, which should attempt to create the table, but not fail
+ try {
+ sink.initForTesting(conn);
+ } catch (Exception e) {
+ fail("Initialization shouldn't fail if table already exists!");
+ }
+ }
+
+ /**
+ * Simple metrics writing and reading check, that uses the standard wrapping in the
+ * {@link PhoenixMetricsWriter}
+ * @throws Exception on failure
+ */
+ @Test
+ public void writeMetrics() throws Exception {
+ // hook up a phoenix sink
+ PhoenixTableMetricsWriter sink = new PhoenixTableMetricsWriter();
+ Connection conn = getConnectionWithoutTracing();
+ sink.initForTesting(conn);
+
+ // create a simple metrics record
+ long traceid = 987654;
+ String description = "Some generic trace";
+ long spanid = 10;
+ long parentid = 11;
+ long startTime = 12;
+ long endTime = 13;
+ String annotation = "test annotation for a span";
+ String hostnameValue = "host-name.value";
+ PhoenixMetricsRecord record =
+ createRecord(traceid, parentid, spanid, description, startTime, endTime,
+ hostnameValue, annotation);
+
+ // actually write the record to the table
+ sink.addMetrics(record);
+ sink.flush();
+
+ // make sure we only get expected stat entry (matcing the trace id), otherwise we could the
+ // stats for the update as well
+ TraceReader reader = new TraceReader(conn);
+ Collection<TraceHolder> traces = reader.readAll(10);
+ assertEquals("Wrong number of traces in the tracing table", 1, traces.size());
+
+ // validate trace
+ TraceHolder trace = traces.iterator().next();
+ // we are just going to get an orphan span b/c we don't send in a parent
+ assertEquals("Didn't get expected orphaned spans!" + trace.orphans, 1, trace.orphans.size());
+
+ assertEquals(traceid, trace.traceid);
+ SpanInfo spanInfo = trace.orphans.get(0);
+ assertEquals(description, spanInfo.description);
+ assertEquals(parentid, spanInfo.getParentIdForTesting());
+ assertEquals(startTime, spanInfo.start);
+ assertEquals(endTime, spanInfo.end);
+ assertEquals(hostnameValue, spanInfo.hostname);
+ assertEquals("Wrong number of tags", 0, spanInfo.tagCount);
+ assertEquals("Wrong number of annotations", 1, spanInfo.annotationCount);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTagImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTagImpl.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTagImpl.java
new file mode 100644
index 0000000..c8e2219
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTagImpl.java
@@ -0,0 +1,52 @@
+/**
+ * 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.phoenix.trace;
+
+import org.apache.phoenix.metrics.PhoenixMetricTag;
+
+/**
+ * Simple Tag implementation for testing
+ */
+public class PhoenixTagImpl implements PhoenixMetricTag {
+
+ private final String name;
+ private final String description;
+ private final String value;
+
+ public PhoenixTagImpl(String name, String description, String value) {
+ super();
+ this.name = name;
+ this.description = description;
+ this.value = value;
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public String description() {
+ return description;
+ }
+
+ @Override
+ public String value() {
+ return value;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTracingEndToEndIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTracingEndToEndIT.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTracingEndToEndIT.java
new file mode 100644
index 0000000..955d640
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/PhoenixTracingEndToEndIT.java
@@ -0,0 +1,401 @@
+/*
+ * 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.phoenix.trace;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.metrics.Metrics;
+import org.apache.phoenix.metrics.TracingTestCompat;
+import org.apache.phoenix.trace.Hadoop1TracingTestEnabler.Hadoop1Disabled;
+import org.apache.phoenix.trace.TraceReader.SpanInfo;
+import org.apache.phoenix.trace.TraceReader.TraceHolder;
+import org.cloudera.htrace.Sampler;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.SpanReceiver;
+import org.cloudera.htrace.Trace;
+import org.cloudera.htrace.TraceScope;
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+/**
+ * Test that the logging sink stores the expected metrics/stats
+ */
+@RunWith(Hadoop1TracingTestEnabler.class)
+@Hadoop1Disabled("tracing")
+public class PhoenixTracingEndToEndIT extends BaseTracingTestIT {
+
+ private static final Log LOG = LogFactory.getLog(PhoenixTracingEndToEndIT.class);
+ private static final int MAX_RETRIES = 10;
+ private final String table = "ENABLED_FOR_LOGGING";
+ private final String index = "ENABALED_FOR_LOGGING_INDEX";
+
+ private static DisableableMetricsWriter sink;
+
+ @BeforeClass
+ public static void setupMetrics() throws Exception {
+ if (shouldEarlyExitForHadoop1Test()) {
+ return;
+ }
+ PhoenixTableMetricsWriter pWriter = new PhoenixTableMetricsWriter();
+ Connection conn = getConnectionWithoutTracing();
+ pWriter.initForTesting(conn);
+ sink = new DisableableMetricsWriter(pWriter);
+
+ TracingTestCompat.registerSink(sink);
+ }
+
+ @After
+ public void cleanup() {
+ sink.disable();
+ sink.clear();
+ sink.enable();
+
+ // LISTENABLE.clearListeners();
+ }
+
+ private static void waitForCommit(CountDownLatch latch) throws SQLException {
+ Connection conn = new CountDownConnection(getConnectionWithoutTracing(), latch);
+ replaceWriterConnection(conn);
+ }
+
+ private static void replaceWriterConnection(Connection conn) throws SQLException {
+ // disable the writer
+ sink.disable();
+
+ // swap the connection for one that listens
+ sink.getDelegate().initForTesting(conn);
+
+ // enable the writer
+ sink.enable();
+ }
+
+ /**
+ * Simple test that we can correctly write spans to the phoenix table
+ * @throws Exception on failure
+ */
+ @Test
+ public void testWriteSpans() throws Exception {
+ // get a receiver for the spans
+ SpanReceiver receiver = TracingCompat.newTraceMetricSource();
+ // which also needs to a source for the metrics system
+ Metrics.getManager().registerSource("testWriteSpans-source", "source for testWriteSpans",
+ receiver);
+
+ // watch our sink so we know when commits happen
+ CountDownLatch latch = new CountDownLatch(1);
+ waitForCommit(latch);
+
+ // write some spans
+ TraceScope trace = Trace.startSpan("Start write test", Sampler.ALWAYS);
+ Span span = trace.getSpan();
+
+ // add a child with some annotations
+ Span child = span.child("child 1");
+ child.addTimelineAnnotation("timeline annotation");
+ TracingCompat.addAnnotation(child, "test annotation", 10);
+ child.stop();
+
+ // sleep a little bit to get some time difference
+ Thread.sleep(100);
+
+ trace.close();
+
+ // pass the trace on
+ receiver.receiveSpan(span);
+
+ // wait for the tracer to actually do the write
+ latch.await();
+
+ // look for the writes to make sure they were made
+ Connection conn = getConnectionWithoutTracing();
+ checkStoredTraces(conn, new TraceChecker() {
+ public boolean foundTrace(TraceHolder trace, SpanInfo info) {
+ if (info.description.equals("child 1")) {
+ assertEquals("Not all annotations present", 1, info.annotationCount);
+ assertEquals("Not all tags present", 1, info.tagCount);
+ boolean found = false;
+ for (String annotation : info.annotations) {
+ if (annotation.startsWith("test annotation")) {
+ found = true;
+ }
+ }
+ assertTrue("Missing the annotations in span: " + info, found);
+ found = false;
+ for (String tag : info.tags) {
+ if (tag.endsWith("timeline annotation")) {
+ found = true;
+ }
+ }
+ assertTrue("Missing the tags in span: " + info, found);
+ return true;
+ }
+ return false;
+ }
+ });
+ }
+
+ /**
+ * Test that span will actually go into the this sink and be written on both side of the wire,
+ * through the indexing code.
+ * @throws Exception
+ */
+ @Test
+ public void testClientServerIndexingTracing() throws Exception {
+ // one call for client side, one call for server side
+ final CountDownLatch updated = new CountDownLatch(2);
+ waitForCommit(updated);
+
+ // separate connection so we don't create extra traces
+ Connection conn = getConnectionWithoutTracing();
+ createTestTable(conn, true);
+
+ // trace the requests we send
+ Connection traceable = getTracingConnection();
+ LOG.debug("Doing dummy the writes to the tracked table");
+ String insert = "UPSERT INTO " + table + " VALUES (?, ?)";
+ PreparedStatement stmt = traceable.prepareStatement(insert);
+ stmt.setString(1, "key1");
+ stmt.setLong(2, 1);
+ // this first trace just does a simple open/close of the span. Its not doing anything
+ // terribly interesting because we aren't auto-committing on the connection, so it just
+ // updates the mutation state and returns.
+ stmt.execute();
+ stmt.setString(1, "key2");
+ stmt.setLong(2, 2);
+ stmt.execute();
+ traceable.commit();
+
+ // wait for the latch to countdown, as the metrics system is time-based
+ LOG.debug("Waiting for latch to complete!");
+ updated.await(200, TimeUnit.SECONDS);// should be way more than GC pauses
+
+ // read the traces back out
+
+ /* Expected:
+ * 1. Single element trace - for first PreparedStatement#execute span
+ * 2. Two element trace for second PreparedStatement#execute span
+ * a. execute call
+ * b. metadata lookup*
+ * 3. Commit trace.
+ * a. Committing to tables
+ * i. Committing to single table
+ * ii. hbase batch write*
+ * i.I. span on server
+ * i.II. building index updates
+ * i.III. waiting for latch
+ * where '*' is a generically named thread (e.g phoenix-1-thread-X)
+ */
+ boolean indexingCompleted = checkStoredTraces(conn, new TraceChecker() {
+ public boolean foundTrace(TraceHolder trace, SpanInfo span) {
+ String traceInfo = trace.toString();
+ // skip logging traces that are just traces about tracing
+ if (traceInfo.contains(TracingCompat.DEFAULT_STATS_TABLE_NAME)) {
+ return false;
+ }
+ if (traceInfo.contains("Completing index")) {
+ return true;
+ }
+ return false;
+ }
+ });
+
+ assertTrue("Never found indexing updates", indexingCompleted);
+ }
+
+ private void createTestTable(Connection conn, boolean withIndex) throws SQLException {
+ // create a dummy table
+ String ddl =
+ "create table if not exists " + table + "(" + "k varchar not null, " + "c1 bigint"
+ + " CONSTRAINT pk PRIMARY KEY (k))";
+ conn.createStatement().execute(ddl);
+
+ // early exit if we don't need to create an index
+ if (!withIndex) {
+ return;
+ }
+ // create an index on the table - we know indexing has some basic tracing
+ ddl = "CREATE INDEX IF NOT EXISTS " + index + " on " + table + " (c1)";
+ conn.createStatement().execute(ddl);
+ conn.commit();
+ }
+
+ @Test
+ public void testScanTracing() throws Exception {
+ // separate connections to minimize amount of traces that are generated
+ Connection traceable = getTracingConnection();
+ Connection conn = getConnectionWithoutTracing();
+
+ // one call for client side, one call for server side
+ CountDownLatch updated = new CountDownLatch(2);
+ waitForCommit(updated);
+
+ // create a dummy table
+ createTestTable(conn, false);
+
+ // update the table, but don't trace these, to simplify the traces we read
+ LOG.debug("Doing dummy the writes to the tracked table");
+ String insert = "UPSERT INTO " + table + " VALUES (?, ?)";
+ PreparedStatement stmt = conn.prepareStatement(insert);
+ stmt.setString(1, "key1");
+ stmt.setLong(2, 1);
+ stmt.execute();
+ conn.commit();
+ conn.rollback();
+
+ // setup for next set of updates
+ stmt.setString(1, "key2");
+ stmt.setLong(2, 2);
+ stmt.execute();
+ conn.commit();
+ conn.rollback();
+
+ // do a scan of the table
+ String read = "SELECT * FROM " + table;
+ ResultSet results = traceable.createStatement().executeQuery(read);
+ assertTrue("Didn't get first result", results.next());
+ assertTrue("Didn't get second result", results.next());
+ results.close();
+
+ assertTrue("Get expected updates to trace table", updated.await(200, TimeUnit.SECONDS));
+ // don't trace reads either
+ boolean tracingComplete = checkStoredTraces(conn, new TraceChecker(){
+
+ @Override
+ public boolean foundTrace(TraceHolder currentTrace) {
+ String traceInfo = currentTrace.toString();
+ return traceInfo.contains("Parallel scanner");
+ }
+ });
+ assertTrue("Didn't find the parallel scanner in the tracing", tracingComplete);
+ }
+
+ @Test
+ public void testScanTracingOnServer() throws Exception {
+ // separate connections to minimize amount of traces that are generated
+ Connection traceable = getTracingConnection();
+ Connection conn = getConnectionWithoutTracing();
+
+ // one call for client side, one call for server side
+ CountDownLatch updated = new CountDownLatch(2);
+ waitForCommit(updated);
+
+ // create a dummy table
+ createTestTable(conn, false);
+
+ // update the table, but don't trace these, to simplify the traces we read
+ LOG.debug("Doing dummy the writes to the tracked table");
+ String insert = "UPSERT INTO " + table + " VALUES (?, ?)";
+ PreparedStatement stmt = conn.prepareStatement(insert);
+ stmt.setString(1, "key1");
+ stmt.setLong(2, 1);
+ stmt.execute();
+ conn.commit();
+ conn.rollback();
+
+ // setup for next set of updates
+ stmt.setString(1, "key2");
+ stmt.setLong(2, 2);
+ stmt.execute();
+ conn.commit();
+ conn.rollback();
+
+ // do a scan of the table
+ String read = "SELECT COUNT(*) FROM " + table;
+ ResultSet results = traceable.createStatement().executeQuery(read);
+ assertTrue("Didn't get count result", results.next());
+ // make sure we got the expected count
+ assertEquals("Didn't get the expected number of row", 2, results.getInt(1));
+ results.close();
+
+ assertTrue("Get expected updates to trace table", updated.await(200, TimeUnit.SECONDS));
+ // don't trace reads either
+ boolean found = checkStoredTraces(conn, new TraceChecker() {
+ public boolean foundTrace(TraceHolder trace) {
+ String traceInfo = trace.toString();
+ return traceInfo.contains(BaseScannerRegionObserver.SCANNER_OPENED_TRACE_INFO);
+ }
+ });
+ assertTrue("Didn't find the parallel scanner in the tracing", found);
+ }
+
+ private boolean checkStoredTraces(Connection conn, TraceChecker checker) throws Exception {
+ TraceReader reader = new TraceReader(conn);
+ int retries = 0;
+ boolean found = false;
+ outer: while (retries < MAX_RETRIES) {
+ Collection<TraceHolder> traces = reader.readAll(100);
+ for (TraceHolder trace : traces) {
+ LOG.info("Got trace: " + trace);
+ found = checker.foundTrace(trace);
+ if (found) {
+ break outer;
+ }
+ for (SpanInfo span : trace.spans) {
+ found = checker.foundTrace(trace, span);
+ if (found) {
+ break outer;
+ }
+ }
+ }
+ LOG.info("====== Waiting for tracing updates to be propagated ========");
+ Thread.sleep(1000);
+ retries++;
+ }
+ return found;
+ }
+
+ private abstract class TraceChecker {
+ public boolean foundTrace(TraceHolder currentTrace) {
+ return false;
+ }
+
+ public boolean foundTrace(TraceHolder currentTrace, SpanInfo currentSpan) {
+ return false;
+ }
+ }
+
+ private static class CountDownConnection extends DelegatingConnection {
+ private CountDownLatch commit;
+
+ public CountDownConnection(Connection conn, CountDownLatch commit) {
+ super(conn);
+ this.commit = commit;
+ }
+
+ @Override
+ public void commit() throws SQLException {
+ commit.countDown();
+ super.commit();
+ }
+
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/it/java/org/apache/phoenix/trace/TraceReaderTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/trace/TraceReaderTest.java b/phoenix-core/src/it/java/org/apache/phoenix/trace/TraceReaderTest.java
new file mode 100644
index 0000000..5ea677e
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/trace/TraceReaderTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.phoenix.trace;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.phoenix.metrics.MetricInfo;
+import org.apache.phoenix.metrics.PhoenixAbstractMetric;
+import org.apache.phoenix.metrics.PhoenixMetricTag;
+import org.apache.phoenix.metrics.PhoenixMetricsRecord;
+import org.apache.phoenix.trace.Hadoop1TracingTestEnabler.Hadoop1Disabled;
+import org.apache.phoenix.trace.TraceReader.SpanInfo;
+import org.apache.phoenix.trace.TraceReader.TraceHolder;
+import org.cloudera.htrace.Span;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+/**
+ * Test that the {@link TraceReader} will correctly read traces written by the
+ * {@link PhoenixTableMetricsWriter}
+ */
+@RunWith(Hadoop1TracingTestEnabler.class)
+@Hadoop1Disabled("tracing")
+public class TraceReaderTest extends BaseTracingTestIT {
+
+ private static final Log LOG = LogFactory.getLog(TraceReaderTest.class);
+
+ @Test
+ public void singleSpan() throws Exception {
+ PhoenixTableMetricsWriter sink = new PhoenixTableMetricsWriter();
+ Properties props = new Properties(TEST_PROPERTIES);
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ sink.initForTesting(conn);
+
+ // create a simple metrics record
+ long traceid = 987654;
+ PhoenixMetricsRecord record =
+ createAndFlush(sink, traceid, Span.ROOT_SPAN_ID, 10, "root", 12, 13,
+ "host-name.value", "test annotation for a span");
+
+ // start a reader
+ validateTraces(Collections.singletonList(record), conn, traceid);
+ }
+
+ private PhoenixMetricsRecord createAndFlush(PhoenixTableMetricsWriter sink, long traceid,
+ long parentid, long spanid, String desc, long startTime, long endTime, String hostname,
+ String... tags) {
+ PhoenixMetricsRecord record =
+ createRecord(traceid, parentid, spanid, desc, startTime, endTime, hostname, tags);
+ sink.addMetrics(record);
+ sink.flush();
+ return record;
+ }
+
+ /**
+ * Test multiple spans, within the same trace. Some spans are independent of the parent span,
+ * some are child spans
+ * @throws Exception on failure
+ */
+ @Test
+ public void testMultipleSpans() throws Exception {
+ // hook up a phoenix sink
+ PhoenixTableMetricsWriter sink = new PhoenixTableMetricsWriter();
+ Connection conn = getConnectionWithoutTracing();
+ sink.initForTesting(conn);
+
+ // create a simple metrics record
+ long traceid = 12345;
+ List<PhoenixMetricsRecord> records = new ArrayList<PhoenixMetricsRecord>();
+ PhoenixMetricsRecord record =
+ createAndFlush(sink, traceid, Span.ROOT_SPAN_ID, 7777, "root", 10, 30,
+ "hostname.value", "root-span tag");
+ records.add(record);
+
+ // then create a child record
+ record =
+ createAndFlush(sink, traceid, 7777, 6666, "c1", 11, 15, "hostname.value",
+ "first child");
+ records.add(record);
+
+ // create a different child
+ record =
+ createAndFlush(sink, traceid, 7777, 5555, "c2", 11, 18, "hostname.value",
+ "second child");
+ records.add(record);
+
+ // create a child of the second child
+ record =
+ createAndFlush(sink, traceid, 5555, 4444, "c3", 12, 16, "hostname.value",
+ "third child");
+ records.add(record);
+
+ // flush all the values to the table
+ sink.flush();
+
+ // start a reader
+ validateTraces(records, conn, traceid);
+ }
+
+ private void validateTraces(List<PhoenixMetricsRecord> records, Connection conn, long traceid)
+ throws Exception {
+ TraceReader reader = new TraceReader(conn);
+ Collection<TraceHolder> traces = reader.readAll(1);
+ assertEquals("Got an unexpected number of traces!", 1, traces.size());
+ // make sure the trace matches what we wrote
+ TraceHolder trace = traces.iterator().next();
+ assertEquals("Got an unexpected traceid", traceid, trace.traceid);
+ assertEquals("Got an unexpected number of spans", records.size(), trace.spans.size());
+
+ validateTrace(records, trace);
+ }
+
+ /**
+ * @param records
+ * @param trace
+ */
+ private void validateTrace(List<PhoenixMetricsRecord> records, TraceHolder trace) {
+ // drop each span into a sorted list so we get the expected ordering
+ Iterator<SpanInfo> spanIter = trace.spans.iterator();
+ for (PhoenixMetricsRecord record : records) {
+ SpanInfo spanInfo = spanIter.next();
+ LOG.info("Checking span:\n" + spanInfo);
+ Iterator<PhoenixAbstractMetric> metricIter = record.metrics().iterator();
+ assertEquals("Got an unexpected span id", metricIter.next().value(), spanInfo.id);
+ long parentId = (Long) metricIter.next().value();
+ if (parentId == Span.ROOT_SPAN_ID) {
+ assertNull("Got a parent, but it was a root span!", spanInfo.parent);
+ } else {
+ assertEquals("Got an unexpected parent span id", parentId, spanInfo.parent.id);
+ }
+ assertEquals("Got an unexpected start time", metricIter.next().value(), spanInfo.start);
+ assertEquals("Got an unexpected end time", metricIter.next().value(), spanInfo.end);
+
+ Iterator<PhoenixMetricTag> tags = record.tags().iterator();
+
+ int annotationCount = 0;
+ while (tags.hasNext()) {
+ // hostname is a tag, so we differentiate it
+ PhoenixMetricTag tag = tags.next();
+ if (tag.name().equals(MetricInfo.HOSTNAME.traceName)) {
+ assertEquals("Didn't store correct hostname value", tag.value(),
+ spanInfo.hostname);
+ } else {
+ int count = annotationCount++;
+ assertEquals("Didn't get expected annotation", count + " - " + tag.value(),
+ spanInfo.annotations.get(count));
+ }
+ }
+ assertEquals("Didn't get expected number of annotations", annotationCount,
+ spanInfo.annotationCount);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/call/CallRunner.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/call/CallRunner.java b/phoenix-core/src/main/java/org/apache/phoenix/call/CallRunner.java
new file mode 100644
index 0000000..7dc90f8
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/call/CallRunner.java
@@ -0,0 +1,66 @@
+/**
+ * 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.phoenix.call;
+
+import java.util.concurrent.Callable;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Helper class to run a Call with a set of {@link CallWrapper}
+ */
+public class CallRunner {
+
+ /**
+ * Helper {@link Callable} that also declares the type of exception it will throw, to help with
+ * type safety/generics for java
+ * @param <V> value type returned
+ * @param <E> type of check exception thrown
+ */
+ public static interface CallableThrowable<V, E extends Exception> extends Callable<V> {
+ @Override
+ public V call() throws E;
+ }
+
+ private static final Log LOG = LogFactory.getLog(CallRunner.class);
+
+ private CallRunner() {
+ // no ctor for util class
+ }
+
+ public static <V, E extends Exception, T extends CallableThrowable<V, E>> V run(T call,
+ CallWrapper... wrappers) throws E {
+ try {
+ for (CallWrapper wrap : wrappers) {
+ wrap.before();
+ }
+ return call.call();
+ } finally {
+ // have to go in reverse, to match the before logic
+ for (int i = wrappers.length - 1; i >= 0; i--) {
+ try {
+ wrappers[i].after();
+ } catch (Exception e) {
+ LOG.error("Failed to complete wrapper " + wrappers[i], e);
+ }
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/call/CallWrapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/call/CallWrapper.java b/phoenix-core/src/main/java/org/apache/phoenix/call/CallWrapper.java
new file mode 100644
index 0000000..b84dd5d
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/call/CallWrapper.java
@@ -0,0 +1,29 @@
+/**
+ * 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.phoenix.call;
+
+/**
+ *
+ */
+public interface CallWrapper {
+
+ public void before();
+
+ public void after();
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index 3ca0ce3..c04511b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -19,12 +19,16 @@ package org.apache.phoenix.coprocessor;
import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.phoenix.trace.util.Tracing;
import org.apache.phoenix.util.ServerUtil;
+import org.cloudera.htrace.Span;
abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
@@ -50,6 +54,17 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
public static final String DATA_TABLE_COLUMNS_TO_JOIN = "_DataTableColumnsToJoin";
public static final String VIEW_CONSTANTS = "_ViewConstants";
+ /** Exposed for testing */
+ public static final String SCANNER_OPENED_TRACE_INFO = "Scanner opened on server";
+ protected Configuration rawConf;
+
+ @Override
+ public void start(CoprocessorEnvironment e) throws IOException {
+ super.start(e);
+ this.rawConf =
+ ((RegionCoprocessorEnvironment) e).getRegionServerServices().getConfiguration();
+ }
+
/**
* Used by logger to identify coprocessor
*/
@@ -66,10 +81,27 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
*
*/
@Override
- public final RegionScanner postScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Scan scan, final RegionScanner s) throws IOException {
+ public final RegionScanner postScannerOpen(
+ final ObserverContext<RegionCoprocessorEnvironment> c, final Scan scan,
+ final RegionScanner s) throws IOException {
+ // turn on tracing, if its enabled
+ final Span child = Tracing.childOnServer(scan, rawConf, SCANNER_OPENED_TRACE_INFO);
try {
- return doPostScannerOpen(c, scan, s);
+ final RegionScanner scanner = doPostScannerOpen(c, scan, s);
+ return new DelegateRegionScanner(scanner) {
+ @Override
+ public void close() throws IOException {
+ if (child != null) {
+ child.stop();
+ }
+ delegate.close();
+ }
+
+ };
} catch (Throwable t) {
+ if (child != null) {
+ child.stop();
+ }
ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionNameAsString(), t);
return null; // impossible
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionScanner.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionScanner.java
new file mode 100644
index 0000000..f88a931
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionScanner.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.phoenix.coprocessor;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+
+public class DelegateRegionScanner implements RegionScanner {
+
+ protected final RegionScanner delegate;
+
+ public DelegateRegionScanner(RegionScanner scanner) {
+ this.delegate = scanner;
+ }
+
+ @Override
+ public HRegionInfo getRegionInfo() {
+ return delegate.getRegionInfo();
+ }
+
+ @Override
+ public boolean isFilterDone() throws IOException {
+ return delegate.isFilterDone();
+ }
+
+ @Override
+ public boolean reseek(byte[] row) throws IOException {
+ return delegate.reseek(row);
+ }
+
+ @Override
+ public long getMvccReadPoint() {
+ return delegate.getMvccReadPoint();
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+
+ public long getMaxResultSize() {
+ return delegate.getMaxResultSize();
+ }
+
+ public boolean next(List<Cell> arg0, int arg1) throws IOException {
+ return delegate.next(arg0, arg1);
+ }
+
+ public boolean next(List<Cell> arg0) throws IOException {
+ return delegate.next(arg0);
+ }
+
+ public boolean nextRaw(List<Cell> arg0, int arg1) throws IOException {
+ return delegate.nextRaw(arg0, arg1);
+ }
+
+ public boolean nextRaw(List<Cell> arg0) throws IOException {
+ return delegate.nextRaw(arg0);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
index be96956..8a270e3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
@@ -58,9 +58,12 @@ import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.trace.TracingIterator;
+import org.apache.phoenix.trace.util.Tracing;
import org.apache.phoenix.util.SQLCloseable;
import org.apache.phoenix.util.SQLCloseables;
import org.apache.phoenix.util.ScanUtil;
+import org.cloudera.htrace.TraceScope;
import com.google.common.collect.Lists;
@@ -203,7 +206,7 @@ public abstract class BasicQueryPlan implements QueryPlan {
}
}
ResultIterator iterator = newIterator();
- return dependencies.isEmpty() ?
+ iterator = dependencies.isEmpty() ?
iterator : new DelegateResultIterator(iterator) {
@Override
public void close() throws SQLException {
@@ -214,6 +217,12 @@ public abstract class BasicQueryPlan implements QueryPlan {
}
}
};
+
+ // wrap the iterator so we start/end tracing as we expect
+ TraceScope scope =
+ Tracing.startNewSpan(context.getConnection(), "Creating basic query for "
+ + getPlanSteps(iterator));
+ return (scope.getSpan() != null) ? new TracingIterator(scope, iterator) : iterator;
}
private void serializeIndexMaintainerIntoScan(Scan scan, PTable dataTable) {
@@ -349,8 +358,12 @@ public abstract class BasicQueryPlan implements QueryPlan {
// Optimize here when getting explain plan, as queries don't get optimized until after compilation
QueryPlan plan = context.getConnection().getQueryServices().getOptimizer().optimize(context.getStatement(), this);
ResultIterator iterator = plan.iterator();
+ return new ExplainPlan(getPlanSteps(iterator));
+ }
+
+ private List<String> getPlanSteps(ResultIterator iterator){
List<String> planSteps = Lists.newArrayListWithExpectedSize(5);
iterator.explain(planSteps);
- return new ExplainPlan(planSteps);
+ return planSteps;
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b7f46c10/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
index 410b8b4..6519b66 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -49,11 +49,14 @@ import org.apache.phoenix.schema.PRow;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.trace.util.Tracing;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.SQLCloseable;
import org.apache.phoenix.util.ServerUtil;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.TraceScope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -333,6 +336,10 @@ public class MutationState implements SQLCloseable {
long[] serverTimeStamps = validate();
Iterator<Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>>> iterator = this.mutations.entrySet().iterator();
List<Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>>> committedList = Lists.newArrayListWithCapacity(this.mutations.size());
+
+ // add tracing for this operation
+ TraceScope trace = Tracing.startNewSpan(connection, "Committing mutations to tables");
+ Span span = trace.getSpan();
while (iterator.hasNext()) {
Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>> entry = iterator.next();
Map<ImmutableBytesPtr,Map<PColumn,byte[]>> valuesMap = entry.getValue();
@@ -348,6 +355,10 @@ public class MutationState implements SQLCloseable {
byte[] htableName = pair.getFirst();
List<Mutation> mutations = pair.getSecond();
+ //create a span per target table
+ //TODO maybe we can be smarter about the table name to string here?
+ Span child = Tracing.child(span,"Writing mutation batch for table: "+Bytes.toString(htableName));
+
int retryCount = 0;
boolean shouldRetry = false;
do {
@@ -358,6 +369,7 @@ public class MutationState implements SQLCloseable {
if (IndexMetaDataCacheClient.useIndexMetadataCache(connection, mutations, tempPtr.getLength())) {
IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, tableRef);
cache = client.addIndexMetadataCache(mutations, tempPtr);
+ child.addTimelineAnnotation("Updated index metadata cache");
uuidValue = cache.getId();
// If we haven't retried yet, retry for this case only, as it's possible that
// a split will occur after we send the index metadata cache to all known
@@ -385,7 +397,9 @@ public class MutationState implements SQLCloseable {
try {
if (logger.isDebugEnabled()) logMutationSize(hTable, mutations);
long startTime = System.currentTimeMillis();
+ child.addTimelineAnnotation("Attempt " + retryCount);
hTable.batch(mutations);
+ child.stop();
shouldRetry = false;
if (logger.isDebugEnabled()) logger.debug("Total time for batch call of " + mutations.size() + " mutations into " + table.getName().getString() + ": " + (System.currentTimeMillis() - startTime) + " ms");
committedList.add(entry);
@@ -396,8 +410,15 @@ public class MutationState implements SQLCloseable {
// Swallow this exception once, as it's possible that we split after sending the index metadata
// and one of the region servers doesn't have it. This will cause it to have it the next go around.
// If it fails again, we don't retry.
- logger.warn("Swallowing exception and retrying after clearing meta cache on connection. " + inferredE);
+ String msg = "Swallowing exception and retrying after clearing meta cache on connection. " + inferredE;
+ logger.warn(msg);
connection.getQueryServices().clearTableRegionCache(htableName);
+
+ // add a new child span as this one failed
+ child.addTimelineAnnotation(msg);
+ child.stop();
+ child = Tracing.child(span,"Failed batch, attempting retry");
+
continue;
}
e = inferredE;
@@ -432,6 +453,7 @@ public class MutationState implements SQLCloseable {
numRows -= entry.getValue().size();
iterator.remove(); // Remove batches as we process them
}
+ trace.close();
assert(numRows==0);
assert(this.mutations.isEmpty());
}