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/e8def027
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/e8def027
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/e8def027

Branch: refs/heads/4.0
Commit: e8def027a9c60e978ac79acbc6c71979e7478a43
Parents: 083c5f2
Author: Jesse Yates <jya...@apache.org>
Authored: Fri Jun 6 16:11:32 2014 -0700
Committer: Jesse Yates <jya...@apache.org>
Committed: Mon Jul 28 06:38:20 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                   |  29 +-
 .../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                  |  45 +++
 .../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, 4477 insertions(+), 55 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/phoenix/blob/e8def027/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index 5f55e29..af6dcb6 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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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/e8def027/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());
     }

Reply via email to