diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/manage/HBaseManager.java b/collector/src/main/java/com/navercorp/pinpoint/collector/manage/HBaseManager.java new file mode 100644 index 000000000..ab2aa7211 --- /dev/null +++ b/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 getCurrentAsyncOpsCountForEachRegionServer() { + return hBaseAsyncOperation.getCurrentOpsCountForEachRegionServer(); + } + + @Override + public Map getAsyncOpsFailedCountForEachRegionServer() { + return hBaseAsyncOperation.getOpsFailedCountForEachRegionServer(); + } + + @Override + public Map getAsyncOpsAverageLatencyForEachRegionServer() { + return hBaseAsyncOperation.getOpsAverageLatencyForEachRegionServer(); + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/manage/HBaseManagerMBean.java b/collector/src/main/java/com/navercorp/pinpoint/collector/manage/HBaseManagerMBean.java new file mode 100644 index 000000000..6c502bc7d --- /dev/null +++ b/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 getCurrentAsyncOpsCountForEachRegionServer(); + + Map getAsyncOpsFailedCountForEachRegionServer(); + + Map getAsyncOpsAverageLatencyForEachRegionServer(); + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/manage/jmx/JMXCollectorManagerList.java b/collector/src/main/java/com/navercorp/pinpoint/collector/manage/jmx/JMXCollectorManagerList.java index 7072375fe..8e232997f 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/manage/jmx/JMXCollectorManagerList.java +++ b/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 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; } diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/monitor/CollectorMetric.java b/collector/src/main/java/com/navercorp/pinpoint/collector/monitor/CollectorMetric.java index c912c30e0..c2344d154 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/monitor/CollectorMetric.java +++ b/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 metrics = hBaseAsyncOperationMetrics.getMetrics(); + for (Map.Entry 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; } + } diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/monitor/HBaseAsyncOperationMetrics.java b/collector/src/main/java/com/navercorp/pinpoint/collector/monitor/HBaseAsyncOperationMetrics.java new file mode 100644 index 000000000..3afd396f3 --- /dev/null +++ b/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 getMetrics() { + if (!hBaseAsyncOperation.isAvailable()) { + return Collections.emptyMap(); + } + + final Map gauges = new HashMap<>(3); + gauges.put(COUNT, new Gauge() { + @Override + public Long getValue() { + return hBaseAsyncOperation.getOpsCount(); + } + }); + gauges.put(REJECTED_COUNT, new Gauge() { + @Override + public Long getValue() { + return hBaseAsyncOperation.getOpsRejectedCount(); + } + }); + gauges.put(FAILED_COUNT, new Gauge() { + @Override + public Long getValue() { + return hBaseAsyncOperation.getOpsFailedCount(); + } + }); + gauges.put(WAITING_COUNT, new Gauge() { + @Override + public Long getValue() { + return hBaseAsyncOperation.getCurrentOpsCount(); + } + }); + gauges.put(AVERAGE_LATENCY, new Gauge() { + @Override + public Long getValue() { + return hBaseAsyncOperation.getOpsAverageLatency(); + } + }); + + return Collections.unmodifiableMap(gauges); + } + +} diff --git a/collector/src/main/resources/applicationContext-collector.xml b/collector/src/main/resources/applicationContext-collector.xml index ca38047a2..57b5c2b32 100644 --- a/collector/src/main/resources/applicationContext-collector.xml +++ b/collector/src/main/resources/applicationContext-collector.xml @@ -75,6 +75,9 @@ + + + diff --git a/collector/src/main/resources/applicationContext-hbase.xml b/collector/src/main/resources/applicationContext-hbase.xml index 9d2bc7aa4..1c8c0bb73 100644 --- a/collector/src/main/resources/applicationContext-hbase.xml +++ b/collector/src/main/resources/applicationContext-hbase.xml @@ -47,9 +47,14 @@ + + + + + diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/DisabledHBaseAsyncOperation.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/DisabledHBaseAsyncOperation.java index 21bfb3750..f4f67a974 100644 --- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/DisabledHBaseAsyncOperation.java +++ b/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 getCurrentOpsCountForEachRegionServer() { + return Collections.emptyMap(); + } + + @Override + public Map getOpsFailedCountForEachRegionServer() { + return Collections.emptyMap(); + } + + @Override + public Map getOpsAverageLatencyForEachRegionServer() { + return Collections.emptyMap(); } } \ No newline at end of file diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperation.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperation.java index 492ad0d5b..dd534910d 100644 --- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperation.java +++ b/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(TableName tableName, final List puts); - Long getCurrentPutOpsCount(); + Long getOpsCount(); + + Long getOpsRejectedCount(); + + Long getCurrentOpsCount(); + + Long getOpsFailedCount(); + + Long getOpsAverageLatency(); + + Map getCurrentOpsCountForEachRegionServer(); + + Map getOpsFailedCountForEachRegionServer(); + + Map getOpsAverageLatencyForEachRegionServer(); } \ No newline at end of file diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperationFactory.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperationFactory.java index c868dae7d..1401678f7 100644 --- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncOperationFactory.java +++ b/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); + } + } \ No newline at end of file diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncTemplate.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncTemplate.java index e2f23e74b..e695fbd7e 100644 --- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAsyncTemplate.java +++ b/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(TableName tableName, List puts) { - return hTableMultiplexer.put(tableName, puts); + opsCount.addAndGet(puts.size()); + + List 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 getCurrentOpsCountForEachRegionServer() { + return hTableMultiplexer.getHTableMultiplexerStatus().getBufferedCounterForEachRegionServer(); + } + + @Override + public Map getOpsFailedCountForEachRegionServer() { + return hTableMultiplexer.getHTableMultiplexerStatus().getFailedCounterForEachRegionServer(); + } + + @Override + public Map getOpsAverageLatencyForEachRegionServer() { + return hTableMultiplexer.getHTableMultiplexerStatus().getAverageLatencyForEachRegionServer(); + } + } \ No newline at end of file diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java index fb992a210..b02527295 100644 --- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java +++ b/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; } diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java index 7a724dd26..a408935e5 100644 --- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java +++ b/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);