You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2019/06/01 06:18:42 UTC
[accumulo] branch 2.0 updated: Clean up and refactor metrics for
maintainability (#1175)
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a commit to branch 2.0
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.0 by this push:
new 07f196d Clean up and refactor metrics for maintainability (#1175)
07f196d is described below
commit 07f196d41d0903faab55373780d35016b5f68896
Author: Christopher Tubbs <ct...@apache.org>
AuthorDate: Sat Jun 1 02:18:37 2019 -0400
Clean up and refactor metrics for maintainability (#1175)
* Create base class for common code
* Remove unneeded shims from legacy metrics
* Inline fields and methods used only once
* Rename classes to be easier to understand
* Retain current metric names and context/record tags
* Standardize metrics initialization
* Consolidate helper classes into abstract base class
* Separate fields/members to one per line for readability and group
fields according to function (gauge vs. stat vs. class helper like log)
---
.../main/java/org/apache/accumulo/proxy/Proxy.java | 10 +-
.../org/apache/accumulo/server/AbstractServer.java | 10 +-
.../apache/accumulo/server/metrics/Metrics.java | 84 +++++++-
.../server/metrics/Metrics2ThriftMetrics.java | 64 -------
.../server/metrics/MetricsSystemHelper.java | 54 ------
.../{MetricsFactory.java => ThriftMetrics.java} | 17 +-
.../accumulo/server/metrics/ThriftMetricsKeys.java | 24 ---
.../apache/accumulo/server/rpc/TServerUtils.java | 88 ++++-----
.../apache/accumulo/server/rpc/TimedProcessor.java | 32 ++--
.../accumulo/server/util/TServerUtilsTest.java | 8 +-
.../apache/accumulo/gc/SimpleGarbageCollector.java | 12 +-
.../java/org/apache/accumulo/master/Master.java | 13 +-
.../accumulo/master/metrics/MasterMetrics.java} | 15 +-
.../master/metrics/MasterMetricsFactory.java | 41 +---
...icationMetrics.java => ReplicationMetrics.java} | 70 ++-----
.../{Metrics2FateMetrics.java => FateMetrics.java} | 74 ++-----
...etricsTest.java => ReplicationMetricsTest.java} | 16 +-
.../org/apache/accumulo/tracer/TraceServer.java | 14 +-
.../org/apache/accumulo/tserver/TabletServer.java | 102 ++++------
.../metrics/Metrics2TabletServerMetrics.java | 124 ------------
.../metrics/Metrics2TabletServerMinCMetrics.java | 76 --------
.../metrics/Metrics2TabletServerScanMetrics.java | 80 --------
.../metrics/Metrics2TabletServerUpdateMetrics.java | 104 ----------
...rUpdateMetricsKeys.java => TServerMetrics.java} | 27 +--
.../tserver/metrics/TabletServerMetrics.java | 89 +++++++++
.../metrics/TabletServerMetricsFactory.java | 65 -------
...tricsKeys.java => TabletServerMinCMetrics.java} | 28 ++-
...tricsKeys.java => TabletServerScanMetrics.java} | 53 +++---
.../tserver/metrics/TabletServerUpdateMetrics.java | 81 ++++++++
.../org/apache/accumulo/tserver/tablet/Tablet.java | 212 +++++++++++++--------
.../accumulo/test/functional/ZombieTServer.java | 4 +-
.../accumulo/test/performance/NullTserver.java | 6 +-
32 files changed, 638 insertions(+), 1059 deletions(-)
diff --git a/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java b/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
index 5268e2a..03d921d 100644
--- a/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
+++ b/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
@@ -35,7 +35,7 @@ import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.HostAndPort;
import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.accumulo.proxy.thrift.AccumuloProxy;
-import org.apache.accumulo.server.metrics.MetricsFactory;
+import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.rpc.SaslServerConnectionParams;
import org.apache.accumulo.server.rpc.ServerAddress;
import org.apache.accumulo.server.rpc.TServerUtils;
@@ -153,8 +153,9 @@ public class Proxy implements KeywordExecutable {
} catch (InterruptedException | IOException e) {
throw new RuntimeException(e);
} finally {
- if (!folder.delete())
+ if (!folder.delete()) {
log.warn("Unexpected error removing {}", folder);
+ }
}
}));
} else if (clientProps == null) {
@@ -201,7 +202,6 @@ public class Proxy implements KeywordExecutable {
// No timeout
final long serverSocketTimeout = 0L;
// Use the new hadoop metrics2 support
- final MetricsFactory metricsFactory = new MetricsFactory();
final String serverName = "Proxy", threadName = "Accumulo Thrift Proxy";
// create the implementation of the proxy interface
@@ -271,8 +271,8 @@ public class Proxy implements KeywordExecutable {
}
// Hook up support for tracing for thrift calls
- TimedProcessor timedProcessor =
- new TimedProcessor(metricsFactory, processor, serverName, threadName);
+ TimedProcessor timedProcessor = new TimedProcessor(
+ Metrics.initSystem(Proxy.class.getSimpleName()), processor, serverName, threadName);
// Create the thrift server with our processor and properties
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 6c894f6..0eb7a61 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
@@ -23,8 +23,9 @@ import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.trace.TraceUtil;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
+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;
@@ -34,6 +35,7 @@ public abstract class AbstractServer implements AutoCloseable, Runnable {
private 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());
@@ -46,7 +48,7 @@ public abstract class AbstractServer implements AutoCloseable, Runnable {
log.info("Version " + Constants.VERSION);
log.info("Instance " + context.getInstanceID());
ServerUtil.init(context, appName);
- MetricsSystemHelper.configure(getClass().getName());
+ 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
@@ -89,6 +91,10 @@ public abstract class AbstractServer implements AutoCloseable, Runnable {
return getContext().getConfiguration();
}
+ public MetricsSystem getMetricsSystem() {
+ return metricsSystem;
+ }
+
@Override
public void close() {
TraceUtil.disable();
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
index aafea71..3ea8c05 100644
--- 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
@@ -16,16 +16,84 @@
*/
package org.apache.accumulo.server.metrics;
-public interface Metrics {
+import org.apache.commons.lang.StringUtils;
+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;
- String PREFIX = "Accumulo";
- String THRIFT_NAME = "Thrift";
- String TSERVER_NAME = "TabletServer";
- String MASTER_NAME = "Master";
+public abstract class Metrics implements MetricsSource {
- void register() throws Exception;
+ private static String processName = "Unknown";
- void add(String name, long time);
+ public static MetricsSystem initSystem(String serviceName) {
+ processName = serviceName;
+ String serviceInstance = System.getProperty("accumulo.metrics.service.instance", "");
+ if (StringUtils.isNotBlank(serviceInstance)) {
+ 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);
+ }
- boolean isEnabled();
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/metrics/Metrics2ThriftMetrics.java b/server/base/src/main/java/org/apache/accumulo/server/metrics/Metrics2ThriftMetrics.java
deleted file mode 100644
index 7908ec3..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/metrics/Metrics2ThriftMetrics.java
+++ /dev/null
@@ -1,64 +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.Interns;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-
-public class Metrics2ThriftMetrics implements Metrics, MetricsSource, ThriftMetricsKeys {
- public static final String CONTEXT = "thrift";
-
- private final MetricsSystem system;
- private final MetricsRegistry registry;
- private final String record, name, desc;
-
- Metrics2ThriftMetrics(MetricsSystem system, String serverName, String threadName) {
- this.system = system;
- this.record = serverName;
- this.name = THRIFT_NAME + ",sub=" + serverName;
- this.desc = "Thrift Server Metrics - " + serverName + " " + threadName;
- this.registry = new MetricsRegistry(Interns.info(name, desc));
- this.registry.tag(MsInfo.ProcessName, MetricsSystemHelper.getProcessName());
- }
-
- @Override
- public void add(String name, long time) {
- registry.add(name, time);
- }
-
- @Override
- public void register() {
- system.register(name, desc, this);
- }
-
- @Override
- public boolean isEnabled() {
- return true;
- }
-
- @Override
- public void getMetrics(MetricsCollector collector, boolean all) {
- MetricsRecordBuilder builder = collector.addRecord(record).setContext(CONTEXT);
-
- registry.snapshot(builder, all);
- }
-}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricsSystemHelper.java b/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricsSystemHelper.java
deleted file mode 100644
index 12e8cfd..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricsSystemHelper.java
+++ /dev/null
@@ -1,54 +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.commons.lang.StringUtils;
-import org.apache.hadoop.metrics2.MetricsSystem;
-import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
-import org.apache.hadoop.metrics2.source.JvmMetrics;
-import org.apache.hadoop.metrics2.source.JvmMetricsInfo;
-
-public class MetricsSystemHelper {
-
- private static String processName = "Unknown";
-
- public static void configure(String serviceName) {
- MetricsSystemHelper.processName = serviceName;
- String serviceInstance = System.getProperty("accumulo.metrics.service.instance", "");
- if (StringUtils.isNotBlank(serviceInstance)) {
- MetricsSystemHelper.processName += serviceInstance;
- }
- }
-
- public static String getProcessName() {
- return MetricsSystemHelper.processName;
- }
-
- private static class MetricsSystemHolder {
- // Singleton, rely on JVM to initialize the MetricsSystem only when it is accessed.
- private static final MetricsSystem metricsSystem =
- DefaultMetricsSystem.initialize(Metrics.PREFIX);
- }
-
- public static MetricsSystem getInstance() {
- if (MetricsSystemHolder.metricsSystem.getSource(JvmMetricsInfo.JvmMetrics.name()) == null) {
- JvmMetrics.create(getProcessName(), "", MetricsSystemHolder.metricsSystem);
- }
- return MetricsSystemHolder.metricsSystem;
- }
-
-}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricsFactory.java b/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetrics.java
similarity index 68%
rename from server/base/src/main/java/org/apache/accumulo/server/metrics/MetricsFactory.java
rename to server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetrics.java
index 6172d01..4ee6240 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricsFactory.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetrics.java
@@ -16,18 +16,19 @@
*/
package org.apache.accumulo.server.metrics;
-import org.apache.hadoop.metrics2.MetricsSystem;
+public class ThriftMetrics extends Metrics {
-public class MetricsFactory {
-
- private final MetricsSystem metricsSystem;
+ public ThriftMetrics(String serverName, String threadName) {
+ super("Thrift,sub=" + serverName, "Thrift Server Metrics - " + serverName + " " + threadName,
+ "thrift", serverName);
+ }
- public MetricsFactory() {
- metricsSystem = MetricsSystemHelper.getInstance();
+ public void addIdle(long time) {
+ getRegistry().add("idle", time);
}
- public Metrics createThriftMetrics(String serverName, String threadName) {
- return new Metrics2ThriftMetrics(metricsSystem, serverName, threadName);
+ public void addExecute(long time) {
+ getRegistry().add("execute", time);
}
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetricsKeys.java b/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetricsKeys.java
deleted file mode 100644
index 8adefdb..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/metrics/ThriftMetricsKeys.java
+++ /dev/null
@@ -1,24 +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;
-
-public interface ThriftMetricsKeys {
-
- String idle = "idle";
- String execute = "execute";
-
-}
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 c196ac7..632b2c7 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
@@ -44,6 +44,7 @@ import org.apache.accumulo.fate.util.LoggingRunnable;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.util.Halt;
import org.apache.accumulo.server.util.time.SimpleTimer;
+import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.security.SaslRpcServer;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.thrift.TProcessor;
@@ -115,9 +116,9 @@ public class TServerUtils {
* @throws UnknownHostException
* when we don't know our own address
*/
- public static ServerAddress startServer(ServerContext service, String hostname,
- Property portHintProperty, TProcessor processor, String serverName, String threadName,
- Property portSearchProperty, Property minThreadProperty,
+ public static ServerAddress startServer(MetricsSystem metricsSystem, ServerContext service,
+ String hostname, Property portHintProperty, TProcessor processor, String serverName,
+ String threadName, Property portSearchProperty, Property minThreadProperty,
Property timeBetweenThreadChecksProperty, Property maxMessageSizeProperty)
throws UnknownHostException {
final AccumuloConfiguration config = service.getConfiguration();
@@ -125,20 +126,24 @@ public class TServerUtils {
final int[] portHint = config.getPort(portHintProperty);
int minThreads = 2;
- if (minThreadProperty != null)
+ if (minThreadProperty != null) {
minThreads = config.getCount(minThreadProperty);
+ }
long timeBetweenThreadChecks = 1000;
- if (timeBetweenThreadChecksProperty != null)
+ if (timeBetweenThreadChecksProperty != null) {
timeBetweenThreadChecks = config.getTimeInMillis(timeBetweenThreadChecksProperty);
+ }
long maxMessageSize = 10 * 1000 * 1000;
- if (maxMessageSizeProperty != null)
+ if (maxMessageSizeProperty != null) {
maxMessageSize = config.getAsBytes(maxMessageSizeProperty);
+ }
boolean portSearch = false;
- if (portSearchProperty != null)
+ if (portSearchProperty != null) {
portSearch = config.getBoolean(portSearchProperty);
+ }
final int simpleTimerThreadpoolSize =
config.getCount(Property.GENERAL_SIMPLETIMER_THREADPOOL_SIZE);
@@ -150,7 +155,8 @@ 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(config, processor, serverName, threadName);
+ TimedProcessor timedProcessor =
+ new TimedProcessor(metricsSystem, config, processor, serverName, threadName);
HostAndPort[] addresses = getHostAndPorts(hostname, portHint);
try {
@@ -272,24 +278,21 @@ public class TServerUtils {
final int executorThreads, int simpleTimerThreads, long timeBetweenThreadChecks) {
final ThreadPoolExecutor pool = new SimpleThreadPool(executorThreads, "ClientPool");
// periodically adjust the number of threads we need by checking how busy our threads are
- SimpleTimer.getInstance(simpleTimerThreads).schedule(new Runnable() {
- @Override
- public void run() {
- // there is a minor race condition between sampling the current state of the thread pool and
- // adjusting it
- // however, this isn't really an issue, since it adjusts periodically anyway
- if (pool.getCorePoolSize() <= pool.getActiveCount()) {
- int larger = pool.getCorePoolSize() + Math.min(pool.getQueue().size(), 2);
- log.info("Increasing server thread pool size on {} to {}", serverName, larger);
- pool.setMaximumPoolSize(larger);
- pool.setCorePoolSize(larger);
- } else {
- if (pool.getCorePoolSize() > pool.getActiveCount() + 3) {
- int smaller = Math.max(executorThreads, pool.getCorePoolSize() - 1);
- if (smaller != pool.getCorePoolSize()) {
- log.info("Decreasing server thread pool size on {} to {}", serverName, smaller);
- pool.setCorePoolSize(smaller);
- }
+ SimpleTimer.getInstance(simpleTimerThreads).schedule(() -> {
+ // there is a minor race condition between sampling the current state of the thread pool and
+ // adjusting it
+ // however, this isn't really an issue, since it adjusts periodically anyway
+ if (pool.getCorePoolSize() <= pool.getActiveCount()) {
+ int larger = pool.getCorePoolSize() + Math.min(pool.getQueue().size(), 2);
+ log.info("Increasing server thread pool size on {} to {}", serverName, larger);
+ pool.setMaximumPoolSize(larger);
+ pool.setCorePoolSize(larger);
+ } else {
+ if (pool.getCorePoolSize() > pool.getActiveCount() + 3) {
+ int smaller = Math.max(executorThreads, pool.getCorePoolSize() - 1);
+ if (smaller != pool.getCorePoolSize()) {
+ log.info("Decreasing server thread pool size on {} to {}", serverName, smaller);
+ pool.setCorePoolSize(smaller);
}
}
}
@@ -537,19 +540,20 @@ public class TServerUtils {
return new ServerAddress(server, address);
}
- public static ServerAddress startTServer(AccumuloConfiguration conf, ThriftServerType serverType,
- TProcessor processor, String serverName, String threadName, int numThreads, int numSTThreads,
- long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams,
- SaslServerConnectionParams saslParams, long serverSocketTimeout, HostAndPort... addresses)
- throws TTransportException {
+ public static ServerAddress startTServer(MetricsSystem metricsSystem, AccumuloConfiguration conf,
+ ThriftServerType serverType, TProcessor processor, String serverName, String threadName,
+ int numThreads, int numSTThreads, 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(conf, processor, serverName, threadName),
- serverName, threadName, numThreads, numSTThreads, timeBetweenThreadChecks, maxMessageSize,
- sslParams, saslParams, serverSocketTimeout, addresses);
+ return startTServer(serverType,
+ new TimedProcessor(metricsSystem, conf, processor, serverName, threadName), serverName,
+ threadName, numThreads, numSTThreads, timeBetweenThreadChecks, maxMessageSize, sslParams,
+ saslParams, serverSocketTimeout, addresses);
}
/**
@@ -631,14 +635,11 @@ public class TServerUtils {
}
final TServer finalServer = serverAddress.server;
- Runnable serveTask = new Runnable() {
- @Override
- public void run() {
- try {
- finalServer.serve();
- } catch (Error e) {
- Halt.halt("Unexpected error in TThreadPoolServer " + e + ", halting.", 1);
- }
+ Runnable serveTask = () -> {
+ try {
+ finalServer.serve();
+ } catch (Error e) {
+ Halt.halt("Unexpected error in TThreadPoolServer " + e + ", halting.", 1);
}
};
@@ -667,8 +668,9 @@ public class TServerUtils {
* The TServer to stop
*/
public static void stopTServer(TServer s) {
- if (s == null)
+ if (s == null) {
return;
+ }
s.stop();
try {
Field f = s.getClass().getDeclaredField("executorService_");
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 caa6341..9505626 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
@@ -17,9 +17,8 @@
package org.apache.accumulo.server.rpc;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.server.metrics.Metrics;
-import org.apache.accumulo.server.metrics.MetricsFactory;
-import org.apache.accumulo.server.metrics.ThriftMetricsKeys;
+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;
@@ -33,20 +32,20 @@ public class TimedProcessor implements TProcessor {
private static final Logger log = LoggerFactory.getLogger(TimedProcessor.class);
private final TProcessor other;
- private final Metrics metrics;
+ private final ThriftMetrics thriftMetrics;
private long idleStart = 0;
- public TimedProcessor(AccumuloConfiguration conf, TProcessor next, String serverName,
- String threadName) {
- this(new MetricsFactory(), next, serverName, threadName);
+ public TimedProcessor(MetricsSystem metricsSystem, AccumuloConfiguration conf, TProcessor next,
+ String serverName, String threadName) {
+ this(metricsSystem, next, serverName, threadName);
}
- public TimedProcessor(MetricsFactory factory, TProcessor next, String serverName,
+ public TimedProcessor(MetricsSystem metricsSystem, TProcessor next, String serverName,
String threadName) {
this.other = next;
- metrics = factory.createThriftMetrics(serverName, threadName);
+ thriftMetrics = new ThriftMetrics(serverName, threadName);
try {
- metrics.register();
+ thriftMetrics.register(metricsSystem);
} catch (Exception e) {
log.error("Exception registering MBean with MBean Server", e);
}
@@ -56,18 +55,13 @@ public class TimedProcessor implements TProcessor {
@Override
public boolean process(TProtocol in, TProtocol out) throws TException {
long now = 0;
- final boolean metricsEnabled = metrics.isEnabled();
- if (metricsEnabled) {
- now = System.currentTimeMillis();
- metrics.add(ThriftMetricsKeys.idle, (now - idleStart));
- }
+ now = System.currentTimeMillis();
+ thriftMetrics.addIdle(now - idleStart);
try {
return other.process(in, out);
} finally {
- if (metricsEnabled) {
- idleStart = System.currentTimeMillis();
- metrics.add(ThriftMetricsKeys.execute, idleStart - now);
- }
+ idleStart = System.currentTimeMillis();
+ thriftMetrics.addExecute(idleStart - now);
}
}
}
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 30a8eeb..4f855db 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
@@ -43,6 +43,7 @@ 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.conf.ServerConfigurationFactory;
+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;
@@ -320,9 +321,10 @@ public class TServerUtilsTest {
// misconfiguration)
String hostname = "localhost";
- return TServerUtils.startServer(ctx, hostname, Property.TSERV_CLIENTPORT, processor,
- "TServerUtilsTest", "TServerUtilsTestThread", Property.TSERV_PORTSEARCH,
- Property.TSERV_MINTHREADS, Property.TSERV_THREADCHECK, Property.GENERAL_MAX_MESSAGE_SIZE);
+ return TServerUtils.startServer(Metrics.initSystem(getClass().getSimpleName()), ctx, hostname,
+ Property.TSERV_CLIENTPORT, processor, "TServerUtilsTest", "TServerUtilsTestThread",
+ Property.TSERV_PORTSEARCH, Property.TSERV_MINTHREADS, Property.TSERV_THREADCHECK,
+ Property.GENERAL_MAX_MESSAGE_SIZE);
}
}
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 755395d..147239a 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
@@ -623,12 +623,12 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface {
HostAndPort[] addresses = TServerUtils.getHostAndPorts(this.opts.getAddress(), port);
long maxMessageSize = getConfiguration().getAsBytes(Property.GENERAL_MAX_MESSAGE_SIZE);
try {
- ServerAddress server =
- TServerUtils.startTServer(getConfiguration(), getContext().getThriftServerType(),
- processor, this.getClass().getSimpleName(), "GC Monitor Service", 2,
- getConfiguration().getCount(Property.GENERAL_SIMPLETIMER_THREADPOOL_SIZE), 1000,
- maxMessageSize, getContext().getServerSslParams(), getContext().getSaslParams(), 0,
- addresses);
+ ServerAddress server = TServerUtils.startTServer(getMetricsSystem(), getConfiguration(),
+ getContext().getThriftServerType(), processor, this.getClass().getSimpleName(),
+ "GC Monitor Service", 2,
+ getConfiguration().getCount(Property.GENERAL_SIMPLETIMER_THREADPOOL_SIZE), 1000,
+ maxMessageSize, getContext().getServerSslParams(), getContext().getSaslParams(), 0,
+ addresses);
log.debug("Starting garbage collector listening on " + server.address);
return server.address;
} catch (Exception ex) {
diff --git a/server/master/src/main/java/org/apache/accumulo/master/Master.java b/server/master/src/main/java/org/apache/accumulo/master/Master.java
index 4eef033..0efd08d 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/Master.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/Master.java
@@ -992,9 +992,10 @@ public class Master extends AbstractServer
}
ServerAddress sa;
try {
- sa = TServerUtils.startServer(context, getHostname(), Property.MASTER_CLIENTPORT, processor,
- "Master", "Master Client Service Handler", null, Property.MASTER_MINTHREADS,
- Property.MASTER_THREADCHECK, Property.GENERAL_MAX_MESSAGE_SIZE);
+ sa = TServerUtils.startServer(getMetricsSystem(), context, getHostname(),
+ Property.MASTER_CLIENTPORT, processor, "Master", "Master Client Service Handler", null,
+ Property.MASTER_MINTHREADS, Property.MASTER_THREADCHECK,
+ Property.GENERAL_MAX_MESSAGE_SIZE);
} catch (UnknownHostException e) {
throw new IllegalStateException("Unable to start server on host " + getHostname(), e);
}
@@ -1150,9 +1151,7 @@ public class Master extends AbstractServer
}, 0, 5000);
// Register metrics modules
- MasterMetricsFactory factory = new MasterMetricsFactory(getConfiguration(), this);
-
- int failureCount = factory.register();
+ int failureCount = new MasterMetricsFactory(getConfiguration()).register(this);
if (failureCount > 0) {
log.info("Failed to register {} metrics modules", failureCount);
@@ -1217,7 +1216,7 @@ public class Master extends AbstractServer
HighlyAvailableServiceWrapper.service(impl, this);
ReplicationCoordinator.Processor<ReplicationCoordinator.Iface> replicationCoordinatorProcessor =
new ReplicationCoordinator.Processor<>(TraceUtil.wrapService(haReplicationProxy));
- ServerAddress replAddress = TServerUtils.startServer(context, getHostname(),
+ ServerAddress replAddress = TServerUtils.startServer(getMetricsSystem(), context, getHostname(),
Property.MASTER_REPLICATION_COORDINATOR_PORT, replicationCoordinatorProcessor,
"Master Replication Coordinator", "Replication Coordinator", null,
Property.MASTER_REPLICATION_COORDINATOR_MINTHREADS,
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java b/server/master/src/main/java/org/apache/accumulo/master/metrics/MasterMetrics.java
similarity index 72%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java
rename to server/master/src/main/java/org/apache/accumulo/master/metrics/MasterMetrics.java
index 8c9ab34..37a68b3 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/metrics/MasterMetrics.java
@@ -14,15 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.accumulo.tserver.metrics;
+package org.apache.accumulo.master.metrics;
-/**
- * Keys for referencing scan metrics
- */
-public interface TabletServerScanMetricsKeys {
+import org.apache.accumulo.server.metrics.Metrics;
+
+public abstract class MasterMetrics extends Metrics {
- String SCAN = "scan";
- String RESULT_SIZE = "result";
- String YIELD = "yield";
+ protected MasterMetrics(String subName, String description, String record) {
+ super("Master,sub=" + subName, description, "master", record);
+ }
}
diff --git a/server/master/src/main/java/org/apache/accumulo/master/metrics/MasterMetricsFactory.java b/server/master/src/main/java/org/apache/accumulo/master/metrics/MasterMetricsFactory.java
index 4a8a163..a22183f 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/metrics/MasterMetricsFactory.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/metrics/MasterMetricsFactory.java
@@ -21,9 +21,7 @@ import static java.util.Objects.requireNonNull;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.master.Master;
-import org.apache.accumulo.master.metrics.fate.Metrics2FateMetrics;
-import org.apache.accumulo.server.metrics.Metrics;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
+import org.apache.accumulo.master.metrics.fate.FateMetrics;
import org.apache.hadoop.metrics2.MetricsSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,62 +38,35 @@ public class MasterMetricsFactory {
private final boolean enableFateMetrics;
private final long fateMinUpdateInterval;
- private final MetricsSystem metricsSystem;
- private final Master master;
-
- public MasterMetricsFactory(AccumuloConfiguration conf, Master master) {
+ public MasterMetricsFactory(AccumuloConfiguration conf) {
requireNonNull(conf, "AccumuloConfiguration must not be null");
-
enableFateMetrics = conf.getBoolean(Property.MASTER_FATE_METRICS_ENABLED);
-
fateMinUpdateInterval = conf.getTimeInMillis(Property.MASTER_FATE_METRICS_MIN_UPDATE_INTERVAL);
-
- this.master = master;
-
- metricsSystem = MetricsSystemHelper.getInstance();
}
- public int register() {
+ public int register(Master master) {
+ MetricsSystem metricsSystem = master.getMetricsSystem();
int failureCount = 0;
try {
-
- Metrics replicationMetrics = createReplicationMetrics();
-
- replicationMetrics.register();
-
+ new ReplicationMetrics(master).register(metricsSystem);
log.info("Registered replication metrics module");
-
} catch (Exception ex) {
failureCount++;
log.error("Failed to register replication metrics", ex);
}
try {
-
if (enableFateMetrics) {
-
- Metrics fateMetrics = createFateMetrics();
-
- fateMetrics.register();
-
+ new FateMetrics(master.getContext(), fateMinUpdateInterval).register(metricsSystem);
log.info("Registered FATE metrics module");
}
} catch (Exception ex) {
failureCount++;
log.error("Failed to register fate metrics", ex);
}
-
return failureCount;
}
- private Metrics createReplicationMetrics() {
- return new Metrics2ReplicationMetrics(master, metricsSystem);
- }
-
- private Metrics createFateMetrics() {
- return new Metrics2FateMetrics(master.getContext(), metricsSystem, fateMinUpdateInterval);
- }
-
}
diff --git a/server/master/src/main/java/org/apache/accumulo/master/metrics/Metrics2ReplicationMetrics.java b/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
similarity index 67%
rename from server/master/src/main/java/org/apache/accumulo/master/metrics/Metrics2ReplicationMetrics.java
rename to server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
index c819f2f..01c1be0 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/metrics/Metrics2ReplicationMetrics.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
@@ -27,94 +27,52 @@ import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.master.Master;
-import org.apache.accumulo.server.metrics.Metrics;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.fs.Path;
-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.Interns;
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 Metrics2ReplicationMetrics implements Metrics, MetricsSource {
- public static final String NAME = MASTER_NAME + ",sub=Replication",
- DESCRIPTION = "Data-Center Replication Metrics", CONTEXT = "master",
- RECORD = "MasterReplication";
- public static final String PENDING_FILES = "filesPendingReplication", NUM_PEERS = "numPeers",
- MAX_REPLICATION_THREADS = "maxReplicationThreads",
- REPLICATION_QUEUE_TIME_QUANTILES = "replicationQueue10m",
- REPLICATION_QUEUE_TIME = "replicationQueue";
+public class ReplicationMetrics extends MasterMetrics {
- private static final Logger log = LoggerFactory.getLogger(Metrics2ReplicationMetrics.class);
+ private static final Logger log = LoggerFactory.getLogger(ReplicationMetrics.class);
private final Master master;
- private final MetricsSystem system;
- private final MetricsRegistry registry;
private final ReplicationUtil replicationUtil;
private final MutableQuantiles replicationQueueTimeQuantiles;
private final MutableStat replicationQueueTimeStat;
private final Map<Path,Long> pathModTimes;
- Metrics2ReplicationMetrics(Master master, MetricsSystem system) {
+ ReplicationMetrics(Master master) {
+ super("Replication", "Data-Center Replication Metrics", "MasterReplication");
this.master = master;
- this.system = system;
pathModTimes = new HashMap<>();
- this.registry = new MetricsRegistry(Interns.info(NAME, DESCRIPTION));
- this.registry.tag(MsInfo.ProcessName, MetricsSystemHelper.getProcessName());
replicationUtil = new ReplicationUtil(master.getContext());
- replicationQueueTimeQuantiles = registry.newQuantiles(REPLICATION_QUEUE_TIME_QUANTILES,
+ MetricsRegistry registry = super.getRegistry();
+ replicationQueueTimeQuantiles = registry.newQuantiles("replicationQueue10m",
"Replication queue time quantiles in milliseconds", "ops", "latency", 600);
- replicationQueueTimeStat = registry.newStat(REPLICATION_QUEUE_TIME,
+ replicationQueueTimeStat = registry.newStat("replicationQueue",
"Replication queue time statistics in milliseconds", "ops", "latency", true);
}
- protected void snapshot() {
+ @Override
+ protected void prepareMetrics() {
+ final String PENDING_FILES = "filesPendingReplication";
// Only add these metrics if the replication table is online and there are peers
if (TableState.ONLINE == Tables.getTableState(master.getContext(), ReplicationTable.ID)
&& !replicationUtil.getPeers().isEmpty()) {
- registry.add(PENDING_FILES, getNumFilesPendingReplication());
+ getRegistry().add(PENDING_FILES, getNumFilesPendingReplication());
addReplicationQueueTimeMetrics();
} else {
- registry.add(PENDING_FILES, 0);
+ getRegistry().add(PENDING_FILES, 0);
}
- registry.add(NUM_PEERS, getNumConfiguredPeers());
- registry.add(MAX_REPLICATION_THREADS, getMaxReplicationThreads());
- }
-
- @Override
- public void getMetrics(MetricsCollector collector, boolean all) {
- MetricsRecordBuilder builder = collector.addRecord(RECORD).setContext(CONTEXT);
-
- snapshot();
-
- registry.snapshot(builder, all);
- replicationQueueTimeQuantiles.snapshot(builder, all);
- replicationQueueTimeStat.snapshot(builder, all);
- }
-
- @Override
- public void register() {
- system.register(NAME, DESCRIPTION, this);
- }
-
- @Override
- public void add(String name, long time) {
- throw new UnsupportedOperationException("add() is not implemented");
- }
-
- @Override
- public boolean isEnabled() {
- return true;
+ getRegistry().add("numPeers", getNumConfiguredPeers());
+ getRegistry().add("maxReplicationThreads", getMaxReplicationThreads());
}
protected long getNumFilesPendingReplication() {
diff --git a/server/master/src/main/java/org/apache/accumulo/master/metrics/fate/Metrics2FateMetrics.java b/server/master/src/main/java/org/apache/accumulo/master/metrics/fate/FateMetrics.java
similarity index 51%
rename from server/master/src/main/java/org/apache/accumulo/master/metrics/fate/Metrics2FateMetrics.java
rename to server/master/src/main/java/org/apache/accumulo/master/metrics/fate/FateMetrics.java
index a0c4383..cf3660c 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/metrics/fate/Metrics2FateMetrics.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/metrics/fate/FateMetrics.java
@@ -19,40 +19,20 @@ package org.apache.accumulo.master.metrics.fate;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.accumulo.master.metrics.MasterMetrics;
import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.server.metrics.Metrics;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
-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.Interns;
import org.apache.hadoop.metrics2.lib.MetricsRegistry;
import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-public class Metrics2FateMetrics implements Metrics, MetricsSource {
-
- private static final Logger log = LoggerFactory.getLogger(Metrics2FateMetrics.class);
+public class FateMetrics extends MasterMetrics {
// limit calls to update fate counters to guard against hammering zookeeper.
private static final long DEFAULT_MIN_REFRESH_DELAY = TimeUnit.SECONDS.toMillis(10);
private volatile long minimumRefreshDelay;
- public static final String NAME = MASTER_NAME + ",sub=Fate";
- public static final String DESCRIPTION = "Fate Metrics";
- public static final String CONTEXT = "master";
- public static final String RECORD = "fate";
- public static final String CUR_FATE_OPS = "currentFateOps";
- public static final String TOTAL_FATE_OPS = "totalFateOps";
- public static final String TOTAL_ZK_CONN_ERRORS = "totalZkConnErrors";
-
private final ServerContext context;
- private final MetricsSystem metricsSystem;
- private final MetricsRegistry registry;
+
private final MutableGaugeLong currentFateOps;
private final MutableGaugeLong zkChildFateOpsTotal;
private final MutableGaugeLong zkConnectionErrorsTotal;
@@ -61,8 +41,8 @@ public class Metrics2FateMetrics implements Metrics, MetricsSource {
private volatile long lastUpdate = 0;
- public Metrics2FateMetrics(final ServerContext context, MetricsSystem metricsSystem,
- final long minimumRefreshDelay) {
+ public FateMetrics(final ServerContext context, final long minimumRefreshDelay) {
+ super("Fate", "Fate Metrics", "fate");
this.context = context;
@@ -70,59 +50,27 @@ public class Metrics2FateMetrics implements Metrics, MetricsSource {
metricValues = new AtomicReference<>(FateMetricValues.updateFromZookeeper(context, null));
- this.metricsSystem = metricsSystem;
- this.registry = new MetricsRegistry(Interns.info(NAME, DESCRIPTION));
- this.registry.tag(MsInfo.ProcessName, MetricsSystemHelper.getProcessName());
-
- currentFateOps = registry.newGauge(CUR_FATE_OPS, "Current number of FATE Ops", 0L);
- zkChildFateOpsTotal = registry.newGauge(TOTAL_FATE_OPS, "Total FATE Ops", 0L);
+ 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(TOTAL_ZK_CONN_ERRORS, "Total ZK Connection Errors", 0L);
+ registry.newGauge("totalZkConnErrors", "Total ZK Connection Errors", 0L);
}
@Override
- public void register() {
- metricsSystem.register(NAME, DESCRIPTION, this);
- }
-
- @Override
- public void add(String name, long time) {
- throw new UnsupportedOperationException("add() is not implemented");
- }
-
- @Override
- public boolean isEnabled() {
- return true;
- }
-
- @Override
- public void getMetrics(MetricsCollector collector, boolean all) {
-
- log.trace("getMetrics called with collector: {}", collector);
-
+ protected void prepareMetrics() {
FateMetricValues fateMetrics = metricValues.get();
-
long now = System.currentTimeMillis();
-
if ((lastUpdate + minimumRefreshDelay) < now) {
-
fateMetrics = FateMetricValues.updateFromZookeeper(context, fateMetrics);
-
metricValues.set(fateMetrics);
-
lastUpdate = now;
-
// update individual gauges that are reported.
currentFateOps.set(fateMetrics.getCurrentFateOps());
zkChildFateOpsTotal.set(fateMetrics.getZkFateChildOpsTotal());
zkConnectionErrorsTotal.set(fateMetrics.getZkConnectionErrors());
-
}
-
- // create the metrics record and publish to the registry.
- MetricsRecordBuilder builder = collector.addRecord(RECORD).setContext(CONTEXT);
- registry.snapshot(builder, all);
-
}
+
}
diff --git a/server/master/src/test/java/org/apache/accumulo/master/metrics/Metrics2ReplicationMetricsTest.java b/server/master/src/test/java/org/apache/accumulo/master/metrics/ReplicationMetricsTest.java
similarity index 87%
rename from server/master/src/test/java/org/apache/accumulo/master/metrics/Metrics2ReplicationMetricsTest.java
rename to server/master/src/test/java/org/apache/accumulo/master/metrics/ReplicationMetricsTest.java
index e2c4d30..1592a90 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/metrics/Metrics2ReplicationMetricsTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/metrics/ReplicationMetricsTest.java
@@ -24,7 +24,6 @@ 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.MetricsSystem;
import org.apache.hadoop.metrics2.lib.MutableQuantiles;
import org.apache.hadoop.metrics2.lib.MutableStat;
import org.easymock.EasyMock;
@@ -32,15 +31,15 @@ import org.junit.Test;
import com.google.common.collect.ImmutableSet;
-public class Metrics2ReplicationMetricsTest {
+public class ReplicationMetricsTest {
private long currentTime = 1000L;
/**
* Extend the class to override the current time for testing
*/
- public class TestMetrics2ReplicationMetrics extends Metrics2ReplicationMetrics {
- TestMetrics2ReplicationMetrics(Master master, MetricsSystem system) {
- super(master, system);
+ public class ReplicationMetricsTestMetrics extends ReplicationMetrics {
+ ReplicationMetricsTestMetrics(Master master) {
+ super(master);
}
@Override
@@ -53,7 +52,6 @@ public class Metrics2ReplicationMetricsTest {
public void testAddReplicationQueueTimeMetrics() throws Exception {
Master master = EasyMock.createMock(Master.class);
ServerContext context = EasyMock.createMock(ServerContext.class);
- MetricsSystem system = EasyMock.createMock(MetricsSystem.class);
VolumeManager fileSystem = EasyMock.createMock(VolumeManager.class);
ReplicationUtil util = EasyMock.createMock(ReplicationUtil.class);
MutableStat stat = EasyMock.createMock(MutableStat.class);
@@ -84,9 +82,9 @@ public class Metrics2ReplicationMetricsTest {
stat.add(currentTime - 100);
EasyMock.expectLastCall();
- EasyMock.replay(master, system, fileSystem, util, stat, quantiles);
+ EasyMock.replay(master, fileSystem, util, stat, quantiles);
- Metrics2ReplicationMetrics metrics = new TestMetrics2ReplicationMetrics(master, system);
+ ReplicationMetrics metrics = new ReplicationMetricsTestMetrics(master);
// Inject our mock objects
replaceField(metrics, "replicationUtil", util);
@@ -97,7 +95,7 @@ public class Metrics2ReplicationMetricsTest {
metrics.addReplicationQueueTimeMetrics();
metrics.addReplicationQueueTimeMetrics();
- EasyMock.verify(master, system, fileSystem, util, stat, quantiles);
+ EasyMock.verify(master, fileSystem, util, stat, quantiles);
}
private void replaceField(Object instance, String fieldName, Object target)
diff --git a/server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java b/server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java
index 20f1d40..fd6b02a 100644
--- a/server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java
+++ b/server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java
@@ -56,7 +56,6 @@ import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.ServerOpts;
import org.apache.accumulo.server.ServerUtil;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
import org.apache.accumulo.server.security.SecurityUtil;
import org.apache.accumulo.server.util.time.SimpleTimer;
import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
@@ -171,8 +170,9 @@ public class TraceServer implements Watcher, AutoCloseable {
}
writer.addMutation(spanMutation);
writer.addMutation(indexMutation);
- if (timeMutation != null)
+ if (timeMutation != null) {
writer.addMutation(timeMutation);
+ }
} catch (MutationsRejectedException exception) {
log.warn("Unable to write mutation to table; discarding span. set log"
+ " level to DEBUG for span information and stacktrace. cause: " + exception);
@@ -382,12 +382,14 @@ public class TraceServer implements Watcher, AutoCloseable {
if (keyTab == null || keyTab.length() == 0) {
keyTab = acuConf.getPath(Property.GENERAL_KERBEROS_KEYTAB);
}
- if (keyTab == null || keyTab.length() == 0)
+ if (keyTab == null || keyTab.length() == 0) {
return;
+ }
String principalConfig = acuConf.get(Property.TRACE_USER);
- if (principalConfig == null || principalConfig.length() == 0)
+ if (principalConfig == null || principalConfig.length() == 0) {
return;
+ }
log.info("Attempting to login as {} with {}", principalConfig, keyTab);
SecurityUtil.serverLogin(acuConf, keyTab, principalConfig);
@@ -407,7 +409,6 @@ public class TraceServer implements Watcher, AutoCloseable {
opts.parseArgs(app, args);
ServerContext context = new ServerContext(opts.getSiteConfiguration());
loginTracer(context.getConfiguration());
- MetricsSystemHelper.configure(TraceServer.class.getSimpleName());
ServerUtil.init(context, app);
try (TraceServer server = new TraceServer(context, opts.getAddress())) {
server.run();
@@ -429,8 +430,9 @@ public class TraceServer implements Watcher, AutoCloseable {
}
if (event.getPath() != null) {
try {
- if (context.getZooReaderWriter().exists(event.getPath(), this))
+ if (context.getZooReaderWriter().exists(event.getPath(), this)) {
return;
+ }
} catch (Exception ex) {
log.error("{}", ex.getMessage(), ex);
}
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 5baa3ca..11df49f 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
@@ -193,7 +193,6 @@ import org.apache.accumulo.server.master.state.TabletLocationState.BadLocationSt
import org.apache.accumulo.server.master.state.TabletStateStore;
import org.apache.accumulo.server.master.state.ZooTabletStateStore;
import org.apache.accumulo.server.master.tableOps.UserCompactionConfig;
-import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.problems.ProblemReport;
import org.apache.accumulo.server.problems.ProblemReports;
import org.apache.accumulo.server.replication.ZooKeeperInitialization;
@@ -230,9 +229,10 @@ import org.apache.accumulo.tserver.log.TabletServerLogger;
import org.apache.accumulo.tserver.mastermessage.MasterMessage;
import org.apache.accumulo.tserver.mastermessage.SplitReportMessage;
import org.apache.accumulo.tserver.mastermessage.TabletStatusMessage;
-import org.apache.accumulo.tserver.metrics.TabletServerMetricsFactory;
-import org.apache.accumulo.tserver.metrics.TabletServerScanMetricsKeys;
-import org.apache.accumulo.tserver.metrics.TabletServerUpdateMetricsKeys;
+import org.apache.accumulo.tserver.metrics.TabletServerMetrics;
+import org.apache.accumulo.tserver.metrics.TabletServerMinCMetrics;
+import org.apache.accumulo.tserver.metrics.TabletServerScanMetrics;
+import org.apache.accumulo.tserver.metrics.TabletServerUpdateMetrics;
import org.apache.accumulo.tserver.replication.ReplicationServicerHandler;
import org.apache.accumulo.tserver.replication.ReplicationWorker;
import org.apache.accumulo.tserver.scan.LookupTask;
@@ -260,6 +260,7 @@ import org.apache.hadoop.fs.FSError;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
import org.apache.thrift.TException;
@@ -288,16 +289,15 @@ public class TabletServer extends AbstractServer {
private final TabletServerLogger logger;
- private final TabletServerMetricsFactory metricsFactory;
- private final Metrics updateMetrics;
- private final Metrics scanMetrics;
- private final Metrics mincMetrics;
+ private final TabletServerUpdateMetrics updateMetrics;
+ private final TabletServerScanMetrics scanMetrics;
+ private final TabletServerMinCMetrics mincMetrics;
- public Metrics getScanMetrics() {
+ public TabletServerScanMetrics getScanMetrics() {
return scanMetrics;
}
- public Metrics getMinCMetrics() {
+ public TabletServerMinCMetrics getMinCMetrics() {
return mincMetrics;
}
@@ -416,10 +416,9 @@ public class TabletServer extends AbstractServer {
this.resourceManager = new TabletServerResourceManager(this, fs, context);
this.security = AuditedSecurityOperation.getInstance(context);
- metricsFactory = new TabletServerMetricsFactory();
- updateMetrics = metricsFactory.createUpdateMetrics();
- scanMetrics = metricsFactory.createScanMetrics();
- mincMetrics = metricsFactory.createMincMetrics();
+ updateMetrics = new TabletServerUpdateMetrics();
+ scanMetrics = new TabletServerScanMetrics();
+ mincMetrics = new TabletServerMinCMetrics();
SimpleTimer.getInstance(aconf).schedule(() -> TabletLocator.clearLocators(),
jitter(TIME_BETWEEN_LOCATOR_CACHE_CLEARS), jitter(TIME_BETWEEN_LOCATOR_CACHE_CLEARS));
walMarker = new WalStateManager(context);
@@ -746,10 +745,8 @@ public class TabletServer extends AbstractServer {
(t2 - ss.startTime) / 1000.0, ss.runStats.toString()));
}
- if (scanMetrics.isEnabled()) {
- scanMetrics.add(TabletServerScanMetricsKeys.SCAN, t2 - ss.startTime);
- scanMetrics.add(TabletServerScanMetricsKeys.RESULT_SIZE, ss.entriesReturned);
- }
+ scanMetrics.addScan(t2 - ss.startTime);
+ scanMetrics.addResult(ss.entriesReturned);
}
}
@@ -903,9 +900,7 @@ public class TabletServer extends AbstractServer {
// Make sure user is real
Durability durability = DurabilityImpl.fromThrift(tdurabilty);
security.authenticateUser(credentials, credentials);
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.PERMISSION_ERRORS, 0);
- }
+ updateMetrics.addPermissionErrors(0);
UpdateSession us = new UpdateSession(
new TservConstraintEnv(getContext(), security, credentials), credentials, durability);
@@ -941,9 +936,7 @@ public class TabletServer extends AbstractServer {
// not serving tablet, so report all mutations as
// failures
us.failures.put(keyExtent, 0L);
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.UNKNOWN_TABLET_ERRORS, 0);
- }
+ updateMetrics.addUnknownTabletErrors(0);
}
} else {
log.warn("Denying access to table {} for user {}", keyExtent.getTableId(), us.getUser());
@@ -951,9 +944,7 @@ public class TabletServer extends AbstractServer {
us.authTimes.addStat(t2 - t1);
us.currentTablet = null;
us.authFailures.put(keyExtent, SecurityErrorCode.PERMISSION_DENIED);
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.PERMISSION_ERRORS, 0);
- }
+ updateMetrics.addPermissionErrors(0);
return;
}
} catch (TableNotFoundException tnfe) {
@@ -962,9 +953,7 @@ public class TabletServer extends AbstractServer {
us.authTimes.addStat(t2 - t1);
us.currentTablet = null;
us.authFailures.put(keyExtent, SecurityErrorCode.TABLE_DOESNT_EXIST);
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.UNKNOWN_TABLET_ERRORS, 0);
- }
+ updateMetrics.addUnknownTabletErrors(0);
return;
} catch (ThriftSecurityException e) {
log.error("Denying permission to check user " + us.getUser() + " with user " + e.getUser(),
@@ -973,9 +962,7 @@ public class TabletServer extends AbstractServer {
us.authTimes.addStat(t2 - t1);
us.currentTablet = null;
us.authFailures.put(keyExtent, e.getCode());
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.PERMISSION_ERRORS, 0);
- }
+ updateMetrics.addPermissionErrors(0);
return;
}
}
@@ -1054,10 +1041,7 @@ public class TabletServer extends AbstractServer {
List<Mutation> mutations = entry.getValue();
if (mutations.size() > 0) {
try {
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.MUTATION_ARRAY_SIZE,
- mutations.size());
- }
+ updateMetrics.addMutationArraySize(mutations.size());
CommitSession commitSession = tablet.prepareMutationsForCommit(us.cenv, mutations);
if (commitSession == null) {
@@ -1076,9 +1060,7 @@ public class TabletServer extends AbstractServer {
} catch (TConstraintViolationException e) {
us.violations.add(e.getViolations());
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.CONSTRAINT_VIOLATIONS, 0);
- }
+ updateMetrics.addConstraintViolations(0);
if (e.getNonViolators().size() > 0) {
// only log and commit mutations if there were some
@@ -1166,23 +1148,15 @@ public class TabletServer extends AbstractServer {
}
private void updateWalogWriteTime(long time) {
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.WALOG_WRITE_TIME, time);
- }
+ updateMetrics.addWalogWriteTime(time);
}
private void updateAvgCommitTime(long time, int size) {
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.COMMIT_TIME,
- (long) ((time) / (double) size));
- }
+ updateMetrics.addCommitTime((long) (time / (double) size));
}
private void updateAvgPrepTime(long time, int size) {
- if (updateMetrics.isEnabled()) {
- updateMetrics.add(TabletServerUpdateMetricsKeys.COMMIT_PREP,
- (long) ((time) / (double) size));
- }
+ updateMetrics.addCommitPrep((long) (time / (double) size));
}
@Override
@@ -2587,8 +2561,8 @@ public class TabletServer extends AbstractServer {
TProcessor processor, String threadName) 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(getContext(), address, portHint, processor,
- this.getClass().getSimpleName(), threadName, Property.TSERV_PORTSEARCH,
+ ServerAddress sp = TServerUtils.startServer(getMetricsSystem(), getContext(), address, portHint,
+ processor, this.getClass().getSimpleName(), threadName, Property.TSERV_PORTSEARCH,
Property.TSERV_MINTHREADS, Property.TSERV_THREADCHECK, maxMessageSizeProperty);
this.server = sp.server;
return sp.address;
@@ -2654,10 +2628,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(getContext(), clientAddress.getHost(),
- Property.REPLICATION_RECEIPT_SERVICE_PORT, processor, "ReplicationServicerHandler",
- "Replication Servicer", Property.TSERV_PORTSEARCH, Property.REPLICATION_MIN_THREADS,
- Property.REPLICATION_THREADCHECK, maxMessageSizeProperty);
+ ServerAddress sp = TServerUtils.startServer(getMetricsSystem(), getContext(),
+ clientAddress.getHost(), Property.REPLICATION_RECEIPT_SERVICE_PORT, processor,
+ "ReplicationServicerHandler", "Replication Servicer", Property.TSERV_PORTSEARCH,
+ Property.REPLICATION_MIN_THREADS, Property.REPLICATION_THREADCHECK, maxMessageSizeProperty);
this.replServer = sp.server;
log.info("Started replication service on {}", sp.address);
@@ -2753,16 +2727,14 @@ public class TabletServer extends AbstractServer {
throw new RuntimeException(e);
}
- Metrics tserverMetrics = metricsFactory.createTabletServerMetrics(this);
-
- // Register MBeans
try {
- tserverMetrics.register();
- mincMetrics.register();
- scanMetrics.register();
- updateMetrics.register();
+ MetricsSystem metricsSystem = getMetricsSystem();
+ new TabletServerMetrics(this).register(metricsSystem);
+ mincMetrics.register(metricsSystem);
+ scanMetrics.register(metricsSystem);
+ updateMetrics.register(metricsSystem);
} catch (Exception e) {
- log.error("Error registering with JMX", e);
+ log.error("Error registering metrics", e);
}
if (authKeyWatcher != null) {
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerMetrics.java
deleted file mode 100644
index 3dde77c..0000000
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerMetrics.java
+++ /dev/null
@@ -1,124 +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;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
-import org.apache.accumulo.tserver.TabletServer;
-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.Interns;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
-
-public class Metrics2TabletServerMetrics
- implements Metrics, MetricsSource, TabletServerMetricsKeys {
- public static final String NAME = TSERVER_NAME + ",sub=General",
- DESCRIPTION = "General TabletServer Metrics", CONTEXT = "tserver", RECORD = "general";
-
- private final TabletServerMetricsUtil util;
- private final MetricsSystem system;
- private final MetricsRegistry registry;
-
- private final MutableGaugeLong entries, entriesInMemory, activeMajcs, queuedMajcs, activeMincs,
- queuedMincs, onlineTablets, openingTablets, unopenedTablets, queries, totalMincs;
-
- // Use TabletServerMetricsFactory
- Metrics2TabletServerMetrics(TabletServer tserver, MetricsSystem system) {
- util = new TabletServerMetricsUtil(tserver);
- this.system = system;
- this.registry = new MetricsRegistry(Interns.info(NAME, DESCRIPTION));
- this.registry.tag(MsInfo.ProcessName, MetricsSystemHelper.getProcessName());
-
- entries = registry.newGauge(Interns.info(ENTRIES, "Number of entries"), 0L);
- entriesInMemory =
- registry.newGauge(Interns.info(ENTRIES_IN_MEM, "Number of entries in memory"), 0L);
- activeMajcs =
- registry.newGauge(Interns.info(ACTIVE_MAJCS, "Number of active major compactions"), 0L);
- queuedMajcs =
- registry.newGauge(Interns.info(QUEUED_MAJCS, "Number of queued major compactions"), 0L);
- activeMincs =
- registry.newGauge(Interns.info(ACTIVE_MINCS, "Number of active minor compactions"), 0L);
- queuedMincs =
- registry.newGauge(Interns.info(QUEUED_MINCS, "Number of queued minor compactions"), 0L);
- onlineTablets = registry.newGauge(Interns.info(ONLINE_TABLETS, "Number of online tablets"), 0L);
- openingTablets =
- registry.newGauge(Interns.info(OPENING_TABLETS, "Number of opening tablets"), 0L);
- unopenedTablets =
- registry.newGauge(Interns.info(UNOPENED_TABLETS, "Number of unopened tablets"), 0L);
- queries = registry.newGauge(Interns.info(QUERIES, "Number of queries"), 0L);
- totalMincs = registry
- .newGauge(Interns.info(TOTAL_MINCS, "Total number of minor compactions performed"), 0L);
- }
-
- @Override
- public void add(String name, long time) {
- throw new UnsupportedOperationException("add() is not implemented");
- }
-
- @Override
- public void register() {
- system.register(NAME, DESCRIPTION, this);
- }
-
- @Override
- public boolean isEnabled() {
- return true;
- }
-
- protected void snapshot() {
-
- 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());
- }
-
- @Override
- public void getMetrics(MetricsCollector collector, boolean all) {
- MetricsRecordBuilder builder = collector.addRecord(RECORD).setContext(CONTEXT);
-
- // Update each MutableMetric with the new value
- snapshot();
-
- // Add then all to the builder
- registry.snapshot(builder, all);
-
- // TODO Some day, MetricsRegistry will also support the MetricsGaugeDouble or allow us to
- // instantiate it directly
- builder.addGauge(Interns.info(FILES_PER_TABLET, "Number of files per tablet"),
- util.getAverageFilesPerTablet());
- builder.addGauge(Interns.info(HOLD_TIME, "Time commits held"), util.getHoldTime());
- builder.addGauge(Interns.info(INGEST_RATE, "Ingest rate (entries/sec)"), util.getIngest());
- builder.addGauge(Interns.info(INGEST_BYTE_RATE, "Ingest rate (bytes/sec)"),
- util.getIngestByteRate());
- builder.addGauge(Interns.info(QUERY_RATE, "Query rate (entries/sec)"), util.getQueryRate());
- builder.addGauge(Interns.info(QUERY_BYTE_RATE, "Query rate (bytes/sec)"),
- util.getQueryByteRate());
- builder.addGauge(Interns.info(SCANNED_RATE, "Scanned rate"), util.getScannedRate());
- }
-}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerMinCMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerMinCMetrics.java
deleted file mode 100644
index 1427b7c..0000000
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerMinCMetrics.java
+++ /dev/null
@@ -1,76 +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;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
-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.Interns;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableStat;
-
-public class Metrics2TabletServerMinCMetrics
- implements Metrics, MetricsSource, TabletServerMinCMetricsKeys {
- public static final String NAME = TSERVER_NAME + ",sub=MinorCompactions",
- DESCRIPTION = "TabletServer Minor Compaction Metrics", CONTEXT = "tserver",
- RECORD = "MinorCompactions";
-
- private final MetricsSystem system;
- private final MetricsRegistry registry;
- private final MutableStat activeMinc, queuedMinc;
-
- // Use TabletServerMetricsFactory
- Metrics2TabletServerMinCMetrics(MetricsSystem system) {
- this.system = system;
- this.registry = new MetricsRegistry(Interns.info(NAME, DESCRIPTION));
- this.registry.tag(MsInfo.ProcessName, MetricsSystemHelper.getProcessName());
-
- activeMinc = registry.newStat(MINC, "Minor compactions", "Ops", "Count", true);
- queuedMinc = registry.newStat(QUEUE, "Queued minor compactions", "Ops", "Count", true);
- }
-
- @Override
- public void add(String name, long value) {
- if (MINC.equals(name)) {
- activeMinc.add(value);
- } else if (QUEUE.equals(name)) {
- queuedMinc.add(value);
- }
- }
-
- @Override
- public void register() {
- system.register(NAME, DESCRIPTION, this);
- }
-
- @Override
- public boolean isEnabled() {
- return true;
- }
-
- @Override
- public void getMetrics(MetricsCollector collector, boolean all) {
- MetricsRecordBuilder builder = collector.addRecord(RECORD).setContext(CONTEXT);
-
- registry.snapshot(builder, all);
- }
-
-}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java
deleted file mode 100644
index 294960b..0000000
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java
+++ /dev/null
@@ -1,80 +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;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
-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.Interns;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableStat;
-
-public class Metrics2TabletServerScanMetrics
- implements Metrics, MetricsSource, TabletServerScanMetricsKeys {
- public static final String NAME = TSERVER_NAME + ",sub=Scans",
- DESCRIPTION = "TabletServer Scan Metrics", CONTEXT = "tserver", RECORD = "Scans";
-
- private final MetricsSystem system;
- private final MetricsRegistry registry;
- private final MutableStat scans, resultsPerScan, yields;
-
- // Use TabletServerMetricsFactory
- Metrics2TabletServerScanMetrics(MetricsSystem system) {
- this.system = system;
- this.registry = new MetricsRegistry(Interns.info(NAME, DESCRIPTION));
- this.registry.tag(MsInfo.ProcessName, MetricsSystemHelper.getProcessName());
-
- scans = registry.newStat(SCAN, "Scans", "Ops", "Count", true);
- resultsPerScan = registry.newStat(RESULT_SIZE, "Results per scan", "Ops", "Count", true);
- yields = registry.newStat(YIELD, "Yields", "Ops", "Count", true);
- }
-
- @Override
- public void add(String name, long value) {
- if (SCAN.equals(name)) {
- scans.add(value);
- } else if (RESULT_SIZE.equals(name)) {
- resultsPerScan.add(value);
- } else if (YIELD.equals(name)) {
- yields.add(value);
- } else {
- throw new RuntimeException("Could not find metric to update for name " + name);
- }
- }
-
- @Override
- public void register() {
- system.register(NAME, DESCRIPTION, this);
- }
-
- @Override
- public boolean isEnabled() {
- return true;
- }
-
- @Override
- public void getMetrics(MetricsCollector collector, boolean all) {
- MetricsRecordBuilder builder = collector.addRecord(RECORD).setContext(CONTEXT);
-
- registry.snapshot(builder, all);
- }
-
-}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerUpdateMetrics.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerUpdateMetrics.java
deleted file mode 100644
index 3063997..0000000
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerUpdateMetrics.java
+++ /dev/null
@@ -1,104 +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;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
-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.Interns;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableCounterLong;
-import org.apache.hadoop.metrics2.lib.MutableStat;
-
-public class Metrics2TabletServerUpdateMetrics
- implements Metrics, MetricsSource, TabletServerUpdateMetricsKeys {
- public static final String NAME = TSERVER_NAME + ",sub=Updates",
- DESCRIPTION = "TabletServer Update Metrics", CONTEXT = "tserver", RECORD = "Updates";
-
- private final MetricsSystem system;
- private final MetricsRegistry registry;
-
- private final MutableCounterLong permissionErrorsCounter, unknownTabletErrorsCounter,
- constraintViolationsCounter;
- private final MutableStat commitPrepStat, walogWriteTimeStat, commitTimeStat,
- mutationArraySizeStat;
-
- // Use TabletServerMetricsFactory
- Metrics2TabletServerUpdateMetrics(MetricsSystem system) {
- this.system = system;
- this.registry = new MetricsRegistry(Interns.info(NAME, DESCRIPTION));
- this.registry.tag(MsInfo.ProcessName, MetricsSystemHelper.getProcessName());
-
- permissionErrorsCounter =
- registry.newCounter(Interns.info(PERMISSION_ERRORS, "Permission Errors"), 0L);
- unknownTabletErrorsCounter =
- registry.newCounter(Interns.info(UNKNOWN_TABLET_ERRORS, "Unknown Tablet Errors"), 0L);
- constraintViolationsCounter =
- registry.newCounter(Interns.info(CONSTRAINT_VIOLATIONS, "Table Constraint Violations"), 0L);
-
- commitPrepStat =
- registry.newStat(COMMIT_PREP, "preparing to commit mutations", "Ops", "Time", true);
- walogWriteTimeStat =
- registry.newStat(WALOG_WRITE_TIME, "writing mutations to WAL", "Ops", "Time", true);
- commitTimeStat = registry.newStat(COMMIT_TIME, "committing mutations", "Ops", "Time", true);
- mutationArraySizeStat =
- registry.newStat(MUTATION_ARRAY_SIZE, "mutation array", "ops", "Size", true);
- }
-
- @Override
- public void add(String name, long value) {
- if (PERMISSION_ERRORS.equals(name)) {
- permissionErrorsCounter.incr(value);
- } else if (UNKNOWN_TABLET_ERRORS.equals(name)) {
- unknownTabletErrorsCounter.incr(value);
- } else if (MUTATION_ARRAY_SIZE.equals(name)) {
- mutationArraySizeStat.add(value);
- } else if (COMMIT_PREP.equals(name)) {
- commitPrepStat.add(value);
- } else if (CONSTRAINT_VIOLATIONS.equals(name)) {
- constraintViolationsCounter.incr(value);
- } else if (WALOG_WRITE_TIME.equals(name)) {
- walogWriteTimeStat.add(value);
- } else if (COMMIT_TIME.equals(name)) {
- commitTimeStat.add(value);
- } else {
- throw new RuntimeException("Cannot process metric with name " + name);
- }
- }
-
- @Override
- public void register() {
- system.register(NAME, DESCRIPTION, this);
- }
-
- @Override
- public boolean isEnabled() {
- return true;
- }
-
- @Override
- public void getMetrics(MetricsCollector collector, boolean all) {
- MetricsRecordBuilder builder = collector.addRecord(RECORD).setContext(CONTEXT);
-
- registry.snapshot(builder, all);
- }
-
-}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetricsKeys.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TServerMetrics.java
similarity index 52%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetricsKeys.java
rename to server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TServerMetrics.java
index 5df2a56..ff08ce6 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetricsKeys.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TServerMetrics.java
@@ -16,17 +16,22 @@
*/
package org.apache.accumulo.tserver.metrics;
-/**
- * Keys to identify which update metric is being altered
- */
-public interface TabletServerUpdateMetricsKeys {
+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);
+ }
- String PERMISSION_ERRORS = "permissionErrors";
- String UNKNOWN_TABLET_ERRORS = "unknownTabletErrors";
- String MUTATION_ARRAY_SIZE = "mutationArraysSize";
- String COMMIT_PREP = "commitPrep";
- String CONSTRAINT_VIOLATIONS = "constraintViolations";
- String WALOG_WRITE_TIME = "waLogWriteTime";
- String COMMIT_TIME = "commitTime";
+ 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
new file mode 100644
index 0000000..2bbaf8d
--- /dev/null
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java
@@ -0,0 +1,89 @@
+/*
+ * 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.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 {
+
+ private final TabletServerMetricsUtil util;
+
+ 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;
+
+ 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());
+ }
+
+ @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());
+ }
+}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsFactory.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsFactory.java
deleted file mode 100644
index aaa7b73..0000000
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsFactory.java
+++ /dev/null
@@ -1,65 +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;
-import org.apache.accumulo.server.metrics.MetricsSystemHelper;
-import org.apache.accumulo.tserver.TabletServer;
-import org.apache.hadoop.metrics2.MetricsSystem;
-
-/**
- * Factory to create Metrics instances for various TabletServer functions.
- *
- * Necessary shim to support both the custom JMX metrics from <1.7.0 and the new Hadoop Metrics2
- * implementations. Refactoring with 2.0 may make this class unnecessary - keeping for now to
- * minimize code chages as metrics is evaluated.
- */
-public class TabletServerMetricsFactory {
- private final MetricsSystem metricsSystem;
-
- public TabletServerMetricsFactory() {
- metricsSystem = MetricsSystemHelper.getInstance();
- }
-
- /**
- * Create Metrics to track MinorCompactions
- */
- public Metrics createMincMetrics() {
- return new Metrics2TabletServerMinCMetrics(metricsSystem);
- }
-
- /**
- * Create Metrics to track TabletServer state
- */
- public Metrics createTabletServerMetrics(TabletServer tserver) {
- return new Metrics2TabletServerMetrics(tserver, metricsSystem);
- }
-
- /**
- * Create Metrics to track scans
- */
- public Metrics createScanMetrics() {
- return new Metrics2TabletServerScanMetrics(metricsSystem);
- }
-
- /**
- * Create Metrics to track updates (writes)
- */
- public Metrics createUpdateMetrics() {
- return new Metrics2TabletServerUpdateMetrics(metricsSystem);
- }
-}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetricsKeys.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetrics.java
similarity index 55%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetricsKeys.java
rename to server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetrics.java
index 29d4846..488ff64 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetricsKeys.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMinCMetrics.java
@@ -16,12 +16,28 @@
*/
package org.apache.accumulo.tserver.metrics;
-/**
- * Keys to reference minor compaction metrics
- */
-public interface TabletServerMinCMetricsKeys {
+import org.apache.hadoop.metrics2.lib.MetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableStat;
+
+public class TabletServerMinCMetrics extends TServerMetrics {
+
+ private final MutableStat activeMinc;
+ private final MutableStat queuedMinc;
+
+ public TabletServerMinCMetrics() {
+ super("MinorCompactions");
+
+ MetricsRegistry registry = super.getRegistry();
+ activeMinc = registry.newStat("minc", "Minor compactions", "Ops", "Count", true);
+ queuedMinc = registry.newStat("queue", "Queued minor compactions", "Ops", "Count", true);
+ }
+
+ public void addActive(long value) {
+ activeMinc.add(value);
+ }
- String MINC = "minc";
- String QUEUE = "queue";
+ public void addQueued(long value) {
+ queuedMinc.add(value);
+ }
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsKeys.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java
similarity index 50%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsKeys.java
rename to server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java
index b755597..a801171 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsKeys.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java
@@ -16,27 +16,34 @@
*/
package org.apache.accumulo.tserver.metrics;
-/**
- * Keys for general tablet server metrics
- */
-public interface TabletServerMetricsKeys {
-
- String ENTRIES = "entries";
- String ENTRIES_IN_MEM = "entriesInMem";
- String HOLD_TIME = "holdTime";
- String FILES_PER_TABLET = "filesPerTablet";
- String ACTIVE_MAJCS = "activeMajCs";
- String QUEUED_MAJCS = "queuedMajCs";
- String ACTIVE_MINCS = "activeMinCs";
- String QUEUED_MINCS = "queuedMinCs";
- String ONLINE_TABLETS = "onlineTablets";
- String OPENING_TABLETS = "openingTablets";
- String UNOPENED_TABLETS = "unopenedTablets";
- String QUERIES = "queries";
- String TOTAL_MINCS = "totalMinCs";
- String INGEST_RATE = "ingestRate";
- String INGEST_BYTE_RATE = "ingestByteRate";
- String QUERY_RATE = "queryRate";
- String QUERY_BYTE_RATE = "queryByteRate";
- String SCANNED_RATE = "scannedRate";
+import org.apache.hadoop.metrics2.lib.MetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableStat;
+
+public class TabletServerScanMetrics extends TServerMetrics {
+
+ private final MutableStat scans;
+ private final MutableStat resultsPerScan;
+ private final MutableStat yields;
+
+ public TabletServerScanMetrics() {
+ super("Scans");
+
+ 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);
+ }
+
+ public void addScan(long value) {
+ scans.add(value);
+ }
+
+ public void addResult(long value) {
+ resultsPerScan.add(value);
+ }
+
+ public void addYield(long value) {
+ yields.add(value);
+ }
+
}
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
new file mode 100644
index 0000000..f24a271
--- /dev/null
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerUpdateMetrics.java
@@ -0,0 +1,81 @@
+/*
+ * 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.hadoop.metrics2.lib.MetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableCounterLong;
+import org.apache.hadoop.metrics2.lib.MutableStat;
+
+public class TabletServerUpdateMetrics extends TServerMetrics {
+
+ private final MutableCounterLong permissionErrorsCounter;
+ private final MutableCounterLong unknownTabletErrorsCounter;
+ private final MutableCounterLong constraintViolationsCounter;
+
+ private final MutableStat commitPrepStat;
+ private final MutableStat walogWriteTimeStat;
+ private final MutableStat commitTimeStat;
+ private final MutableStat mutationArraySizeStat;
+
+ 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);
+ }
+
+ public void addPermissionErrors(long value) {
+ permissionErrorsCounter.incr(value);
+ }
+
+ public void addUnknownTabletErrors(long value) {
+ unknownTabletErrorsCounter.incr(value);
+ }
+
+ public void addMutationArraySize(long value) {
+ mutationArraySizeStat.add(value);
+ }
+
+ public void addCommitPrep(long value) {
+ commitPrepStat.add(value);
+ }
+
+ public void addConstraintViolations(long value) {
+ constraintViolationsCounter.incr(value);
+ }
+
+ public void addWalogWriteTime(long value) {
+ walogWriteTimeStat.add(value);
+ }
+
+ public void addCommitTime(long value) {
+ commitTimeStat.add(value);
+ }
+
+}
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 2b25943..f71c4f5 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
@@ -103,7 +103,6 @@ import org.apache.accumulo.server.fs.VolumeUtil;
import org.apache.accumulo.server.fs.VolumeUtil.TabletFiles;
import org.apache.accumulo.server.master.state.TServerInstance;
import org.apache.accumulo.server.master.tableOps.UserCompactionConfig;
-import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.problems.ProblemReport;
import org.apache.accumulo.server.problems.ProblemReports;
import org.apache.accumulo.server.problems.ProblemType;
@@ -135,8 +134,7 @@ import org.apache.accumulo.tserver.compaction.WriteParameters;
import org.apache.accumulo.tserver.constraints.ConstraintChecker;
import org.apache.accumulo.tserver.log.DfsLogger;
import org.apache.accumulo.tserver.mastermessage.TabletStatusMessage;
-import org.apache.accumulo.tserver.metrics.TabletServerMinCMetricsKeys;
-import org.apache.accumulo.tserver.metrics.TabletServerScanMetricsKeys;
+import org.apache.accumulo.tserver.metrics.TabletServerMinCMetrics;
import org.apache.accumulo.tserver.tablet.Compactor.CompactionCanceledException;
import org.apache.accumulo.tserver.tablet.Compactor.CompactionEnv;
import org.apache.commons.codec.DecoderException;
@@ -194,7 +192,7 @@ public class Tablet {
private final Set<ScanDataSource> activeScans = new HashSet<>();
- private static enum CloseState {
+ private enum CloseState {
OPEN, CLOSING, CLOSED, COMPLETE
}
@@ -214,7 +212,7 @@ public class Tablet {
// finish
private final CompactionWaitInfo compactionWaitInfo = new CompactionWaitInfo();
- static enum CompactionState {
+ enum CompactionState {
WAITING_TO_START, IN_PROGRESS
}
@@ -292,10 +290,11 @@ public class Tablet {
}
if (files == null) {
- if (location.getName().startsWith(Constants.CLONE_PREFIX))
+ if (location.getName().startsWith(Constants.CLONE_PREFIX)) {
log.debug("Tablet {} had no dir, creating {}", extent, location); // its a clone dir...
- else
+ } else {
log.warn("Tablet {} had no dir, creating {}", extent, location);
+ }
getTabletServer().getFileSystem().mkdirs(location);
}
@@ -398,9 +397,9 @@ public class Tablet {
@Override
public void propertyChanged(String prop) {
- if (prop.startsWith(Property.TABLE_CONSTRAINT_PREFIX.getKey()))
+ if (prop.startsWith(Property.TABLE_CONSTRAINT_PREFIX.getKey())) {
reloadConstraints();
- else if (prop.equals(Property.TABLE_DEFAULT_SCANTIME_VISIBILITY.getKey())) {
+ } else if (prop.equals(Property.TABLE_DEFAULT_SCANTIME_VISIBILITY.getKey())) {
try {
log.info("Default security labels changed for extent: {}", extent);
setupDefaultSecurityLabels(extent);
@@ -431,8 +430,9 @@ public class Tablet {
final CommitSession commitSession = getTabletMemory().getCommitSession();
try {
Set<String> absPaths = new HashSet<>();
- for (FileRef ref : datafiles.keySet())
+ for (FileRef ref : datafiles.keySet()) {
absPaths.add(ref.path().toString());
+ }
tabletServer.recover(this.getTabletServer().getFileSystem(), extent, logEntries, absPaths,
m -> {
@@ -581,8 +581,9 @@ public class Tablet {
boolean tabletClosed = false;
Set<ByteSequence> cfset = null;
- if (columnSet.size() > 0)
+ if (columnSet.size() > 0) {
cfset = LocalityGroupUtil.families(columnSet);
+ }
long returnTime = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(batchTimeOut);
if (batchTimeOut <= 0 || batchTimeOut == Long.MAX_VALUE) {
@@ -606,10 +607,11 @@ public class Tablet {
int entriesAdded = 0;
try {
- if (cfset != null)
+ if (cfset != null) {
mmfi.seek(range, cfset, true);
- else
+ } else {
mmfi.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
+ }
while (mmfi.hasTop()) {
if (yield.hasYielded()) {
@@ -652,9 +654,7 @@ public class Tablet {
addUnfinishedRange(lookupResult, range, yieldPosition, false);
log.debug("Scan yield detected at position " + yieldPosition);
- Metrics scanMetrics = getTabletServer().getScanMetrics();
- if (scanMetrics.isEnabled())
- scanMetrics.add(TabletServerScanMetricsKeys.YIELD, 1);
+ getTabletServer().getScanMetrics().addYield(1);
}
} catch (TooManyFilesException tmfe) {
// treat this as a closed tablet, and let the client retry
@@ -693,14 +693,16 @@ public class Tablet {
private void handleTabletClosedDuringScan(List<KVEntry> results, LookupResult lookupResult,
boolean exceededMemoryUsage, Range range, int entriesAdded) {
- if (exceededMemoryUsage)
+ if (exceededMemoryUsage) {
throw new IllegalStateException(
"Tablet " + extent + "should not exceed memory usage or close, not both");
+ }
- if (entriesAdded > 0)
+ if (entriesAdded > 0) {
addUnfinishedRange(lookupResult, range, results.get(results.size() - 1).getKey(), false);
- else
+ } else {
lookupResult.unfinishedRanges.add(range);
+ }
lookupResult.closed = true;
}
@@ -774,8 +776,9 @@ public class Tablet {
synchronized (this) {
queryCount += results.size();
- if (result != null)
+ if (result != null) {
queryBytes += result.dataSize;
+ }
}
}
}
@@ -853,14 +856,13 @@ public class Tablet {
}
log.debug("Scan yield detected at position " + continueKey);
- Metrics scanMetrics = getTabletServer().getScanMetrics();
- if (scanMetrics.isEnabled())
- scanMetrics.add(TabletServerScanMetricsKeys.YIELD, 1);
+ getTabletServer().getScanMetrics().addYield(1);
} else if (!iter.hasTop()) {
// end of tablet has been reached
continueKey = null;
- if (results.size() == 0)
+ if (results.size() == 0) {
results = null;
+ }
}
return new Batch(skipContinueKey, results, continueKey, resultBytes);
@@ -910,8 +912,9 @@ public class Tablet {
count = memTable.getNumEntries();
DataFileValue dfv = null;
- if (mergeFile != null)
+ if (mergeFile != null) {
dfv = getDatafileManager().getDatafileSizes().get(mergeFile);
+ }
MinorCompactor compactor = new MinorCompactor(tabletServer, this, memTable, mergeFile, dfv,
tmpDatafile, mincReason, tableConfiguration);
@@ -938,15 +941,14 @@ public class Tablet {
if (!failed) {
lastMinorCompactionFinishTime = System.currentTimeMillis();
}
- Metrics minCMetrics = getTabletServer().getMinCMetrics();
- if (minCMetrics.isEnabled())
- minCMetrics.add(TabletServerMinCMetricsKeys.MINC, (lastMinorCompactionFinishTime - start));
+ TabletServerMinCMetrics minCMetrics = getTabletServer().getMinCMetrics();
+ minCMetrics.addActive(lastMinorCompactionFinishTime - start);
if (hasQueueTime) {
timer.updateTime(Operation.MINOR, queued, start, count, failed);
- if (minCMetrics.isEnabled())
- minCMetrics.add(TabletServerMinCMetricsKeys.QUEUE, (start - queued));
- } else
+ minCMetrics.addQueued(start - queued);
+ } else {
timer.updateTime(Operation.MINOR, start, failed);
+ }
}
}
@@ -981,21 +983,25 @@ public class Tablet {
// only want one thing at a time to update flush ID to ensure that metadata table and tablet
// in memory state are consistent
- if (updatingFlushID)
+ if (updatingFlushID) {
return;
+ }
- if (lastFlushID >= tableFlushID)
+ if (lastFlushID >= tableFlushID) {
return;
+ }
- if (isClosing() || isClosed() || getTabletMemory().memoryReservedForMinC())
+ if (isClosing() || isClosed() || getTabletMemory().memoryReservedForMinC()) {
return;
+ }
if (getTabletMemory().getMemTable().getNumEntries() == 0) {
lastFlushID = tableFlushID;
updatingFlushID = true;
updateMetadata = true;
- } else
+ } else {
initiateMinor = true;
+ }
}
if (updateMetadata) {
@@ -1003,8 +1009,9 @@ public class Tablet {
// a race condition
MetadataTableUtil.updateTabletFlushID(extent, tableFlushID, context,
getTabletServer().getLock());
- } else if (initiateMinor)
+ } else if (initiateMinor) {
initiateMinorCompaction(tableFlushID, MinorCompactionReason.USER);
+ }
} finally {
if (updateMetadata) {
@@ -1046,16 +1053,18 @@ public class Tablet {
return false;
}
MinorCompactionTask mct = createMinorCompactionTask(flushId, mincReason);
- if (mct == null)
+ if (mct == null) {
return false;
+ }
mct.run();
return true;
}
boolean initiateMinorCompaction(long flushId, MinorCompactionReason mincReason) {
MinorCompactionTask mct = createMinorCompactionTask(flushId, mincReason);
- if (mct == null)
+ if (mct == null) {
return false;
+ }
getTabletResources().executeMinorCompaction(mct);
return true;
}
@@ -1080,12 +1089,14 @@ public class Tablet {
logMessage.append(extent);
logMessage.append(" closeState " + closeState);
logMessage.append(" majorCompactionState " + majorCompactionState);
- if (getTabletMemory() != null)
+ if (getTabletMemory() != null) {
logMessage.append(" tabletMemory.memoryReservedForMinC() "
+ getTabletMemory().memoryReservedForMinC());
- if (getTabletMemory() != null && getTabletMemory().getMemTable() != null)
+ }
+ if (getTabletMemory() != null && getTabletMemory().getMemTable() != null) {
logMessage.append(" tabletMemory.getMemTable().getNumEntries() "
+ getTabletMemory().getMemTable().getNumEntries());
+ }
logMessage.append(" updatingFlushID " + updatingFlushID);
return null;
@@ -1096,8 +1107,9 @@ public class Tablet {
}
} finally {
// log outside of sync block
- if (logMessage != null && log.isDebugEnabled())
+ if (logMessage != null && log.isDebugEnabled()) {
log.debug("{}", logMessage);
+ }
}
log.debug(String.format("MinC initiate lock %.2f secs", (t2 - t1) / 1000.0));
@@ -1206,8 +1218,9 @@ public class Tablet {
Violations more = cc.check(cenv, mutation);
if (more != null) {
violations.add(more);
- if (violators == null)
+ if (violators == null) {
violators = new ArrayList<>();
+ }
violators.add(mutation);
}
}
@@ -1231,8 +1244,9 @@ public class Tablet {
// if everything is a violation, then it is expected that
// code calling this will not log or commit
commitSession = finishPreparingMutations(time);
- if (commitSession == null)
+ if (commitSession == null) {
return null;
+ }
}
throw new TConstraintViolationException(violations, violators, nonViolators, commitSession);
@@ -1267,8 +1281,9 @@ public class Tablet {
+ extent);
}
writesInProgress--;
- if (writesInProgress == 0)
+ if (writesInProgress == 0) {
this.notifyAll();
+ }
}
public synchronized void abortCommit(CommitSession commitSession) {
@@ -1467,8 +1482,9 @@ public class Tablet {
tableConfiguration.getNamespaceConfiguration().removeObserver(configObserver);
tableConfiguration.removeObserver(configObserver);
- if (completeClose)
+ if (completeClose) {
closeState = CloseState.COMPLETE;
+ }
}
private void closeConsistencyCheck() {
@@ -1566,10 +1582,12 @@ public class Tablet {
*
*/
public boolean needsMajorCompaction(MajorCompactionReason reason) {
- if (isMajorCompactionRunning())
+ if (isMajorCompactionRunning()) {
return false;
- if (reason == MajorCompactionReason.CHOP || reason == MajorCompactionReason.USER)
+ }
+ if (reason == MajorCompactionReason.CHOP || reason == MajorCompactionReason.USER) {
return true;
+ }
return getTabletResources().needsMajorCompaction(getDatafileManager().getDatafileSizes(),
reason);
}
@@ -1583,8 +1601,9 @@ public class Tablet {
public long estimateTabletSize() {
long size = 0L;
- for (DataFileValue sz : getDatafileManager().getDatafileSizes().values())
+ for (DataFileValue sz : getDatafileManager().getDatafileSizes().values()) {
size += sz.getSize();
+ }
return size;
}
@@ -1653,10 +1672,11 @@ public class Tablet {
}
if (candidate.compareRow(lastRow) != 0) {
// we should use this ratio in split size estimations
- if (log.isTraceEnabled())
+ if (log.isTraceEnabled()) {
log.trace(
String.format("Splitting at %6.2f instead of .5, row at .5 is same as end row%n",
keys.firstKey()));
+ }
return new SplitRowSpec(keys.firstKey(), candidate.getRow());
}
@@ -1792,8 +1812,9 @@ public class Tablet {
*
*/
public synchronized boolean needsSplit() {
- if (isClosing() || isClosed())
+ if (isClosing() || isClosed()) {
return false;
+ }
return findSplitRow(getDatafileManager().getFiles()) != null;
}
@@ -1905,8 +1926,9 @@ public class Tablet {
// deletes
Set<FileRef> droppedFiles = new HashSet<>();
droppedFiles.addAll(inputFiles);
- if (plan != null)
+ if (plan != null) {
droppedFiles.addAll(plan.deleteFiles);
+ }
propogateDeletes = !(droppedFiles.equals(allFiles.keySet()));
log.debug("Major compaction plan: {} propagate deletes : {}", plan, propogateDeletes);
filesToCompact = new HashMap<>(allFiles);
@@ -1953,9 +1975,10 @@ public class Tablet {
compactionIterators = compactionId.getSecond().getIterators();
synchronized (this) {
- if (lastCompactID >= compactionId.getFirst())
+ if (lastCompactID >= compactionId.getFirst()) {
// already compacted
return majCStats;
+ }
}
}
@@ -2006,9 +2029,10 @@ public class Tablet {
HashMap<FileRef,DataFileValue> copy =
new HashMap<>(getDatafileManager().getDatafileSizes());
- if (!copy.keySet().containsAll(smallestFiles))
+ if (!copy.keySet().containsAll(smallestFiles)) {
throw new IllegalStateException("Cannot find data file values for " + smallestFiles
+ " on " + extent + " during MajC");
+ }
copy.keySet().retainAll(smallestFiles);
@@ -2056,20 +2080,26 @@ public class Tablet {
protected AccumuloConfiguration createTableConfiguration(TableConfiguration base,
CompactionPlan plan) {
- if (plan == null || plan.writeParameters == null)
+ if (plan == null || plan.writeParameters == null) {
return base;
+ }
WriteParameters p = plan.writeParameters;
ConfigurationCopy result = new ConfigurationCopy(base);
- if (p.getHdfsBlockSize() > 0)
+ if (p.getHdfsBlockSize() > 0) {
result.set(Property.TABLE_FILE_BLOCK_SIZE, "" + p.getHdfsBlockSize());
- if (p.getBlockSize() > 0)
+ }
+ if (p.getBlockSize() > 0) {
result.set(Property.TABLE_FILE_COMPRESSED_BLOCK_SIZE, "" + p.getBlockSize());
- if (p.getIndexBlockSize() > 0)
+ }
+ if (p.getIndexBlockSize() > 0) {
result.set(Property.TABLE_FILE_COMPRESSED_BLOCK_SIZE_INDEX, "" + p.getIndexBlockSize());
- if (p.getCompressType() != null)
+ }
+ if (p.getCompressType() != null) {
result.set(Property.TABLE_FILE_COMPRESSION_TYPE, p.getCompressType());
- if (p.getReplication() != 0)
+ }
+ if (p.getReplication() != 0) {
result.set(Property.TABLE_FILE_REPLICATION, "" + p.getReplication());
+ }
return result;
}
@@ -2086,10 +2116,12 @@ public class Tablet {
PriorityQueue<Pair<FileRef,Long>> fileHeap =
new PriorityQueue<>(filesToCompact.size(), (o1, o2) -> {
- if (o1.getSecond().equals(o2.getSecond()))
+ if (o1.getSecond().equals(o2.getSecond())) {
return o1.getFirst().compareTo(o2.getFirst());
- if (o1.getSecond() < o2.getSecond())
+ }
+ if (o1.getSecond() < o2.getSecond()) {
return -1;
+ }
return 1;
});
@@ -2168,8 +2200,9 @@ public class Tablet {
}
}
long count = 0;
- if (majCStats != null)
+ if (majCStats != null) {
count = majCStats.getEntriesRead();
+ }
timer.updateTime(Operation.MAJOR, queued, start, count, !success);
return majCStats;
@@ -2276,9 +2309,9 @@ public class Tablet {
long t1 = System.currentTimeMillis();
// choose a split point
SplitRowSpec splitPoint;
- if (sp == null)
+ if (sp == null) {
splitPoint = findSplitRow(getDatafileManager().getFiles());
- else {
+ } else {
Text tsp = new Text(sp);
splitPoint = new SplitRowSpec(
FileUtil.estimatePercentageLTE(context, tabletDirectory, extent.getPrevEndRow(),
@@ -2543,9 +2576,10 @@ public class Tablet {
logLock.lock();
synchronized (this) {
- if (removingLogs)
+ if (removingLogs) {
throw new IllegalStateException(
"Attempted to clear logs when removal of logs in progress on " + extent);
+ }
for (DfsLogger logger : otherLogs) {
otherLogsCopy.add(logger.toString());
@@ -2563,8 +2597,9 @@ public class Tablet {
// finishClearingUnusedLogs() calls rebuildReferencedLogs(). See the comments in
// rebuildReferencedLogs() for more info.
- if (unusedLogs.size() > 0)
+ if (unusedLogs.size() > 0) {
removingLogs = true;
+ }
}
// do debug logging outside tablet lock
@@ -2614,24 +2649,28 @@ public class Tablet {
boolean addToOther;
- if (memTable == getTabletMemory().getMinCMemTable())
+ if (memTable == getTabletMemory().getMinCMemTable()) {
addToOther = true;
- else if (memTable == getTabletMemory().getMemTable())
+ } else if (memTable == getTabletMemory().getMemTable()) {
addToOther = false;
- else
+ } else {
throw new IllegalArgumentException("Passed in memtable that is not in use for " + extent);
+ }
if (mincFinish) {
- if (addToOther)
+ if (addToOther) {
throw new IllegalStateException("Adding to other logs for mincFinish on " + extent);
- if (otherLogs.size() != 0)
+ }
+ if (otherLogs.size() != 0) {
throw new IllegalStateException("Expect other logs to be 0 when minC finish, but its "
+ otherLogs + " for " + extent);
+ }
// when writing a minc finish event, there is no need to add the log to metadata
// if nothing has been logged for the tablet since the minor compaction started
- if (currentLogs.size() == 0)
+ if (currentLogs.size() == 0) {
return !releaseLock;
+ }
}
boolean added;
@@ -2655,8 +2694,9 @@ public class Tablet {
return !releaseLock;
}
} finally {
- if (releaseLock)
+ if (releaseLock) {
logLock.unlock();
+ }
}
}
@@ -2673,11 +2713,12 @@ public class Tablet {
String clazzName = strategyConfig.getClassName();
try {
Class<? extends CompactionStrategy> clazz;
- if (context != null && !context.equals(""))
+ if (context != null && !context.equals("")) {
clazz = AccumuloVFSClassLoader.getContextManager().loadClass(context, clazzName,
CompactionStrategy.class);
- else
+ } else {
clazz = AccumuloVFSClassLoader.loadClass(clazzName, CompactionStrategy.class);
+ }
CompactionStrategy strategy = clazz.newInstance();
strategy.init(strategyConfig.getOptions());
return strategy;
@@ -2690,14 +2731,16 @@ public class Tablet {
boolean updateMetadata = false;
synchronized (this) {
- if (lastCompactID >= compactionId)
+ if (lastCompactID >= compactionId) {
return;
+ }
if (isMinorCompactionRunning()) {
// want to wait for running minc to finish before starting majc, see ACCUMULO-3041
if (compactionWaitInfo.compactionID == compactionId) {
- if (lastFlushID == compactionWaitInfo.flushID)
+ if (lastFlushID == compactionWaitInfo.flushID) {
return;
+ }
} else {
compactionWaitInfo.compactionID = compactionId;
compactionWaitInfo.flushID = lastFlushID;
@@ -2706,8 +2749,9 @@ public class Tablet {
}
if (isClosing() || isClosed() || majorCompactionQueued.contains(MajorCompactionReason.USER)
- || isMajorCompactionRunning())
+ || isMajorCompactionRunning()) {
return;
+ }
CompactionStrategyConfig strategyConfig = compactionConfig.getCompactionStrategy();
CompactionStrategy strategy = createCompactionStrategy(strategyConfig);
@@ -2771,8 +2815,9 @@ public class Tablet {
public void updatePersistedTime(long bulkTime, Map<FileRef,DataFileValue> paths, long tid) {
synchronized (timeLock) {
- if (bulkTime > persistedTime)
+ if (bulkTime > persistedTime) {
persistedTime = bulkTime;
+ }
MetadataTableUtil.updateTabletDataFile(tid, extent, paths,
tabletTime.getMetadataValue(persistedTime), getTabletServer().getContext(),
@@ -2784,8 +2829,9 @@ public class Tablet {
public void updateTabletDataFile(long maxCommittedTime, FileRef newDatafile, FileRef absMergeFile,
DataFileValue dfv, Set<String> unusedWalLogs, Set<FileRef> filesInUseByScans, long flushId) {
synchronized (timeLock) {
- if (maxCommittedTime > persistedTime)
+ if (maxCommittedTime > persistedTime) {
persistedTime = maxCommittedTime;
+ }
String time = tabletTime.getMetadataValue(persistedTime);
MasterMetadataUtil.updateTabletDataFile(getTabletServer().getContext(), extent, newDatafile,
@@ -2835,8 +2881,9 @@ public class Tablet {
}
public synchronized void setLastCompactionID(Long compactionId) {
- if (compactionId != null)
+ if (compactionId != null) {
this.lastCompactID = compactionId;
+ }
}
public void minorCompactionWaitingToStart() {
@@ -2883,9 +2930,10 @@ public class Tablet {
} else {
lowDirectory = "/" + Constants.GENERATED_TABLET_DIRECTORY_PREFIX + namer.getNextName();
Path lowDirectoryPath = new Path(volume + "/" + tableId + "/" + lowDirectory);
- if (fs.exists(lowDirectoryPath))
+ if (fs.exists(lowDirectoryPath)) {
throw new IllegalStateException("Attempting to create tablet dir for tableID " + tableId
+ " and dir exists when it should not: " + lowDirectoryPath);
+ }
if (fs.mkdirs(lowDirectoryPath)) {
FileSystem lowDirectoryFs = fs.getVolumeByPath(lowDirectoryPath).getFileSystem();
return lowDirectoryPath
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 194aec2..79a9291 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
@@ -41,6 +41,7 @@ import org.apache.accumulo.fate.zookeeper.ZooLock.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;
@@ -104,7 +105,8 @@ public class ZombieTServer {
TransactionWatcher watcher = new TransactionWatcher(context);
final ThriftClientHandler tch = new ThriftClientHandler(context, watcher);
Processor<Iface> processor = new Processor<>(tch);
- ServerAddress serverPort = TServerUtils.startTServer(context.getConfiguration(),
+ ServerAddress serverPort = TServerUtils.startTServer(
+ Metrics.initSystem(ZombieTServer.class.getSimpleName()), context.getConfiguration(),
ThriftServerType.CUSTOM_HS_HA, processor, "ZombieTServer", "walking dead", 2, 1, 1000,
10 * 1024 * 1024, null, null, -1, HostAndPort.fromParts("0.0.0.0", port));
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 089e051..be015dd 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
@@ -71,6 +71,7 @@ import org.apache.accumulo.server.master.state.MetaDataStateStore;
import org.apache.accumulo.server.master.state.MetaDataTableScanner;
import org.apache.accumulo.server.master.state.TServerInstance;
import org.apache.accumulo.server.master.state.TabletLocationState;
+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;
@@ -296,8 +297,9 @@ public class NullTserver {
TransactionWatcher watcher = new TransactionWatcher(context);
ThriftClientHandler tch = new ThriftClientHandler(context, watcher);
Processor<Iface> processor = new Processor<>(tch);
- TServerUtils.startTServer(context.getConfiguration(), ThriftServerType.CUSTOM_HS_HA, processor,
- "NullTServer", "null tserver", 2, 1, 1000, 10 * 1024 * 1024, null, null, -1,
+ TServerUtils.startTServer(Metrics.initSystem(NullTserver.class.getSimpleName()),
+ context.getConfiguration(), ThriftServerType.CUSTOM_HS_HA, processor, "NullTServer",
+ "null tserver", 2, 1, 1000, 10 * 1024 * 1024, null, null, -1,
HostAndPort.fromParts("0.0.0.0", opts.port));
HostAndPort addr = HostAndPort.fromParts(InetAddress.getLocalHost().getHostName(), opts.port);