You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2021/10/27 12:54:39 UTC
[accumulo] branch main updated: Replace hadoop metrics with
micrometer (#2305)
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new b5ca158 Replace hadoop metrics with micrometer (#2305)
b5ca158 is described below
commit b5ca15806659efbbd6f5d158be8648cc2cd13593
Author: Dave Marion <dl...@apache.org>
AuthorDate: Wed Oct 27 08:54:16 2021 -0400
Replace hadoop metrics with micrometer (#2305)
Replaced Hadoop Metrics2 framework with Micrometer.
Closes #1134
Co-authored-by: domgarguilo <do...@gmail.com>
Co-authored-by: EdColeman <de...@etcoleman.com>
Co-authored-by: Manno15 <je...@gmail.com>
Co-authored-by: Luke Foster <84...@users.noreply.github.com>
---
.../templates/hadoop-metrics2-accumulo.properties | 95 ----
assemble/pom.xml | 23 +
assemble/src/main/assemblies/component.xml | 3 +
core/pom.xml | 4 +
.../org/apache/accumulo/core/conf/Property.java | 9 +-
.../core/metrics/MeterRegistryFactory.java | 13 +-
.../accumulo/core/metrics/MetricsProducer.java | 611 +++++++++++++++++++++
.../apache/accumulo/core/metrics/MetricsUtil.java | 136 +++++
.../accumulo/core/util/threads/ThreadPools.java | 9 +-
pom.xml | 8 +
server/base/pom.xml | 4 +
.../org/apache/accumulo/server/AbstractServer.java | 12 +-
.../apache/accumulo/server/metrics/Metrics.java | 114 ----
.../accumulo/server/metrics/ThriftMetrics.java | 28 +-
.../apache/accumulo/server/rpc/TServerUtils.java | 32 +-
.../apache/accumulo/server/rpc/TimedProcessor.java | 23 +-
.../accumulo/server/util/TServerUtilsTest.java | 8 +-
.../base/src/test/resources/log4j2-test.properties | 1 -
.../coordinator/CompactionCoordinator.java | 10 +-
.../org/apache/accumulo/compactor/Compactor.java | 10 +-
server/gc/pom.xml | 4 +
.../apache/accumulo/gc/SimpleGarbageCollector.java | 41 +-
.../apache/accumulo/gc/metrics/GcCycleMetrics.java | 10 +-
.../org/apache/accumulo/gc/metrics/GcMetrics.java | 143 ++---
.../accumulo/gc/metrics/GcMetricsFactory.java | 62 ---
server/manager/pom.xml | 4 +
.../java/org/apache/accumulo/manager/Manager.java | 30 +-
.../accumulo/manager/metrics/ManagerMetrics.java | 23 +-
.../manager/metrics/ManagerMetricsFactory.java | 74 ---
.../manager/metrics/ReplicationMetrics.java | 75 +--
.../accumulo/manager/metrics/fate/FateMetrics.java | 217 ++++----
.../manager/metrics/ReplicationMetricsTest.java | 43 +-
.../resources/hadoop-metrics2-accumulo.properties | 60 --
server/tserver/pom.xml | 4 +
.../apache/accumulo/tserver/BusiestTracker.java | 2 +-
.../org/apache/accumulo/tserver/TabletServer.java | 70 +--
.../metrics/CompactionExecutorsMetrics.java | 104 ++--
.../accumulo/tserver/metrics/TServerMetrics.java | 39 --
.../tserver/metrics/TabletServerMetrics.java | 110 ++--
.../tserver/metrics/TabletServerMetricsUtil.java | 36 +-
.../tserver/metrics/TabletServerMinCMetrics.java | 33 +-
.../tserver/metrics/TabletServerScanMetrics.java | 37 +-
.../tserver/metrics/TabletServerUpdateMetrics.java | 81 +--
.../org/apache/accumulo/tserver/tablet/Tablet.java | 59 +-
test/pom.xml | 18 +-
.../compaction/ExternalCompactionMetricsIT.java | 185 +++++++
.../test/compaction/ExternalCompaction_1_IT.java | 101 ----
.../test/functional/FateConcurrencyIT.java | 1 +
.../accumulo/test/functional/GcMetricsIT.java | 197 -------
.../accumulo/test/functional/ManagerMetricsIT.java | 301 ----------
.../accumulo/test/functional/ZombieTServer.java | 8 +-
.../accumulo/test/metrics/MetricsFileTailer.java | 360 ------------
.../apache/accumulo/test/metrics/MetricsIT.java | 158 ++++++
.../test/metrics/TestStatsDRegistryFactory.java | 96 ++++
.../accumulo/test/metrics/TestStatsDSink.java | 140 +++++
.../accumulo/test/metrics/fate/FateMetricsIT.java | 312 -----------
.../test/metrics/fate/InMemTestCollector.java | 146 -----
.../accumulo/test/performance/NullTserver.java | 8 +-
.../resources/hadoop-metrics2-accumulo.properties | 61 --
test/src/main/resources/log4j2-test.properties | 23 +
.../test/metrics/MetricsFileTailerTest.java | 98 ----
61 files changed, 2068 insertions(+), 2659 deletions(-)
diff --git a/assemble/conf/templates/hadoop-metrics2-accumulo.properties b/assemble/conf/templates/hadoop-metrics2-accumulo.properties
deleted file mode 100644
index 98ecfd9..0000000
--- a/assemble/conf/templates/hadoop-metrics2-accumulo.properties
+++ /dev/null
@@ -1,95 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-
-# Poll collectors every thirty seconds
-*.period=30
-
-#
-# MetricSink configuration
-#
-# Uncomment a sink (and configuration) to enable it. By default,
-# no sinks are enabled.
-#
-
-#
-# Configure file logging
-#
-
-# File sink for all metrics
-# accumulo.sink.file-all.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-all.filename=all.metrics
-
-# File sink for tserver metrics
-# accumulo.sink.file-tserver.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-tserver.context=tserver
-# accumulo.sink.file-tserver.filename=tserver.metrics
-
-# File sink for manager metrics
-# accumulo.sink.file-manager.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-manager.context=manager
-# accumulo.sink.file-manager.filename=manager.metrics
-
-# File sink for thrift server metrics
-# accumulo.sink.file-thrift.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-thrift.context=thrift
-# accumulo.sink.file-thrift.filename=thrift.metrics
-
-# File sink for gc metrics
-# accumulo.sink.file-gc.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-gc.context=accgc
-# accumulo.sink.file-gc.filename=accgc.metrics
-
-#
-# Configure Graphite
-#
-
-# Configure all sources to be written to a Graphite instance
-# accumulo.sink.graphite.class=org.apache.hadoop.metrics2.sink.GraphiteSink
-# accumulo.sink.graphite.server_host=localhost
-# accumulo.sink.graphite.server_port=2003
-
-#
-# Configure Ganglia
-#
-
-# for Ganglia 3.0 support
-# accumulo.sink.ganglia.class=org.apache.hadoop.metrics2.sink.ganglia.GangliaSink30
-
-# for Ganglia 3.1 support
-# accumulo.sink.ganglia.class=org.apache.hadoop.metrics2.sink.ganglia.GangliaSink31
-
-# accumulo.sink.ganglia.period=10
-
-# default for supportsparse is false
-# accumulo.sink.ganglia.supportsparse=true
-
-# accumulo.sink.ganglia.slope=jvm.metrics.gcCount=zero,jvm.metrics.memHeapUsedM=both
-# accumulo.sink.ganglia.dmax=jvm.metrics.threadsBlocked=70,jvm.metrics.memHeapUsedM=40
-
-# Tag values to use for the ganglia prefix. If not defined no tags are used.
-# If '*' all tags are used. If specifying multiple tags separate them with
-# commas. Note that the last segment of the property name is the context name.
-#
-# accumulo.sink.ganglia.tagsForPrefix.jvm=ProcesName
-# accumulo.sink.ganglia.tagsForPrefix.tserver=
-# accumulo.sink.ganglia.tagsForPrefix.manager=
-# accumulo.sink.ganglia.tagsForPrefix.thrift=
-
-# Ganglia host(s)
-# accumulo.sink.ganglia.servers=yourgangliahost_1:8649,yourgangliahost_2:8649
diff --git a/assemble/pom.xml b/assemble/pom.xml
index 4ed88f8..19b8fb9 100644
--- a/assemble/pom.xml
+++ b/assemble/pom.xml
@@ -130,6 +130,11 @@
<optional>true</optional>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
<groupId>jakarta.activation</groupId>
<artifactId>jakarta.activation-api</artifactId>
<optional>true</optional>
@@ -451,6 +456,12 @@
<optional>true</optional>
</dependency>
<dependency>
+ <groupId>org.hdrhistogram</groupId>
+ <artifactId>HdrHistogram</artifactId>
+ <version>2.1.12</version>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
<groupId>org.hibernate.validator</groupId>
<artifactId>hibernate-validator</artifactId>
<optional>true</optional>
@@ -471,6 +482,18 @@
<optional>true</optional>
</dependency>
<dependency>
+ <groupId>org.latencyutils</groupId>
+ <artifactId>LatencyUtils</artifactId>
+ <version>2.0.3</version>
+ <optional>true</optional>
+ <exclusions>
+ <exclusion>
+ <groupId>org.hdrhistogram</groupId>
+ <artifactId>HdrHistogram</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<optional>true</optional>
diff --git a/assemble/src/main/assemblies/component.xml b/assemble/src/main/assemblies/component.xml
index 2033fe1..6093753 100644
--- a/assemble/src/main/assemblies/component.xml
+++ b/assemble/src/main/assemblies/component.xml
@@ -49,6 +49,7 @@
<include>commons-io:commons-io</include>
<include>commons-logging:commons-logging</include>
<include>com.sun.istack:istack-commons-runtime</include>
+ <include>io.micrometer:micrometer-core</include>
<include>jakarta.activation:jakarta.activation-api</include>
<include>jakarta.annotation:jakarta.annotation-api</include>
<include>jakarta.el:jakarta.el-api</include>
@@ -99,10 +100,12 @@
<include>org.glassfish.jersey.inject:jersey-hk2</include>
<include>org.glassfish.jersey.media:jersey-media-jaxb</include>
<include>org.glassfish.jersey.media:jersey-media-json-jackson</include>
+ <include>org.hdrhistogram:HdrHistogram</include>
<include>org.hibernate.validator:hibernate-validator</include>
<include>org.javassist:javassist</include>
<include>org.jboss.logging:jboss-logging</include>
<include>org.jline:jline</include>
+ <include>org.latencyutils:LatencyUtils</include>
<include>org.slf4j:slf4j-api</include>
<include>org.yaml:snakeyaml</include>
</includes>
diff --git a/core/pom.xml b/core/pom.xml
index d0c83a2..d0e88a1 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -70,6 +70,10 @@
<artifactId>commons-logging</artifactId>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-start</artifactId>
</dependency>
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index 0d21807..af0a487 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -267,7 +267,12 @@ public enum Property {
PropertyType.TIMEDURATION,
"The maximum amount of time that a Scanner should wait before retrying a failed RPC",
"1.7.3"),
-
+ GENERAL_MICROMETER_ENABLED("general.micrometer.enabled", "false", PropertyType.BOOLEAN,
+ "Enables metrics functionality using Micrometer", "2.1.0"),
+ GENERAL_MICROMETER_JVM_METRICS_ENABLED("general.micrometer.jvm.metrics.enabled", "false",
+ PropertyType.BOOLEAN, "Enables JVM metrics functionality using Micrometer", "2.1.0"),
+ GENERAL_MICROMETER_FACTORY("general.micrometer.factory", "", PropertyType.CLASSNAME,
+ "Name of class that implements MeterRegistryFactory", "2.1.0"),
// properties that are specific to manager server behavior
MANAGER_PREFIX("manager.", null, PropertyType.PREFIX,
"Properties in this category affect the behavior of the manager server. "
@@ -332,6 +337,7 @@ public enum Property {
MANAGER_WALOG_CLOSER_IMPLEMETATION("manager.walog.closer.implementation",
"org.apache.accumulo.server.manager.recovery.HadoopLogCloser", PropertyType.CLASSNAME,
"A class that implements a mechanism to steal write access to a write-ahead log", "1.5.0"),
+ @Deprecated
MANAGER_FATE_METRICS_ENABLED("manager.fate.metrics.enabled", "true", PropertyType.BOOLEAN,
"Enable reporting of FATE metrics in JMX (and logging with Hadoop Metrics2", "1.9.3"),
MANAGER_FATE_METRICS_MIN_UPDATE_INTERVAL("manager.fate.metrics.min.update.interval", "60s",
@@ -767,6 +773,7 @@ public enum Property {
+ " and possibly compacted. Legal values are: compact - which both flushes and compacts the"
+ " metadata; flush - which flushes only (compactions may be triggered if required); or none",
"1.10.0"),
+ @Deprecated
GC_METRICS_ENABLED("gc.metrics.enabled", "true", PropertyType.BOOLEAN,
"Enable detailed gc metrics reporting with hadoop metrics.", "1.10.0"),
diff --git a/test/src/main/java/org/apache/accumulo/test/metrics/MetricsTestSinkProperties.java b/core/src/main/java/org/apache/accumulo/core/metrics/MeterRegistryFactory.java
similarity index 67%
rename from test/src/main/java/org/apache/accumulo/test/metrics/MetricsTestSinkProperties.java
rename to core/src/main/java/org/apache/accumulo/core/metrics/MeterRegistryFactory.java
index 4e137f0..8f08e80 100644
--- a/test/src/main/java/org/apache/accumulo/test/metrics/MetricsTestSinkProperties.java
+++ b/core/src/main/java/org/apache/accumulo/core/metrics/MeterRegistryFactory.java
@@ -16,15 +16,10 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.test.metrics;
+package org.apache.accumulo.core.metrics;
-/**
- * common properties used with metrics configuration.
- */
-public class MetricsTestSinkProperties {
-
- public static final String METRICS_PROP_FILENAME = "hadoop-metrics2-accumulo.properties";
- public static final String ACC_GC_SINK_PREFIX = "accumulo.sink.file-gc";
- public static final String ACC_MANAGER_SINK_PREFIX = "accumulo.sink.file-manager";
+import io.micrometer.core.instrument.MeterRegistry;
+public interface MeterRegistryFactory {
+ MeterRegistry create();
}
diff --git a/core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java b/core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java
new file mode 100644
index 0000000..e332270
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java
@@ -0,0 +1,611 @@
+/*
+ * 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.accumulo.core.metrics;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.micrometer.core.instrument.MeterRegistry;
+
+/**
+ * Prior to 2.1.0 Accumulo used the <a href=
+ * "https://hadoop.apache.org/docs/current/api/org/apache/hadoop/metrics2/package-summary.html">Hadoop
+ * Metrics2</a> framework. In 2.1.0 Accumulo migrated away from the Metrics2 framework to
+ * <a href="https://micrometer.io/">Micrometer</a>. Micrometer suggests using a particular
+ * <a href="https://micrometer.io/docs/concepts#_naming_meters">naming convention</a> for the
+ * metrics. The table below contains a mapping of the old to new metric names.
+ *
+ * <table border="1">
+ * <caption>Summary of Metric Changes</caption> <!-- fate -->
+ * <tr>
+ * <th>Old Name</th>
+ * <th>Hadoop Metrics2 Type</th>
+ * <th>New Name</th>
+ * <th>Micrometer Type</th>
+ * <th>Notes</th>
+ * </tr>
+ * <tr>
+ * <td>currentFateOps</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TOTAL_IN_PROGRESS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>FateTxOpType_{name}</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TYPE_IN_PROGRESS}</td>
+ * <td>Gauge</td>
+ * <td>Previously there was a metric per operation type with the count of in-progress transactions
+ * of that type. Now there is one metric and the type is in the tag op.type</td>
+ * </tr>
+ * <tr>
+ * <td>totalFateOps</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_OPS_ACTIVITY}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>totalZkConnErrors</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_ERRORS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>FateTxState_NEW</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TX}</td>
+ * <td>Gauge</td>
+ * <td>The state is now in a tag: state=new</td>
+ * </tr>
+ * <tr>
+ * <td>FateTxState_IN_PROGRESS</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TX}</td>
+ * <td>Gauge</td>
+ * <td>The state is now in a tag: state=in.progress</td>
+ * </tr>
+ * <tr>
+ * <td>FateTxState_FAILED_IN_PROGRESS</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TX}</td>
+ * <td>Gauge</td>
+ * <td>The state is now in a tag: state=failed.in.progress</td>
+ * </tr>
+ * <tr>
+ * <td>FateTxState_FAILED</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TX}</td>
+ * <td>Gauge</td>
+ * <td>The state is now in a tag: state=failed</td>
+ * </tr>
+ * <tr>
+ * <td>FateTxState_SUCCESSFUL</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TX}</td>
+ * <td>Gauge</td>
+ * <td>The state is now in a tag: state=successful</td>
+ * </tr>
+ * <tr>
+ * <td>FateTxState_UNKNOWN</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_FATE_TX}</td>
+ * <td>Gauge</td>
+ * <td>The state is now in a tag: state=unknown</td>
+ * </tr>
+ * <!-- garbage collection -->
+ * <tr>
+ * <td>AccGcStarted</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_STARTED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcFinished</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_FINISHED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcCandidates</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_CANDIDATES}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcInUse</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_IN_USE}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcDeleted</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_DELETED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcErrors</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_ERRORS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcWalStarted</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_WAL_STARTED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcWalFinished</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_WAL_FINISHED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcWalCandidates</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_WAL_CANDIDATES}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcWalInUse</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_WAL_IN_USE}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcWalDeleted</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_WAL_DELETED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcWalErrors</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_WAL_ERRORS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcPosOpDuration</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_POST_OP_DURATION}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>AccGcRunCycleCount</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_GC_RUN_CYCLE}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <!-- tablet server -->
+ * <tr>
+ * <td>entries</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_ENTRIES}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>entriesInMem</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_MEM_ENTRIES}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>activeMajCs</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_MAJC_RUNNING}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>queuedMajCs</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_MAJC_QUEUED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>activeMinCs</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_MINC_RUNNING}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>queuedMinCs</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_MINC_QUEUED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>totalMinCs</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_MINC_TOTAL}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>onlineTablets</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_TABLETS_ONLINE}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>openingTablets</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_TABLETS_OPENING}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>unopenedTablets</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_TABLETS_UNOPENED}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>filesPerTablet</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_TABLETS_FILES}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>queries</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_QUERIES}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>scannedRate</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_SCANNED_ENTRIES}</td>
+ * <td>Gauge</td>
+ * <td>Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be
+ * derived</td>
+ * </tr>
+ * <tr>
+ * <td>queryRate</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_SCAN_RESULTS}</td>
+ * <td>Gauge</td>
+ * <td>Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be
+ * derived</td>
+ * </tr>
+ * <tr>
+ * <td>queryByteRate</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_SCAN_RESULTS_BYTES}</td>
+ * <td>Gauge</td>
+ * <td>Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be
+ * derived</td>
+ * </tr>
+ * <tr>
+ * <td>ingestRate</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_INGEST_MUTATIONS}</td>
+ * <td>Gauge</td>
+ * <td>Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be
+ * derived</td>
+ * </tr>
+ * <tr>
+ * <td>ingestByteRate</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_INGEST_BYTES}</td>
+ * <td>Gauge</td>
+ * <td>Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be
+ * derived</td>
+ * </tr>
+ * <tr>
+ * <td>holdTime</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_TSERVER_HOLD}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <!-- scans -->
+ * <tr>
+ * <td>scan</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_SCAN}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>result</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_SCAN_RESULTS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>yield</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_SCAN_YIELDS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <!-- major compactions -->
+ * <tr>
+ * <td>{i|e}_{compactionServiceName}_{executor_name}_queued</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_MAJC_QUEUED}</td>
+ * <td>Gauge</td>
+ * <td>The compaction service information is in a tag:
+ * id={i|e}_{compactionServiceName}_{executor_name}</td>
+ * </tr>
+ * <tr>
+ * <td>{i|e}_{compactionServiceName}_{executor_name}_running</td>
+ * <td>Gauge</td>
+ * <td>{@link #METRICS_MAJC_RUNNING}</td>
+ * <td>Gauge</td>
+ * <td>The compaction service information is in a tag:
+ * id={i|e}_{compactionServiceName}_{executor_name}</td>
+ * </tr>
+ * <!-- minor compactions -->
+ * <tr>
+ * <td>Queue</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_MINC_QUEUED}</td>
+ * <td>Timer</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>Minc</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_MINC_RUNNING}</td>
+ * <td>Timer</td>
+ * <td></td>
+ * </tr>
+ * <!-- replication -->
+ * <tr>
+ * <td>ReplicationQueue</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_REPLICATION_QUEUE}</td>
+ * <td>Timer</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>ReplicationQueue10m</td>
+ * <td>Quantiles</td>
+ * <td>N/A</td>
+ * <td>N/A</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>filesPendingReplication</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_REPLICATION_PENDING_FILES}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>maxReplicationThreads</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_REPLICATION_THREADS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>numPeers</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_REPLICATION_PEERS}</td>
+ * <td>Gauge</td>
+ * <td></td>
+ * </tr>
+ * <!-- Updates (ingest) -->
+ * <tr>
+ * <td>permissionErrors</td>
+ * <td>Counter</td>
+ * <td>{@link #METRICS_UPDATE_ERRORS}</td>
+ * <td>Gauge</td>
+ * <td>Type is stored in tag: type=permission</td>
+ * </tr>
+ * <tr>
+ * <td>unknownTabletErrors</td>
+ * <td>Counter</td>
+ * <td>{@link #METRICS_UPDATE_ERRORS}</td>
+ * <td>Gauge</td>
+ * <td>Type is stored in tag: type=unknown.tablet</td>
+ * </tr>
+ * <tr>
+ * <td>constraintViolations</td>
+ * <td>Counter</td>
+ * <td>{@link #METRICS_UPDATE_ERRORS}</td>
+ * <td>Gauge</td>
+ * <td>Type is stored in tag: type=constraint.violation</td>
+ * </tr>
+ * <tr>
+ * <td>commitPrep</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_UPDATE_COMMIT_PREP}</td>
+ * <td>Timer</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>commitTime</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_UPDATE_COMMIT}</td>
+ * <td>Timer</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>waLogWriteTime</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_UPDATE_WALOG_WRITE}</td>
+ * <td>Timer</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>mutationArraysSize</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_UPDATE_MUTATION_ARRAY_SIZE}</td>
+ * <td>Distribution Summary</td>
+ * <td></td>
+ * </tr>
+ * <!-- Thrift -->
+ * <tr>
+ * <td>idle</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_THRIFT_IDLE}</td>
+ * <td>Distribution Summary</td>
+ * <td></td>
+ * </tr>
+ * <tr>
+ * <td>execute</td>
+ * <td>Stat</td>
+ * <td>{@link #METRICS_THRIFT_EXECUTE}</td>
+ * <td>Distribution Summary</td>
+ * <td></td>
+ * </tr>
+ * </table>
+ *
+ * @since 2.1.0
+ */
+public interface MetricsProducer {
+
+ Logger LOG = LoggerFactory.getLogger(MetricsProducer.class);
+
+ String METRICS_FATE_PREFIX = "accumulo.fate.";
+ String METRICS_FATE_TYPE_IN_PROGRESS = METRICS_FATE_PREFIX + "ops.in_progress_by_type";
+ String METRICS_FATE_TOTAL_IN_PROGRESS = METRICS_FATE_PREFIX + "ops.in_progress";
+ String METRICS_FATE_OPS_ACTIVITY = METRICS_FATE_PREFIX + "ops.activity";
+ String METRICS_FATE_ERRORS = METRICS_FATE_PREFIX + "errors";
+ String METRICS_FATE_TX = METRICS_FATE_PREFIX + "tx";
+
+ String METRICS_GC_PREFIX = "accumulo.gc.";
+ String METRICS_GC_STARTED = METRICS_GC_PREFIX + "started";
+ String METRICS_GC_FINISHED = METRICS_GC_PREFIX + "finished";
+ String METRICS_GC_CANDIDATES = METRICS_GC_PREFIX + "candidates";
+ String METRICS_GC_IN_USE = METRICS_GC_PREFIX + "in.use";
+ String METRICS_GC_DELETED = METRICS_GC_PREFIX + "deleted";
+ String METRICS_GC_ERRORS = METRICS_GC_PREFIX + "errors";
+ String METRICS_GC_WAL_STARTED = METRICS_GC_PREFIX + "wal.started";
+ String METRICS_GC_WAL_FINISHED = METRICS_GC_PREFIX + "wal.finished";
+ String METRICS_GC_WAL_CANDIDATES = METRICS_GC_PREFIX + "wal.candidates";
+ String METRICS_GC_WAL_IN_USE = METRICS_GC_PREFIX + "wal.in.use";
+ String METRICS_GC_WAL_DELETED = METRICS_GC_PREFIX + "wal.deleted";
+ String METRICS_GC_WAL_ERRORS = METRICS_GC_PREFIX + "wal.errors";
+ String METRICS_GC_POST_OP_DURATION = METRICS_GC_PREFIX + "post.op.duration";
+ String METRICS_GC_RUN_CYCLE = METRICS_GC_PREFIX + "run.cycle";
+
+ String METRICS_MAJC_PREFIX = "accumulo.tserver.compactions.majc.";
+ String METRICS_MAJC_QUEUED = METRICS_MAJC_PREFIX + "queued";
+ String METRICS_MAJC_RUNNING = METRICS_MAJC_PREFIX + "running";
+
+ String METRICS_MINC_PREFIX = "accumulo.tserver.compactions.minc.";
+ String METRICS_MINC_QUEUED = METRICS_MINC_PREFIX + "queued";
+ String METRICS_MINC_RUNNING = METRICS_MINC_PREFIX + "running";
+
+ String METRICS_REPLICATION_PREFIX = "accumulo.replication.";
+ String METRICS_REPLICATION_QUEUE = METRICS_REPLICATION_PREFIX + "queue";
+ String METRICS_REPLICATION_PENDING_FILES = METRICS_REPLICATION_PREFIX + "files.pending";
+ String METRICS_REPLICATION_PEERS = METRICS_REPLICATION_PREFIX + "peers";
+ String METRICS_REPLICATION_THREADS = METRICS_REPLICATION_PREFIX + "threads";
+
+ String METRICS_SCAN = "accumulo.tserver.scans";
+ String METRICS_SCAN_RESULTS = METRICS_SCAN + ".result";
+ String METRICS_SCAN_YIELDS = METRICS_SCAN + ".yields";
+
+ String METRICS_TSERVER_PREFIX = "accumulo.tserver.";
+ String METRICS_TSERVER_ENTRIES = METRICS_TSERVER_PREFIX + "entries";
+ String METRICS_TSERVER_MEM_ENTRIES = METRICS_TSERVER_PREFIX + "entries.mem";
+ String METRICS_TSERVER_MAJC_QUEUED = METRICS_TSERVER_PREFIX + "majc.queued";
+ String METRICS_TSERVER_MAJC_RUNNING = METRICS_TSERVER_PREFIX + "majc.running";
+ String METRICS_TSERVER_MINC_QUEUED = METRICS_TSERVER_PREFIX + "minc.queued";
+ String METRICS_TSERVER_MINC_RUNNING = METRICS_TSERVER_PREFIX + "minc.running";
+ String METRICS_TSERVER_MINC_TOTAL = METRICS_TSERVER_PREFIX + "minc.total";
+ String METRICS_TSERVER_TABLETS_ONLINE = METRICS_TSERVER_PREFIX + "tablets.online";
+ String METRICS_TSERVER_TABLETS_OPENING = METRICS_TSERVER_PREFIX + "tablets.opening";
+ String METRICS_TSERVER_TABLETS_UNOPENED = METRICS_TSERVER_PREFIX + "tablets.unopened";
+ String METRICS_TSERVER_QUERIES = METRICS_TSERVER_PREFIX + "queries";
+ String METRICS_TSERVER_TABLETS_FILES = METRICS_TSERVER_PREFIX + "tablets.files";
+ String METRICS_TSERVER_HOLD = METRICS_TSERVER_PREFIX + "hold";
+ String METRICS_TSERVER_INGEST_MUTATIONS = METRICS_TSERVER_PREFIX + "ingest.mutations";
+ String METRICS_TSERVER_INGEST_BYTES = METRICS_TSERVER_PREFIX + "ingest.bytes";
+ String METRICS_TSERVER_SCAN_RESULTS = METRICS_TSERVER_PREFIX + "scan.results";
+ String METRICS_TSERVER_SCAN_RESULTS_BYTES = METRICS_TSERVER_PREFIX + "scan.results.bytes";
+ String METRICS_TSERVER_SCANNED_ENTRIES = METRICS_TSERVER_PREFIX + "scan.scanned.entries";
+
+ String METRICS_THRIFT_PREFIX = "accumulo.thrift.";
+ String METRICS_THRIFT_EXECUTE = METRICS_THRIFT_PREFIX + "execute";
+ String METRICS_THRIFT_IDLE = METRICS_THRIFT_PREFIX + "idle";
+
+ String METRICS_UPDATE_PREFIX = "accumulo.tserver.updates.";
+ String METRICS_UPDATE_ERRORS = METRICS_UPDATE_PREFIX + "error";
+ String METRICS_UPDATE_COMMIT = METRICS_UPDATE_PREFIX + "commit";
+ String METRICS_UPDATE_COMMIT_PREP = METRICS_UPDATE_COMMIT + ".prep";
+ String METRICS_UPDATE_WALOG_WRITE = METRICS_UPDATE_PREFIX + "walog.write";
+ String METRICS_UPDATE_MUTATION_ARRAY_SIZE = METRICS_UPDATE_PREFIX + "mutation.arrays.size";
+
+ /**
+ * Build Micrometer Meter objects and register them with the registry
+ */
+ void registerMetrics(MeterRegistry registry);
+
+ /**
+ * Returns a mapping of metric field value to metric field name.
+ *
+ * @return map of field names to variable names.
+ */
+ default Map<String,String> getMetricFields() {
+ Map<String,String> fields = new HashMap<>();
+ for (Field f : MetricsProducer.class.getDeclaredFields()) {
+ if (Modifier.isStatic(f.getModifiers()) && f.getType().equals(String.class)
+ && !f.getName().contains("PREFIX")) {
+ try {
+
+ fields.put((String) f.get(MetricsProducer.class), f.getName());
+ } catch (IllegalArgumentException | IllegalAccessException e) {
+ // this shouldn't happen, but lets log it anyway
+ LOG.error("Error getting metric value for field: " + f.getName());
+ }
+ }
+ }
+ return fields;
+ }
+}
diff --git a/core/src/main/java/org/apache/accumulo/core/metrics/MetricsUtil.java b/core/src/main/java/org/apache/accumulo/core/metrics/MetricsUtil.java
new file mode 100644
index 0000000..c887923
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/metrics/MetricsUtil.java
@@ -0,0 +1,136 @@
+/*
+ * 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.accumulo.core.metrics;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.accumulo.core.classloader.ClassLoaderUtil;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.util.HostAndPort;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.binder.jvm.ClassLoaderMetrics;
+import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics;
+import io.micrometer.core.instrument.binder.system.ProcessorMetrics;
+import io.micrometer.core.instrument.config.MeterFilter;
+import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
+
+public class MetricsUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MetricsUtil.class);
+
+ private static JvmGcMetrics gc;
+ private static List<Tag> commonTags;
+
+ public static void initializeMetrics(final AccumuloConfiguration conf, final String appName,
+ final HostAndPort address) throws Exception {
+ initializeMetrics(conf.getBoolean(Property.GENERAL_MICROMETER_ENABLED),
+ conf.getBoolean(Property.GENERAL_MICROMETER_JVM_METRICS_ENABLED),
+ conf.get(Property.GENERAL_MICROMETER_FACTORY), appName, address);
+ }
+
+ private static void initializeMetrics(boolean enabled, boolean jvmMetricsEnabled,
+ String factoryClass, String appName, HostAndPort address) throws Exception {
+
+ LOG.info("initializing metrics, enabled:{}, class:{}", enabled, factoryClass);
+
+ if (enabled && factoryClass != null && !factoryClass.isEmpty()) {
+
+ String processName = appName;
+ String serviceInstance = System.getProperty("accumulo.metrics.service.instance", "");
+ if (!serviceInstance.isBlank()) {
+ processName += serviceInstance;
+ }
+
+ List<Tag> tags = new ArrayList<>();
+ tags.add(Tag.of("process.name", processName));
+
+ if (address != null) {
+ tags.add(Tag.of("Address", address.toString()));
+ }
+ commonTags = Collections.unmodifiableList(tags);
+
+ // Configure replication metrics to display percentiles and change its expiry to 10 mins
+ MeterFilter replicationFilter = new MeterFilter() {
+ @Override
+ public DistributionStatisticConfig configure(Meter.Id id,
+ DistributionStatisticConfig config) {
+ if (id.getName().equals("replicationQueue")) {
+ return DistributionStatisticConfig.builder().percentiles(0.5, 0.75, 0.9, 0.95, 0.99)
+ .expiry(Duration.ofMinutes(10)).build().merge(config);
+ }
+ return config;
+ }
+ };
+
+ Class<? extends MeterRegistryFactory> clazz =
+ ClassLoaderUtil.loadClass(factoryClass, MeterRegistryFactory.class);
+ MeterRegistryFactory factory = clazz.getDeclaredConstructor().newInstance();
+
+ MeterRegistry registry = factory.create();
+ registry.config().commonTags(commonTags);
+ registry.config().meterFilter(replicationFilter);
+ Metrics.addRegistry(registry);
+
+ if (jvmMetricsEnabled) {
+ new ClassLoaderMetrics(commonTags).bindTo(Metrics.globalRegistry);
+ new JvmMemoryMetrics(commonTags).bindTo(Metrics.globalRegistry);
+ gc = new JvmGcMetrics(commonTags);
+ gc.bindTo(Metrics.globalRegistry);
+ new ProcessorMetrics(commonTags).bindTo(Metrics.globalRegistry);
+ new JvmThreadMetrics(commonTags).bindTo(Metrics.globalRegistry);
+ }
+ }
+ }
+
+ public static void initializeProducers(MetricsProducer... producer) {
+ for (MetricsProducer p : producer) {
+ p.registerMetrics(Metrics.globalRegistry);
+ LOG.info("Metric producer {} initialize", p.getClass().getSimpleName());
+ }
+ }
+
+ public static void addExecutorServiceMetrics(ExecutorService executor, String name) {
+ new ExecutorServiceMetrics(executor, name, commonTags).bindTo(Metrics.globalRegistry);
+ }
+
+ public static List<Tag> getCommonTags() {
+ return commonTags;
+ }
+
+ public static void close() {
+ if (gc != null) {
+ gc.close();
+ }
+ }
+
+}
diff --git a/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPools.java b/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPools.java
index 4633903..4f082df 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPools.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPools.java
@@ -30,6 +30,7 @@ import java.util.function.IntSupplier;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -186,6 +187,7 @@ public class ThreadPools {
if (timeOut > 0) {
result.allowCoreThreadTimeOut(true);
}
+ MetricsUtil.addExecutorServiceMetrics(result, name);
return result;
}
@@ -206,12 +208,15 @@ public class ThreadPools {
public static ScheduledThreadPoolExecutor createScheduledExecutorService(int numThreads,
final String name, OptionalInt priority, boolean enableTracing) {
+ ScheduledThreadPoolExecutor result = null;
if (enableTracing) {
- return new TracingScheduledThreadPoolExecutor(numThreads,
+ result = new TracingScheduledThreadPoolExecutor(numThreads,
new NamedThreadFactory(name, priority));
} else {
- return new ScheduledThreadPoolExecutor(numThreads, new NamedThreadFactory(name, priority));
+ result = new ScheduledThreadPoolExecutor(numThreads, new NamedThreadFactory(name, priority));
}
+ MetricsUtil.addExecutorServiceMetrics(result, name);
+ return result;
}
}
diff --git a/pom.xml b/pom.xml
index 93265d8..02adb09 100644
--- a/pom.xml
+++ b/pom.xml
@@ -140,6 +140,7 @@
<maven.compiler.target>11</maven.compiler.target>
<!-- surefire/failsafe plugin option -->
<maven.test.redirectTestOutputToFile>true</maven.test.redirectTestOutputToFile>
+ <micrometer.metrics.version>1.7.5</micrometer.metrics.version>
<powermock.version>2.0.9</powermock.version>
<!-- timestamp for reproducible outputs, updated on release by the release plugin -->
<project.build.outputTimestamp>2020-12-17T22:06:50Z</project.build.outputTimestamp>
@@ -579,6 +580,13 @@
<scope>import</scope>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-bom</artifactId>
+ <version>${micrometer.metrics.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ <dependency>
<groupId>jakarta.platform</groupId>
<artifactId>jakarta.jakartaee-bom</artifactId>
<version>9.1.0-RC1</version>
diff --git a/server/base/pom.xml b/server/base/pom.xml
index eca7744..c8f3e2c 100644
--- a/server/base/pom.xml
+++ b/server/base/pom.xml
@@ -57,6 +57,10 @@
<artifactId>commons-codec</artifactId>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-core</artifactId>
</dependency>
diff --git a/server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java b/server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java
index 49e03b0..7fa05fe 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java
@@ -24,20 +24,18 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.classloader.ClassLoaderUtil;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.trace.TraceUtil;
-import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.security.SecurityUtil;
-import org.apache.hadoop.metrics2.MetricsSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public abstract class AbstractServer implements AutoCloseable, Runnable {
private final ServerContext context;
- private final String applicationName;
+ protected final String applicationName;
private final String hostname;
private final Logger log;
- private final MetricsSystem metricsSystem;
protected AbstractServer(String appName, ServerOpts opts, String[] args) {
this.log = LoggerFactory.getLogger(getClass().getName());
@@ -51,7 +49,6 @@ public abstract class AbstractServer implements AutoCloseable, Runnable {
log.info("Instance " + context.getInstanceID());
context.init(appName);
ClassLoaderUtil.initContextFactory(context.getConfiguration());
- this.metricsSystem = Metrics.initSystem(getClass().getSimpleName());
TraceUtil.enableServerTraces(hostname, appName, context.getConfiguration());
if (context.getSaslParams() != null) {
// Server-side "client" check to make sure we're logged in as a user we expect to be
@@ -94,13 +91,10 @@ public abstract class AbstractServer implements AutoCloseable, Runnable {
return getContext().getConfiguration();
}
- public MetricsSystem getMetricsSystem() {
- return metricsSystem;
- }
-
@Override
public void close() {
TraceUtil.disable();
+ MetricsUtil.close();
}
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/metrics/Metrics.java b/server/base/src/main/java/org/apache/accumulo/server/metrics/Metrics.java
deleted file mode 100644
index 7739243..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/metrics/Metrics.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.server.metrics;
-
-import org.apache.hadoop.metrics2.MetricsCollector;
-import org.apache.hadoop.metrics2.MetricsRecordBuilder;
-import org.apache.hadoop.metrics2.MetricsSource;
-import org.apache.hadoop.metrics2.MetricsSystem;
-import org.apache.hadoop.metrics2.impl.MsInfo;
-import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
-import org.apache.hadoop.metrics2.lib.Interns;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.source.JvmMetrics;
-import org.apache.hadoop.metrics2.source.JvmMetricsInfo;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Accumulo will search for a file named hadoop-metrics-accumulo.properties on the Accumulo
- * classpath to configute the hadoop metrics2 system. The hadoop metrics system publishes to jmx and
- * can be configured, via a configuration file, to publish to other metric collection systems
- * (files,...)
- * <p>
- * A note on naming: The naming for jmx vs the hadoop metrics systems are slightly different. Hadoop
- * metrics records will start with CONTEXT.RECORD, for example, accgc.AccGcCycleMetrics. The context
- * parameter value is also used by the configuration file for sink configuration.
- * <p>
- * In JMX, the hierarchy is: Hadoop..Accumulo..[jmxName]..[processName]..attributes..[name]
- * <p>
- * For jvm metrics, the hierarchy is Hadoop..Accumulo..JvmMetrics..attributes..[name]
- */
-public abstract class Metrics implements MetricsSource {
-
- private static String processName = "Unknown";
-
- public static MetricsSystem initSystem(String serviceName) {
- processName = serviceName;
- String serviceInstance = System.getProperty("accumulo.metrics.service.instance", "");
- if (!serviceInstance.isBlank()) {
- processName += serviceInstance;
- }
-
- // create a new one if needed
- MetricsSystem ms = DefaultMetricsSystem.initialize("Accumulo");
- if (ms.getSource(JvmMetricsInfo.JvmMetrics.name()) == null) {
- JvmMetrics.create(processName, "", ms);
- }
- return ms;
- }
-
- private final String name;
- private final String description;
- private final String context;
- private final String record;
- private final MetricsRegistry registry;
- private final Logger log;
-
- protected Metrics(String name, String description, String context, String record) {
- this.log = LoggerFactory.getLogger(getClass());
- this.name = name;
- this.description = description;
- this.context = context;
- this.record = record;
- this.registry = new MetricsRegistry(Interns.info(name, description));
- this.registry.tag(MsInfo.ProcessName, processName);
- }
-
- public void register(MetricsSystem system) {
- system.register(name, description, this);
- }
-
- protected final MetricsRegistry getRegistry() {
- return registry;
- }
-
- /**
- * Runs prior to {@link #getMetrics(MetricsCollector, boolean)} in order to prepare metrics in the
- * {@link MetricsRegistry} to be published.
- */
- protected void prepareMetrics() {}
-
- /**
- * Append any additional metrics directly to the builder when
- * {@link #getMetrics(MetricsCollector, boolean)} is called, after any metrics in the
- * {@link MetricsRegistry} have already been added.
- */
- protected void getMoreMetrics(MetricsRecordBuilder builder, boolean all) {}
-
- @Override
- public final void getMetrics(MetricsCollector collector, boolean all) {
- log.trace("getMetrics called with collector: {} (all: {})", collector, all);
- prepareMetrics();
- MetricsRecordBuilder builder = collector.addRecord(record).setContext(context);
- registry.snapshot(builder, all);
- getMoreMetrics(builder, all);
- }
-
-}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetrics.java b/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetrics.java
index 5f01b134..6712ded 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetrics.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetrics.java
@@ -18,19 +18,37 @@
*/
package org.apache.accumulo.server.metrics;
-public class ThriftMetrics extends Metrics {
+import org.apache.accumulo.core.metrics.MetricsProducer;
+
+import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.MeterRegistry;
+
+public class ThriftMetrics implements MetricsProducer {
+
+ private final String serverName;
+ private final String threadName;
+ private DistributionSummary idle;
+ private DistributionSummary execute;
public ThriftMetrics(String serverName, String threadName) {
- super("Thrift,sub=" + serverName, "Thrift Server Metrics - " + serverName + " " + threadName,
- "thrift", serverName);
+ this.serverName = serverName;
+ this.threadName = threadName;
}
public void addIdle(long time) {
- getRegistry().add("idle", time);
+ idle.record(time);
}
public void addExecute(long time) {
- getRegistry().add("execute", time);
+ execute.record(time);
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ idle = DistributionSummary.builder(METRICS_THRIFT_IDLE).baseUnit("ms")
+ .tags("server", serverName, "thread", threadName).register(registry);
+ execute = DistributionSummary.builder(METRICS_THRIFT_EXECUTE).baseUnit("ms")
+ .tags("server", serverName, "thread", threadName).register(registry);
}
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java
index ed32d3b..5e54bfe 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java
@@ -52,7 +52,6 @@ import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.server.ServerContext;
-import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.security.SaslRpcServer;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.thrift.TProcessor;
@@ -136,11 +135,11 @@ public class TServerUtils {
* @throws UnknownHostException
* when we don't know our own address
*/
- public static ServerAddress startServer(MetricsSystem metricsSystem, ServerContext context,
- String hostname, Property portHintProperty, TProcessor processor, String serverName,
- String threadName, Property portSearchProperty, Property minThreadProperty,
- Property threadTimeOutProperty, Property timeBetweenThreadChecksProperty,
- Property maxMessageSizeProperty) throws UnknownHostException {
+ public static ServerAddress startServer(ServerContext context, String hostname,
+ Property portHintProperty, TProcessor processor, String serverName, String threadName,
+ Property portSearchProperty, Property minThreadProperty, Property threadTimeOutProperty,
+ Property timeBetweenThreadChecksProperty, Property maxMessageSizeProperty)
+ throws UnknownHostException {
final AccumuloConfiguration config = context.getConfiguration();
final IntStream portHint = config.getPortStream(portHintProperty);
@@ -179,8 +178,7 @@ public class TServerUtils {
// create the TimedProcessor outside the port search loop so we don't try to
// register the same
// metrics mbean more than once
- TimedProcessor timedProcessor =
- new TimedProcessor(metricsSystem, config, processor, serverName, threadName);
+ TimedProcessor timedProcessor = new TimedProcessor(config, processor, serverName, threadName);
HostAndPort[] addresses = getHostAndPorts(hostname, portHint);
try {
@@ -575,21 +573,19 @@ public class TServerUtils {
return new ServerAddress(server, address);
}
- public static ServerAddress startTServer(MetricsSystem metricsSystem,
- final AccumuloConfiguration conf, ThriftServerType serverType, TProcessor processor,
- String serverName, String threadName, int numThreads, long threadTimeOut,
- long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams,
- SaslServerConnectionParams saslParams, long serverSocketTimeout, HostAndPort... addresses)
- throws TTransportException {
+ public static ServerAddress startTServer(final AccumuloConfiguration conf,
+ ThriftServerType serverType, TProcessor processor, String serverName, String threadName,
+ int numThreads, long threadTimeOut, long timeBetweenThreadChecks, long maxMessageSize,
+ SslConnectionParams sslParams, SaslServerConnectionParams saslParams,
+ long serverSocketTimeout, HostAndPort... addresses) throws TTransportException {
if (serverType == ThriftServerType.SASL) {
processor = updateSaslProcessor(serverType, processor);
}
- return startTServer(serverType,
- new TimedProcessor(metricsSystem, conf, processor, serverName, threadName), serverName,
- threadName, numThreads, threadTimeOut, conf, timeBetweenThreadChecks, maxMessageSize,
- sslParams, saslParams, serverSocketTimeout, addresses);
+ return startTServer(serverType, new TimedProcessor(conf, processor, serverName, threadName),
+ serverName, threadName, numThreads, threadTimeOut, conf, timeBetweenThreadChecks,
+ maxMessageSize, sslParams, saslParams, serverSocketTimeout, addresses);
}
/**
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java
index 453d9b8..83c2a72 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java
@@ -19,45 +19,36 @@
package org.apache.accumulo.server.rpc;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.server.metrics.ThriftMetrics;
-import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.thrift.TException;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TProtocol;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* A {@link TProcessor} which tracks the duration of an RPC and adds it to the metrics subsystem.
*/
public class TimedProcessor implements TProcessor {
- private static final Logger log = LoggerFactory.getLogger(TimedProcessor.class);
private final TProcessor other;
private final ThriftMetrics thriftMetrics;
private long idleStart = 0;
- public TimedProcessor(MetricsSystem metricsSystem, AccumuloConfiguration conf, TProcessor next,
- String serverName, String threadName) {
- this(metricsSystem, next, serverName, threadName);
+ public TimedProcessor(AccumuloConfiguration conf, TProcessor next, String serverName,
+ String threadName) {
+ this(next, serverName, threadName);
}
- public TimedProcessor(MetricsSystem metricsSystem, TProcessor next, String serverName,
- String threadName) {
+ public TimedProcessor(TProcessor next, String serverName, String threadName) {
this.other = next;
thriftMetrics = new ThriftMetrics(serverName, threadName);
- try {
- thriftMetrics.register(metricsSystem);
- } catch (Exception e) {
- log.error("Exception registering MBean with MBean Server", e);
- }
+ MetricsUtil.initializeProducers(thriftMetrics);
idleStart = System.currentTimeMillis();
}
@Override
public void process(TProtocol in, TProtocol out) throws TException {
- long now = 0;
- now = System.currentTimeMillis();
+ long now = System.currentTimeMillis();
thriftMetrics.addIdle(now - idleStart);
try {
other.process(in, out);
diff --git a/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java b/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java
index ba4c300..877bdcc 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java
@@ -44,7 +44,6 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.client.ClientServiceHandler;
-import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.rpc.ServerAddress;
import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.rpc.ThriftServerType;
@@ -350,10 +349,9 @@ public class TServerUtilsTest {
// misconfiguration)
String hostname = "localhost";
- return TServerUtils.startServer(Metrics.initSystem(getClass().getSimpleName()), context,
- hostname, Property.TSERV_CLIENTPORT, processor, "TServerUtilsTest",
- "TServerUtilsTestThread", Property.TSERV_PORTSEARCH, Property.TSERV_MINTHREADS,
- Property.TSERV_MINTHREADS_TIMEOUT, Property.TSERV_THREADCHECK,
+ return TServerUtils.startServer(context, hostname, Property.TSERV_CLIENTPORT, processor,
+ "TServerUtilsTest", "TServerUtilsTestThread", Property.TSERV_PORTSEARCH,
+ Property.TSERV_MINTHREADS, Property.TSERV_MINTHREADS_TIMEOUT, Property.TSERV_THREADCHECK,
Property.GENERAL_MAX_MESSAGE_SIZE);
}
diff --git a/server/base/src/test/resources/log4j2-test.properties b/server/base/src/test/resources/log4j2-test.properties
index d0ff316..dcf46b3 100644
--- a/server/base/src/test/resources/log4j2-test.properties
+++ b/server/base/src/test/resources/log4j2-test.properties
@@ -32,4 +32,3 @@ logger.01.level = error
rootLogger.level = info
rootLogger.appenderRef.console.ref = STDOUT
-
diff --git a/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java b/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java
index 14abd6d..05e8e7d 100644
--- a/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java
+++ b/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java
@@ -46,6 +46,7 @@ import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
import org.apache.accumulo.core.metadata.TServerInstance;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
import org.apache.accumulo.core.tabletserver.thrift.TCompactionQueueSummary;
@@ -217,7 +218,7 @@ public class CompactionCoordinator extends AbstractServer
Property maxMessageSizeProperty =
(aconf.get(Property.COMPACTION_COORDINATOR_MAX_MESSAGE_SIZE) != null
? Property.COMPACTION_COORDINATOR_MAX_MESSAGE_SIZE : Property.GENERAL_MAX_MESSAGE_SIZE);
- ServerAddress sp = TServerUtils.startServer(getMetricsSystem(), getContext(), getHostname(),
+ ServerAddress sp = TServerUtils.startServer(getContext(), getHostname(),
Property.COMPACTION_COORDINATOR_CLIENTPORT, processor, this.getClass().getSimpleName(),
"Thrift Client Server", Property.COMPACTION_COORDINATOR_THRIFTCLIENT_PORTSEARCH,
Property.COMPACTION_COORDINATOR_MINTHREADS,
@@ -244,6 +245,13 @@ public class CompactionCoordinator extends AbstractServer
throw new IllegalStateException("Exception getting Coordinator lock", e);
}
+ try {
+ MetricsUtil.initializeMetrics(getContext().getConfiguration(), this.applicationName,
+ clientAddress);
+ } catch (Exception e1) {
+ LOG.error("Error initializing metrics, metrics will not be emitted.", e1);
+ }
+
// On a re-start of the coordinator it's possible that external compactions are in-progress.
// Attempt to get the running compactions on the compactors and then resolve which tserver
// the external compaction came from to re-populate the RUNNING collection.
diff --git a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
index ec25495..ec5dc08 100644
--- a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
+++ b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
@@ -58,6 +58,7 @@ import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
import org.apache.accumulo.core.tabletserver.thrift.ActiveCompaction;
@@ -315,7 +316,7 @@ public class Compactor extends AbstractServer implements CompactorService.Iface
final CompactorService.Processor<Iface> processor = new CompactorService.Processor<>(rpcProxy);
Property maxMessageSizeProperty = (aconf.get(Property.COMPACTOR_MAX_MESSAGE_SIZE) != null
? Property.COMPACTOR_MAX_MESSAGE_SIZE : Property.GENERAL_MAX_MESSAGE_SIZE);
- ServerAddress sp = TServerUtils.startServer(getMetricsSystem(), getContext(), getHostname(),
+ ServerAddress sp = TServerUtils.startServer(getContext(), getHostname(),
Property.COMPACTOR_CLIENTPORT, processor, this.getClass().getSimpleName(),
"Thrift Client Server", Property.COMPACTOR_PORTSEARCH, Property.COMPACTOR_MINTHREADS,
Property.COMPACTOR_MINTHREADS_TIMEOUT, Property.COMPACTOR_THREADCHECK,
@@ -623,6 +624,13 @@ public class Compactor extends AbstractServer implements CompactorService.Iface
throw new RuntimeException("Error registering compactor in ZooKeeper", e);
}
+ try {
+ MetricsUtil.initializeMetrics(getContext().getConfiguration(), this.applicationName,
+ clientAddress);
+ } catch (Exception e1) {
+ LOG.error("Error initializing metrics, metrics will not be emitted.", e1);
+ }
+
LOG.info("Compactor started, waiting for work");
try {
diff --git a/server/gc/pom.xml b/server/gc/pom.xml
index 22c2bca..f052148 100644
--- a/server/gc/pom.xml
+++ b/server/gc/pom.xml
@@ -45,6 +45,10 @@
<artifactId>protobuf-java</artifactId>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-core</artifactId>
</dependency>
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index 2c67b7c..b6a7c3f 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@ -63,6 +63,7 @@ import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.BlipSection;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
@@ -81,7 +82,7 @@ import org.apache.accumulo.fate.zookeeper.ServiceLock;
import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason;
import org.apache.accumulo.fate.zookeeper.ServiceLock.LockWatcher;
import org.apache.accumulo.gc.metrics.GcCycleMetrics;
-import org.apache.accumulo.gc.metrics.GcMetricsFactory;
+import org.apache.accumulo.gc.metrics.GcMetrics;
import org.apache.accumulo.gc.replication.CloseWriteAheadLogReferences;
import org.apache.accumulo.server.AbstractServer;
import org.apache.accumulo.server.ServerOpts;
@@ -122,25 +123,11 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
private final GcCycleMetrics gcCycleMetrics = new GcCycleMetrics();
- public static void main(String[] args) throws Exception {
- try (SimpleGarbageCollector gc = new SimpleGarbageCollector(new ServerOpts(), args)) {
- gc.runServer();
- }
- }
-
SimpleGarbageCollector(ServerOpts opts, String[] args) {
super("gc", opts, args);
final AccumuloConfiguration conf = getConfiguration();
- boolean gcMetricsRegistered = new GcMetricsFactory(conf).register(this);
-
- if (gcMetricsRegistered) {
- log.info("gc metrics modules registered with metrics system");
- } else {
- log.info("Failed to register gc metrics module");
- }
-
final long gcDelay = conf.getTimeInMillis(Property.GC_CYCLE_DELAY);
final String useFullCompaction = conf.get(Property.GC_USE_FULL_COMPACTION);
@@ -152,6 +139,12 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
log.info("gc post metadata action: {}", useFullCompaction);
}
+ public static void main(String[] args) throws Exception {
+ try (SimpleGarbageCollector gc = new SimpleGarbageCollector(new ServerOpts(), args)) {
+ gc.runServer();
+ }
+ }
+
/**
* Gets the delay before the first collection.
*
@@ -384,7 +377,8 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
deleteThreadPool.shutdown();
try {
- while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) {}
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
} catch (InterruptedException e1) {
log.error("{}", e1.getMessage(), e1);
}
@@ -458,14 +452,23 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
// old data files to be unused
log.info("Trying to acquire ZooKeeper lock for garbage collector");
+ HostAndPort address = startStatsService();
+
try {
- getZooLock(startStatsService());
+ getZooLock(address);
} catch (Exception ex) {
log.error("{}", ex.getMessage(), ex);
System.exit(1);
}
try {
+ MetricsUtil.initializeMetrics(getContext().getConfiguration(), this.applicationName, address);
+ MetricsUtil.initializeProducers(new GcMetrics(this));
+ } catch (Exception e1) {
+ log.error("Error initializing metrics, metrics will not be emitted.", e1);
+ }
+
+ try {
long delay = getStartDelay();
log.debug("Sleeping for {} milliseconds before beginning garbage collection cycles", delay);
Thread.sleep(delay);
@@ -578,6 +581,7 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
}
}
try {
+
gcCycleMetrics.incrementRunCycleCount();
long gcDelay = getConfiguration().getTimeInMillis(Property.GC_CYCLE_DELAY);
log.debug("Sleeping for {} milliseconds", gcDelay);
@@ -654,7 +658,7 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
HostAndPort[] addresses = TServerUtils.getHostAndPorts(getHostname(), port);
long maxMessageSize = getConfiguration().getAsBytes(Property.GENERAL_MAX_MESSAGE_SIZE);
try {
- ServerAddress server = TServerUtils.startTServer(getMetricsSystem(), getConfiguration(),
+ ServerAddress server = TServerUtils.startTServer(getConfiguration(),
getContext().getThriftServerType(), processor, this.getClass().getSimpleName(),
"GC Monitor Service", 2, ThreadPools.DEFAULT_TIMEOUT_MILLISECS, 1000, maxMessageSize,
getContext().getServerSslParams(), getContext().getSaslParams(), 0, addresses);
@@ -750,4 +754,5 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
public GcCycleMetrics getGcCycleMetrics() {
return gcCycleMetrics;
}
+
}
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcCycleMetrics.java b/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcCycleMetrics.java
index 3205f7f..e067749 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcCycleMetrics.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcCycleMetrics.java
@@ -29,11 +29,13 @@ import org.apache.accumulo.core.gc.thrift.GcCycleStats;
*/
public class GcCycleMetrics {
- private AtomicReference<GcCycleStats> lastCollect = new AtomicReference<>(new GcCycleStats());
- private AtomicReference<GcCycleStats> lastWalCollect = new AtomicReference<>(new GcCycleStats());
+ private final AtomicReference<GcCycleStats> lastCollect =
+ new AtomicReference<>(new GcCycleStats());
+ private final AtomicReference<GcCycleStats> lastWalCollect =
+ new AtomicReference<>(new GcCycleStats());
- private AtomicLong postOpDurationNanos = new AtomicLong(0);
- private AtomicLong runCycleCount = new AtomicLong(0);
+ private final AtomicLong postOpDurationNanos = new AtomicLong(0);
+ private final AtomicLong runCycleCount = new AtomicLong(0);
public GcCycleMetrics() {}
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcMetrics.java b/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcMetrics.java
index fdf2a9b..3d76502 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcMetrics.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcMetrics.java
@@ -20,111 +20,60 @@ package org.apache.accumulo.gc.metrics;
import java.util.concurrent.TimeUnit;
-import org.apache.accumulo.core.gc.thrift.GcCycleStats;
+import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.gc.SimpleGarbageCollector;
-import org.apache.accumulo.server.metrics.Metrics;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
-/**
- * Expected to be instantiated with GcMetricsFactory. This will configure both jmx and the hadoop
- * metrics systems. The naming convention, in hadoop metrics2, the records will appear as
- * CONTEXT.RECORD (accgc.AccGcCycleMetrics). The value for context is also used by the configuration
- * file for sink configuration.
- */
-public class GcMetrics extends Metrics {
-
- // use common prefix, different that just gc, to prevent confusion with jvm gc metrics.
- public static final String GC_METRIC_PREFIX = "AccGc";
-
- private static final String jmxName = "GarbageCollector";
- private static final String description = "Accumulo garbage collection metrics";
- private static final String record = "AccGcCycleMetrics";
-
- private final SimpleGarbageCollector gc;
-
- // metrics gauges / counters.
- private final MutableGaugeLong gcStarted;
- private final MutableGaugeLong gcFinished;
- private final MutableGaugeLong gcCandidates;
- private final MutableGaugeLong gcInUse;
- private final MutableGaugeLong gcDeleted;
- private final MutableGaugeLong gcErrors;
-
- private final MutableGaugeLong walStarted;
- private final MutableGaugeLong walFinished;
- private final MutableGaugeLong walCandidates;
- private final MutableGaugeLong walInUse;
- private final MutableGaugeLong walDeleted;
- private final MutableGaugeLong walErrors;
-
- private final MutableGaugeLong postOpDuration;
- private final MutableGaugeLong runCycleCount;
-
- GcMetrics(final SimpleGarbageCollector gc) {
- super(jmxName + ",sub=" + gc.getClass().getSimpleName(), description, "accgc", record);
- this.gc = gc;
-
- MetricsRegistry registry = super.getRegistry();
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
- gcStarted = registry.newGauge(GC_METRIC_PREFIX + "Started",
- "Timestamp GC file collection cycle started", 0L);
- gcFinished = registry.newGauge(GC_METRIC_PREFIX + "Finished",
- "Timestamp GC file collect cycle finished", 0L);
- gcCandidates = registry.newGauge(GC_METRIC_PREFIX + "Candidates",
- "Number of files that are candidates for deletion", 0L);
- gcInUse =
- registry.newGauge(GC_METRIC_PREFIX + "InUse", "Number of candidate files still in use", 0L);
- gcDeleted =
- registry.newGauge(GC_METRIC_PREFIX + "Deleted", "Number of candidate files deleted", 0L);
- gcErrors =
- registry.newGauge(GC_METRIC_PREFIX + "Errors", "Number of candidate deletion errors", 0L);
+public class GcMetrics implements MetricsProducer {
- walStarted = registry.newGauge(GC_METRIC_PREFIX + "WalStarted",
- "Timestamp GC WAL collection started", 0L);
- walFinished = registry.newGauge(GC_METRIC_PREFIX + "WalFinished",
- "Timestamp GC WAL collection finished", 0L);
- walCandidates = registry.newGauge(GC_METRIC_PREFIX + "WalCandidates",
- "Number of files that are candidates for deletion", 0L);
- walInUse = registry.newGauge(GC_METRIC_PREFIX + "WalInUse",
- "Number of wal file candidates that are still in use", 0L);
- walDeleted = registry.newGauge(GC_METRIC_PREFIX + "WalDeleted",
- "Number of candidate wal files deleted", 0L);
- walErrors = registry.newGauge(GC_METRIC_PREFIX + "WalErrors",
- "Number candidate wal file deletion errors", 0L);
-
- postOpDuration = registry.newGauge(GC_METRIC_PREFIX + "PostOpDuration",
- "GC metadata table post operation duration in milliseconds", 0L);
-
- runCycleCount = registry.newGauge(GC_METRIC_PREFIX + "RunCycleCount",
- "gauge incremented each gc cycle run, rest on process start", 0L);
+ private final GcCycleMetrics metricValues;
+ public GcMetrics(SimpleGarbageCollector gc) {
+ // Updated during each cycle of SimpleGC
+ metricValues = gc.getGcCycleMetrics();
}
@Override
- protected void prepareMetrics() {
-
- GcCycleMetrics values = gc.getGcCycleMetrics();
-
- GcCycleStats lastFileCollect = values.getLastCollect();
+ public void registerMetrics(MeterRegistry registry) {
+ Gauge.builder(METRICS_GC_STARTED, metricValues, v -> v.getLastCollect().getStarted())
+ .description("Timestamp GC file collection cycle started").register(registry);
+ Gauge.builder(METRICS_GC_FINISHED, metricValues, v -> v.getLastCollect().getFinished())
+ .description("Timestamp GC file collect cycle finished").register(registry);
+ Gauge.builder(METRICS_GC_CANDIDATES, metricValues, v -> v.getLastCollect().getCandidates())
+ .description("Number of files that are candidates for deletion").register(registry);
+ Gauge.builder(METRICS_GC_IN_USE, metricValues, v -> v.getLastCollect().getInUse())
+ .description("Number of candidate files still in use").register(registry);
+ Gauge.builder(METRICS_GC_DELETED, metricValues, v -> v.getLastCollect().getDeleted())
+ .description("Number of candidate files deleted").register(registry);
+ Gauge.builder(METRICS_GC_ERRORS, metricValues, v -> v.getLastCollect().getErrors())
+ .description("Number of candidate deletion errors").register(registry);
+
+ // WAL metrics Gauges
+ Gauge.builder(METRICS_GC_WAL_STARTED, metricValues, v -> v.getLastWalCollect().getStarted())
+ .description("Timestamp GC WAL collection cycle started").register(registry);
+ Gauge.builder(METRICS_GC_WAL_FINISHED, metricValues, v -> v.getLastWalCollect().getFinished())
+ .description("Timestamp GC WAL collect cycle finished").register(registry);
+ Gauge
+ .builder(METRICS_GC_WAL_CANDIDATES, metricValues,
+ v -> v.getLastWalCollect().getCandidates())
+ .description("Number of files that are candidates for deletion").register(registry);
+ Gauge.builder(METRICS_GC_WAL_IN_USE, metricValues, v -> v.getLastWalCollect().getInUse())
+ .description("Number of wal file candidates that are still in use").register(registry);
+ Gauge.builder(METRICS_GC_WAL_DELETED, metricValues, v -> v.getLastWalCollect().getDeleted())
+ .description("Number of candidate wal files deleted").register(registry);
+ Gauge.builder(METRICS_GC_WAL_ERRORS, metricValues, v -> v.getLastWalCollect().getErrors())
+ .description("Number candidate wal file deletion errors").register(registry);
+ Gauge
+ .builder(METRICS_GC_POST_OP_DURATION, metricValues,
+ v -> TimeUnit.NANOSECONDS.toMillis(v.getPostOpDurationNanos()))
+ .description("GC metadata table post operation duration in milliseconds")
+ .register(registry);
+ Gauge.builder(METRICS_GC_RUN_CYCLE, metricValues, GcCycleMetrics::getRunCycleCount)
+ .description("gauge incremented each gc cycle run, rest on process start")
+ .register(registry);
- gcStarted.set(lastFileCollect.getStarted());
- gcFinished.set(lastFileCollect.getFinished());
- gcCandidates.set(lastFileCollect.getCandidates());
- gcInUse.set(lastFileCollect.getInUse());
- gcDeleted.set(lastFileCollect.getDeleted());
- gcErrors.set(lastFileCollect.getErrors());
-
- GcCycleStats lastWalCollect = values.getLastWalCollect();
-
- walStarted.set(lastWalCollect.getStarted());
- walFinished.set(lastWalCollect.getFinished());
- walCandidates.set(lastWalCollect.getCandidates());
- walInUse.set(lastWalCollect.getInUse());
- walDeleted.set(lastWalCollect.getDeleted());
- walErrors.set(lastWalCollect.getErrors());
-
- postOpDuration.set(TimeUnit.NANOSECONDS.toMillis(values.getPostOpDurationNanos()));
- runCycleCount.set(values.getRunCycleCount());
}
+
}
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcMetricsFactory.java b/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcMetricsFactory.java
deleted file mode 100644
index ce79cce..0000000
--- a/server/gc/src/main/java/org/apache/accumulo/gc/metrics/GcMetricsFactory.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.gc.metrics;
-
-import static java.util.Objects.requireNonNull;
-
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.gc.SimpleGarbageCollector;
-import org.apache.hadoop.metrics2.MetricsSystem;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class GcMetricsFactory {
-
- private final static Logger log = LoggerFactory.getLogger(GcMetricsFactory.class);
-
- private boolean enableMetrics;
-
- public GcMetricsFactory(AccumuloConfiguration conf) {
- requireNonNull(conf, "AccumuloConfiguration must not be null");
- enableMetrics = conf.getBoolean(Property.GC_METRICS_ENABLED);
- }
-
- public boolean register(SimpleGarbageCollector gc) {
-
- if (!enableMetrics) {
- log.info("Accumulo gc metrics are disabled. To enable, set {} in configuration",
- Property.GC_METRICS_ENABLED);
- return false;
- }
-
- try {
-
- MetricsSystem metricsSystem = gc.getMetricsSystem();
-
- new GcMetrics(gc).register(metricsSystem);
-
- return true;
-
- } catch (Exception ex) {
- log.error("Failed to register accumulo gc metrics", ex);
- return false;
- }
- }
-}
diff --git a/server/manager/pom.xml b/server/manager/pom.xml
index 70f1b46..d8432f5 100644
--- a/server/manager/pom.xml
+++ b/server/manager/pom.xml
@@ -61,6 +61,10 @@
<artifactId>commons-io</artifactId>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-core</artifactId>
</dependency>
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
index e9fa3c0..f07192c 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
@@ -81,6 +81,7 @@ import org.apache.accumulo.core.metadata.TabletLocationState;
import org.apache.accumulo.core.metadata.TabletState;
import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.replication.thrift.ReplicationCoordinator;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.spi.balancer.BalancerEnvironment;
@@ -104,7 +105,7 @@ import org.apache.accumulo.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.fate.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy;
-import org.apache.accumulo.manager.metrics.ManagerMetricsFactory;
+import org.apache.accumulo.manager.metrics.ManagerMetrics;
import org.apache.accumulo.manager.recovery.RecoveryManager;
import org.apache.accumulo.manager.replication.ManagerReplicationCoordinator;
import org.apache.accumulo.manager.replication.ReplicationDriver;
@@ -1024,10 +1025,10 @@ public class Manager extends AbstractServer
}
ServerAddress sa;
try {
- sa = TServerUtils.startServer(getMetricsSystem(), context, getHostname(),
- Property.MANAGER_CLIENTPORT, processor, "Manager", "Manager Client Service Handler", null,
- Property.MANAGER_MINTHREADS, Property.MANAGER_MINTHREADS_TIMEOUT,
- Property.MANAGER_THREADCHECK, Property.GENERAL_MAX_MESSAGE_SIZE);
+ sa = TServerUtils.startServer(context, getHostname(), Property.MANAGER_CLIENTPORT, processor,
+ "Manager", "Manager Client Service Handler", null, Property.MANAGER_MINTHREADS,
+ Property.MANAGER_MINTHREADS_TIMEOUT, Property.MANAGER_THREADCHECK,
+ Property.GENERAL_MAX_MESSAGE_SIZE);
} catch (UnknownHostException e) {
throw new IllegalStateException("Unable to start server on host " + getHostname(), e);
}
@@ -1041,6 +1042,14 @@ public class Manager extends AbstractServer
throw new IllegalStateException("Exception getting manager lock", e);
}
+ try {
+ MetricsUtil.initializeMetrics(getContext().getConfiguration(), this.applicationName,
+ sa.getAddress());
+ ManagerMetrics.init(getConfiguration(), this);
+ } catch (Exception e1) {
+ log.error("Error initializing metrics, metrics will not be emitted.", e1);
+ }
+
recoveryManager = new RecoveryManager(this, TIME_TO_CACHE_RECOVERY_WAL_EXISTENCE);
context.getTableManager().addObserver(this);
@@ -1191,15 +1200,6 @@ public class Manager extends AbstractServer
}
}, 0, 5000, TimeUnit.MILLISECONDS);
- // Register metrics modules
- int failureCount = new ManagerMetricsFactory(getConfiguration()).register(this);
-
- if (failureCount > 0) {
- log.info("Failed to register {} metrics modules", failureCount);
- } else {
- log.info("All metrics modules registered");
- }
-
// checking stored user hashes if any of them uses an outdated algorithm
security.validateStoredUserCreditentials();
@@ -1343,7 +1343,7 @@ public class Manager extends AbstractServer
HighlyAvailableServiceWrapper.service(impl, this);
ReplicationCoordinator.Processor<ReplicationCoordinator.Iface> replicationCoordinatorProcessor =
new ReplicationCoordinator.Processor<>(TraceUtil.wrapService(haReplicationProxy));
- ServerAddress replAddress = TServerUtils.startServer(getMetricsSystem(), context, getHostname(),
+ ServerAddress replAddress = TServerUtils.startServer(context, getHostname(),
Property.MANAGER_REPLICATION_COORDINATOR_PORT, replicationCoordinatorProcessor,
"Manager Replication Coordinator", "Replication Coordinator", null,
Property.MANAGER_REPLICATION_COORDINATOR_MINTHREADS, null,
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetrics.java b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetrics.java
index 1430386..2f6d693 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetrics.java
@@ -18,12 +18,27 @@
*/
package org.apache.accumulo.manager.metrics;
-import org.apache.accumulo.server.metrics.Metrics;
+import static java.util.Objects.requireNonNull;
-public abstract class ManagerMetrics extends Metrics {
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.metrics.MetricsUtil;
+import org.apache.accumulo.manager.Manager;
+import org.apache.accumulo.manager.metrics.fate.FateMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
- protected ManagerMetrics(String subName, String description, String record) {
- super("Manager,sub=" + subName, description, "manager", record);
+public class ManagerMetrics {
+
+ private final static Logger log = LoggerFactory.getLogger(ManagerMetrics.class);
+
+ public static void init(AccumuloConfiguration conf, Manager m) {
+ requireNonNull(conf, "AccumuloConfiguration must not be null");
+ MetricsUtil.initializeProducers(new ReplicationMetrics(m));
+ log.info("Registered replication metrics module");
+ MetricsUtil.initializeProducers(new FateMetrics(m.getContext(),
+ conf.getTimeInMillis(Property.MANAGER_FATE_METRICS_MIN_UPDATE_INTERVAL)));
+ log.info("Registered FATE metrics module");
}
}
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetricsFactory.java b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetricsFactory.java
deleted file mode 100644
index 8861f47..0000000
--- a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetricsFactory.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.manager.metrics;
-
-import static java.util.Objects.requireNonNull;
-
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.manager.Manager;
-import org.apache.accumulo.manager.metrics.fate.FateMetrics;
-import org.apache.hadoop.metrics2.MetricsSystem;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Provide manager metrics configuration. Currently this is replication and FATE metrics. Metrics
- * can be configured using hadoop metrics2 Fate metrics must be enabled via configuration file
- * (default is disabled)
- */
-public class ManagerMetricsFactory {
-
- private final static Logger log = LoggerFactory.getLogger(ManagerMetricsFactory.class);
-
- private final boolean enableFateMetrics;
- private final long fateMinUpdateInterval;
-
- public ManagerMetricsFactory(AccumuloConfiguration conf) {
- requireNonNull(conf, "AccumuloConfiguration must not be null");
- enableFateMetrics = conf.getBoolean(Property.MANAGER_FATE_METRICS_ENABLED);
- fateMinUpdateInterval = conf.getTimeInMillis(Property.MANAGER_FATE_METRICS_MIN_UPDATE_INTERVAL);
- }
-
- public int register(Manager manager) {
- MetricsSystem metricsSystem = manager.getMetricsSystem();
-
- int failureCount = 0;
-
- try {
- new ReplicationMetrics(manager).register(metricsSystem);
- log.info("Registered replication metrics module");
- } catch (Exception ex) {
- failureCount++;
- log.error("Failed to register replication metrics", ex);
- }
-
- try {
- if (enableFateMetrics) {
- new FateMetrics(manager.getContext(), fateMinUpdateInterval).register(metricsSystem);
- log.info("Registered FATE metrics module");
- }
- } catch (Exception ex) {
- failureCount++;
- log.error("Failed to register fate metrics", ex);
- }
- return failureCount;
- }
-
-}
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ReplicationMetrics.java b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ReplicationMetrics.java
index 2778daf..e983505 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ReplicationMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ReplicationMetrics.java
@@ -19,62 +19,62 @@
package org.apache.accumulo.manager.metrics;
import java.io.IOException;
+import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.accumulo.core.clientImpl.Tables;
import org.apache.accumulo.core.manager.state.tables.TableState;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.manager.Manager;
import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableQuantiles;
-import org.apache.hadoop.metrics2.lib.MutableStat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class ReplicationMetrics extends ManagerMetrics {
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+
+public class ReplicationMetrics implements MetricsProducer {
private static final Logger log = LoggerFactory.getLogger(ReplicationMetrics.class);
private final Manager manager;
private final ReplicationUtil replicationUtil;
- private final MutableQuantiles replicationQueueTimeQuantiles;
- private final MutableStat replicationQueueTimeStat;
private final Map<Path,Long> pathModTimes;
+ private Timer replicationQueueTimer;
+ private AtomicLong pendingFiles;
+ private AtomicInteger numPeers;
+ private AtomicInteger maxReplicationThreads;
+
ReplicationMetrics(Manager manager) {
- super("Replication", "Data-Center Replication Metrics", "ManagerReplication");
this.manager = manager;
-
pathModTimes = new HashMap<>();
-
replicationUtil = new ReplicationUtil(manager.getContext());
- MetricsRegistry registry = super.getRegistry();
- replicationQueueTimeQuantiles = registry.newQuantiles("replicationQueue10m",
- "Replication queue time quantiles in milliseconds", "ops", "latency", 600);
- replicationQueueTimeStat = registry.newStat("replicationQueue",
- "Replication queue time statistics in milliseconds", "ops", "latency", true);
}
- @Override
- protected void prepareMetrics() {
- final String PENDING_FILES = "filesPendingReplication";
+ protected void update() {
// Only add these metrics if the replication table is online and there are peers
if (TableState.ONLINE == Tables.getTableState(manager.getContext(), ReplicationTable.ID)
&& !replicationUtil.getPeers().isEmpty()) {
- getRegistry().add(PENDING_FILES, getNumFilesPendingReplication());
+ pendingFiles.set(getNumFilesPendingReplication());
addReplicationQueueTimeMetrics();
} else {
- getRegistry().add(PENDING_FILES, 0);
+ pendingFiles.set(0);
}
-
- getRegistry().add("numPeers", getNumConfiguredPeers());
- getRegistry().add("maxReplicationThreads", getMaxReplicationThreads());
+ numPeers.set(getNumConfiguredPeers());
+ maxReplicationThreads.set(getMaxReplicationThreads());
}
protected long getNumFilesPendingReplication() {
@@ -112,7 +112,7 @@ public class ReplicationMetrics extends ManagerMetrics {
// We'll take a snap of the current time and use this as a diff between any deleted
// file's modification time and now. The reported latency will be off by at most a
// number of seconds equal to the metric polling period
- long currentTime = getCurrentTime();
+ long currentTime = System.currentTimeMillis();
// Iterate through all the pending paths and update the mod time if we don't know it yet
for (Path path : paths) {
@@ -122,8 +122,8 @@ public class ReplicationMetrics extends ManagerMetrics {
manager.getVolumeManager().getFileStatus(path).getModificationTime());
} catch (IOException e) {
// Ignore all IOExceptions
- // Either the system is unavailable or the file was deleted
- // since the initial scan and this check
+ // Either the system is unavailable, or the file was deleted since the initial scan and
+ // this check
log.trace(
"Failed to get file status for {}, file system is unavailable or it does not exist",
path);
@@ -140,20 +140,33 @@ public class ReplicationMetrics extends ManagerMetrics {
return;
}
- replicationQueueTimeStat.resetMinMax();
-
for (Path path : deletedPaths) {
// Remove this path and add the latency
Long modTime = pathModTimes.remove(path);
if (modTime != null) {
long diff = Math.max(0, currentTime - modTime);
- replicationQueueTimeQuantiles.add(diff);
- replicationQueueTimeStat.add(diff);
+ // micrometer timer
+ replicationQueueTimer.record(Duration.ofMillis(diff));
}
}
}
- protected long getCurrentTime() {
- return System.currentTimeMillis();
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ replicationQueueTimer = registry.timer(METRICS_REPLICATION_QUEUE, MetricsUtil.getCommonTags());
+ pendingFiles = registry.gauge(METRICS_REPLICATION_PENDING_FILES, MetricsUtil.getCommonTags(),
+ new AtomicLong(0));
+ numPeers = registry.gauge(METRICS_REPLICATION_PEERS, MetricsUtil.getCommonTags(),
+ new AtomicInteger(0));
+ maxReplicationThreads = registry.gauge(METRICS_REPLICATION_THREADS, MetricsUtil.getCommonTags(),
+ new AtomicInteger(0));
+
+ ScheduledExecutorService scheduler =
+ ThreadPools.createScheduledExecutorService(1, "replicationMetricsPoller", false);
+ Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdownNow));
+ long minimumRefreshDelay = TimeUnit.SECONDS.toMillis(5);
+ scheduler.scheduleAtFixedRate(this::update, minimumRefreshDelay, minimumRefreshDelay,
+ TimeUnit.MILLISECONDS);
}
+
}
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/FateMetrics.java b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/FateMetrics.java
index 202641f..dcd5eea4 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/FateMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/FateMetrics.java
@@ -18,63 +18,58 @@
*/
package org.apache.accumulo.manager.metrics.fate;
-import java.util.Map;
-import java.util.TreeMap;
+import java.util.Map.Entry;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.metrics.MetricsUtil;
+import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.fate.ReadOnlyTStore;
import org.apache.accumulo.fate.ZooStore;
-import org.apache.accumulo.manager.metrics.ManagerMetrics;
import org.apache.accumulo.server.ServerContext;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class FateMetrics extends ManagerMetrics {
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tags;
+
+public class FateMetrics implements MetricsProducer {
private static final Logger log = LoggerFactory.getLogger(FateMetrics.class);
// limit calls to update fate counters to guard against hammering zookeeper.
- private static final long DEFAULT_MIN_REFRESH_DELAY = TimeUnit.SECONDS.toMillis(10);
- private long minimumRefreshDelay;
-
- private static final String FATE_TX_STATE_METRIC_PREFIX = "FateTxState_";
- private static final String FATE_OP_TYPE_METRIC_PREFIX = "FateTxOpType_";
-
- private final MutableGaugeLong currentFateOps;
- private final MutableGaugeLong zkChildFateOpsTotal;
- private final MutableGaugeLong zkConnectionErrorsTotal;
-
- private final Map<String,MutableGaugeLong> fateTypeCounts = new TreeMap<>();
- private final Map<String,MutableGaugeLong> fateOpCounts = new TreeMap<>();
+ private static final long DEFAULT_MIN_REFRESH_DELAY = TimeUnit.SECONDS.toMillis(5);
- /*
- * lock should be used to guard read and write access to metricValues and the lastUpdate
- * timestamp.
- */
- private final Lock metricsValuesLock = new ReentrantLock();
- private FateMetricValues metricValues;
- private volatile long lastUpdate = 0;
+ private static final String OP_TYPE_TAG = "op.type";
private final ServerContext context;
private final ReadOnlyTStore<FateMetrics> zooStore;
private final String fateRootPath;
+ private final long refreshDelay;
+
+ private AtomicLong totalCurrentOpsGauge;
+ private AtomicLong totalOpsGauge;
+ private AtomicLong fateErrorsGauge;
+ private AtomicLong newTxGauge;
+ private AtomicLong inProgressTxGauge;
+ private AtomicLong failedInProgressTxGauge;
+ private AtomicLong failedTxGauge;
+ private AtomicLong successfulTxGauge;
+ private AtomicLong unknownTxGauge;
public FateMetrics(final ServerContext context, final long minimumRefreshDelay) {
- super("Fate", "Fate Metrics", "fate");
this.context = context;
- fateRootPath = context.getZooKeeperRoot() + Constants.ZFATE;
+ this.fateRootPath = context.getZooKeeperRoot() + Constants.ZFATE;
+ this.refreshDelay = Math.max(DEFAULT_MIN_REFRESH_DELAY, minimumRefreshDelay);
try {
-
- zooStore = new ZooStore<>(fateRootPath, context.getZooReaderWriter());
-
+ this.zooStore = new ZooStore<>(fateRootPath, context.getZooReaderWriter());
} catch (KeeperException ex) {
throw new IllegalStateException(
"FATE Metrics - Failed to create zoo store - metrics unavailable", ex);
@@ -84,103 +79,83 @@ public class FateMetrics extends ManagerMetrics {
"FATE Metrics - Interrupt received while initializing zoo store");
}
- this.minimumRefreshDelay = Math.max(DEFAULT_MIN_REFRESH_DELAY, minimumRefreshDelay);
-
- // lock not necessary in constructor.
- metricValues = FateMetricValues.getFromZooKeeper(context, fateRootPath, zooStore);
-
- MetricsRegistry registry = super.getRegistry();
-
- currentFateOps = registry.newGauge("currentFateOps", "Current number of FATE Ops", 0L);
- zkChildFateOpsTotal = registry.newGauge("totalFateOps", "Total FATE Ops", 0L);
- zkConnectionErrorsTotal =
- registry.newGauge("totalZkConnErrors", "Total ZK Connection Errors", 0L);
+ }
- for (ReadOnlyTStore.TStatus t : ReadOnlyTStore.TStatus.values()) {
- MutableGaugeLong g = registry.newGauge(FATE_TX_STATE_METRIC_PREFIX + t.name().toUpperCase(),
- "Transaction count for " + t.name() + " transactions", 0L);
- fateTypeCounts.put(t.name(), g);
+ private void update() {
+
+ FateMetricValues metricValues =
+ FateMetricValues.getFromZooKeeper(context, fateRootPath, zooStore);
+
+ totalCurrentOpsGauge.set(metricValues.getCurrentFateOps());
+ totalOpsGauge.set(metricValues.getZkFateChildOpsTotal());
+ fateErrorsGauge.set(metricValues.getZkConnectionErrors());
+
+ for (Entry<String,Long> vals : metricValues.getTxStateCounters().entrySet()) {
+ switch (ReadOnlyTStore.TStatus.valueOf(vals.getKey())) {
+ case NEW:
+ newTxGauge.set(vals.getValue());
+ break;
+ case IN_PROGRESS:
+ inProgressTxGauge.set(vals.getValue());
+ break;
+ case FAILED_IN_PROGRESS:
+ failedInProgressTxGauge.set(vals.getValue());
+ break;
+ case FAILED:
+ failedTxGauge.set(vals.getValue());
+ break;
+ case SUCCESSFUL:
+ successfulTxGauge.set(vals.getValue());
+ break;
+ case UNKNOWN:
+ default:
+ log.warn("Unhandled status type: {}", vals.getKey());
+ unknownTxGauge.set(vals.getValue());
+ }
}
- }
- /**
- * For testing only: allow refresh delay to be set to any value, over riding the enforced minimum.
- *
- * @param minimumRefreshDelay
- * set new min refresh value, in seconds.
- */
- public void overrideRefresh(final long minimumRefreshDelay) {
- long delay = Math.max(0, minimumRefreshDelay);
- this.minimumRefreshDelay = TimeUnit.SECONDS.toMillis(delay);
+ metricValues.getOpTypeCounters().forEach((name, count) -> {
+ Metrics.gauge(METRICS_FATE_TYPE_IN_PROGRESS, Tags.of(OP_TYPE_TAG, name), count);
+ });
}
@Override
- protected void prepareMetrics() {
-
- metricsValuesLock.lock();
- try {
-
- long now = System.currentTimeMillis();
-
- if ((lastUpdate + minimumRefreshDelay) < now) {
- metricValues = FateMetricValues.getFromZooKeeper(context, fateRootPath, zooStore);
- lastUpdate = now;
-
- recordValues();
+ public void registerMetrics(final MeterRegistry registry) {
+ totalCurrentOpsGauge = registry.gauge(METRICS_FATE_TOTAL_IN_PROGRESS,
+ MetricsUtil.getCommonTags(), new AtomicLong(0));
+ totalOpsGauge =
+ registry.gauge(METRICS_FATE_OPS_ACTIVITY, MetricsUtil.getCommonTags(), new AtomicLong(0));
+ fateErrorsGauge = registry.gauge(METRICS_FATE_ERRORS,
+ Tags.concat(MetricsUtil.getCommonTags(), "type", "zk.connection"), new AtomicLong(0));
+ newTxGauge = registry.gauge(METRICS_FATE_TX, Tags.concat(MetricsUtil.getCommonTags(), "state",
+ ReadOnlyTStore.TStatus.NEW.name().toLowerCase()), new AtomicLong(0));
+ inProgressTxGauge = registry.gauge(METRICS_FATE_TX, Tags.concat(MetricsUtil.getCommonTags(),
+ "state", ReadOnlyTStore.TStatus.IN_PROGRESS.name().toLowerCase()), new AtomicLong(0));
+ failedInProgressTxGauge =
+ registry.gauge(METRICS_FATE_TX, Tags.concat(MetricsUtil.getCommonTags(), "state",
+ ReadOnlyTStore.TStatus.FAILED_IN_PROGRESS.name().toLowerCase()), new AtomicLong(0));
+ failedTxGauge = registry.gauge(METRICS_FATE_TX, Tags.concat(MetricsUtil.getCommonTags(),
+ "state", ReadOnlyTStore.TStatus.FAILED.name().toLowerCase()), new AtomicLong(0));
+ successfulTxGauge = registry.gauge(METRICS_FATE_TX, Tags.concat(MetricsUtil.getCommonTags(),
+ "state", ReadOnlyTStore.TStatus.SUCCESSFUL.name().toLowerCase()), new AtomicLong(0));
+ unknownTxGauge = registry.gauge(METRICS_FATE_TX, Tags.concat(MetricsUtil.getCommonTags(),
+ "state", ReadOnlyTStore.TStatus.UNKNOWN.name().toLowerCase()), new AtomicLong(0));
+
+ update();
+
+ // get fate status is read only operation - no reason to be nice on shutdown.
+ ScheduledExecutorService scheduler =
+ ThreadPools.createScheduledExecutorService(1, "fateMetricsPoller", false);
+ Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdownNow));
+
+ scheduler.scheduleAtFixedRate(() -> {
+ try {
+ update();
+ } catch (Exception ex) {
+ log.info("Failed to update fate metrics due to exception", ex);
}
- } finally {
- metricsValuesLock.unlock();
- }
- }
-
- /**
- * Update the metrics gauges from the measured values - this method assumes that concurrent access
- * is controlled externally to this method with the metricsValueLock, and that the lock has been
- * acquired before calling this method, and then released by the caller.
- */
- private void recordValues() {
-
- // update individual gauges that are reported.
- currentFateOps.set(metricValues.getCurrentFateOps());
- zkChildFateOpsTotal.set(metricValues.getZkFateChildOpsTotal());
- zkConnectionErrorsTotal.set(metricValues.getZkConnectionErrors());
-
- // the number FATE Tx states (NEW< IN_PROGRESS...) are fixed - the underlying
- // getTxStateCounters call will return a current valid count for each possible state.
- Map<String,Long> states = metricValues.getTxStateCounters();
-
- states.forEach((key,
- value) -> fateTypeCounts.computeIfAbsent(key,
- v -> super.getRegistry().newGauge(metricNameHelper(FATE_TX_STATE_METRIC_PREFIX, key),
- "By transaction state count for " + key, value))
- .set(value));
-
- // the op types are dynamic and the metric gauges generated when first seen. After
- // that the values need to be cleared and set to any new values present. This is so
- // that the metrics system will report "known" values once seen. In operation, the
- // number of types will be a fairly small set and should populate a consistent set
- // with normal operations.
-
- // clear current values.
- fateOpCounts.forEach((key, value) -> value.set(0));
-
- // update new counts, create new gauge if first time seen.
- Map<String,Long> opTypes = metricValues.getOpTypeCounters();
-
- log.trace("OP Counts Before: prev {}, updates {}", fateOpCounts, opTypes);
-
- opTypes.forEach((key,
- value) -> fateOpCounts.computeIfAbsent(key,
- gauge -> super.getRegistry().newGauge(metricNameHelper(FATE_OP_TYPE_METRIC_PREFIX, key),
- "By transaction op type count for " + key, value))
- .set(value));
-
- log.trace("OP Counts After: prev {}, updates {}", fateOpCounts, opTypes);
-
- }
+ }, refreshDelay, refreshDelay, TimeUnit.MILLISECONDS);
- private String metricNameHelper(final String prefix, final String name) {
- return prefix + name;
}
}
diff --git a/server/manager/src/test/java/org/apache/accumulo/manager/metrics/ReplicationMetricsTest.java b/server/manager/src/test/java/org/apache/accumulo/manager/metrics/ReplicationMetricsTest.java
index 486bc30..d14c831 100644
--- a/server/manager/src/test/java/org/apache/accumulo/manager/metrics/ReplicationMetricsTest.java
+++ b/server/manager/src/test/java/org/apache/accumulo/manager/metrics/ReplicationMetricsTest.java
@@ -19,7 +19,10 @@
package org.apache.accumulo.manager.metrics;
import java.lang.reflect.Field;
+import java.time.Duration;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.accumulo.manager.Manager;
import org.apache.accumulo.server.ServerContext;
@@ -27,13 +30,13 @@ import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.metrics2.lib.MutableQuantiles;
-import org.apache.hadoop.metrics2.lib.MutableStat;
import org.easymock.EasyMock;
import org.junit.Test;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+
public class ReplicationMetricsTest {
- private long currentTime = 1000L;
/**
* Extend the class to override the current time for testing
@@ -43,10 +46,6 @@ public class ReplicationMetricsTest {
super(manager);
}
- @Override
- public long getCurrentTime() {
- return currentTime;
- }
}
@Test
@@ -55,14 +54,23 @@ public class ReplicationMetricsTest {
ServerContext context = EasyMock.createMock(ServerContext.class);
VolumeManager fileSystem = EasyMock.createMock(VolumeManager.class);
ReplicationUtil util = EasyMock.createMock(ReplicationUtil.class);
- MutableStat stat = EasyMock.createMock(MutableStat.class);
- MutableQuantiles quantiles = EasyMock.createMock(MutableQuantiles.class);
+ MeterRegistry meterRegistry = EasyMock.createMock(MeterRegistry.class);
+ Timer timer = EasyMock.createMock(Timer.class);
Path path1 = new Path("hdfs://localhost:9000/accumulo/wal/file1");
Path path2 = new Path("hdfs://localhost:9000/accumulo/wal/file2");
// First call will initialize the map of paths to modification time
EasyMock.expect(manager.getContext()).andReturn(context).anyTimes();
+ EasyMock.expect(meterRegistry.timer("replicationQueue")).andReturn(timer).anyTimes();
+ EasyMock.expect(meterRegistry.gauge(EasyMock.eq("filesPendingReplication"),
+ EasyMock.anyObject(AtomicLong.class))).andReturn(new AtomicLong(0)).anyTimes();
+ EasyMock
+ .expect(
+ meterRegistry.gauge(EasyMock.eq("numPeers"), EasyMock.anyObject(AtomicInteger.class)))
+ .andReturn(new AtomicInteger(0)).anyTimes();
+ EasyMock.expect(meterRegistry.gauge(EasyMock.eq("maxReplicationThreads"),
+ EasyMock.anyObject(AtomicInteger.class))).andReturn(new AtomicInteger(0)).anyTimes();
EasyMock.expect(util.getPendingReplicationPaths()).andReturn(Set.of(path1, path2));
EasyMock.expect(manager.getVolumeManager()).andReturn(fileSystem);
EasyMock.expect(fileSystem.getFileStatus(path1)).andReturn(createStatus(100));
@@ -72,31 +80,22 @@ public class ReplicationMetricsTest {
// Second call will recognize the missing path1 and add the latency stat
EasyMock.expect(util.getPendingReplicationPaths()).andReturn(Set.of(path2));
- // Expect a call to reset the min/max
- stat.resetMinMax();
- EasyMock.expectLastCall();
-
- // Expect the calls of adding the stats
- quantiles.add(currentTime - 100);
- EasyMock.expectLastCall();
-
- stat.add(currentTime - 100);
+ timer.record(EasyMock.isA(Duration.class));
EasyMock.expectLastCall();
- EasyMock.replay(manager, fileSystem, util, stat, quantiles);
+ EasyMock.replay(manager, fileSystem, util, meterRegistry, timer);
ReplicationMetrics metrics = new ReplicationMetricsTestMetrics(manager);
// Inject our mock objects
replaceField(metrics, "replicationUtil", util);
- replaceField(metrics, "replicationQueueTimeQuantiles", quantiles);
- replaceField(metrics, "replicationQueueTimeStat", stat);
+ replaceField(metrics, "replicationQueueTimer", timer);
// Two calls to this will initialize the map and then add metrics
metrics.addReplicationQueueTimeMetrics();
metrics.addReplicationQueueTimeMetrics();
- EasyMock.verify(manager, fileSystem, util, stat, quantiles);
+ EasyMock.verify(manager, fileSystem, util, meterRegistry, timer);
}
private void replaceField(Object instance, String fieldName, Object target)
diff --git a/server/manager/src/test/resources/hadoop-metrics2-accumulo.properties b/server/manager/src/test/resources/hadoop-metrics2-accumulo.properties
deleted file mode 100644
index 9689807..0000000
--- a/server/manager/src/test/resources/hadoop-metrics2-accumulo.properties
+++ /dev/null
@@ -1,60 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-
-# Poll collectors every thirty seconds
-*.period=5
-
-#
-# MetricSink configuration
-#
-# Uncomment a sink (and configuration) to enable it. By default,
-# no sinks are enabled.
-#
-
-#
-# Configure file logging
-#
-
-# File sink for all metrics
-accumulo.sink.file-all.class=org.apache.hadoop.metrics2.sink.FileSink
-accumulo.sink.file-all.filename=./target/it.all.metrics
-
-accumulo.sink.file-gc.class=org.apache.hadoop.metrics2.sink.FileSink
-accumulo.sink.file-gc.context=accgc
-accumulo.sink.file-gc.filename=./target/accgc.metrics
-accumulo.sink.file-gc.period=5
-
-# File sink for tserver metrics
-# accumulo.sink.file-tserver.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-tserver.context=tserver
-# accumulo.sink.file-tserver.filename=tserver.metrics
-
-# File sink for manager metrics
-accumulo.sink.file-manager.class=org.apache.hadoop.metrics2.sink.FileSink
-accumulo.sink.file-manager.context=manager
-accumulo.sink.file-manager.filename=./target/manager.metrics
-accumulo.sink.file-manager.period=5
-
-accumulo.jmx.manager.context=manager
-
-# File sink for thrift server metrics
-# accumulo.sink.file-thrift.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-thrift.context=thrift
-# accumulo.sink.file-thrift.filename=thrift.metrics
-
diff --git a/server/tserver/pom.xml b/server/tserver/pom.xml
index 69bb4b0..84ee8d1 100644
--- a/server/tserver/pom.xml
+++ b/server/tserver/pom.xml
@@ -57,6 +57,10 @@
<artifactId>commons-io</artifactId>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-core</artifactId>
</dependency>
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/BusiestTracker.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/BusiestTracker.java
index f78c3e1..1ccd8c0 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/BusiestTracker.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/BusiestTracker.java
@@ -91,7 +91,7 @@ public abstract class BusiestTracker {
return new BusiestTracker(numBusiestTabletsToLog) {
@Override
protected long extractCount(Tablet tablet) {
- return tablet.totalQueries();
+ return tablet.totalQueriesResults();
}
};
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index a9f7a43..6128240 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -73,6 +73,7 @@ import org.apache.accumulo.core.metadata.TServerInstance;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.Location;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType;
+import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.replication.ReplicationConstants;
import org.apache.accumulo.core.replication.thrift.ReplicationServicer;
import org.apache.accumulo.core.rpc.ThriftUtil;
@@ -153,7 +154,6 @@ import org.apache.accumulo.tserver.tablet.TabletData;
import org.apache.commons.collections4.map.LRUMap;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.thrift.TException;
import org.apache.thrift.TProcessor;
import org.apache.thrift.TServiceClient;
@@ -176,10 +176,11 @@ public class TabletServer extends AbstractServer {
final TabletServerLogger logger;
- final TabletServerUpdateMetrics updateMetrics;
- final TabletServerScanMetrics scanMetrics;
- final TabletServerMinCMetrics mincMetrics;
- final CompactionExecutorsMetrics ceMetrics;
+ private TabletServerMetrics metrics;
+ TabletServerUpdateMetrics updateMetrics;
+ TabletServerScanMetrics scanMetrics;
+ TabletServerMinCMetrics mincMetrics;
+ CompactionExecutorsMetrics ceMetrics;
public TabletServerScanMetrics getScanMetrics() {
return scanMetrics;
@@ -346,10 +347,6 @@ public class TabletServer extends AbstractServer {
this.resourceManager = new TabletServerResourceManager(context);
this.security = AuditedSecurityOperation.getInstance(context);
- updateMetrics = new TabletServerUpdateMetrics();
- scanMetrics = new TabletServerScanMetrics();
- mincMetrics = new TabletServerMinCMetrics();
- ceMetrics = new CompactionExecutorsMetrics();
context.getScheduledExecutor().scheduleWithFixedDelay(TabletLocator::clearLocators, jitter(),
jitter(), TimeUnit.MILLISECONDS);
walMarker = new WalStateManager(context);
@@ -542,10 +539,10 @@ public class TabletServer extends AbstractServer {
throws UnknownHostException {
Property maxMessageSizeProperty = (conf.get(Property.TSERV_MAX_MESSAGE_SIZE) != null
? Property.TSERV_MAX_MESSAGE_SIZE : Property.GENERAL_MAX_MESSAGE_SIZE);
- ServerAddress sp = TServerUtils.startServer(getMetricsSystem(), getContext(), address,
- Property.TSERV_CLIENTPORT, processor, this.getClass().getSimpleName(),
- "Thrift Client Server", Property.TSERV_PORTSEARCH, Property.TSERV_MINTHREADS,
- Property.TSERV_MINTHREADS_TIMEOUT, Property.TSERV_THREADCHECK, maxMessageSizeProperty);
+ ServerAddress sp = TServerUtils.startServer(getContext(), address, Property.TSERV_CLIENTPORT,
+ processor, this.getClass().getSimpleName(), "Thrift Client Server",
+ Property.TSERV_PORTSEARCH, Property.TSERV_MINTHREADS, Property.TSERV_MINTHREADS_TIMEOUT,
+ Property.TSERV_THREADCHECK, maxMessageSizeProperty);
this.server = sp.server;
return sp.address;
}
@@ -614,11 +611,10 @@ public class TabletServer extends AbstractServer {
Property maxMessageSizeProperty =
getConfiguration().get(Property.TSERV_MAX_MESSAGE_SIZE) != null
? Property.TSERV_MAX_MESSAGE_SIZE : Property.GENERAL_MAX_MESSAGE_SIZE;
- ServerAddress sp =
- TServerUtils.startServer(getMetricsSystem(), getContext(), clientAddress.getHost(),
- Property.REPLICATION_RECEIPT_SERVICE_PORT, processor, "ReplicationServicerHandler",
- "Replication Servicer", Property.TSERV_PORTSEARCH, Property.REPLICATION_MIN_THREADS,
- null, Property.REPLICATION_THREADCHECK, maxMessageSizeProperty);
+ ServerAddress sp = TServerUtils.startServer(getContext(), clientAddress.getHost(),
+ Property.REPLICATION_RECEIPT_SERVICE_PORT, processor, "ReplicationServicerHandler",
+ "Replication Servicer", Property.TSERV_PORTSEARCH, Property.REPLICATION_MIN_THREADS, null,
+ Property.REPLICATION_THREADCHECK, maxMessageSizeProperty);
this.replServer = sp.server;
log.info("Started replication service on {}", sp.address);
@@ -714,17 +710,6 @@ public class TabletServer extends AbstractServer {
throw new RuntimeException(e);
}
- try {
- MetricsSystem metricsSystem = getMetricsSystem();
- new TabletServerMetrics(this).register(metricsSystem);
- mincMetrics.register(metricsSystem);
- scanMetrics.register(metricsSystem);
- updateMetrics.register(metricsSystem);
- ceMetrics.register(metricsSystem);
- } catch (Exception e) {
- log.error("Error registering metrics", e);
- }
-
if (authKeyWatcher != null) {
log.info("Seeding ZooKeeper watcher for authentication keys");
try {
@@ -739,6 +724,27 @@ public class TabletServer extends AbstractServer {
}
}
+ try {
+ clientAddress = startTabletClientService();
+ } catch (UnknownHostException e1) {
+ throw new RuntimeException("Failed to start the tablet client service", e1);
+ }
+ announceExistence();
+
+ try {
+ MetricsUtil.initializeMetrics(context.getConfiguration(), this.applicationName,
+ clientAddress);
+ } catch (Exception e1) {
+ log.error("Error initializing metrics, metrics will not be emitted.", e1);
+ }
+
+ metrics = new TabletServerMetrics(this);
+ updateMetrics = new TabletServerUpdateMetrics();
+ scanMetrics = new TabletServerScanMetrics();
+ mincMetrics = new TabletServerMinCMetrics();
+ ceMetrics = new CompactionExecutorsMetrics();
+ MetricsUtil.initializeProducers(metrics, updateMetrics, scanMetrics, mincMetrics, ceMetrics);
+
this.compactionManager = new CompactionManager(new Iterable<Compactable>() {
@Override
public Iterator<Compactable> iterator() {
@@ -749,12 +755,6 @@ public class TabletServer extends AbstractServer {
compactionManager.start();
try {
- clientAddress = startTabletClientService();
- } catch (UnknownHostException e1) {
- throw new RuntimeException("Failed to start the tablet client service", e1);
- }
- announceExistence();
- try {
walMarker.initWalMarker(getTabletSession());
} catch (Exception e) {
log.error("Unable to create WAL marker node in zookeeper", e);
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/CompactionExecutorsMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/CompactionExecutorsMetrics.java
index 3d7f221..e5d79ad 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/CompactionExecutorsMetrics.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/CompactionExecutorsMetrics.java
@@ -24,110 +24,125 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import java.util.function.Supplier;
+import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
+import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.tserver.compactions.CompactionManager.ExtCompMetric;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableGaugeInt;
import com.google.common.collect.Sets;
-public class CompactionExecutorsMetrics extends TServerMetrics {
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tags;
+
+public class CompactionExecutorsMetrics implements MetricsProducer {
- private volatile List<CeMetrics> ceml = List.of();
- private Map<CompactionExecutorId,CeMetrics> metrics = new HashMap<>();
- private Map<CompactionExecutorId,ExMetrics> exMetrics = new HashMap<>();
private volatile Supplier<Collection<ExtCompMetric>> externalMetricsSupplier;
+ private volatile List<CeMetrics> ceMetricsList = List.of();
+ private final Map<CompactionExecutorId,CeMetrics> ceMetricsMap = new HashMap<>();
+ private final Map<CompactionExecutorId,ExMetrics> exCeMetricsMap = new HashMap<>();
+ private MeterRegistry registry = null;
+
private static class CeMetrics {
- MutableGaugeInt queuedGauge;
- MutableGaugeInt runningGauge;
+ AtomicInteger queued;
+ AtomicInteger running;
IntSupplier runningSupplier;
IntSupplier queuedSupplier;
}
private static class ExMetrics {
- MutableGaugeInt queuedGauge;
- MutableGaugeInt runningGauge;
-
+ AtomicInteger queued;
+ AtomicInteger running;
}
public CompactionExecutorsMetrics() {
- super("compactionExecutors");
+
+ ScheduledExecutorService scheduler =
+ ThreadPools.createScheduledExecutorService(1, "compactionExecutorsMetricsPoller", false);
+ Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdownNow));
+ long minimumRefreshDelay = TimeUnit.SECONDS.toMillis(5);
+ scheduler.scheduleAtFixedRate(this::update, minimumRefreshDelay, minimumRefreshDelay,
+ TimeUnit.MILLISECONDS);
+
}
public synchronized AutoCloseable addExecutor(CompactionExecutorId ceid,
IntSupplier runningSupplier, IntSupplier queuedSupplier) {
- MetricsRegistry registry = super.getRegistry();
+ synchronized (ceMetricsMap) {
- synchronized (metrics) {
- CeMetrics cem = metrics.computeIfAbsent(ceid, id -> {
+ CeMetrics cem = ceMetricsMap.computeIfAbsent(ceid, id -> {
CeMetrics m = new CeMetrics();
- m.queuedGauge = registry.newGauge(id.canonical().replace('.', '_') + "_queued",
- "Queued compactions for executor " + id, 0);
- m.runningGauge = registry.newGauge(id.canonical().replace('.', '_') + "_running",
- "Running compactions for executor " + id, 0);
+ if (registry != null) {
+ m.queued = registry.gauge(METRICS_MAJC_QUEUED, Tags.of("id", ceid.canonical()),
+ new AtomicInteger(0));
+ m.running = registry.gauge(METRICS_MAJC_RUNNING, Tags.of("id", ceid.canonical()),
+ new AtomicInteger(0));
+ }
return m;
});
cem.runningSupplier = runningSupplier;
cem.queuedSupplier = queuedSupplier;
- ceml = List.copyOf(metrics.values());
+ ceMetricsList = List.copyOf(ceMetricsMap.values());
return () -> {
+
cem.runningSupplier = () -> 0;
cem.queuedSupplier = () -> 0;
- cem.runningGauge.set(0);
- cem.queuedGauge.set(0);
+ cem.running.set(0);
+ cem.queued.set(0);
};
}
}
- @Override
- public void prepareMetrics() {
+ public void update() {
if (externalMetricsSupplier != null) {
Set<CompactionExecutorId> seenIds = new HashSet<>();
- MetricsRegistry registry = super.getRegistry();
-
- synchronized (exMetrics) {
+ synchronized (exCeMetricsMap) {
externalMetricsSupplier.get().forEach(ecm -> {
seenIds.add(ecm.ceid);
- ExMetrics exm = exMetrics.computeIfAbsent(ecm.ceid, id -> {
+ ExMetrics exm = exCeMetricsMap.computeIfAbsent(ecm.ceid, id -> {
ExMetrics m = new ExMetrics();
- m.queuedGauge = registry.newGauge(id.canonical().replace('.', '_') + "_queued",
- "Queued compactions for executor " + id, 0);
- m.runningGauge = registry.newGauge(id.canonical().replace('.', '_') + "_running",
- "Running compactions for executor " + id, 0);
+ if (registry != null) {
+ m.queued = registry.gauge(METRICS_MAJC_QUEUED, Tags.of("id", ecm.ceid.canonical()),
+ new AtomicInteger(0));
+ m.running = registry.gauge(METRICS_MAJC_RUNNING, Tags.of("id", ecm.ceid.canonical()),
+ new AtomicInteger(0));
+ }
return m;
});
- exm.queuedGauge.set(ecm.queued);
- exm.runningGauge.set(ecm.running);
+ exm.queued.set(ecm.queued);
+ exm.running.set(ecm.running);
});
- Sets.difference(exMetrics.keySet(), seenIds).forEach(unusedId -> {
- ExMetrics exm = exMetrics.get(unusedId);
- exm.queuedGauge.set(0);
- exm.runningGauge.set(0);
+ Sets.difference(exCeMetricsMap.keySet(), seenIds).forEach(unusedId -> {
+ ExMetrics exm = exCeMetricsMap.get(unusedId);
+ exm.queued.set(0);
+ exm.running.set(0);
});
}
}
- ceml.forEach(cem -> {
- cem.runningGauge.set(cem.runningSupplier.getAsInt());
- cem.queuedGauge.set(cem.queuedSupplier.getAsInt());
+ ceMetricsList.forEach(cem -> {
+ cem.running.set(cem.runningSupplier.getAsInt());
+ cem.queued.set(cem.queuedSupplier.getAsInt());
});
}
@@ -135,4 +150,11 @@ public class CompactionExecutorsMetrics extends TServerMetrics {
this.externalMetricsSupplier = ems;
}
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ // Meters are registered dynamically. Save off the reference to the
+ // registry to use at that time.
+ this.registry = registry;
+ }
+
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TServerMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TServerMetrics.java
deleted file mode 100644
index 0f68c4f..0000000
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TServerMetrics.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.tserver.metrics;
-
-import org.apache.accumulo.server.metrics.Metrics;
-
-abstract class TServerMetrics extends Metrics {
-
- protected static final String TSERVER_NAME = "TabletServer";
-
- protected TServerMetrics(String record) {
- // this capitalization thing is just to preserve the capitalization difference between the
- // "general" record and the "TabletServer,sub=General" metrics name that existed in 1.9 without
- // duplicating too much code; the description, however, did change between 1.9 and 2.0
- super("TabletServer,sub=" + capitalize(record),
- "TabletServer " + capitalize(record) + " Metrics", "tserver", record);
- }
-
- private static final String capitalize(String word) {
- return word.substring(0, 1).toUpperCase() + word.substring(1);
- }
-
-}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java
index c732fcc..6576a3d 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java
@@ -18,74 +18,68 @@
*/
package org.apache.accumulo.tserver.metrics;
+import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.tserver.TabletServer;
-import org.apache.hadoop.metrics2.MetricsRecordBuilder;
-import org.apache.hadoop.metrics2.lib.Interns;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
-public class TabletServerMetrics extends TServerMetrics {
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
- private final TabletServerMetricsUtil util;
+public class TabletServerMetrics implements MetricsProducer {
- private final MutableGaugeLong entries;
- private final MutableGaugeLong entriesInMemory;
- private final MutableGaugeLong activeMajcs;
- private final MutableGaugeLong queuedMajcs;
- private final MutableGaugeLong activeMincs;
- private final MutableGaugeLong queuedMincs;
- private final MutableGaugeLong onlineTablets;
- private final MutableGaugeLong openingTablets;
- private final MutableGaugeLong unopenedTablets;
- private final MutableGaugeLong queries;
- private final MutableGaugeLong totalMincs;
+ private final TabletServerMetricsUtil util;
public TabletServerMetrics(TabletServer tserver) {
- super("general");
util = new TabletServerMetricsUtil(tserver);
-
- MetricsRegistry registry = super.getRegistry();
- entries = registry.newGauge("entries", "Number of entries", 0L);
- entriesInMemory = registry.newGauge("entriesInMem", "Number of entries in memory", 0L);
- activeMajcs = registry.newGauge("activeMajCs", "Number of active major compactions", 0L);
- queuedMajcs = registry.newGauge("queuedMajCs", "Number of queued major compactions", 0L);
- activeMincs = registry.newGauge("activeMinCs", "Number of active minor compactions", 0L);
- queuedMincs = registry.newGauge("queuedMinCs", "Number of queued minor compactions", 0L);
- onlineTablets = registry.newGauge("onlineTablets", "Number of online tablets", 0L);
- openingTablets = registry.newGauge("openingTablets", "Number of opening tablets", 0L);
- unopenedTablets = registry.newGauge("unopenedTablets", "Number of unopened tablets", 0L);
- queries = registry.newGauge("queries", "Number of queries", 0L);
- totalMincs = registry.newGauge("totalMinCs", "Total number of minor compactions performed", 0L);
}
@Override
- protected void prepareMetrics() {
- entries.set(util.getEntries());
- entriesInMemory.set(util.getEntriesInMemory());
- activeMajcs.set(util.getMajorCompactions());
- queuedMajcs.set(util.getMajorCompactionsQueued());
- activeMincs.set(util.getMinorCompactions());
- queuedMincs.set(util.getMinorCompactionsQueued());
- onlineTablets.set(util.getOnlineCount());
- openingTablets.set(util.getOpeningCount());
- unopenedTablets.set(util.getUnopenedCount());
- queries.set(util.getQueries());
- totalMincs.set(util.getTotalMinorCompactions());
- }
+ public void registerMetrics(MeterRegistry registry) {
+ Gauge.builder(METRICS_TSERVER_ENTRIES, util, TabletServerMetricsUtil::getEntries)
+ .description("Number of entries").register(registry);
+ Gauge.builder(METRICS_TSERVER_MEM_ENTRIES, util, TabletServerMetricsUtil::getEntriesInMemory)
+ .description("Number of entries in memory").register(registry);
+ Gauge.builder(METRICS_TSERVER_MAJC_RUNNING, util, TabletServerMetricsUtil::getMajorCompactions)
+ .description("Number of active major compactions").register(registry);
+ Gauge
+ .builder(METRICS_TSERVER_MAJC_QUEUED, util,
+ TabletServerMetricsUtil::getMajorCompactionsQueued)
+ .description("Number of queued major compactions").register(registry);
+ Gauge.builder(METRICS_TSERVER_MINC_RUNNING, util, TabletServerMetricsUtil::getMinorCompactions)
+ .description("Number of active minor compactions").register(registry);
+ Gauge
+ .builder(METRICS_TSERVER_MINC_QUEUED, util,
+ TabletServerMetricsUtil::getMinorCompactionsQueued)
+ .description("Number of queued minor compactions").register(registry);
+ Gauge.builder(METRICS_TSERVER_TABLETS_ONLINE, util, TabletServerMetricsUtil::getOnlineCount)
+ .description("Number of online tablets").register(registry);
+ Gauge.builder(METRICS_TSERVER_TABLETS_OPENING, util, TabletServerMetricsUtil::getOpeningCount)
+ .description("Number of opening tablets").register(registry);
+ Gauge.builder(METRICS_TSERVER_TABLETS_UNOPENED, util, TabletServerMetricsUtil::getUnopenedCount)
+ .description("Number of unopened tablets").register(registry);
+ Gauge.builder(METRICS_TSERVER_QUERIES, util, TabletServerMetricsUtil::getLookupCount)
+ .description("Number of queries").register(registry);
+ Gauge
+ .builder(METRICS_TSERVER_MINC_TOTAL, util,
+ TabletServerMetricsUtil::getTotalMinorCompactions)
+ .description("Total number of minor compactions performed").register(registry);
- @Override
- protected void getMoreMetrics(MetricsRecordBuilder builder, boolean all) {
- // TODO Some day, MetricsRegistry will also support the MetricsGaugeDouble or allow us to
- // instantiate it directly
- builder.addGauge(Interns.info("filesPerTablet", "Number of files per tablet"),
- util.getAverageFilesPerTablet());
- builder.addGauge(Interns.info("holdTime", "Time commits held"), util.getHoldTime());
- builder.addGauge(Interns.info("ingestRate", "Ingest rate (entries/sec)"), util.getIngest());
- builder.addGauge(Interns.info("ingestByteRate", "Ingest rate (bytes/sec)"),
- util.getIngestByteRate());
- builder.addGauge(Interns.info("queryRate", "Query rate (entries/sec)"), util.getQueryRate());
- builder.addGauge(Interns.info("queryByteRate", "Query rate (bytes/sec)"),
- util.getQueryByteRate());
- builder.addGauge(Interns.info("scannedRate", "Scanned rate"), util.getScannedRate());
+ Gauge
+ .builder(METRICS_TSERVER_TABLETS_FILES, util,
+ TabletServerMetricsUtil::getAverageFilesPerTablet)
+ .description("Number of files per tablet").register(registry);
+ Gauge.builder(METRICS_TSERVER_HOLD, util, TabletServerMetricsUtil::getHoldTime)
+ .description("Time commits held").register(registry);
+ Gauge.builder(METRICS_TSERVER_INGEST_MUTATIONS, util, TabletServerMetricsUtil::getIngestCount)
+ .description("Ingest rate (entries/sec)").register(registry);
+ Gauge.builder(METRICS_TSERVER_INGEST_BYTES, util, TabletServerMetricsUtil::getIngestByteCount)
+ .description("Ingest rate (bytes/sec)").register(registry);
+ Gauge.builder(METRICS_TSERVER_SCAN_RESULTS, util, TabletServerMetricsUtil::getQueryResultCount)
+ .description("Query rate (entries/sec)").register(registry);
+ Gauge
+ .builder(METRICS_TSERVER_SCAN_RESULTS_BYTES, util,
+ TabletServerMetricsUtil::getQueryByteCount)
+ .description("Query rate (bytes/sec)").register(registry);
+ Gauge.builder(METRICS_TSERVER_SCANNED_ENTRIES, util, TabletServerMetricsUtil::getScannedCount)
+ .description("Scanned rate").register(registry);
}
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsUtil.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsUtil.java
index 022285c..e102751 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsUtil.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsUtil.java
@@ -50,42 +50,34 @@ public class TabletServerMetricsUtil {
return result;
}
- public double getIngest() {
+ public double getIngestCount() {
double result = 0;
for (Tablet tablet : tserver.getOnlineTablets().values()) {
- result += tablet.ingestRate();
+ result += tablet.totalIngest();
}
return result;
}
- public double getIngestByteRate() {
+ public double getIngestByteCount() {
double result = 0;
for (Tablet tablet : tserver.getOnlineTablets().values()) {
- result += tablet.ingestByteRate();
+ result += tablet.totalIngestBytes();
}
return result;
}
- public double getQueryRate() {
+ public double getQueryByteCount() {
double result = 0;
for (Tablet tablet : tserver.getOnlineTablets().values()) {
- result += tablet.queryRate();
+ result += tablet.totalQueryResultsBytes();
}
return result;
}
- public double getQueryByteRate() {
+ public double getScannedCount() {
double result = 0;
for (Tablet tablet : tserver.getOnlineTablets().values()) {
- result += tablet.queryByteRate();
- }
- return result;
- }
-
- public double getScannedRate() {
- double result = 0;
- for (Tablet tablet : tserver.getOnlineTablets().values()) {
- result += tablet.scanRate();
+ result += tablet.totalScannedCount();
}
return result;
}
@@ -126,10 +118,18 @@ public class TabletServerMetricsUtil {
return tserver.getOpeningCount();
}
- public long getQueries() {
+ public long getLookupCount() {
+ long result = 0;
+ for (Tablet tablet : tserver.getOnlineTablets().values()) {
+ result += tablet.totalLookupCount();
+ }
+ return result;
+ }
+
+ public long getQueryResultCount() {
long result = 0;
for (Tablet tablet : tserver.getOnlineTablets().values()) {
- result += tablet.totalQueries();
+ result += tablet.totalQueriesResults();
}
return result;
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetrics.java
index de520d0..8174f44 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetrics.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetrics.java
@@ -18,28 +18,35 @@
*/
package org.apache.accumulo.tserver.metrics;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableStat;
+import java.time.Duration;
-public class TabletServerMinCMetrics extends TServerMetrics {
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.metrics.MetricsUtil;
- private final MutableStat activeMinc;
- private final MutableStat queuedMinc;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
- public TabletServerMinCMetrics() {
- super("MinorCompactions");
+public class TabletServerMinCMetrics implements MetricsProducer {
- MetricsRegistry registry = super.getRegistry();
- activeMinc = registry.newStat("minc", "Minor compactions", "Ops", "Count", true);
- queuedMinc = registry.newStat("queue", "Queued minor compactions", "Ops", "Count", true);
- }
+ private Timer activeMinc;
+ private Timer queuedMinc;
public void addActive(long value) {
- activeMinc.add(value);
+ activeMinc.record(Duration.ofMillis(value));
}
public void addQueued(long value) {
- queuedMinc.add(value);
+ queuedMinc.record(Duration.ofMillis(value));
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ activeMinc = Timer.builder(METRICS_MINC_RUNNING).description("Minor compactions time active")
+ .tags(MetricsUtil.getCommonTags()).register(registry);
+
+ queuedMinc =
+ Timer.builder(METRICS_MINC_QUEUED).description("Queued minor compactions time queued")
+ .tags(MetricsUtil.getCommonTags()).register(registry);
}
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java
index 2c856f9..592058b 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java
@@ -18,34 +18,39 @@
*/
package org.apache.accumulo.tserver.metrics;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableStat;
+import java.time.Duration;
-public class TabletServerScanMetrics extends TServerMetrics {
+import org.apache.accumulo.core.metrics.MetricsProducer;
- private final MutableStat scans;
- private final MutableStat resultsPerScan;
- private final MutableStat yields;
+import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
- public TabletServerScanMetrics() {
- super("Scans");
+public class TabletServerScanMetrics implements MetricsProducer {
- MetricsRegistry registry = super.getRegistry();
- scans = registry.newStat("scan", "Scans", "Ops", "Count", true);
- resultsPerScan = registry.newStat("result", "Results per scan", "Ops", "Count", true);
- yields = registry.newStat("yield", "Yields", "Ops", "Count", true);
- }
+ private Timer scans;
+ private DistributionSummary resultsPerScan;
+ private DistributionSummary yields;
public void addScan(long value) {
- scans.add(value);
+ scans.record(Duration.ofMillis(value));
}
public void addResult(long value) {
- resultsPerScan.add(value);
+ resultsPerScan.record(value);
}
public void addYield(long value) {
- yields.add(value);
+ yields.record(value);
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ scans = Timer.builder(METRICS_SCAN).description("Scans").register(registry);
+ resultsPerScan = DistributionSummary.builder(METRICS_SCAN_RESULTS)
+ .description("Results per scan").register(registry);
+ yields =
+ DistributionSummary.builder(METRICS_SCAN_YIELDS).description("yields").register(registry);
}
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetrics.java
index b231e8a..c81e87b 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetrics.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetrics.java
@@ -18,66 +18,67 @@
*/
package org.apache.accumulo.tserver.metrics;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableCounterLong;
-import org.apache.hadoop.metrics2.lib.MutableStat;
+import java.time.Duration;
-public class TabletServerUpdateMetrics extends TServerMetrics {
+import org.apache.accumulo.core.metrics.MetricsProducer;
- private final MutableCounterLong permissionErrorsCounter;
- private final MutableCounterLong unknownTabletErrorsCounter;
- private final MutableCounterLong constraintViolationsCounter;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
- private final MutableStat commitPrepStat;
- private final MutableStat walogWriteTimeStat;
- private final MutableStat commitTimeStat;
- private final MutableStat mutationArraySizeStat;
+public class TabletServerUpdateMetrics implements MetricsProducer {
- public TabletServerUpdateMetrics() {
- super("Updates");
-
- MetricsRegistry registry = super.getRegistry();
- permissionErrorsCounter = registry.newCounter("permissionErrors", "Permission Errors", 0L);
- unknownTabletErrorsCounter =
- registry.newCounter("unknownTabletErrors", "Unknown Tablet Errors", 0L);
- constraintViolationsCounter =
- registry.newCounter("constraintViolations", "Table Constraint Violations", 0L);
-
- commitPrepStat =
- registry.newStat("commitPrep", "preparing to commit mutations", "Ops", "Time", true);
- walogWriteTimeStat =
- registry.newStat("waLogWriteTime", "writing mutations to WAL", "Ops", "Time", true);
- commitTimeStat = registry.newStat("commitTime", "committing mutations", "Ops", "Time", true);
- mutationArraySizeStat =
- registry.newStat("mutationArraysSize", "mutation array", "ops", "Size", true);
- }
+ private Counter permissionErrorsCounter;
+ private Counter unknownTabletErrorsCounter;
+ private Counter constraintViolationsCounter;
+ private Timer commitPrepStat;
+ private Timer walogWriteTimeStat;
+ private Timer commitTimeStat;
+ private DistributionSummary mutationArraySizeStat;
public void addPermissionErrors(long value) {
- permissionErrorsCounter.incr(value);
+ permissionErrorsCounter.increment(value);
}
public void addUnknownTabletErrors(long value) {
- unknownTabletErrorsCounter.incr(value);
+ unknownTabletErrorsCounter.increment(value);
}
- public void addMutationArraySize(long value) {
- mutationArraySizeStat.add(value);
+ public void addConstraintViolations(long value) {
+ constraintViolationsCounter.increment(value);
}
public void addCommitPrep(long value) {
- commitPrepStat.add(value);
- }
-
- public void addConstraintViolations(long value) {
- constraintViolationsCounter.incr(value);
+ commitPrepStat.record(Duration.ofMillis(value));
}
public void addWalogWriteTime(long value) {
- walogWriteTimeStat.add(value);
+ walogWriteTimeStat.record(Duration.ofMillis(value));
}
public void addCommitTime(long value) {
- commitTimeStat.add(value);
+ commitTimeStat.record(Duration.ofMillis(value));
+ }
+
+ public void addMutationArraySize(long value) {
+ mutationArraySizeStat.record(value);
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ permissionErrorsCounter = registry.counter(METRICS_UPDATE_ERRORS, "type", "permission");
+ unknownTabletErrorsCounter = registry.counter(METRICS_UPDATE_ERRORS, "type", "unknown.tablet");
+ constraintViolationsCounter =
+ registry.counter(METRICS_UPDATE_ERRORS, "type", "constraint.violation");
+ commitPrepStat = Timer.builder(METRICS_UPDATE_COMMIT_PREP)
+ .description("preparing to commit mutations").register(registry);
+ walogWriteTimeStat = Timer.builder(METRICS_UPDATE_WALOG_WRITE)
+ .description("writing mutations to WAL").register(registry);
+ commitTimeStat =
+ Timer.builder(METRICS_UPDATE_COMMIT).description("committing mutations").register(registry);
+ mutationArraySizeStat = DistributionSummary.builder(METRICS_UPDATE_MUTATION_ARRAY_SIZE)
+ .description("mutation array").register(registry);
}
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index 709ed43..0d324e4 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
@@ -208,17 +208,26 @@ public class Tablet {
private final TabletStatsKeeper timer = new TabletStatsKeeper();
- private final Rate queryRate = new Rate(0.95);
- private long queryCount = 0;
+ /**
+ * Counts are maintained in this object and reported out with the Micrometer metrics via
+ * TabletServerMetricsUtil
+ */
+ private long lookupCount = 0;
+ private long queryResultCount = 0;
+ private long queryResultBytes = 0;
+ private long ingestCount = 0;
+ private long ingestBytes = 0;
+ private final AtomicLong scannedCount = new AtomicLong(0);
+ /**
+ * Rates are calculated here in the Tablet for use in the Monitor but we do not emit them as
+ * metrics. Rates can be calculated from the "Count" metrics above by downstream systems.
+ */
+ private final Rate queryRate = new Rate(0.95);
private final Rate queryByteRate = new Rate(0.95);
- private long queryBytes = 0;
-
private final Rate ingestRate = new Rate(0.95);
- private long ingestCount = 0;
-
private final Rate ingestByteRate = new Rate(0.95);
- private long ingestBytes = 0;
+ private final Rate scannedRate = new Rate(0.95);
private final Deriver<byte[]> defaultSecurityLabel;
@@ -228,9 +237,6 @@ public class Tablet {
private volatile long numEntries = 0;
private volatile long numEntriesInMemory = 0;
- private final Rate scannedRate = new Rate(0.95);
- private final AtomicLong scannedCount = new AtomicLong(0);
-
// Files that are currently in the process of bulk importing. Access to this is protected by the
// tablet lock.
private final Set<TabletFile> bulkImporting = new HashSet<>();
@@ -663,6 +669,7 @@ public class Tablet {
try {
SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
+ lookupCount++;
result = lookup(iter, ranges, results, scanParams, maxResultSize);
return result;
} catch (IOException ioe) {
@@ -674,9 +681,9 @@ public class Tablet {
dataSource.close(false);
synchronized (this) {
- queryCount += results.size();
+ queryResultCount += results.size();
if (result != null) {
- queryBytes += result.dataSize;
+ queryResultBytes += result.dataSize;
}
}
}
@@ -1774,18 +1781,34 @@ public class Tablet {
return scannedRate.rate();
}
- public long totalQueries() {
- return this.queryCount;
+ public long totalQueriesResults() {
+ return this.queryResultCount;
}
public long totalIngest() {
return this.ingestCount;
}
+ public long totalIngestBytes() {
+ return this.ingestBytes;
+ }
+
+ public long totalQueryResultsBytes() {
+ return this.queryResultBytes;
+ }
+
+ public long totalScannedCount() {
+ return this.scannedCount.get();
+ }
+
+ public long totalLookupCount() {
+ return this.lookupCount;
+ }
+
// synchronized?
public void updateRates(long now) {
- queryRate.update(now, queryCount);
- queryByteRate.update(now, queryBytes);
+ queryRate.update(now, queryResultCount);
+ queryByteRate.update(now, queryResultBytes);
ingestRate.update(now, ingestCount);
ingestByteRate.update(now, ingestBytes);
scannedRate.update(now, scannedCount.get());
@@ -2140,8 +2163,8 @@ public class Tablet {
}
public synchronized void updateQueryStats(int size, long numBytes) {
- queryCount += size;
- queryBytes += numBytes;
+ queryResultCount += size;
+ queryResultBytes += numBytes;
}
public void updateTimer(Operation operation, long queued, long start, long count,
diff --git a/test/pom.xml b/test/pom.xml
index ad49aa1..20a3ab0 100644
--- a/test/pom.xml
+++ b/test/pom.xml
@@ -63,6 +63,15 @@
<artifactId>commons-io</artifactId>
</dependency>
<dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-registry-statsd</artifactId>
+ <version>${micrometer.metrics.version}</version>
+ </dependency>
+ <dependency>
<groupId>jakarta.servlet</groupId>
<artifactId>jakarta.servlet-api</artifactId>
</dependency>
@@ -308,15 +317,6 @@
</execution>
</executions>
</plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <excludes>
- <exclude>hadoop-metrics2-accumulo.properties</exclude>
- </excludes>
- </configuration>
- </plugin>
</plugins>
</build>
<profiles>
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
new file mode 100644
index 0000000..62f433c
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
@@ -0,0 +1,185 @@
+/*
+ * 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.accumulo.test.compaction;
+
+import static org.apache.accumulo.minicluster.ServerType.TABLET_SERVER;
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.compactor.Compactor;
+import org.apache.accumulo.coordinator.CompactionCoordinator;
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.util.threads.Threads;
+import org.apache.accumulo.fate.util.UtilWaitThread;
+import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl.ProcessInfo;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.miniclusterImpl.ProcessReference;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class ExternalCompactionMetricsIT extends AccumuloClusterHarness
+ implements MiniClusterConfigurationCallback {
+
+ private static TestStatsDSink sink;
+
+ @BeforeClass
+ public static void before() throws Exception {
+ sink = new TestStatsDSink();
+ }
+
+ @AfterClass
+ public static void after() throws Exception {
+ if (sink != null) {
+ sink.close();
+ }
+ }
+
+ @Override
+ public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration coreSite) {
+ ExternalCompactionUtils.configureMiniCluster(cfg, coreSite);
+
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY, TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+ }
+
+ @Test
+ public void testMetrics() throws Exception {
+ Collection<ProcessReference> tservers =
+ ((MiniAccumuloClusterImpl) getCluster()).getProcesses().get(ServerType.TABLET_SERVER);
+ assertEquals(2, tservers.size());
+ // kill one tserver so that queue metrics are not spread across tservers
+ ((MiniAccumuloClusterImpl) getCluster()).killProcess(TABLET_SERVER, tservers.iterator().next());
+ ProcessInfo c1 = null, c2 = null, coord = null;
+ String[] names = getUniqueNames(2);
+ try (final AccumuloClient client =
+ Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
+ String table1 = names[0];
+ ExternalCompactionUtils.createTable(client, table1, "cs1", 5);
+
+ String table2 = names[1];
+ ExternalCompactionUtils.createTable(client, table2, "cs2", 10);
+
+ ExternalCompactionUtils.writeData(client, table1);
+ ExternalCompactionUtils.writeData(client, table2);
+
+ final LinkedBlockingQueue<Metric> queueMetrics = new LinkedBlockingQueue<>();
+ final AtomicBoolean shutdownTailer = new AtomicBoolean(false);
+
+ Thread thread = Threads.createThread("metric-tailer", () -> {
+ while (!shutdownTailer.get()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String s : statsDMetrics) {
+ if (shutdownTailer.get()) {
+ break;
+ }
+ if (s.startsWith(MetricsProducer.METRICS_MAJC_QUEUED)) {
+ queueMetrics.add(TestStatsDSink.parseStatsDMetric(s));
+ }
+ }
+ }
+ });
+ thread.start();
+
+ ExternalCompactionUtils.compact(client, table1, 7, "DCQ1", false);
+ ExternalCompactionUtils.compact(client, table2, 13, "DCQ2", false);
+
+ boolean sawDCQ1_5 = false;
+ boolean sawDCQ2_10 = false;
+
+ // wait until expected number of queued are seen in metrics
+ while (!sawDCQ1_5 || !sawDCQ2_10) {
+ Metric qm = queueMetrics.take();
+ sawDCQ1_5 |= match(qm, "DCQ1", "5");
+ sawDCQ2_10 |= match(qm, "DCQ2", "10");
+ }
+
+ // start compactors
+ c1 = ((MiniAccumuloClusterImpl) getCluster()).exec(Compactor.class, "-q", "DCQ1");
+ c2 = ((MiniAccumuloClusterImpl) getCluster()).exec(Compactor.class, "-q", "DCQ2");
+ coord = ((MiniAccumuloClusterImpl) getCluster()).exec(CompactionCoordinator.class);
+
+ boolean sawDCQ1_0 = false;
+ boolean sawDCQ2_0 = false;
+
+ // wait until queued goes to zero in metrics
+ while (!sawDCQ1_0 || !sawDCQ2_0) {
+ Metric qm = queueMetrics.take();
+ sawDCQ1_0 |= match(qm, "DCQ1", "0");
+ sawDCQ2_0 |= match(qm, "DCQ2", "0");
+ }
+
+ shutdownTailer.set(true);
+ thread.join();
+
+ // Wait for all external compactions to complete
+ long count;
+ do {
+ UtilWaitThread.sleep(100);
+ try (TabletsMetadata tm = getCluster().getServerContext().getAmple().readTablets()
+ .forLevel(DataLevel.USER).fetch(ColumnType.ECOMP).build()) {
+ count = tm.stream().flatMap(t -> t.getExternalCompactions().keySet().stream()).count();
+ }
+ } while (count > 0);
+
+ ExternalCompactionUtils.verify(client, table1, 7);
+ ExternalCompactionUtils.verify(client, table2, 13);
+
+ } finally {
+ ExternalCompactionUtils.stopProcesses(c1, c2, coord);
+ // We stopped the TServer and started our own, restart the original TabletServers
+ ((MiniAccumuloClusterImpl) getCluster()).getClusterControl().start(ServerType.TABLET_SERVER);
+ }
+ }
+
+ private static boolean match(Metric input, String queue, String value) {
+ if (input.getTags() != null) {
+ String id = input.getTags().get("id");
+ if (id != null && id.equals("e." + queue) && input.getValue().equals(value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+}
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java
index 0923489..56caa4b 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java
@@ -24,23 +24,17 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Optional;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -69,7 +63,6 @@ import org.apache.accumulo.core.iterators.Filter;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
-import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionFinalState;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionFinalState.FinalState;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
@@ -85,9 +78,6 @@ import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl.ProcessInfo;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
-import org.apache.accumulo.miniclusterImpl.ProcessReference;
-import org.apache.commons.io.input.Tailer;
-import org.apache.commons.io.input.TailerListenerAdapter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.junit.Test;
@@ -523,95 +513,4 @@ public class ExternalCompaction_1_IT extends AccumuloClusterHarness
}
}
- private static Optional<String> extract(String input, String regex) {
- Pattern pattern = Pattern.compile(regex);
- Matcher matcher = pattern.matcher(input);
- if (matcher.matches()) {
- return Optional.of(matcher.group(1));
- }
-
- return Optional.empty();
- }
-
- @Test
- public void testMetrics() throws Exception {
- Collection<ProcessReference> tservers =
- ((MiniAccumuloClusterImpl) getCluster()).getProcesses().get(ServerType.TABLET_SERVER);
- assertEquals(2, tservers.size());
- // kill one tserver so that queue metrics are not spread across tservers
- ((MiniAccumuloClusterImpl) getCluster()).killProcess(TABLET_SERVER, tservers.iterator().next());
- ProcessInfo c1 = null, c2 = null, coord = null;
- String[] names = getUniqueNames(2);
- try (final AccumuloClient client =
- Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- String table1 = names[0];
- ExternalCompactionUtils.createTable(client, table1, "cs1", 5);
-
- String table2 = names[1];
- ExternalCompactionUtils.createTable(client, table2, "cs2", 10);
-
- ExternalCompactionUtils.writeData(client, table1);
- ExternalCompactionUtils.writeData(client, table2);
-
- LinkedBlockingQueue<String> queueMetrics = new LinkedBlockingQueue<>();
-
- Tailer tailer =
- Tailer.create(new File("./target/tserver.metrics"), new TailerListenerAdapter() {
- @Override
- public void handle(final String line) {
- extract(line, ".*(DCQ1_queued=[0-9]+).*").ifPresent(queueMetrics::add);
- extract(line, ".*(DCQ2_queued=[0-9]+).*").ifPresent(queueMetrics::add);
- }
- });
-
- ExternalCompactionUtils.compact(client, table1, 7, "DCQ1", false);
- ExternalCompactionUtils.compact(client, table2, 13, "DCQ2", false);
-
- boolean sawDCQ1_5 = false;
- boolean sawDCQ2_10 = false;
-
- // wait until expected number of queued are seen in metrics
- while (!sawDCQ1_5 || !sawDCQ2_10) {
- String qm = queueMetrics.take();
- sawDCQ1_5 |= qm.equals("DCQ1_queued=5");
- sawDCQ2_10 |= qm.equals("DCQ2_queued=10");
- }
-
- // start compactors
- c1 = ((MiniAccumuloClusterImpl) getCluster()).exec(Compactor.class, "-q", "DCQ1");
- c2 = ((MiniAccumuloClusterImpl) getCluster()).exec(Compactor.class, "-q", "DCQ2");
- coord = ((MiniAccumuloClusterImpl) getCluster()).exec(CompactionCoordinator.class);
-
- boolean sawDCQ1_0 = false;
- boolean sawDCQ2_0 = false;
-
- // wait until queued goes to zero in metrics
- while (!sawDCQ1_0 || !sawDCQ2_0) {
- String qm = queueMetrics.take();
- sawDCQ1_0 |= qm.equals("DCQ1_queued=0");
- sawDCQ2_0 |= qm.equals("DCQ2_queued=0");
- }
-
- tailer.stop();
-
- // Wait for all external compactions to complete
- long count;
- do {
- UtilWaitThread.sleep(100);
- try (TabletsMetadata tm = getCluster().getServerContext().getAmple().readTablets()
- .forLevel(DataLevel.USER).fetch(ColumnType.ECOMP).build()) {
- count = tm.stream().flatMap(t -> t.getExternalCompactions().keySet().stream()).count();
- }
- } while (count > 0);
-
- ExternalCompactionUtils.verify(client, table1, 7);
- ExternalCompactionUtils.verify(client, table2, 13);
-
- } finally {
- ExternalCompactionUtils.stopProcesses(c1, c2, coord);
- // We stopped the TServer and started our own, restart the original TabletServers
- ((MiniAccumuloClusterImpl) getCluster()).getClusterControl().start(ServerType.TABLET_SERVER);
- }
- }
-
}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/FateConcurrencyIT.java b/test/src/main/java/org/apache/accumulo/test/functional/FateConcurrencyIT.java
index abfb355..39bb3d7 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/FateConcurrencyIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/FateConcurrencyIT.java
@@ -507,5 +507,6 @@ public class FateConcurrencyIT extends AccumuloClusterHarness {
t.getTableName());
}
});
+
}
}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/GcMetricsIT.java b/test/src/main/java/org/apache/accumulo/test/functional/GcMetricsIT.java
deleted file mode 100644
index 5d76a21..0000000
--- a/test/src/main/java/org/apache/accumulo/test/functional/GcMetricsIT.java
+++ /dev/null
@@ -1,197 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.test.functional;
-
-import static java.util.stream.Collectors.toMap;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assume.assumeTrue;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.Objects;
-import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.Stream;
-
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.gc.metrics.GcMetrics;
-import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
-import org.apache.accumulo.test.metrics.MetricsFileTailer;
-import org.apache.hadoop.conf.Configuration;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Functional test that uses a hadoop metrics 2 file sink to read published metrics for
- * verification.
- */
-public class GcMetricsIT extends ConfigurableMacBase {
-
- private static final Logger log = LoggerFactory.getLogger(GcMetricsIT.class);
- private static final int NUM_TAIL_ATTEMPTS = 20;
- private static final long TAIL_DELAY = 5_000;
- private static final Pattern metricLinePattern = Pattern.compile("^(?<timestamp>\\d+).*");
- private static final String[] EXPECTED_METRIC_KEYS = new String[] {"AccGcCandidates",
- "AccGcDeleted", "AccGcErrors", "AccGcFinished", "AccGcInUse", "AccGcPostOpDuration",
- "AccGcRunCycleCount", "AccGcStarted", "AccGcWalCandidates", "AccGcWalDeleted",
- "AccGcWalErrors", "AccGcWalFinished", "AccGcWalInUse", "AccGcWalStarted"};
-
- @Override
- protected void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
- cfg.setProperty(Property.GC_METRICS_ENABLED, "true");
- cfg.setProperty(Property.GC_CYCLE_START, "5s");
- cfg.setProperty(Property.GC_CYCLE_DELAY, "15s");
- }
-
- @Override
- protected int defaultTimeoutSeconds() {
- return 4 * 60;
- }
-
- private final MetricsFileTailer gcTail = new MetricsFileTailer("accumulo.sink.file-gc");
-
- @Before
- public void startTailer() {
- gcTail.startDaemonThread();
- }
-
- @After
- public void stopTailer() {
- gcTail.close();
- }
-
- @Test
- public void gcMetricsPublished() throws Exception {
- assumeTrue("gc metrics are disabled with GC_METRICS_ENABLED=false",
- cluster.getSiteConfiguration().getBoolean(Property.GC_METRICS_ENABLED));
-
- // uncomment for manual jmx / jconsole validation - not for automated testing
- // Thread.sleep(320_000);
-
- long testStart = System.currentTimeMillis();
- log.debug("Test started: {}", testStart);
-
- LineUpdate firstGc = captureMetricsAfterTimestamp(testStart);
- log.debug("First captured metrics finished at {}", firstGc.gcWalFinished);
-
- LineUpdate secondGc = captureMetricsAfterTimestamp(firstGc.gcWalFinished);
- log.debug("Second captured metrics finished at {}", secondGc.gcWalFinished);
-
- firstGc.compareWithSubsequentRun(secondGc);
- }
-
- private static class LineUpdate {
- private final long gcStarted;
- private final long gcFinished;
- private final long gcWalStarted;
- private final long gcWalFinished;
- private final long gcRunCycleCount;
- private final Map<String,Long> values;
-
- LineUpdate(String line) {
- values = parseLine(line);
- gcStarted = values.get("AccGcStarted");
- gcFinished = values.get("AccGcFinished");
- gcWalStarted = values.get("AccGcWalStarted");
- gcWalFinished = values.get("AccGcWalFinished");
- gcRunCycleCount = values.get("AccGcRunCycleCount");
-
- // ensure internal consistency
- assertTrue(Stream.of(EXPECTED_METRIC_KEYS).allMatch(values::containsKey));
- assertTrue(gcStarted <= gcFinished);
- assertTrue(gcWalStarted <= gcWalFinished);
- }
-
- void compareWithSubsequentRun(LineUpdate update) {
- log.debug("First run: {}, Second run: {}", values, update.values);
- assertTrue("first gc should finish before second starts", gcFinished < update.gcStarted);
- assertTrue("first gcWal should finish before second starts",
- gcWalFinished < update.gcWalStarted);
- assertTrue("cycle count should increment", gcRunCycleCount < update.gcRunCycleCount);
- }
-
- /**
- * The hadoop metrics file sink published records as a line with comma separated key=value
- * pairs. This method parses the line and extracts the key, value pair from metrics that start
- * with AccGc and returns them in a sorted map.
- *
- * @return a map of the metrics that start with AccGc
- */
- private static Map<String,Long> parseLine(String line) {
- log.debug("Line received:{}", line);
- Map<String,Long> values = Collections.emptyMap();
- if (line != null) {
- values = Stream.of(line.split(",")).map(String::trim)
- .filter(s -> s.startsWith(GcMetrics.GC_METRIC_PREFIX)).map(s -> s.split("="))
- .collect(toMap(a -> a[0], a -> Long.parseLong(a[1]), (a, b) -> {
- throw new IllegalArgumentException("Metric field found with two values '" + a
- + "' and '" + b + "' in GC Metrics line: " + line);
- }, TreeMap::new));
- }
- log.debug("Mapped values:{}", values);
- return values;
- }
- }
-
- private LineUpdate captureMetricsAfterTimestamp(final long timestamp) throws Exception {
- for (int count = 0; count < NUM_TAIL_ATTEMPTS; count++) {
- String line = gcTail.getLast();
- // check for valid metrics line update since the previous update
- if (line != null && isValidRecentMetricsLine(line, timestamp)) {
- LineUpdate lineUpdate = new LineUpdate(line);
- // capture metrics for a regular gc event and a wal gc event that both occurred
- // after the specified timestamp, and not just the same entry in the metrics file
- // from a subsequent poll that represents the previous gc events;
- // note that regular gc events can be updated in the metrics before the subsequent
- // wal gc event that follows it, so the wal gc timestamps may be earlier than the
- // regular gc timestamps; that's okay, as long as they both occur after the provided
- // timestamp, and are internally consistent (start times < finish times)
- if (timestamp < lineUpdate.gcStarted && timestamp < lineUpdate.gcWalStarted) {
- return lineUpdate;
- }
- }
- Thread.sleep(TAIL_DELAY);
- }
- throw new IllegalStateException(
- String.format("File source update not received after %d tries in %d sec", NUM_TAIL_ATTEMPTS,
- TimeUnit.MILLISECONDS.toSeconds(TAIL_DELAY * NUM_TAIL_ATTEMPTS)));
- }
-
- private boolean isValidRecentMetricsLine(final String line, final long prevTimestamp) {
- if (Objects.isNull(line)) {
- return false;
- }
- Matcher m = metricLinePattern.matcher(line);
- if (m.matches()) {
- try {
- long timestamp = Long.parseLong(m.group("timestamp"));
- return timestamp > prevTimestamp;
- } catch (NumberFormatException ex) {
- log.debug("Could not parse timestamp from line '{}", line);
- return false;
- }
- }
- return false;
- }
-}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ManagerMetricsIT.java b/test/src/main/java/org/apache/accumulo/test/functional/ManagerMetricsIT.java
deleted file mode 100644
index 40e9666..0000000
--- a/test/src/main/java/org/apache/accumulo/test/functional/ManagerMetricsIT.java
+++ /dev/null
@@ -1,301 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.test.functional;
-
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assume.assumeTrue;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-
-import org.apache.accumulo.core.client.Accumulo;
-import org.apache.accumulo.core.client.AccumuloClient;
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.harness.AccumuloClusterHarness;
-import org.apache.accumulo.test.metrics.MetricsFileTailer;
-import org.apache.accumulo.test.util.SlowOps;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Functional test that uses a hadoop metrics 2 file sink to read published metrics for
- * verification.
- */
-public class ManagerMetricsIT extends AccumuloClusterHarness {
-
- @Override
- public boolean canRunTest(ClusterType type) {
- return type == ClusterType.MINI;
- }
-
- private static final Logger log = LoggerFactory.getLogger(ManagerMetricsIT.class);
-
- private AccumuloClient accumuloClient;
-
- private static final int NUM_TAIL_ATTEMPTS = 20;
- private static final long TAIL_DELAY = 5_000;
-
- // number of tables / concurrent compactions used during testing.
- private final int tableCount = 1;
-
- private long maxWait;
-
- private static final Set<String> REQUIRED_METRIC_KEYS =
- Set.of("currentFateOps", "totalFateOps", "totalZkConnErrors", "FateTxState_NEW",
- "FateTxState_IN_PROGRESS", "FateTxState_FAILED_IN_PROGRESS", "FateTxState_FAILED",
- "FateTxState_SUCCESSFUL", "FateTxState_UNKNOWN");
-
- private static final Set<String> OPTIONAL_METRIC_KEYS = Set.of("FateTxOpType_CompactRange");
-
- private final MetricsFileTailer metricsTail = new MetricsFileTailer("accumulo.sink.file-manager");
-
- @Before
- public void setup() {
- accumuloClient = Accumulo.newClient().from(getClientProps()).build();
- maxWait = defaultTimeoutSeconds() <= 0 ? 60_000 : ((defaultTimeoutSeconds() * 1000) / 2);
- metricsTail.startDaemonThread();
- }
-
- @After
- public void cleanup() {
- metricsTail.close();
- }
-
- @Override
- protected int defaultTimeoutSeconds() {
- return 4 * 60;
- }
-
- /**
- * Validates that the expected metrics are published - this excludes the dynamic metrics derived
- * from operation types.
- */
- @Test
- public void metricsPublished() throws AccumuloException, AccumuloSecurityException {
-
- assumeTrue(accumuloClient.instanceOperations().getSystemConfiguration()
- .get(Property.MANAGER_FATE_METRICS_ENABLED.getKey()).compareTo("true") == 0);
-
- log.trace("Client started, properties:{}", accumuloClient.properties());
-
- MetricsFileTailer.LineUpdate firstUpdate =
- metricsTail.waitForUpdate(-1, NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- Map<String,Long> firstSeenMap = parseLine(firstUpdate.getLine());
-
- log.info("L:{}", firstUpdate.getLine());
- log.info("Expected: ({})", REQUIRED_METRIC_KEYS.size());
- log.info("M({}):{}", firstSeenMap.size(), firstSeenMap);
-
- assertTrue(lookForExpectedKeys(firstSeenMap));
- sanity(firstSeenMap);
-
- MetricsFileTailer.LineUpdate nextUpdate =
- metricsTail.waitForUpdate(firstUpdate.getLastUpdate(), NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- Map<String,Long> updateSeenMap = parseLine(nextUpdate.getLine());
-
- log.debug("Line received:{}", nextUpdate.getLine());
- log.trace("Mapped values:{}", updateSeenMap);
-
- assertTrue(lookForExpectedKeys(updateSeenMap));
- sanity(updateSeenMap);
-
- validate(firstSeenMap, updateSeenMap);
- }
-
- /**
- * Run a few compactions - this should trigger the a dynamic op type to be included in the
- * metrics.
- */
- @Test
- public void compactionMetrics() throws AccumuloSecurityException, AccumuloException {
-
- assumeTrue(accumuloClient.instanceOperations().getSystemConfiguration()
- .get(Property.MANAGER_FATE_METRICS_ENABLED.getKey()).compareTo("true") == 0);
-
- MetricsFileTailer.LineUpdate firstUpdate =
- metricsTail.waitForUpdate(-1, NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- List<SlowOps> tables = new ArrayList<>();
-
- for (int i = 0; i < tableCount; i++) {
- String uniqueName = getUniqueNames(1)[0] + "_" + i;
- SlowOps.setExpectedCompactions(accumuloClient, tableCount);
- SlowOps gen = new SlowOps(accumuloClient, uniqueName, maxWait);
- tables.add(gen);
- gen.startCompactTask();
- }
-
- // check file tailer here....
- MetricsFileTailer.LineUpdate nextUpdate =
- metricsTail.waitForUpdate(firstUpdate.getLastUpdate(), NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- log.info("Received metrics {}", nextUpdate);
-
- Map<String,String> results = blockForRequiredTables();
-
- assertFalse(results.isEmpty());
- log.error("IN_PROGRESS: {}", results.get("FateTxState_IN_PROGRESS"));
-
- assertTrue(Long.parseLong(results.get("FateTxState_IN_PROGRESS")) >= tableCount);
- assertTrue(Long.parseLong(results.get("FateTxOpType_CompactRange")) >= tableCount);
-
- for (String k : OPTIONAL_METRIC_KEYS) {
- assertTrue(results.containsKey(k));
- assertTrue(Long.parseLong(results.get(k)) >= tableCount);
- }
-
- for (SlowOps t : tables) {
- try {
- accumuloClient.tableOperations().cancelCompaction(t.getTableName());
- // block if compaction still running
- boolean cancelled = t.blockWhileCompactionRunning();
- if (!cancelled) {
- log.info("Failed to cancel compaction during multiple compaction test clean-up for {}",
- t.getTableName());
- }
- } catch (AccumuloSecurityException | TableNotFoundException | AccumuloException ex) {
- log.debug("Exception thrown during multiple table test clean-up", ex);
- }
- }
-
- // wait for one more metrics update after compactions cancelled.
- MetricsFileTailer.LineUpdate update =
- metricsTail.waitForUpdate(0L, NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- metricsTail.waitForUpdate(update.getLastUpdate(), NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- results = metricsTail.parseLine("");
-
- log.info("Received metrics {}", results);
-
- }
-
- private Map<String,String> blockForRequiredTables() {
-
- MetricsFileTailer.LineUpdate update =
- metricsTail.waitForUpdate(0L, NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- for (int i = 0; i < 20; i++) {
-
- update = metricsTail.waitForUpdate(update.getLastUpdate(), NUM_TAIL_ATTEMPTS, TAIL_DELAY);
-
- log.info("Received metrics update {}", update);
-
- Map<String,String> results = metricsTail.parseLine("");
-
- if (results != null && !results.isEmpty()
- && Long.parseLong(results.get("currentFateOps")) >= tableCount) {
- log.info("Found required number of fate operations");
- return results;
- }
- try {
- Thread.sleep(10_000);
- } catch (InterruptedException iex) {
- Thread.currentThread().interrupt();
- return Collections.emptyMap();
- }
-
- }
- return Collections.emptyMap();
- }
-
- /**
- * Validate metrics for consistency withing a run cycle.
- *
- * @param values
- * map of values from one run cycle.
- */
- private void sanity(final Map<String,Long> values) {
-
- assertTrue(values.get("currentFateOps") <= values.get("totalFateOps"));
-
- long total = values.entrySet().stream().filter(x -> x.getKey().startsWith("FateTxState_"))
- .mapToLong(Map.Entry::getValue).sum();
-
- assertTrue(total >= values.get("currentFateOps"));
- }
-
- /**
- * A series of sanity checks for the metrics between different update cycles, some values should
- * be at least different, and some of the checks can include ordering.
- *
- * @param firstSeen
- * map of first metric update
- * @param nextSeen
- * map of a later metric update.
- */
- private void validate(Map<String,Long> firstSeen, Map<String,Long> nextSeen) {
- // total fate ops should not decrease.
- assertTrue(firstSeen.get("totalFateOps") <= nextSeen.get("totalFateOps"));
- }
-
- /**
- * The hadoop metrics file sink published records as a line with comma separated key=value pairs.
- * This method parses the line and extracts the key, value pair from metrics that start with AccGc
- * and returns them in a sort map.
- *
- * @param line
- * a line from the metrics system file sink.
- * @return a map of the metrics that start with AccGc
- */
- private Map<String,Long> parseLine(final String line) {
-
- if (line == null) {
- return Collections.emptyMap();
- }
-
- Map<String,Long> m = new TreeMap<>();
-
- String[] csvTokens = line.split(",");
-
- for (String token : csvTokens) {
- token = token.trim();
- String[] parts = token.split("=");
- if (REQUIRED_METRIC_KEYS.contains(parts[0])) {
- m.put(parts[0], Long.parseLong(parts[1]));
- }
- }
- return m;
- }
-
- private boolean lookForExpectedKeys(final Map<String,Long> received) {
-
- for (String e : REQUIRED_METRIC_KEYS) {
- if (!received.containsKey(e)) {
- log.info("Couldn't find {}", e);
- return false;
- }
- }
-
- return true;
- }
-}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java b/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
index df4bcc4..2001353 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
@@ -45,7 +45,6 @@ import org.apache.accumulo.fate.zookeeper.ServiceLock.LockWatcher;
import org.apache.accumulo.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.rpc.ServerAddress;
import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.rpc.ThriftServerType;
@@ -110,10 +109,9 @@ public class ZombieTServer {
final ThriftClientHandler tch = new ThriftClientHandler(context, watcher);
Processor<Iface> processor = new Processor<>(tch);
ServerAddress serverPort =
- TServerUtils.startTServer(Metrics.initSystem(ZombieTServer.class.getSimpleName()),
- context.getConfiguration(), ThriftServerType.CUSTOM_HS_HA, processor, "ZombieTServer",
- "walking dead", 2, ThreadPools.DEFAULT_TIMEOUT_MILLISECS, 1000, 10 * 1024 * 1024, null,
- null, -1, HostAndPort.fromParts("0.0.0.0", port));
+ TServerUtils.startTServer(context.getConfiguration(), ThriftServerType.CUSTOM_HS_HA,
+ processor, "ZombieTServer", "walking dead", 2, ThreadPools.DEFAULT_TIMEOUT_MILLISECS,
+ 1000, 10 * 1024 * 1024, null, null, -1, HostAndPort.fromParts("0.0.0.0", port));
String addressString = serverPort.address.toString();
var zLockPath =
diff --git a/test/src/main/java/org/apache/accumulo/test/metrics/MetricsFileTailer.java b/test/src/main/java/org/apache/accumulo/test/metrics/MetricsFileTailer.java
deleted file mode 100644
index b13ba7e..0000000
--- a/test/src/main/java/org/apache/accumulo/test/metrics/MetricsFileTailer.java
+++ /dev/null
@@ -1,360 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.test.metrics;
-
-import static java.nio.charset.StandardCharsets.UTF_8;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.RandomAccessFile;
-import java.net.URL;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.commons.configuration2.Configuration;
-import org.apache.commons.configuration2.PropertiesConfiguration;
-import org.apache.commons.configuration2.ex.ConfigurationException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
-
-/**
- * This class allows testing of the publishing to the hadoop metrics system by processing a file for
- * metric records (written as a line.) The file should be configured using the hadoop metrics
- * properties as a file based sink with the prefix that is provided on instantiation of the
- * instance.
- *
- * This class will simulate tail-ing a file and is intended to be run in a separate thread. When the
- * underlying file has data written, the vaule returned by getLastUpdate will change, and the last
- * line can be retrieved with getLast().
- */
-public class MetricsFileTailer implements AutoCloseable {
-
- private static final Logger log = LoggerFactory.getLogger(MetricsFileTailer.class);
-
- private static final int BUFFER_SIZE = 4;
-
- private final String metricsPrefix;
-
- private final Lock lock = new ReentrantLock();
- private final AtomicBoolean running = new AtomicBoolean(Boolean.FALSE);
-
- private AtomicLong lastUpdate = new AtomicLong(0);
- private long startTime = System.nanoTime();
-
- private int lineCounter = 0;
- private String[] lineBuffer = new String[BUFFER_SIZE];
-
- private final String metricsFilename;
-
- /**
- * Create an instance that will tail a metrics file. The filename / path is determined by the
- * hadoop-metrics-accumulo.properties sink configuration for the metrics prefix that is provided.
- *
- * @param metricsPrefix
- * the prefix in the metrics configuration.
- */
- public MetricsFileTailer(final String metricsPrefix) {
-
- this.metricsPrefix = metricsPrefix;
-
- Configuration sub = loadMetricsConfig();
-
- // dump received configuration keys received.
- if (log.isTraceEnabled()) {
- Iterator<String> keys = sub.getKeys();
- while (keys.hasNext()) {
- log.trace("configuration key:{}", keys.next());
- }
- }
-
- if (sub.containsKey("filename")) {
- metricsFilename = sub.getString("filename");
- } else {
- metricsFilename = "";
- }
-
- }
-
- /**
- * Create an instance by specifying a file directly instead of using the metrics configuration -
- * mainly for testing.
- *
- * @param metricsPrefix
- * generally can be ignored.
- * @param filename
- * the path / file to be monitored.
- */
- MetricsFileTailer(final String metricsPrefix, final String filename) {
- this.metricsPrefix = metricsPrefix;
- metricsFilename = filename;
- }
-
- /**
- * Look for the accumulo metrics configuration file on the classpath and return the subset for the
- * http sink.
- *
- * @return a configuration with http sink properties.
- */
- @SuppressFBWarnings(value = "URLCONNECTION_SSRF_FD",
- justification = "url specified by test code, not unchecked user input")
- private Configuration loadMetricsConfig() {
- final URL propUrl =
- getClass().getClassLoader().getResource(MetricsTestSinkProperties.METRICS_PROP_FILENAME);
-
- if (propUrl == null) {
- throw new IllegalStateException(
- "Could not find " + MetricsTestSinkProperties.METRICS_PROP_FILENAME + " on classpath");
- }
-
- // Read data from this file
- var config = new PropertiesConfiguration();
- try (var reader = new InputStreamReader(propUrl.openStream(), UTF_8)) {
- config.read(reader);
- } catch (ConfigurationException | IOException e) {
- throw new IllegalStateException(
- String.format("Could not find configuration file \'%s\' on classpath",
- MetricsTestSinkProperties.METRICS_PROP_FILENAME));
- }
-
- final Configuration sub = config.subset(metricsPrefix);
-
- if (log.isTraceEnabled()) {
- log.trace("Config {}", config);
- Iterator<String> iterator = sub.getKeys();
- while (iterator.hasNext()) {
- String key = iterator.next();
- log.trace("'{}'='{}'", key, sub.getProperty(key));
- }
- }
-
- return sub;
- }
-
- /**
- * Creates a marker value that increases each time a new line is detected. Clients can use this to
- * determine if a call to getLast() will return a new value. However, this value is <b>NOT</b> a
- * timestamp and should not be interpreted as such. Furthermore, it does not indicate that the
- * metrics being reported have changed, only that a new metrics poll took place and was written to
- * the file. So, if clients need to observe new metrics from a new event, they need to parse the
- * line themselves to look for changed metrics values.
- *
- * @return a marker value set when a new line is available.
- */
- public long getLastUpdate() {
- return lastUpdate.get();
- }
-
- /**
- * Get the last line seen in the file.
- *
- * @return the last line from the file.
- */
- public String getLast() {
- lock.lock();
- try {
-
- int last = (lineCounter % BUFFER_SIZE) - 1;
- if (last < 0) {
- last = BUFFER_SIZE - 1;
- }
- return lineBuffer[last];
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * The hadoop metrics file sink published records as a line with comma separated key=value pairs.
- * This method parses the line and extracts the key, value pair from metrics that start with an
- * optional prefix and returns them in a sort map. If the prefix is null or empty, all keys are
- * accepted.
- *
- * @param prefix
- * optional filter - include metrics that start with provided value..
- * @return a map of the metrics that start with AccGc
- */
- public Map<String,String> parseLine(final String prefix) {
-
- String line = getLast();
-
- if (line == null) {
- return Collections.emptyMap();
- }
-
- Map<String,String> m = new TreeMap<>();
-
- String[] csvTokens = line.split(",");
-
- for (String token : csvTokens) {
- token = token.trim();
- if (filter(prefix, token)) {
- String[] parts = token.split("=");
- m.put(parts[0], parts[1]);
- }
- }
- return m;
- }
-
- private boolean filter(final String prefix, final String candidate) {
- if (candidate == null) {
- return false;
- }
-
- if (prefix == null || prefix.isEmpty()) {
- return true;
- }
- return candidate.startsWith(prefix);
- }
-
- /**
- * A loop that polls for changes and when the file changes, put the last line in a buffer that can
- * be retrieved by clients using getLast().
- */
- private void run() {
-
- long filePos = 0;
-
- File f = new File(metricsFilename);
-
- while (running.get()) {
-
- try {
- Thread.sleep(5_000);
- } catch (InterruptedException ex) {
- running.set(Boolean.FALSE);
- Thread.currentThread().interrupt();
- return;
- }
-
- long len = f.length();
-
- try {
-
- // file truncated? reset position
- if (len < filePos) {
- filePos = 0;
- lock.lock();
- try {
- for (int i = 0; i < BUFFER_SIZE; i++) {
- lineBuffer[i] = "";
- }
- lineCounter = 0;
- } finally {
- lock.unlock();
- }
- }
-
- if (len > filePos) {
- // File must have had something added to it!
- RandomAccessFile raf = new RandomAccessFile(f, "r");
- raf.seek(filePos);
- String line;
- lock.lock();
- try {
- while ((line = raf.readLine()) != null) {
- lineBuffer[lineCounter++ % BUFFER_SIZE] = line;
- }
-
- lastUpdate.set(System.nanoTime() - startTime);
-
- } finally {
- lock.unlock();
- }
- filePos = raf.getFilePointer();
- raf.close();
- }
- } catch (Exception ex) {
- log.info("Error processing metrics file {}", metricsFilename, ex);
- }
- }
- }
-
- public void startDaemonThread() {
- if (running.compareAndSet(false, true)) {
- Thread t = new Thread(() -> this.run());
- t.setDaemon(true);
- t.start();
- }
- }
-
- @Override
- public void close() {
- running.set(Boolean.FALSE);
- }
-
- // utilities to block, waiting for update - call from process thread
-
- public static class LineUpdate {
- private final long lastUpdate;
- private final String line;
-
- public LineUpdate(long lastUpdate, String line) {
- this.lastUpdate = lastUpdate;
- this.line = line;
- }
-
- public long getLastUpdate() {
- return lastUpdate;
- }
-
- public String getLine() {
- return line;
- }
-
- @Override
- public String toString() {
- return "LineUpdate{" + "lastUpdate=" + lastUpdate + ", line='" + line + '\'' + '}';
- }
- }
-
- public LineUpdate waitForUpdate(final long prevUpdate, final int maxAttempts, final long delay) {
-
- for (int count = 0; count < maxAttempts; count++) {
-
- String line = getLast();
- long currUpdate = getLastUpdate();
-
- if (line != null && (currUpdate != prevUpdate)) {
- return new LineUpdate(getLastUpdate(), line);
- }
-
- try {
- Thread.sleep(delay);
- } catch (InterruptedException ex) {
- Thread.currentThread().interrupt();
- throw new IllegalStateException(ex);
- }
- }
- // not found - throw exception.
- throw new IllegalStateException(
- String.format("File source update not received after %d tries in %d sec", maxAttempts,
- TimeUnit.MILLISECONDS.toSeconds(delay * maxAttempts)));
- }
-
-}
diff --git a/test/src/main/java/org/apache/accumulo/test/metrics/MetricsIT.java b/test/src/main/java/org/apache/accumulo/test/metrics/MetricsIT.java
new file mode 100644
index 0000000..d891326
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/metrics/MetricsIT.java
@@ -0,0 +1,158 @@
+/*
+ * 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.accumulo.test.metrics;
+
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.ConfigurableMacBase;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.micrometer.core.instrument.MeterRegistry;
+
+public class MetricsIT extends ConfigurableMacBase implements MetricsProducer {
+
+ private static TestStatsDSink sink;
+
+ @BeforeClass
+ public static void before() throws Exception {
+ sink = new TestStatsDSink();
+ }
+
+ @AfterClass
+ public static void after() throws Exception {
+ if (sink != null) {
+ sink.close();
+ }
+ }
+
+ @Override
+ protected void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+ cfg.setProperty(Property.GC_CYCLE_START, "1s");
+ cfg.setProperty(Property.GC_CYCLE_DELAY, "1s");
+ cfg.setProperty(Property.MANAGER_FATE_METRICS_MIN_UPDATE_INTERVAL, "1s");
+
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY, TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+ }
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 4 * 60;
+ }
+
+ @Test
+ public void confirmMetricsPublished() throws Exception {
+ Set<String> flakyMetricNames = new HashSet<>();
+ flakyMetricNames.add(METRICS_GC_WAL_ERRORS);
+
+ Map<String,String> expectedMetricNames = this.getMetricFields();
+ // We might not see these in the course of normal operations
+ expectedMetricNames.remove(METRICS_SCAN_YIELDS);
+ expectedMetricNames.remove(METRICS_UPDATE_ERRORS);
+ expectedMetricNames.remove(METRICS_REPLICATION_QUEUE);
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProperties()).build()) {
+ String tableName = this.getClass().getSimpleName();
+ client.tableOperations().create(tableName);
+ BatchWriterConfig config = new BatchWriterConfig();
+ config.setMaxMemory(0);
+ try (BatchWriter writer = client.createBatchWriter(tableName, config)) {
+ Mutation m = new Mutation("row");
+ m.put("cf", "cq", new Value("value"));
+ writer.addMutation(m);
+ }
+ client.tableOperations().flush(tableName);
+ try (BatchWriter writer = client.createBatchWriter(tableName, config)) {
+ Mutation m = new Mutation("row");
+ m.put("cf", "cq", new Value("value"));
+ writer.addMutation(m);
+ }
+ client.tableOperations().flush(tableName);
+ client.tableOperations().compact(tableName, new CompactionConfig().setWait(true));
+ client.tableOperations().delete(tableName);
+ while (client.tableOperations().exists(tableName)) {
+ Thread.sleep(1000);
+ }
+ }
+
+ Thread.sleep(30000);
+ cluster.stop();
+
+ Map<String,String> seenMetricNames = new HashMap<>();
+
+ List<String> statsDMetrics = sink.getLines();
+ while (!statsDMetrics.isEmpty()) {
+ for (String s : statsDMetrics) {
+ if (!s.startsWith("accumulo")) {
+ continue;
+ }
+ Metric m = TestStatsDSink.parseStatsDMetric(s);
+ boolean hasBeenSeen = seenMetricNames.containsKey(m.getName());
+ boolean isExpectedMetricName = expectedMetricNames.containsKey(m.getName());
+ if (!hasBeenSeen && isExpectedMetricName) {
+ String expectedValue = expectedMetricNames.remove(m.getName());
+ seenMetricNames.put(m.getName(), expectedValue);
+ } else if (!hasBeenSeen && !isExpectedMetricName) {
+ if (!flakyMetricNames.contains(m.getName())) {
+ fail("Found accumulo metric not in expectedMetricNames: " + m.getName());
+ }
+ }
+ if (expectedMetricNames.isEmpty()) {
+ break;
+ }
+ }
+ statsDMetrics = sink.getLines();
+ }
+ if (!expectedMetricNames.isEmpty()) {
+ fail("Did not see all metric names, missing: " + expectedMetricNames.values());
+ }
+
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ // unused
+ }
+
+}
diff --git a/test/src/main/java/org/apache/accumulo/test/metrics/TestStatsDRegistryFactory.java b/test/src/main/java/org/apache/accumulo/test/metrics/TestStatsDRegistryFactory.java
new file mode 100644
index 0000000..674165d
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/metrics/TestStatsDRegistryFactory.java
@@ -0,0 +1,96 @@
+/*
+ * 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.accumulo.test.metrics;
+
+import java.time.Duration;
+
+import org.apache.accumulo.core.metrics.MeterRegistryFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.statsd.StatsdConfig;
+import io.micrometer.statsd.StatsdFlavor;
+import io.micrometer.statsd.StatsdMeterRegistry;
+import io.micrometer.statsd.StatsdProtocol;
+
+public class TestStatsDRegistryFactory implements MeterRegistryFactory {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TestStatsDRegistryFactory.class);
+
+ public static final String SERVER_HOST = "test.meter.registry.host";
+ public static final String SERVER_PORT = "test.meter.registry.port";
+
+ @Override
+ public MeterRegistry create() {
+
+ String host = System.getProperty(SERVER_HOST, null);
+ String port = System.getProperty(SERVER_PORT, null);
+
+ if (host == null || port == null) {
+ throw new IllegalArgumentException("Host and Port cannot be null");
+ }
+
+ LOG.info("host: {}, port:{}", host, port);
+
+ StatsdConfig config = new StatsdConfig() {
+ @Override
+ public StatsdFlavor flavor() {
+ return StatsdFlavor.DATADOG;
+ }
+
+ @Override
+ public boolean enabled() {
+ return true;
+ }
+
+ @Override
+ public String host() {
+ return host;
+ }
+
+ @Override
+ public int port() {
+ return Integer.parseInt(port);
+ }
+
+ @Override
+ public StatsdProtocol protocol() {
+ return StatsdProtocol.UDP;
+ }
+
+ @Override
+ public Duration pollingFrequency() {
+ return Duration.ofSeconds(3);
+ }
+
+ @Override
+ public boolean buffered() {
+ return false;
+ }
+
+ @Override
+ public String get(String key) {
+ return null;
+ }
+ };
+ return StatsdMeterRegistry.builder(config).build();
+ }
+
+}
diff --git a/test/src/main/java/org/apache/accumulo/test/metrics/TestStatsDSink.java b/test/src/main/java/org/apache/accumulo/test/metrics/TestStatsDSink.java
new file mode 100644
index 0000000..b0ec38e
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/metrics/TestStatsDSink.java
@@ -0,0 +1,140 @@
+/*
+ * 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.accumulo.test.metrics;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.accumulo.core.util.threads.Threads;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestStatsDSink implements Closeable {
+
+ public static class Metric {
+ private final String name;
+ private final String value;
+ private final String type;
+ private final Map<String,String> tags = new HashMap<>();
+
+ public Metric(String name, String value, String type) {
+ super();
+ this.name = name;
+ this.value = value;
+ this.type = type;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public Map<String,String> getTags() {
+ return tags;
+ }
+
+ @Override
+ public String toString() {
+ return "Metric [name=" + name + ", value=" + value + ", type=" + type + ", tags=" + tags
+ + "]";
+ }
+ }
+
+ public static Metric parseStatsDMetric(String line) {
+ int idx = line.indexOf(':');
+ String name = line.substring(0, idx);
+ int idx2 = line.indexOf('|');
+ String value = line.substring(idx + 1, idx2);
+ int idx3 = line.indexOf('|', idx2 + 1);
+ String type = line.substring(idx2 + 1, idx3);
+ int idx4 = line.indexOf('#');
+ String tags = line.substring(idx4 + 1);
+ Metric m = new Metric(name, value, type);
+ String[] tag = tags.split(",");
+ for (String t : tag) {
+ String[] p = t.split(":");
+ m.getTags().put(p[0], p[1]);
+ }
+ return m;
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(TestStatsDSink.class);
+
+ private final DatagramSocket sock;
+ private final LinkedBlockingQueue<String> received = new LinkedBlockingQueue<>();
+
+ public TestStatsDSink() throws SocketException {
+ sock = new DatagramSocket();
+ int len = sock.getReceiveBufferSize();
+ Threads.createThread("test-server-thread", () -> {
+ while (!sock.isClosed()) {
+ byte[] buf = new byte[len];
+ DatagramPacket packet = new DatagramPacket(buf, len);
+ try {
+ sock.receive(packet);
+ received.add(new String(packet.getData()));
+ } catch (IOException e) {
+ if (!sock.isClosed()) {
+ LOG.error("Error receiving packet", e);
+ sock.close();
+ }
+ }
+ }
+
+ }).start();
+ }
+
+ public String getHost() {
+ return sock.getLocalAddress().getHostAddress();
+ }
+
+ public int getPort() {
+ return sock.getLocalPort();
+ }
+
+ public List<String> getLines() {
+ List<String> metrics = new ArrayList<>(received.size());
+ received.drainTo(metrics);
+ return metrics;
+ }
+
+ public String getLine() {
+ return received.remove();
+ }
+
+ @Override
+ public void close() {
+ sock.close();
+ }
+}
diff --git a/test/src/main/java/org/apache/accumulo/test/metrics/fate/FateMetricsIT.java b/test/src/main/java/org/apache/accumulo/test/metrics/fate/FateMetricsIT.java
deleted file mode 100644
index 74587e7..0000000
--- a/test/src/main/java/org/apache/accumulo/test/metrics/fate/FateMetricsIT.java
+++ /dev/null
@@ -1,312 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.test.metrics.fate;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-import java.util.List;
-
-import org.apache.accumulo.core.Constants;
-import org.apache.accumulo.fate.ReadOnlyTStore;
-import org.apache.accumulo.fate.Repo;
-import org.apache.accumulo.fate.ZooStore;
-import org.apache.accumulo.fate.zookeeper.ZooReaderWriter;
-import org.apache.accumulo.manager.Manager;
-import org.apache.accumulo.manager.metrics.fate.FateMetrics;
-import org.apache.accumulo.manager.tableOps.ManagerRepo;
-import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.test.categories.ZooKeeperTestingServerTests;
-import org.apache.accumulo.test.zookeeper.ZooKeeperTestingServer;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.Stat;
-import org.easymock.EasyMock;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Test FATE metrics using stubs and in-memory version of supporting infrastructure - a test
- * zookeeper server is used an the FATE repos are stubs, but this should represent the metrics
- * collection execution without needed to stand up a mini cluster to exercise these execution paths.
- */
-@Category({ZooKeeperTestingServerTests.class})
-public class FateMetricsIT {
-
- public static final String INSTANCE_ID = "1234";
- public static final String MOCK_ZK_ROOT = "/accumulo/" + INSTANCE_ID;
- public static final String A_FAKE_SECRET = "aPasswd";
- private static final Logger log = LoggerFactory.getLogger(FateMetricsIT.class);
- private static ZooKeeperTestingServer szk = null;
- private ZooStore<Manager> zooStore = null;
- private ZooKeeper zookeeper = null;
- private ServerContext context = null;
- private Manager manager;
-
- @BeforeClass
- public static void setupZk() {
- // using default zookeeper port - we don't have a full configuration
- szk = new ZooKeeperTestingServer();
- szk.initPaths(MOCK_ZK_ROOT);
- }
-
- @AfterClass
- public static void shutdownZK() throws Exception {
- szk.close();
- }
-
- /**
- * Instantiate a test zookeeper and setup mocks for Manager and Context. The test zookeeper is
- * used create a ZooReaderWriter. The zookeeper used in tests needs to be the one from the
- * zooReaderWriter, not the test server, because the zooReaderWriter sets up ACLs.
- *
- * @throws Exception
- * any exception is a test failure.
- */
- @Before
- public void init() throws Exception {
-
- ZooReaderWriter zooReaderWriter = new ZooReaderWriter(szk.getConn(), 10_0000, A_FAKE_SECRET);
- zookeeper = zooReaderWriter.getZooKeeper();
-
- clear(MOCK_ZK_ROOT);
-
- zooStore = new ZooStore<>(MOCK_ZK_ROOT + Constants.ZFATE, zooReaderWriter);
-
- manager = EasyMock.createMock(Manager.class);
- context = EasyMock.createMock(ServerContext.class);
-
- EasyMock.expect(context.getZooReaderWriter()).andReturn(zooReaderWriter).anyTimes();
- EasyMock.expect(context.getZooKeeperRoot()).andReturn(MOCK_ZK_ROOT).anyTimes();
-
- EasyMock.replay(manager, context);
- }
-
- @After
- public void cleanup() throws Exception {
- clear(MOCK_ZK_ROOT);
- }
-
- /**
- * Validate that the expected metrics values are present in the metrics collector output
- */
- @Test
- public void noFates() {
-
- FateMetrics metrics = new FateMetrics(context, 10);
- metrics.overrideRefresh(0);
-
- InMemTestCollector collector = new InMemTestCollector();
-
- metrics.getMetrics(collector, true);
-
- log.debug("Collector: {}", collector);
-
- assertTrue(collector.contains("currentFateOps"));
- assertTrue(collector.contains("totalFateOps"));
- assertTrue(collector.contains("totalZkConnErrors"));
-
- // Transaction STATES - defined by TStatus.
- assertTrue(collector.contains("FateTxState_NEW"));
- assertTrue(collector.contains("FateTxState_IN_PROGRESS"));
- assertTrue(collector.contains("FateTxState_FAILED_IN_PROGRESS"));
- assertTrue(collector.contains("FateTxState_FAILED"));
- assertTrue(collector.contains("FateTxState_SUCCESSFUL"));
- assertTrue(collector.contains("FateTxState_UNKNOWN"));
-
- // metrics derived from operation types when see - none should have been seen.
- assertFalse(collector.contains("FateTxOpType_FakeOp"));
-
- assertEquals(0L, collector.getValue("FateTxState_IN_PROGRESS"));
- assertEquals(0L, collector.getValue("currentFateOps"));
-
- EasyMock.verify(manager);
-
- }
-
- /**
- * Seed a fake FAKE fate that has a reserved transaction id. This sets a "new" status, but the
- * repo and debug props are not yet set. Verify the the metric collection handles partial
- * transaction states.
- */
- @Test
- public void fateNewStatus() {
-
- long tx1Id = zooStore.create();
- log.debug("ZooStore tx1 id {}", tx1Id);
-
- FateMetrics metrics = new FateMetrics(context, 10);
- metrics.overrideRefresh(0);
-
- InMemTestCollector collector = new InMemTestCollector();
-
- metrics.getMetrics(collector, true);
-
- log.debug("Collector: {}", collector);
-
- assertTrue(collector.contains("FateTxState_NEW"));
- assertEquals(1L, collector.getValue("FateTxState_NEW"));
- assertEquals(1L, collector.getValue("currentFateOps"));
-
- assertTrue(collector.contains("FateTxState_IN_PROGRESS"));
- assertEquals(0L, collector.getValue("FateTxState_IN_PROGRESS"));
-
- EasyMock.verify(manager);
-
- }
-
- /**
- * Seeds the zoo store with a "fake" repo operation with a step, and sets the prop_debug field.
- * This emulates the actions performed with {@link org.apache.accumulo.fate.Fate} for what is
- * expected in zookeeper / the zoo store for an IN_PROGRESS transaction.
- *
- * @throws Exception
- * any exception is a test failure.
- */
- @Test
- public void oneInProgress() throws Exception {
-
- long tx1Id = seedTransaction();
-
- log.debug("FATE tx: {}", prettyStat(
- zookeeper.exists(MOCK_ZK_ROOT + "/fate/" + String.format("tx_%016x", tx1Id), false)));
-
- FateMetrics metrics = new FateMetrics(context, 10);
- metrics.overrideRefresh(0);
-
- InMemTestCollector collector = new InMemTestCollector();
-
- metrics.getMetrics(collector, true);
-
- log.debug("Collector: {}", collector);
-
- assertTrue(collector.contains("FateTxState_IN_PROGRESS"));
- assertEquals(1L, collector.getValue("FateTxState_IN_PROGRESS"));
- assertEquals(1L, collector.getValue("FateTxOpType_FakeOp"));
-
- EasyMock.verify(manager);
- }
-
- private long seedTransaction() throws Exception {
-
- long txId = zooStore.create();
- zooStore.reserve(txId);
-
- zooStore.setStatus(txId, ReadOnlyTStore.TStatus.IN_PROGRESS);
-
- Repo<Manager> repo = new FakeOp();
- zooStore.push(txId, repo);
-
- Repo<Manager> step = new FakeOpStep1();
- zooStore.push(txId, step);
-
- zooStore.setProperty(txId, "debug", repo.getDescription());
-
- zooStore.unreserve(txId, 50);
-
- return txId;
- }
-
- /**
- * builds on the "in progress" transaction - when a transaction completes, the op type metric
- * should not reflect the previous operation that was "in progress".
- *
- * @throws Exception
- * any exception is a test failure.
- */
- @Test
- public void typeClears() throws Exception {
- long txId = seedTransaction();
-
- zooStore.reserve(txId);
- zooStore.setStatus(txId, ReadOnlyTStore.TStatus.SUCCESSFUL);
- zooStore.unreserve(txId, 50);
-
- FateMetrics metrics = new FateMetrics(context, 10);
- metrics.overrideRefresh(0);
-
- InMemTestCollector collector = new InMemTestCollector();
-
- metrics.getMetrics(collector, true);
-
- assertEquals(0L, collector.getValue("FateTxState_IN_PROGRESS"));
- assertEquals(1L, collector.getValue("FateTxState_SUCCESSFUL"));
- assertNull(collector.getValue("FateTxOpType_FakeOp"));
-
- }
-
- String prettyStat(final Stat stat) {
-
- if (stat == null) {
- return "{Stat:[null]}";
- }
-
- return "{Stat:[" + "czxid:" + stat.getCzxid() + ", mzxid:" + stat.getMzxid() + ", ctime: "
- + stat.getCtime() + ", mtime: " + stat.getMtime() + ", version: " + stat.getVersion()
- + ", cversion: " + stat.getCversion() + ", aversion: " + stat.getAversion()
- + ", eph owner: " + stat.getEphemeralOwner() + ", dataLength: " + stat.getDataLength()
- + ", numChildren: " + stat.getNumChildren() + ", pzxid: " + stat.getPzxid() + "}";
-
- }
-
- private void clear(String path) throws Exception {
-
- log.debug("clean up - search: {}", path);
-
- List<String> children = zookeeper.getChildren(path, false);
-
- if (children.isEmpty()) {
- log.debug("clean up - delete path: {}", path);
-
- if (!path.equals(MOCK_ZK_ROOT)) {
- zookeeper.delete(path, -1);
- }
- return;
- }
-
- for (String cp : children) {
- clear(path + "/" + cp);
- }
- }
-
- private static class FakeOp extends ManagerRepo {
- private static final long serialVersionUID = -1L;
-
- @Override
- public Repo<Manager> call(long tid, Manager environment) {
- return null;
- }
- }
-
- private static class FakeOpStep1 extends ManagerRepo {
- private static final long serialVersionUID = -1L;
-
- @Override
- public Repo<Manager> call(long tid, Manager environment) {
- return null;
- }
- }
-}
diff --git a/test/src/main/java/org/apache/accumulo/test/metrics/fate/InMemTestCollector.java b/test/src/main/java/org/apache/accumulo/test/metrics/fate/InMemTestCollector.java
deleted file mode 100644
index 741f917..0000000
--- a/test/src/main/java/org/apache/accumulo/test/metrics/fate/InMemTestCollector.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.test.metrics.fate;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.hadoop.metrics2.AbstractMetric;
-import org.apache.hadoop.metrics2.MetricsCollector;
-import org.apache.hadoop.metrics2.MetricsInfo;
-import org.apache.hadoop.metrics2.MetricsRecordBuilder;
-import org.apache.hadoop.metrics2.MetricsTag;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-class InMemTestCollector implements MetricsCollector {
-
- private static final Logger log = LoggerFactory.getLogger(InMemTestCollector.class);
-
- final private Map<String,Number> measurements = new HashMap<>();
-
- final MetricsRecordBuilder builder = new FakeMetricsBuilder(measurements);
-
- @Override
- public MetricsRecordBuilder addRecord(String s) {
- log.debug("RecordBuilder: Adding string: {}", s);
- return builder;
- }
-
- @Override
- public MetricsRecordBuilder addRecord(MetricsInfo metricsInfo) {
- log.debug("FC: Add info {}", metricsInfo);
- return builder;
- }
-
- boolean contains(final String name) {
- return measurements.containsKey(name);
- }
-
- Number getValue(final String name) {
- return measurements.get(name);
- }
-
- @Override
- public String toString() {
- return "InMemTestCollector{" + "measurements=" + measurements + '}';
- }
-
- private static class FakeMetricsBuilder extends MetricsRecordBuilder {
-
- final Map<String,Number> measurements;
-
- FakeMetricsBuilder(Map<String,Number> measurements) {
- this.measurements = measurements;
- }
-
- @Override
- public MetricsRecordBuilder tag(MetricsInfo metricsInfo, String s) {
- log.debug("Add tag {}: {}", metricsInfo, s);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder add(MetricsTag metricsTag) {
- log.debug("Add tag {}", metricsTag);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder add(AbstractMetric abstractMetric) {
- log.debug("Add abstractMetric: {}", abstractMetric);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder setContext(String s) {
- log.debug("Set context: {}", s);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder addCounter(MetricsInfo metricsInfo, int i) {
- log.debug("add int counter: {}: {}", metricsInfo, i);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder addCounter(MetricsInfo metricsInfo, long l) {
- log.debug("add long counter: {}: {}", metricsInfo, l);
- measurements.put(metricsInfo.name(), l);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder addGauge(MetricsInfo metricsInfo, int i) {
- log.debug("add int gauge: {}: {}", metricsInfo, i);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder addGauge(MetricsInfo metricsInfo, long l) {
- log.debug("add long gauge: {}: {}", metricsInfo.name(), l);
- measurements.put(metricsInfo.name(), l);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder addGauge(MetricsInfo metricsInfo, float v) {
- log.debug("add float gauge: {}: {}", metricsInfo, v);
- return this;
- }
-
- @Override
- public MetricsRecordBuilder addGauge(MetricsInfo metricsInfo, double v) {
- log.debug("add double gauge: {}: {}", metricsInfo, v);
- return this;
- }
-
- @Override
- public MetricsCollector endRecord() {
- log.debug("end record called");
- return super.endRecord();
- }
-
- @Override
- public MetricsCollector parent() {
- return null;
- }
- }
-}
diff --git a/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java b/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java
index 37f137b..597dce5 100644
--- a/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java
+++ b/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java
@@ -79,7 +79,6 @@ import org.apache.accumulo.server.client.ClientServiceHandler;
import org.apache.accumulo.server.manager.state.Assignment;
import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
import org.apache.accumulo.server.manager.state.TabletStateStore;
-import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.rpc.ThriftServerType;
import org.apache.accumulo.server.zookeeper.TransactionWatcher;
@@ -329,10 +328,9 @@ public class NullTserver {
TransactionWatcher watcher = new TransactionWatcher(context);
ThriftClientHandler tch = new ThriftClientHandler(context, watcher);
Processor<Iface> processor = new Processor<>(tch);
- TServerUtils.startTServer(Metrics.initSystem(NullTserver.class.getSimpleName()),
- context.getConfiguration(), ThriftServerType.CUSTOM_HS_HA, processor, "NullTServer",
- "null tserver", 2, ThreadPools.DEFAULT_TIMEOUT_MILLISECS, 1000, 10 * 1024 * 1024, null,
- null, -1, HostAndPort.fromParts("0.0.0.0", opts.port));
+ TServerUtils.startTServer(context.getConfiguration(), ThriftServerType.CUSTOM_HS_HA, processor,
+ "NullTServer", "null tserver", 2, ThreadPools.DEFAULT_TIMEOUT_MILLISECS, 1000,
+ 10 * 1024 * 1024, null, null, -1, HostAndPort.fromParts("0.0.0.0", opts.port));
HostAndPort addr = HostAndPort.fromParts(InetAddress.getLocalHost().getHostName(), opts.port);
diff --git a/test/src/main/resources/hadoop-metrics2-accumulo.properties b/test/src/main/resources/hadoop-metrics2-accumulo.properties
deleted file mode 100644
index cacd779..0000000
--- a/test/src/main/resources/hadoop-metrics2-accumulo.properties
+++ /dev/null
@@ -1,61 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-
-# Poll collectors period in seconds
-*.period=5
-
-#
-# MetricSink configuration
-#
-# Uncomment a sink (and configuration) to enable it. By default,
-# no sinks are enabled.
-#
-
-#
-# Configure file logging
-#
-
-# File sink for all metrics
-accumulo.sink.file-all.class=org.apache.hadoop.metrics2.sink.FileSink
-accumulo.sink.file-all.filename=./target/it.all.metrics
-
-accumulo.sink.file-gc.class=org.apache.hadoop.metrics2.sink.FileSink
-accumulo.sink.file-gc.context=accgc
-accumulo.sink.file-gc.filename=./target/accgc.metrics
-accumulo.sink.file-gc.period=5
-
-# File sink for tserver metrics
-accumulo.sink.file-tserver.class=org.apache.hadoop.metrics2.sink.FileSink
-accumulo.sink.file-tserver.context=tserver
-accumulo.sink.file-tserver.filename=./target/tserver.metrics
-accumulo.sink.file-tserver.period=5
-
-# File sink for manager metrics
-accumulo.sink.file-manager.class=org.apache.hadoop.metrics2.sink.FileSink
-accumulo.sink.file-manager.context=manager
-accumulo.sink.file-manager.filename=./target/manager.metrics
-accumulo.sink.file-manager.period=5
-
-accumulo.jmx.manager.context=manager
-
-# File sink for thrift server metrics
-# accumulo.sink.file-thrift.class=org.apache.hadoop.metrics2.sink.FileSink
-# accumulo.sink.file-thrift.context=thrift
-# accumulo.sink.file-thrift.filename=thrift.metrics
-
diff --git a/test/src/main/resources/log4j2-test.properties b/test/src/main/resources/log4j2-test.properties
index b61cdc1..9124914 100644
--- a/test/src/main/resources/log4j2-test.properties
+++ b/test/src/main/resources/log4j2-test.properties
@@ -146,6 +146,29 @@ logger.36.level = error
logger.37.name = org.eclipse.jetty
logger.37.level = warn
+property.metricsFilename = ./target/test-metrics
+
+# appender.metrics.type = Console
+appender.metrics.type = RollingFile
+appender.metrics.name = LoggingMetricsOutput
+appender.metrics.fileName = ${metricsFilename}.metrics
+appender.metrics.filePattern = ${metricsFilename}-%d{MM-dd-yy-HH}-%i.metrics.gz
+appender.metrics.layout.type = PatternLayout
+appender.metrics.layout.pattern = METRICS: %d{ISO8601}, %m%n
+appender.metrics.policies.type = Policies
+appender.metrics.policies.time.type = TimeBasedTriggeringPolicy
+appender.metrics.policies.time.interval = 1
+appender.metrics.policies.time.modulate = true
+appender.metrics.policies.size.type = SizeBasedTriggeringPolicy
+appender.metrics.policies.size.size=100MB
+appender.metrics.strategy.type = DefaultRolloverStrategy
+appender.metrics.strategy.max = 10
+
+logger.metrics.name = org.apache.accumulo.metrics
+logger.metrics.level = info
+logger.metrics.additivity = false
+logger.metrics.appenderRef.metrics.ref = LoggingMetricsOutput
+
rootLogger.level = debug
rootLogger.appenderRef.console.ref = STDOUT
diff --git a/test/src/test/java/org/apache/accumulo/test/metrics/MetricsFileTailerTest.java b/test/src/test/java/org/apache/accumulo/test/metrics/MetricsFileTailerTest.java
deleted file mode 100644
index 6288738..0000000
--- a/test/src/test/java/org/apache/accumulo/test/metrics/MetricsFileTailerTest.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.test.metrics;
-
-import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.junit.Assert.assertTrue;
-
-import java.io.FileWriter;
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.nio.file.FileSystems;
-import java.nio.file.Files;
-
-import org.junit.AfterClass;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class MetricsFileTailerTest {
-
- private static final Logger log = LoggerFactory.getLogger(MetricsFileTailerTest.class);
-
- private static final String TEST_OUTFILE_NAME = "/tmp/testfile.txt";
- private static final String SUCCESS = "success";
-
- @AfterClass
- public static void cleanup() {
- try {
- Files.deleteIfExists(FileSystems.getDefault().getPath(TEST_OUTFILE_NAME));
- } catch (IOException ex) {
- log.trace("Failed to clean-up test file " + TEST_OUTFILE_NAME, ex);
- }
- }
-
- /**
- * Create a file tailer and then write some lines and validate the tailer returns the last line.
- */
- @Test
- public void fileUpdates() throws InterruptedException {
-
- boolean passed = false;
- try (MetricsFileTailer tailer = new MetricsFileTailer("foo", TEST_OUTFILE_NAME)) {
- tailer.startDaemonThread();
-
- long lastUpdate = tailer.getLastUpdate();
-
- writeToFile();
-
- int count = 0;
- while (count++ < 5) {
- if (lastUpdate == tailer.getLastUpdate()) {
- log.trace("no change");
- } else {
- lastUpdate = tailer.getLastUpdate();
- log.trace("{} - {}", tailer.getLastUpdate(), tailer.getLast());
- if (SUCCESS.compareTo(tailer.getLast()) == 0) {
- passed = true;
- break;
- }
- }
- Thread.sleep(5_000);
- }
-
- }
- assertTrue(passed);
- }
-
- /**
- * Simulate write record(s) to the file.
- */
- private void writeToFile() {
- try (FileWriter writer = new FileWriter(TEST_OUTFILE_NAME, UTF_8, true);
- PrintWriter printWriter = new PrintWriter(writer)) {
- printWriter.println("foo");
- // needs to be last line for test to pass
- printWriter.println(SUCCESS);
- printWriter.flush();
- } catch (IOException ex) {
- throw new IllegalStateException("failed to write data to test file", ex);
- }
- }
-}