Support the async put API provided by HTableMultiplexer to improve the write throughput in collector. #1683

- support to reuse connection
 - added HBaseAsyncOperationMetric to CollectorMetric
 - show  HBaseAsyncOperationMetric to use JMX

collector/src/main/java/com/navercorp/pinpoint/collector/manage/HBaseManager.java

@@ -0,0 +1,41 @@
+package com.navercorp.pinpoint.collector.manage;
+
+import com.navercorp.pinpoint.common.hbase.HBaseAsyncOperation;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.Map;
+
+/**
+ * @Author Taejin Koo
+ */
+public class HBaseManager extends AbstractCollectorManager implements HBaseManagerMBean {
+
+    @Autowired
+    private HBaseAsyncOperation hBaseAsyncOperation;
+
+    @Override
+    public Long getAsyncOpsCount() {
+        return hBaseAsyncOperation.getOpsCount();
+    }
+
+    @Override
+    public Long getAsyncOpsRejectedCount() {
+        return hBaseAsyncOperation.getOpsRejectedCount();
+    }
+
+    @Override
+    public Map<String, Long> getCurrentAsyncOpsCountForEachRegionServer() {
+        return hBaseAsyncOperation.getCurrentOpsCountForEachRegionServer();
+    }
+
+    @Override
+    public Map<String, Long> getAsyncOpsFailedCountForEachRegionServer() {
+        return hBaseAsyncOperation.getOpsFailedCountForEachRegionServer();
+    }
+
+    @Override
+    public Map<String, Long> getAsyncOpsAverageLatencyForEachRegionServer() {
+        return hBaseAsyncOperation.getOpsAverageLatencyForEachRegionServer();
+    }
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/manage/HBaseManagerMBean.java

@@ -0,0 +1,20 @@
+package com.navercorp.pinpoint.collector.manage;
+
+import java.util.Map;
+
+/**
+ * @Author Taejin Koo
+ */
+public interface HBaseManagerMBean {
+
+    Long getAsyncOpsCount();
+
+    Long getAsyncOpsRejectedCount();
+
+    Map<String, Long> getCurrentAsyncOpsCountForEachRegionServer();
+
+    Map<String, Long> getAsyncOpsFailedCountForEachRegionServer();
+
+    Map<String, Long> getAsyncOpsAverageLatencyForEachRegionServer();
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/manage/jmx/JMXCollectorManagerList.java

@@ -16,19 +16,19 @@
 
 package com.navercorp.pinpoint.collector.manage.jmx;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
 import com.navercorp.pinpoint.collector.manage.ClusterManager;
+import com.navercorp.pinpoint.collector.manage.CollectorManager;
+import com.navercorp.pinpoint.collector.manage.HBaseManager;
+import com.navercorp.pinpoint.collector.manage.HandlerManager;
 import com.navercorp.pinpoint.rpc.util.ListUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 
-import com.navercorp.pinpoint.collector.manage.CollectorManager;
-import com.navercorp.pinpoint.collector.manage.HandlerManager;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 /**
  * @author Taejin Koo
@@ -46,6 +46,9 @@ public class JMXCollectorManagerList {
     @Autowired
     private ClusterManager clusterManager;
 
+    @Autowired
+    private HBaseManager hBaseManager;
+
     public List<CollectorManager> getSupportList() {
         if (!isActive) {
             logger.warn("not activing jmx api for admin.");
@@ -56,6 +59,7 @@ public class JMXCollectorManagerList {
 
         ListUtils.addIfValueNotNull(supportManagerList, handlerManager);
         ListUtils.addIfValueNotNull(supportManagerList, clusterManager);
+        ListUtils.addIfValueNotNull(supportManagerList, hBaseManager);
 
         return supportManagerList;
     }

collector/src/main/java/com/navercorp/pinpoint/collector/monitor/CollectorMetric.java

@@ -17,6 +17,7 @@
 package com.navercorp.pinpoint.collector.monitor;
 
 import com.codahale.metrics.JvmAttributeGaugeSet;
+import com.codahale.metrics.Metric;
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.ScheduledReporter;
 import com.codahale.metrics.Slf4jReporter;
@@ -31,6 +32,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -46,11 +48,13 @@ public class CollectorMetric {
     @Autowired
     private MetricRegistry metricRegistry;
 
+    @Autowired(required = false)
+    private HBaseAsyncOperationMetrics hBaseAsyncOperationMetrics;
+
     private ScheduledReporter reporter;
 
     private final boolean isEnable = isEnable0(REPORTER_LOGGER_NAME);
 
-
     @PostConstruct
     public void start() {
         initRegistry();
@@ -76,8 +80,14 @@ public class CollectorMetric {
         metricRegistry.register("jvm.vm", new JvmAttributeGaugeSet());
         metricRegistry.register("jvm.garbage-collectors", new GarbageCollectorMetricSet());
         metricRegistry.register("jvm.thread-states", new ThreadStatesGaugeSet());
-    }
 
+        if (hBaseAsyncOperationMetrics != null) {
+            Map<String, Metric> metrics = hBaseAsyncOperationMetrics.getMetrics();
+            for (Map.Entry<String, Metric> metric : metrics.entrySet()) {
+                metricRegistry.register(metric.getKey(), metric.getValue());
+            }
+        }
+    }
 
     private void initReporters() {
         Slf4jReporter.Builder builder = Slf4jReporter.forRegistry(metricRegistry);
@@ -99,4 +109,5 @@ public class CollectorMetric {
         reporter.stop();
         reporter = null;
     }
+
 }

collector/src/main/java/com/navercorp/pinpoint/collector/monitor/HBaseAsyncOperationMetrics.java

@@ -0,0 +1,74 @@
+package com.navercorp.pinpoint.collector.monitor;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricSet;
+import com.navercorp.pinpoint.common.hbase.HBaseAsyncOperation;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @Author Taejin Koo
+ */
+public class HBaseAsyncOperationMetrics implements MetricSet {
+
+    private static final String HBASE_ASYNC_OPS = "hbase.async.ops";
+    private static final String COUNT = HBASE_ASYNC_OPS + ".count";
+    private static final String REJECTED_COUNT = HBASE_ASYNC_OPS + ".rejected.count";
+    private static final String FAILED_COUNT = HBASE_ASYNC_OPS + ".failed.count";
+    private static final String WAITING_COUNT = HBASE_ASYNC_OPS + ".waiting.count";
+    private static final String AVERAGE_LATENCY = HBASE_ASYNC_OPS + ".latency.value";
+
+    private final HBaseAsyncOperation hBaseAsyncOperation;
+
+    public HBaseAsyncOperationMetrics(HBaseAsyncOperation hBaseAsyncOperation) {
+        if (hBaseAsyncOperation == null) {
+            throw new NullPointerException("null");
+        }
+        this.hBaseAsyncOperation = hBaseAsyncOperation;
+    }
+
+    @Override
+    public Map<String, Metric> getMetrics() {
+        if (!hBaseAsyncOperation.isAvailable()) {
+            return Collections.emptyMap();
+        }
+
+        final Map<String, Metric> gauges = new HashMap<>(3);
+        gauges.put(COUNT, new Gauge<Long>() {
+            @Override
+            public Long getValue() {
+                return hBaseAsyncOperation.getOpsCount();
+            }
+        });
+        gauges.put(REJECTED_COUNT, new Gauge<Long>() {
+            @Override
+            public Long getValue() {
+                return hBaseAsyncOperation.getOpsRejectedCount();
+            }
+        });
+        gauges.put(FAILED_COUNT, new Gauge<Long>() {
+            @Override
+            public Long getValue() {
+                return hBaseAsyncOperation.getOpsFailedCount();
+            }
+        });
+        gauges.put(WAITING_COUNT, new Gauge<Long>() {
+            @Override
+            public Long getValue() {
+                return hBaseAsyncOperation.getCurrentOpsCount();
+            }
+        });
+        gauges.put(AVERAGE_LATENCY, new Gauge<Long>() {
+            @Override
+            public Long getValue() {
+                return hBaseAsyncOperation.getOpsAverageLatency();
+            }
+        });
+
+        return Collections.unmodifiableMap(gauges);
+    }
+
+}

collector/src/main/resources/applicationContext-collector.xml

@@ -75,6 +75,9 @@
         <constructor-arg ref="targetClusterPointRepository"/>
     </bean>
 
+    <bean id="hBaseManager" class="com.navercorp.pinpoint.collector.manage.HBaseManager">
+    </bean>
+
     <!-- DispatchHandler-related Beans -->
     <bean id="tcpDispatchHandler" class="com.navercorp.pinpoint.collector.receiver.TcpDispatchHandler"/>
     <bean id="tcpDispatchHandlerWrapper" class="com.navercorp.pinpoint.collector.receiver.DispatchHandlerWrapper">

collector/src/main/resources/applicationContext-hbase.xml

@@ -47,9 +47,14 @@
     </bean>
 
     <bean id="asyncOperation" class="com.navercorp.pinpoint.common.hbase.HBaseAsyncOperationFactory" factory-method="create">
+        <constructor-arg value="#{connectionFactory.getConnection()}"/>
         <constructor-arg ref="hbaseConfiguration"/>
     </bean>
 
+    <bean id="asyncOperationMetrics" class="com.navercorp.pinpoint.collector.monitor.HBaseAsyncOperationMetrics">
+        <constructor-arg ref="asyncOperation"/>
+    </bean>
+
     <bean id="hbaseTemplate" class="com.navercorp.pinpoint.common.hbase.HbaseTemplate2" destroy-method="destroy">
         <property name="configuration" ref="hbaseConfiguration"/>
         <property name="tableFactory" ref="connectionFactory"/>

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/DisabledHBaseAsyncOperation.java

@@ -3,7 +3,9 @@ package com.navercorp.pinpoint.common.hbase;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Put;
 
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 /**
  * @Author Taejin Koo
@@ -28,8 +30,43 @@ public class DisabledHBaseAsyncOperation implements HBaseAsyncOperation {
     }
 
     @Override
-    public Long getCurrentPutOpsCount() {
-        return 0L;
+    public Long getOpsCount() {
+        return -1L;
+    }
+
+    @Override
+    public Long getOpsRejectedCount() {
+        return -1L;
+    }
+
+    @Override
+    public Long getCurrentOpsCount() {
+        return -1L;
+    }
+
+    @Override
+    public Long getOpsFailedCount() {
+        return -1L;
+    }
+
+    @Override
+    public Long getOpsAverageLatency() {
+        return -1L;
+    }
+
+    @Override
+    public Map<String, Long> getCurrentOpsCountForEachRegionServer() {
+        return Collections.emptyMap();
+    }
+
+    @Override
+    public Map<String, Long> getOpsFailedCountForEachRegionServer() {
+        return Collections.emptyMap();
+    }
+
+    @Override
+    public Map<String, Long> getOpsAverageLatencyForEachRegionServer() {
+        return Collections.emptyMap();
     }
 
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperation.java

@@ -4,6 +4,7 @@ import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Put;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * @Author Taejin Koo
@@ -16,6 +17,20 @@ public interface HBaseAsyncOperation {
 
     List<Put> put(TableName tableName, final List<Put> puts);
 
-    Long getCurrentPutOpsCount();
+    Long getOpsCount();
+
+    Long getOpsRejectedCount();
+
+    Long getCurrentOpsCount();
+
+    Long getOpsFailedCount();
+
+    Long getOpsAverageLatency();
+
+    Map<String, Long> getCurrentOpsCountForEachRegionServer();
+
+    Map<String, Long> getOpsFailedCountForEachRegionServer();
+
+    Map<String, Long> getOpsAverageLatencyForEachRegionServer();
 
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperationFactory.java

@@ -1,6 +1,7 @@
 package com.navercorp.pinpoint.common.hbase;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.Connection;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,4 +45,23 @@ public class HBaseAsyncOperationFactory {
         return new HBaseAsyncTemplate(configuration, queueSize);
     }
 
+    public static HBaseAsyncOperation create(Connection connection, Configuration configuration) throws IOException {
+        boolean enableAsyncMethod = configuration.getBoolean(ENABLE_ASYNC_METHOD, DEFAULT_ENABLE_ASYNC_METHOD);
+        if (!enableAsyncMethod) {
+            return DisabledHBaseAsyncOperation.INSTANCE;
+        }
+
+        int queueSize = configuration.getInt(ASYNC_IN_QUEUE_SIZE, DEFAULT_ASYNC_IN_QUEUE_SIZE);
+
+        if (configuration.get(ASYNC_PERIODIC_FLUSH_TIME, null) == null) {
+            configuration.setInt(ASYNC_PERIODIC_FLUSH_TIME, DEFAULT_ASYNC_PERIODIC_FLUSH_TIME);
+        }
+
+        if (configuration.get(ASYNC_RETRY_COUNT, null) == null) {
+            configuration.setInt(ASYNC_RETRY_COUNT, DEFAULT_ASYNC_RETRY_COUNT);
+        }
+
+        return new HBaseAsyncTemplate(connection, configuration, queueSize);
+    }
+
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncTemplate.java

@@ -2,11 +2,14 @@ package com.navercorp.pinpoint.common.hbase;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.HTableMultiplexer;
 import org.apache.hadoop.hbase.client.Put;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @Author Taejin Koo
@@ -14,11 +17,17 @@ import java.util.List;
 public class HBaseAsyncTemplate implements HBaseAsyncOperation {
 
     private final HTableMultiplexer hTableMultiplexer;
+    private final AtomicInteger opsCount = new AtomicInteger();
+    private final AtomicInteger opsRejectCount = new AtomicInteger();
 
     public HBaseAsyncTemplate(Configuration conf, int perRegionServerBufferQueueSize) throws IOException {
         this.hTableMultiplexer = new HTableMultiplexer(conf, perRegionServerBufferQueueSize);
     }
 
+    public HBaseAsyncTemplate(Connection connection, Configuration conf, int perRegionServerBufferQueueSize) throws IOException {
+        this.hTableMultiplexer = new HTableMultiplexer(connection, conf, perRegionServerBufferQueueSize);
+    }
+
     @Override
     public boolean isAvailable() {
         return true;
@@ -26,17 +35,64 @@ public class HBaseAsyncTemplate implements HBaseAsyncOperation {
 
     @Override
     public boolean put(TableName tableName, Put put) {
-        return hTableMultiplexer.put(tableName, put);
+        opsCount.incrementAndGet();
+
+        boolean success = hTableMultiplexer.put(tableName, put);
+        if (!success) {
+            opsRejectCount.incrementAndGet();
+        }
+        return success;
     }
 
     @Override
     public List<Put> put(TableName tableName, List<Put> puts) {
-        return hTableMultiplexer.put(tableName, puts);
+        opsCount.addAndGet(puts.size());
+
+        List<Put> rejectPuts = hTableMultiplexer.put(tableName, puts);
+        if (rejectPuts != null && rejectPuts.size() > 0) {
+            opsRejectCount.addAndGet(rejectPuts.size());
+        }
+        return rejectPuts;
     }
 
     @Override
-    public Long getCurrentPutOpsCount() {
+    public Long getOpsCount() {
+        return opsCount.longValue();
+    }
+
+    @Override
+    public Long getOpsRejectedCount() {
+        return opsRejectCount.longValue();
+    }
+
+    @Override
+    public Long getCurrentOpsCount() {
         return hTableMultiplexer.getHTableMultiplexerStatus().getTotalBufferedCounter();
     }
 
+    @Override
+    public Long getOpsFailedCount() {
+        return hTableMultiplexer.getHTableMultiplexerStatus().getTotalFailedCounter();
+    }
+
+    @Override
+    public Long getOpsAverageLatency() {
+        return hTableMultiplexer.getHTableMultiplexerStatus().getOverallAverageLatency();
+    }
+
+    @Override
+    public Map<String, Long> getCurrentOpsCountForEachRegionServer() {
+        return hTableMultiplexer.getHTableMultiplexerStatus().getBufferedCounterForEachRegionServer();
+    }
+
+    @Override
+    public Map<String, Long> getOpsFailedCountForEachRegionServer() {
+        return hTableMultiplexer.getHTableMultiplexerStatus().getFailedCounterForEachRegionServer();
+    }
+
+    @Override
+    public Map<String, Long> getOpsAverageLatencyForEachRegionServer() {
+        return hTableMultiplexer.getHTableMultiplexerStatus().getAverageLatencyForEachRegionServer();
+    }
+
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java

@@ -144,9 +144,9 @@ public class HbaseTemplate2 extends HbaseAccessor implements HbaseOperations2, I
         stopWatch.start();
 
         while (true) {
-            Long currentPutOpsCount = asyncOperation.getCurrentPutOpsCount();
+            Long currentPutOpsCount = asyncOperation.getCurrentOpsCount();
             logger.warn("count " + currentPutOpsCount);
-            if (currentPutOpsCount == 0L) {
+            if (currentPutOpsCount <= 0L) {
                 return true;
             }
 

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java

@@ -62,6 +62,10 @@ public class PooledHTableFactory implements TableFactory, DisposableBean {
         }
     }
 
+    public Connection getConnection() {
+        return connection;
+    }
+
     private ExecutorService createExecutorService(int poolSize, int workQueueMaxSize, boolean prestartThreadPool) {
 
         logger.info("create HConnectionThreadPoolExecutor poolSize:{}, workerQueueMaxSize:{}", poolSize, workQueueMaxSize);