You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by do...@apache.org on 2024/09/13 21:19:12 UTC
(accumulo) branch main updated (233baddbe8 -> 3fc54839b7)
This is an automated email from the ASF dual-hosted git repository.
domgarguilo pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 233baddbe8 Merge branch '3.1'
add 5e09d985e7 Improve readability of timekeeping code in idle server check (#4737)
add 173a447929 Dynamically generate metrics documentation (#4850)
new 3fc54839b7 Merge Dynamically generate metrics documentation (#4850) into main
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/pom.xml | 15 +
.../org/apache/accumulo/core/metrics/Metric.java | 300 ++++++++
.../accumulo/core/metrics/MetricsDocGen.java | 110 +++
.../accumulo/core/metrics/MetricsProducer.java | 789 +--------------------
.../org/apache/accumulo/server/AbstractServer.java | 28 +-
.../server/compaction/PausedCompactionMetrics.java | 7 +-
.../accumulo/server/metrics/ProcessMetrics.java | 7 +-
.../accumulo/server/metrics/ThriftMetrics.java | 8 +-
.../accumulo/server/metrics/ThriftMetricsTest.java | 8 +-
.../org/apache/accumulo/compactor/Compactor.java | 9 +-
.../org/apache/accumulo/gc/metrics/GcMetrics.java | 43 +-
.../coordinator/CompactionCoordinator.java | 6 +-
.../compaction/coordinator/QueueMetrics.java | 19 +-
.../accumulo/manager/metrics/BalancerMetrics.java | 4 +-
.../accumulo/manager/metrics/ManagerMetrics.java | 12 +-
.../accumulo/manager/metrics/fate/FateMetrics.java | 22 +-
.../manager/metrics/fate/meta/MetaFateMetrics.java | 8 +-
.../apache/accumulo/tserver/BlockCacheMetrics.java | 29 +-
.../apache/accumulo/tserver/ScanServerMetrics.java | 17 +-
.../tserver/metrics/TabletServerMetrics.java | 58 +-
.../tserver/metrics/TabletServerMinCMetrics.java | 7 +-
.../tserver/metrics/TabletServerScanMetrics.java | 49 +-
.../tserver/metrics/TabletServerUpdateMetrics.java | 22 +-
.../org/apache/accumulo/test/ZombieScanIT.java | 4 +-
.../compaction/BadCompactionServiceConfigIT.java | 6 +-
.../CompactionPriorityQueueMetricsIT.java | 29 +-
.../compaction/ExternalCompactionMetricsIT.java | 3 +-
.../compaction/ExternalCompactionProgressIT.java | 19 +-
.../test/functional/IdleProcessMetricsIT.java | 9 +-
.../test/functional/MemoryStarvedMajCIT.java | 14 +-
.../test/functional/MemoryStarvedMinCIT.java | 14 +-
.../test/functional/MemoryStarvedScanIT.java | 11 +-
.../test/functional/OnDemandTabletUnloadingIT.java | 4 +-
.../apache/accumulo/test/metrics/MetricsIT.java | 93 ++-
34 files changed, 776 insertions(+), 1007 deletions(-)
create mode 100644 core/src/main/java/org/apache/accumulo/core/metrics/Metric.java
create mode 100644 core/src/main/java/org/apache/accumulo/core/metrics/MetricsDocGen.java
(accumulo) 01/01: Merge Dynamically generate metrics documentation (#4850) into main
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
domgarguilo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 3fc54839b74bbeef63898c049f50fb9ae0755ef9
Merge: 233baddbe8 173a447929
Author: Dom Garguilo <do...@apache.org>
AuthorDate: Fri Sep 13 17:18:12 2024 -0400
Merge Dynamically generate metrics documentation (#4850) into main
core/pom.xml | 15 +
.../org/apache/accumulo/core/metrics/Metric.java | 300 ++++++++
.../accumulo/core/metrics/MetricsDocGen.java | 110 +++
.../accumulo/core/metrics/MetricsProducer.java | 789 +--------------------
.../org/apache/accumulo/server/AbstractServer.java | 28 +-
.../server/compaction/PausedCompactionMetrics.java | 7 +-
.../accumulo/server/metrics/ProcessMetrics.java | 7 +-
.../accumulo/server/metrics/ThriftMetrics.java | 8 +-
.../accumulo/server/metrics/ThriftMetricsTest.java | 8 +-
.../org/apache/accumulo/compactor/Compactor.java | 9 +-
.../org/apache/accumulo/gc/metrics/GcMetrics.java | 43 +-
.../coordinator/CompactionCoordinator.java | 6 +-
.../compaction/coordinator/QueueMetrics.java | 19 +-
.../accumulo/manager/metrics/BalancerMetrics.java | 4 +-
.../accumulo/manager/metrics/ManagerMetrics.java | 12 +-
.../accumulo/manager/metrics/fate/FateMetrics.java | 22 +-
.../manager/metrics/fate/meta/MetaFateMetrics.java | 8 +-
.../apache/accumulo/tserver/BlockCacheMetrics.java | 29 +-
.../apache/accumulo/tserver/ScanServerMetrics.java | 17 +-
.../tserver/metrics/TabletServerMetrics.java | 58 +-
.../tserver/metrics/TabletServerMinCMetrics.java | 7 +-
.../tserver/metrics/TabletServerScanMetrics.java | 49 +-
.../tserver/metrics/TabletServerUpdateMetrics.java | 22 +-
.../org/apache/accumulo/test/ZombieScanIT.java | 4 +-
.../compaction/BadCompactionServiceConfigIT.java | 6 +-
.../CompactionPriorityQueueMetricsIT.java | 29 +-
.../compaction/ExternalCompactionMetricsIT.java | 3 +-
.../compaction/ExternalCompactionProgressIT.java | 19 +-
.../test/functional/IdleProcessMetricsIT.java | 9 +-
.../test/functional/MemoryStarvedMajCIT.java | 14 +-
.../test/functional/MemoryStarvedMinCIT.java | 14 +-
.../test/functional/MemoryStarvedScanIT.java | 11 +-
.../test/functional/OnDemandTabletUnloadingIT.java | 4 +-
.../apache/accumulo/test/metrics/MetricsIT.java | 93 ++-
34 files changed, 776 insertions(+), 1007 deletions(-)
diff --cc core/src/main/java/org/apache/accumulo/core/metrics/Metric.java
index 0000000000,9850d1a064..e131af1205
mode 000000,100644..100644
--- a/core/src/main/java/org/apache/accumulo/core/metrics/Metric.java
+++ b/core/src/main/java/org/apache/accumulo/core/metrics/Metric.java
@@@ -1,0 -1,276 +1,300 @@@
+ /*
+ * 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
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+ package org.apache.accumulo.core.metrics;
+
+ import java.util.HashMap;
+ import java.util.Map;
+
+ public enum Metric {
+ // General Server Metrics
+ SERVER_IDLE("accumulo.server.idle", MetricType.GAUGE,
+ "Indicates if the server is idle or not. The value will be 1 when idle and 0 when not idle.",
+ MetricCategory.GENERAL_SERVER),
+ LOW_MEMORY("accumulo.detected.low.memory", MetricType.GAUGE,
+ "Reports 1 when process memory usage is above threshold, 0 when memory is okay.",
+ MetricCategory.GENERAL_SERVER),
+
+ // Compactor Metrics
+ COMPACTOR_MAJC_STUCK("accumulo.compactor.majc.stuck", MetricType.LONG_TASK_TIMER, "",
+ MetricCategory.COMPACTOR),
+ COMPACTOR_ENTRIES_READ("accumulo.compactor.entries.read", MetricType.FUNCTION_COUNTER,
+ "Number of entries read by all threads performing compactions.", MetricCategory.COMPACTOR),
+ COMPACTOR_ENTRIES_WRITTEN("accumulo.compactor.entries.written", MetricType.FUNCTION_COUNTER,
+ "Number of entries written by all threads performing compactions.", MetricCategory.COMPACTOR),
++ COMPACTOR_JOB_PRIORITY_QUEUES("accumulo.compactor.queue.count", MetricType.GAUGE,
++ "Number of priority queues for compaction jobs.", MetricCategory.COMPACTOR),
++ COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH("accumulo.compactor.queue.length", MetricType.GAUGE, "",
++ MetricCategory.COMPACTOR),
++ COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED("accumulo.compactor.queue.jobs.dequeued",
++ MetricType.GAUGE, "", MetricCategory.COMPACTOR),
++ COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED("accumulo.compactor.queue.jobs.queued", MetricType.GAUGE,
++ "", MetricCategory.COMPACTOR),
++ COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED("accumulo.compactor.queue.jobs.rejected",
++ MetricType.GAUGE, "", MetricCategory.COMPACTOR),
++ COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY("accumulo.compactor.queue.jobs.priority",
++ MetricType.GAUGE, "", MetricCategory.COMPACTOR),
+
+ // Fate Metrics
+ FATE_TYPE_IN_PROGRESS("accumulo.fate.ops.in.progress.by.type", MetricType.GAUGE,
+ "Number of FATE operations in progress. The type is designated by the `op.type` tag.",
+ MetricCategory.FATE),
+ FATE_OPS("accumulo.fate.ops", MetricType.GAUGE, "Tracks all the current FATE ops in any state.",
+ MetricCategory.FATE),
+ FATE_OPS_ACTIVITY("accumulo.fate.ops.activity", MetricType.GAUGE, "", MetricCategory.FATE),
+ FATE_ERRORS("accumulo.fate.errors", MetricType.GAUGE, "", MetricCategory.FATE),
+ FATE_TX("accumulo.fate.tx", MetricType.GAUGE,
+ "The state is now in a tag (e.g., state=new, state=in.progress, state=failed, etc.).",
+ MetricCategory.FATE),
+
+ // Garbage Collection Metrics
+ GC_STARTED("accumulo.gc.started", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_FINISHED("accumulo.gc.finished", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_CANDIDATES("accumulo.gc.candidates", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_IN_USE("accumulo.gc.in.use", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_DELETED("accumulo.gc.deleted", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_ERRORS("accumulo.gc.errors", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_WAL_STARTED("accumulo.gc.wal.started", MetricType.GAUGE, "",
+ MetricCategory.GARBAGE_COLLECTION),
+ GC_WAL_FINISHED("accumulo.gc.wal.finished", MetricType.GAUGE, "",
+ MetricCategory.GARBAGE_COLLECTION),
+ GC_WAL_CANDIDATES("accumulo.gc.wal.candidates", MetricType.GAUGE, "",
+ MetricCategory.GARBAGE_COLLECTION),
+ GC_WAL_IN_USE("accumulo.gc.wal.in.use", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_WAL_DELETED("accumulo.gc.wal.deleted", MetricType.GAUGE, "",
+ MetricCategory.GARBAGE_COLLECTION),
+ GC_WAL_ERRORS("accumulo.gc.wal.errors", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+ GC_POST_OP_DURATION("accumulo.gc.post.op.duration", MetricType.GAUGE, "",
+ MetricCategory.GARBAGE_COLLECTION),
+ GC_RUN_CYCLE("accumulo.gc.run.cycle", MetricType.GAUGE, "", MetricCategory.GARBAGE_COLLECTION),
+
+ // Tablet Server Metrics
+ TSERVER_ENTRIES("accumulo.tserver.entries", MetricType.GAUGE, "", MetricCategory.TABLET_SERVER),
+ TSERVER_MEM_ENTRIES("accumulo.tserver.entries.mem", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_MAJC_RUNNING("accumulo.tserver.majc.running", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_MAJC_STUCK("accumulo.tserver.majc.stuck", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_MAJC_QUEUED("accumulo.tserver.majc.queued", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_MINC_QUEUED("accumulo.tserver.minc.queued", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_MINC_RUNNING("accumulo.tserver.minc.running", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_MINC_TOTAL("accumulo.tserver.minc.total", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_TABLETS_ONLINE("accumulo.tserver.tablets.online", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_TABLETS_LONG_ASSIGNMENTS("accumulo.tserver.tablets.assignments.warning", MetricType.GAUGE,
+ "", MetricCategory.TABLET_SERVER),
+ TSERVER_TABLETS_OPENING("accumulo.tserver.tablets.opening", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_TABLETS_UNOPENED("accumulo.tserver.tablets.unopened", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_TABLETS_FILES("accumulo.tserver.tablets.files", MetricType.GAUGE, "",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_INGEST_MUTATIONS("accumulo.tserver.ingest.mutations", MetricType.GAUGE,
+ "Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be derived.",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_INGEST_BYTES("accumulo.tserver.ingest.bytes", MetricType.GAUGE,
+ "Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be derived.",
+ MetricCategory.TABLET_SERVER),
+ TSERVER_HOLD("accumulo.tserver.hold", MetricType.GAUGE, "", MetricCategory.TABLET_SERVER),
++ TSERVER_TABLETS_ONLINE_ONDEMAND("accumulo.tserver.tablets.ondemand.online", MetricType.GAUGE, "",
++ MetricCategory.TABLET_SERVER),
++ TSERVER_TABLETS_ONDEMAND_UNLOADED_FOR_MEM("accumulo.tserver.tablets.ondemand.unloaded.lowmem",
++ MetricType.GAUGE, "", MetricCategory.TABLET_SERVER),
+
+ // Scan Metrics
+ SCAN_RESERVATION_TOTAL_TIMER("accumulo.scan.reservation.total.timer", MetricType.TIMER,
+ "Time to reserve a tablet's files for scan.", MetricCategory.SCAN_SERVER),
+ SCAN_RESERVATION_WRITEOUT_TIMER("accumulo.scan.reservation.writeout.timer", MetricType.TIMER,
+ "Time to write out a tablets file reservations for scan", MetricCategory.SCAN_SERVER),
+ SCAN_RESERVATION_CONFLICT_COUNTER("accumulo.scan.reservation.conflict.count", MetricType.COUNTER,
+ "", MetricCategory.SCAN_SERVER),
+ SCAN_BUSY_TIMEOUT_COUNT("accumulo.scan.busy.timeout.count", MetricType.COUNTER,
+ "Count of the scans where a busy timeout happened.", MetricCategory.SCAN_SERVER),
+ SCAN_TABLET_METADATA_CACHE("accumulo.scan.tablet.metadata.cache", MetricType.CACHE,
+ "Scan server tablet cache metrics.", MetricCategory.SCAN_SERVER),
+ SCAN_TIMES("accumulo.scan.times", MetricType.TIMER, "", MetricCategory.SCAN_SERVER),
+ SCAN_OPEN_FILES("accumulo.scan.files.open", MetricType.GAUGE, "", MetricCategory.SCAN_SERVER),
+ SCAN_RESULTS("accumulo.scan.result", MetricType.GAUGE, "", MetricCategory.SCAN_SERVER),
+ SCAN_YIELDS("accumulo.scan.yields", MetricType.GAUGE, "", MetricCategory.SCAN_SERVER),
+ SCAN_START("accumulo.scan.start", MetricType.COUNTER, "", MetricCategory.SCAN_SERVER),
+ SCAN_CONTINUE("accumulo.scan.continue", MetricType.COUNTER, "", MetricCategory.SCAN_SERVER),
+ SCAN_CLOSE("accumulo.scan.close", MetricType.COUNTER, "", MetricCategory.SCAN_SERVER),
+ SCAN_QUERIES("accumulo.scan.queries", MetricType.GAUGE, "", MetricCategory.SCAN_SERVER),
+ SCAN_SCANNED_ENTRIES("accumulo.scan.query.scanned.entries", MetricType.GAUGE,
+ "Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be derived.",
+ MetricCategory.SCAN_SERVER),
+ SCAN_QUERY_SCAN_RESULTS("accumulo.scan.query.results", MetricType.GAUGE,
+ "Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be derived.",
+ MetricCategory.SCAN_SERVER),
+ SCAN_QUERY_SCAN_RESULTS_BYTES("accumulo.scan.query.results.bytes", MetricType.GAUGE,
+ "Prior to 2.1.0 this metric was reported as a rate, it is now the count and the rate can be derived.",
+ MetricCategory.SCAN_SERVER),
+ SCAN_PAUSED_FOR_MEM("accumulo.scan.paused.for.memory", MetricType.COUNTER, "",
+ MetricCategory.SCAN_SERVER),
+ SCAN_RETURN_FOR_MEM("accumulo.scan.return.early.for.memory", MetricType.COUNTER, "",
+ MetricCategory.SCAN_SERVER),
+ SCAN_ZOMBIE_THREADS("accumulo.scan.zombie.threads", MetricType.GAUGE, "",
+ MetricCategory.SCAN_SERVER),
+
+ // Major Compaction Metrics
+ MAJC_QUEUED("accumulo.tserver.compactions.majc.queued", MetricType.GAUGE,
+ "The compaction service information is in a tag: `id={i|e}_{compactionServiceName}_{executor_name}`.",
+ MetricCategory.TABLET_SERVER),
+ MAJC_RUNNING("accumulo.tserver.compactions.majc.running", MetricType.GAUGE,
+ "The compaction service information is in a tag: `id={i|e}_{compactionServiceName}_{executor_name}`.",
+ MetricCategory.TABLET_SERVER),
+ MAJC_PAUSED("accumulo.tserver.compactions.majc.paused", MetricType.COUNTER, "",
+ MetricCategory.TABLET_SERVER),
+
+ // Minor Compaction Metrics
+ MINC_QUEUED("accumulo.tserver.compactions.minc.queued", MetricType.TIMER, "",
+ MetricCategory.TABLET_SERVER),
+ MINC_RUNNING("accumulo.tserver.compactions.minc.running", MetricType.TIMER, "",
+ MetricCategory.TABLET_SERVER),
+ MINC_PAUSED("accumulo.tserver.compactions.minc.paused", MetricType.COUNTER, "",
+ MetricCategory.TABLET_SERVER),
+
+ // Updates (Ingest) Metrics
+ UPDATE_ERRORS("accumulo.tserver.updates.error", MetricType.GAUGE,
+ "Type is stored in a tag (e.g., type=permission, type=unknown.tablet, type=constraint.violation).",
+ MetricCategory.TABLET_SERVER),
+ UPDATE_COMMIT("accumulo.tserver.updates.commit", MetricType.TIMER, "",
+ MetricCategory.TABLET_SERVER),
+ UPDATE_COMMIT_PREP("accumulo.tserver.updates.commit.prep", MetricType.TIMER, "",
+ MetricCategory.TABLET_SERVER),
+ UPDATE_WALOG_WRITE("accumulo.tserver.updates.walog.write", MetricType.TIMER, "",
+ MetricCategory.TABLET_SERVER),
+ UPDATE_MUTATION_ARRAY_SIZE("accumulo.tserver.updates.mutation.arrays.size",
+ MetricType.DISTRIBUTION_SUMMARY, "", MetricCategory.TABLET_SERVER),
+
+ // Thrift Metrics
+
+ THRIFT_IDLE("accumulo.thrift.idle", MetricType.DISTRIBUTION_SUMMARY, "", MetricCategory.THRIFT),
+ THRIFT_EXECUTE("accumulo.thrift.execute", MetricType.DISTRIBUTION_SUMMARY, "",
+ MetricCategory.THRIFT),
+
+ // Block Cache Metrics
+ BLOCKCACHE_INDEX_HITCOUNT("accumulo.blockcache.index.hitcount", MetricType.FUNCTION_COUNTER, "",
+ MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_INDEX_REQUESTCOUNT("accumulo.blockcache.index.requestcount",
+ MetricType.FUNCTION_COUNTER, "", MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_INDEX_EVICTIONCOUNT("accumulo.blockcache.index.evictioncount",
+ MetricType.FUNCTION_COUNTER, "", MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_DATA_HITCOUNT("accumulo.blockcache.data.hitcount", MetricType.FUNCTION_COUNTER, "",
+ MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_DATA_REQUESTCOUNT("accumulo.blockcache.data.requestcount", MetricType.FUNCTION_COUNTER,
+ "", MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_DATA_EVICTIONCOUNT("accumulo.blockcache.data.evictioncount",
+ MetricType.FUNCTION_COUNTER, "", MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_SUMMARY_HITCOUNT("accumulo.blockcache.summary.hitcount", MetricType.FUNCTION_COUNTER,
+ "", MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_SUMMARY_REQUESTCOUNT("accumulo.blockcache.summary.requestcount",
+ MetricType.FUNCTION_COUNTER, "", MetricCategory.BLOCK_CACHE),
+ BLOCKCACHE_SUMMARY_EVICTIONCOUNT("accumulo.blockcache.summary.evictioncount",
+ MetricType.FUNCTION_COUNTER, "", MetricCategory.BLOCK_CACHE),
+
+ // Manager Metrics
+ MANAGER_BALANCER_MIGRATIONS_NEEDED("accumulo.manager.balancer.migrations.needed",
+ MetricType.GAUGE,
+ "The number of migrations that need to complete before the system is balanced.",
- MetricCategory.MANAGER);
++ MetricCategory.MANAGER),
++ MANAGER_ROOT_TGW_ERRORS("accumulo.manager.tabletmgmt.root.errors", MetricType.GAUGE, "",
++ MetricCategory.MANAGER),
++ MANAGER_META_TGW_ERRORS("accumulo.manager.tabletmgmt.meta.errors", MetricType.GAUGE, "",
++ MetricCategory.MANAGER),
++ MANAGER_USER_TGW_ERRORS("accumulo.manager.tabletmgmt.user.errors", MetricType.GAUGE, "",
++ MetricCategory.MANAGER),
++ MANAGER_COMPACTION_SVC_ERRORS("accumulo.manager.compaction.svc.misconfigured", MetricType.GAUGE,
++ "", MetricCategory.MANAGER);
+
+ private final String name;
+ private final MetricType type;
+ private final String description;
+ private final MetricCategory category;
+
+ private static final Map<String,Metric> NAME_TO_ENUM_MAP = new HashMap<>();
+
+ static {
+ for (Metric metric : values()) {
+ NAME_TO_ENUM_MAP.put(metric.name, metric);
+ }
+ }
+
+ Metric(String name, MetricType type, String description, MetricCategory category) {
+ this.name = name;
+ this.type = type;
+ this.description = description;
+ this.category = category;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public MetricType getType() {
+ return type;
+ }
+
+ public String getDescription() {
+ return description;
+ }
+
+ public MetricCategory getCategory() {
+ return category;
+ }
+
+ public enum MetricType {
+ GAUGE, COUNTER, TIMER, LONG_TASK_TIMER, DISTRIBUTION_SUMMARY, FUNCTION_COUNTER, CACHE
+ }
+
+ public enum MetricCategory {
+ GENERAL_SERVER("General Server Metrics"),
+ COMPACTOR("Compactor Metrics"),
+ FATE("Fate Metrics"),
+ GARBAGE_COLLECTION("Garbage Collection Metrics"),
+ TABLET_SERVER("Tablet Server Metrics"),
+ SCAN_SERVER("Scan Server Metrics"),
+ THRIFT("Thrift Metrics"),
+ BLOCK_CACHE("Block Cache Metrics"),
+ MANAGER("Manager Metrics");
+
+ private final String sectionTitle;
+
+ MetricCategory(String sectionTitle) {
+ this.sectionTitle = sectionTitle;
+ }
+
+ public String getSectionTitle() {
+ return sectionTitle;
+ }
+ }
+
+ public static Metric fromName(String name) {
+ Metric metric = NAME_TO_ENUM_MAP.get(name);
+ if (metric == null) {
+ throw new IllegalArgumentException("No enum constant for metric name: " + name);
+ }
+ return metric;
+ }
+
+ }
diff --cc core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java
index 84c2bc55cb,48e1e759ff..7ef301ffb1
--- a/core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java
+++ b/core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java
@@@ -18,14 -18,9 +18,6 @@@
*/
package org.apache.accumulo.core.metrics;
- import java.lang.reflect.Field;
- import java.lang.reflect.Modifier;
- import java.util.HashMap;
- import java.util.Map;
-
--import org.slf4j.Logger;
--import org.slf4j.LoggerFactory;
--
import io.micrometer.core.instrument.MeterRegistry;
/**
@@@ -665,138 -34,9 +31,6 @@@
* @since 2.1.0
*/
public interface MetricsProducer {
--
-- Logger LOG = LoggerFactory.getLogger(MetricsProducer.class);
-
- String METRICS_LOW_MEMORY = "accumulo.detected.low.memory";
- String METRICS_SERVER_IDLE = "accumulo.server.idle";
-
- String METRICS_COMPACTOR_PREFIX = "accumulo.compactor.";
- String METRICS_COMPACTOR_MAJC_STUCK = METRICS_COMPACTOR_PREFIX + "majc.stuck";
- String METRICS_COMPACTOR_QUEUE_PREFIX = METRICS_COMPACTOR_PREFIX + "queue.";
- String METRICS_COMPACTOR_JOB_PRIORITY_QUEUES = METRICS_COMPACTOR_QUEUE_PREFIX + "count";
- String METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH = METRICS_COMPACTOR_QUEUE_PREFIX + "length";
- String METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED =
- METRICS_COMPACTOR_QUEUE_PREFIX + "jobs.dequeued";
- String METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED =
- METRICS_COMPACTOR_QUEUE_PREFIX + "jobs.queued";
- String METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED =
- METRICS_COMPACTOR_QUEUE_PREFIX + "jobs.rejected";
- String METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY =
- METRICS_COMPACTOR_QUEUE_PREFIX + "jobs.priority";
- String METRICS_COMPACTOR_ENTRIES_READ = METRICS_COMPACTOR_PREFIX + "entries.read";
- String METRICS_COMPACTOR_ENTRIES_WRITTEN = METRICS_COMPACTOR_PREFIX + "entries.written";
-
- String METRICS_FATE_PREFIX = "accumulo.fate.";
- String METRICS_FATE_TYPE_IN_PROGRESS = METRICS_FATE_PREFIX + "ops.in.progress.by.type";
- String METRICS_FATE_OPS = METRICS_FATE_PREFIX + "ops";
- String METRICS_FATE_OPS_ACTIVITY = METRICS_FATE_PREFIX + "ops.activity";
- String METRICS_FATE_ERRORS = METRICS_FATE_PREFIX + "errors";
- String METRICS_FATE_TX = METRICS_FATE_PREFIX + "tx";
-
- String METRICS_GC_PREFIX = "accumulo.gc.";
- String METRICS_GC_STARTED = METRICS_GC_PREFIX + "started";
- String METRICS_GC_FINISHED = METRICS_GC_PREFIX + "finished";
- String METRICS_GC_CANDIDATES = METRICS_GC_PREFIX + "candidates";
- String METRICS_GC_IN_USE = METRICS_GC_PREFIX + "in.use";
- String METRICS_GC_DELETED = METRICS_GC_PREFIX + "deleted";
- String METRICS_GC_ERRORS = METRICS_GC_PREFIX + "errors";
- String METRICS_GC_WAL_STARTED = METRICS_GC_PREFIX + "wal.started";
- String METRICS_GC_WAL_FINISHED = METRICS_GC_PREFIX + "wal.finished";
- String METRICS_GC_WAL_CANDIDATES = METRICS_GC_PREFIX + "wal.candidates";
- String METRICS_GC_WAL_IN_USE = METRICS_GC_PREFIX + "wal.in.use";
- String METRICS_GC_WAL_DELETED = METRICS_GC_PREFIX + "wal.deleted";
- String METRICS_GC_WAL_ERRORS = METRICS_GC_PREFIX + "wal.errors";
- String METRICS_GC_POST_OP_DURATION = METRICS_GC_PREFIX + "post.op.duration";
- String METRICS_GC_RUN_CYCLE = METRICS_GC_PREFIX + "run.cycle";
-
- String METRICS_MANAGER_PREFIX = "accumulo.manager.";
- String METRICS_MANAGER_ROOT_TGW_ERRORS = METRICS_MANAGER_PREFIX + "tabletmgmt.root.errors";
- String METRICS_MANAGER_META_TGW_ERRORS = METRICS_MANAGER_PREFIX + "tabletmgmt.meta.errors";
- String METRICS_MANAGER_USER_TGW_ERRORS = METRICS_MANAGER_PREFIX + "tabletmgmt.user.errors";
- String METRICS_MANAGER_COMPACTION_SVC_ERRORS =
- METRICS_MANAGER_PREFIX + "compaction.svc.misconfigured";
-
- String METRICS_MAJC_PREFIX = "accumulo.compactions.majc.";
- String METRICS_MAJC_QUEUED = METRICS_MAJC_PREFIX + "queued";
- String METRICS_MAJC_RUNNING = METRICS_MAJC_PREFIX + "running";
- String METRICS_MAJC_PAUSED = METRICS_MAJC_PREFIX + "paused";
-
- String METRICS_MINC_PREFIX = "accumulo.tserver.compactions.minc.";
- String METRICS_MINC_QUEUED = METRICS_MINC_PREFIX + "queued";
- String METRICS_MINC_RUNNING = METRICS_MINC_PREFIX + "running";
- String METRICS_MINC_PAUSED = METRICS_MINC_PREFIX + "paused";
-
- String METRICS_SCAN_PREFIX = "accumulo.scan.";
- String METRICS_SCAN_TIMES = METRICS_SCAN_PREFIX + "times";
- String METRICS_SCAN_OPEN_FILES = METRICS_SCAN_PREFIX + "files.open";
- String METRICS_SCAN_RESULTS = METRICS_SCAN_PREFIX + "result";
- String METRICS_SCAN_YIELDS = METRICS_SCAN_PREFIX + "yields";
- String METRICS_SCAN_START = METRICS_SCAN_PREFIX + "start";
- String METRICS_SCAN_CONTINUE = METRICS_SCAN_PREFIX + "continue";
- String METRICS_SCAN_CLOSE = METRICS_SCAN_PREFIX + "close";
- String METRICS_SCAN_RESERVATION_TOTAL_TIMER = METRICS_SCAN_PREFIX + "reservation.total.timer";
- String METRICS_SCAN_RESERVATION_WRITEOUT_TIMER =
- METRICS_SCAN_PREFIX + "reservation.writeout.timer";
- String METRICS_SCAN_BUSY_TIMEOUT_COUNTER = METRICS_SCAN_PREFIX + "busy.timeout.count";
- String METRICS_SCAN_RESERVATION_CONFLICT_COUNTER =
- METRICS_SCAN_PREFIX + "reservation.conflict.count";
- String METRICS_SCAN_QUERIES = METRICS_SCAN_PREFIX + "queries";
- String METRICS_SCAN_QUERY_SCAN_RESULTS = METRICS_SCAN_PREFIX + "query.results";
- String METRICS_SCAN_QUERY_SCAN_RESULTS_BYTES = METRICS_SCAN_PREFIX + "query.results.bytes";
- String METRICS_SCAN_SCANNED_ENTRIES = METRICS_SCAN_PREFIX + "query.scanned.entries";
- String METRICS_SCAN_PAUSED_FOR_MEM = METRICS_SCAN_PREFIX + "paused.for.memory";
- String METRICS_SCAN_RETURN_FOR_MEM = METRICS_SCAN_PREFIX + "return.early.for.memory";
-
- String METRICS_SCAN_ZOMBIE_THREADS = METRICS_SCAN_PREFIX + "zombie.threads";
- String METRICS_SCAN_TABLET_METADATA_CACHE = METRICS_SCAN_PREFIX + "tablet.metadata.cache";
-
- String METRICS_TSERVER_PREFIX = "accumulo.tserver.";
- String METRICS_TSERVER_ENTRIES = METRICS_TSERVER_PREFIX + "entries";
- String METRICS_TSERVER_MEM_ENTRIES = METRICS_TSERVER_PREFIX + "entries.mem";
- String METRICS_TSERVER_MINC_QUEUED = METRICS_TSERVER_PREFIX + "minc.queued";
- String METRICS_TSERVER_MINC_RUNNING = METRICS_TSERVER_PREFIX + "minc.running";
- String METRICS_TSERVER_MINC_TOTAL = METRICS_TSERVER_PREFIX + "minc.total";
- String METRICS_TSERVER_TABLETS_LONG_ASSIGNMENTS =
- METRICS_TSERVER_PREFIX + "tablets.assignments.warning";
- String METRICS_TSERVER_TABLETS_ONLINE_ONDEMAND =
- METRICS_TSERVER_PREFIX + "tablets.ondemand.online";
- String METRICS_TSERVER_TABLETS_ONDEMAND_UNLOADED_FOR_MEM =
- METRICS_TSERVER_PREFIX + "tablets.ondemand.unloaded.lowmem";
- String METRICS_TSERVER_TABLETS_ONLINE = METRICS_TSERVER_PREFIX + "tablets.online";
- String METRICS_TSERVER_TABLETS_OPENING = METRICS_TSERVER_PREFIX + "tablets.opening";
- String METRICS_TSERVER_TABLETS_UNOPENED = METRICS_TSERVER_PREFIX + "tablets.unopened";
- String METRICS_TSERVER_TABLETS_FILES = METRICS_TSERVER_PREFIX + "tablets.files";
- String METRICS_TSERVER_HOLD = METRICS_TSERVER_PREFIX + "hold";
- String METRICS_TSERVER_INGEST_MUTATIONS = METRICS_TSERVER_PREFIX + "ingest.mutations";
- String METRICS_TSERVER_INGEST_BYTES = METRICS_TSERVER_PREFIX + "ingest.bytes";
-
- String METRICS_THRIFT_PREFIX = "accumulo.thrift.";
- String METRICS_THRIFT_EXECUTE = METRICS_THRIFT_PREFIX + "execute";
- String METRICS_THRIFT_IDLE = METRICS_THRIFT_PREFIX + "idle";
-
- String METRICS_UPDATE_PREFIX = "accumulo.tserver.updates.";
- String METRICS_UPDATE_ERRORS = METRICS_UPDATE_PREFIX + "error";
- String METRICS_UPDATE_COMMIT = METRICS_UPDATE_PREFIX + "commit";
- String METRICS_UPDATE_COMMIT_PREP = METRICS_UPDATE_COMMIT + ".prep";
- String METRICS_UPDATE_WALOG_WRITE = METRICS_UPDATE_PREFIX + "walog.write";
- String METRICS_UPDATE_MUTATION_ARRAY_SIZE = METRICS_UPDATE_PREFIX + "mutation.arrays.size";
-
- String METRICS_BLOCKCACHE_PREFIX = "accumulo.blockcache.";
- String METRICS_BLOCKCACHE_INDEX_HITCOUNT = METRICS_BLOCKCACHE_PREFIX + "index.hitcount";
- String METRICS_BLOCKCACHE_INDEX_REQUESTCOUNT = METRICS_BLOCKCACHE_PREFIX + "index.requestcount";
- String METRICS_BLOCKCACHE_INDEX_EVICTIONCOUNT = METRICS_BLOCKCACHE_PREFIX + "index.evictioncount";
- String METRICS_BLOCKCACHE_DATA_HITCOUNT = METRICS_BLOCKCACHE_PREFIX + "data.hitcount";
- String METRICS_BLOCKCACHE_DATA_REQUESTCOUNT = METRICS_BLOCKCACHE_PREFIX + "data.requestcount";
- String METRICS_BLOCKCACHE_DATA_EVICTIONCOUNT = METRICS_BLOCKCACHE_PREFIX + "data.evictioncount";
- String METRICS_BLOCKCACHE_SUMMARY_HITCOUNT = METRICS_BLOCKCACHE_PREFIX + "summary.hitcount";
- String METRICS_BLOCKCACHE_SUMMARY_REQUESTCOUNT =
- METRICS_BLOCKCACHE_PREFIX + "summary.requestcount";
- String METRICS_BLOCKCACHE_SUMMARY_EVICTIONCOUNT =
- METRICS_BLOCKCACHE_PREFIX + "summary.evictioncount";
-
- String METRICS_MANAGER_BALANCER_MIGRATIONS_NEEDED = "accumulo.manager.balancer.migrations.needed";
--
/**
* Build Micrometer Meter objects and register them with the registry
*/
diff --cc server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java
index be9456f152,47fe34bf26..ba621f6b01
--- a/server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java
@@@ -18,19 -18,19 +18,21 @@@
*/
package org.apache.accumulo.server;
+ import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
import java.util.concurrent.ScheduledFuture;
- import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.classloader.ClassLoaderUtil;
import org.apache.accumulo.core.cli.ConfigOpts;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.core.trace.TraceUtil;
+ import org.apache.accumulo.core.util.Timer;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.server.mem.LowMemoryDetector;
import org.apache.accumulo.server.metrics.ProcessMetrics;
@@@ -45,13 -45,11 +47,13 @@@ public abstract class AbstractServer im
private final ServerContext context;
protected final String applicationName;
private final String hostname;
+ private final String resourceGroup;
private final ProcessMetrics processMetrics;
- protected final long idleReportingPeriodNanos;
- private volatile long idlePeriodStartNanos = 0L;
+ protected final long idleReportingPeriodMillis;
+ private volatile Timer idlePeriodTimer = null;
- protected AbstractServer(String appName, ConfigOpts opts, String[] args) {
+ protected AbstractServer(String appName, ConfigOpts opts,
+ Function<SiteConfiguration,ServerContext> serverContextFactory, String[] args) {
this.applicationName = appName;
opts.parseArgs(appName, args);
var siteConfig = opts.getSiteConfiguration();
diff --cc server/base/src/main/java/org/apache/accumulo/server/metrics/ProcessMetrics.java
index e110e4d5a8,83def19228..01e973290f
--- a/server/base/src/main/java/org/apache/accumulo/server/metrics/ProcessMetrics.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/metrics/ProcessMetrics.java
@@@ -18,6 -18,10 +18,9 @@@
*/
package org.apache.accumulo.server.metrics;
+ import static org.apache.accumulo.core.metrics.Metric.LOW_MEMORY;
+ import static org.apache.accumulo.core.metrics.Metric.SERVER_IDLE;
+
-import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.accumulo.core.metrics.MetricsProducer;
@@@ -37,8 -41,8 +40,8 @@@ public class ProcessMetrics implements
@Override
public void registerMetrics(MeterRegistry registry) {
- registry.gauge(METRICS_LOW_MEMORY, this, this::lowMemDetected);
- registry.gauge(METRICS_SERVER_IDLE, isIdle, AtomicInteger::get);
- registry.gauge(LOW_MEMORY.getName(), List.of(), this, this::lowMemDetected);
++ registry.gauge(LOW_MEMORY.getName(), this, this::lowMemDetected);
+ registry.gauge(SERVER_IDLE.getName(), isIdle, AtomicInteger::get);
}
private int lowMemDetected(ProcessMetrics processMetrics) {
diff --cc server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
index ea8a90bfe4,4658bd8a5a..26b5b812b8
--- a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
+++ b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
@@@ -19,6 -19,11 +19,9 @@@
package org.apache.accumulo.compactor;
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
-import static java.nio.charset.StandardCharsets.UTF_8;
-import static java.util.concurrent.TimeUnit.MINUTES;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_ENTRIES_READ;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_ENTRIES_WRITTEN;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_MAJC_STUCK;
import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
import java.io.IOException;
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
index 85b0492860,0000000000..1941a762df
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
@@@ -1,1239 -1,0 +1,1241 @@@
+/*
+ * 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
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.manager.compaction.coordinator;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.toList;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.COMPACTED;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.ECOMP;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.FILES;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOCATION;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.OPID;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.SELECTED;
++import static org.apache.accumulo.core.metrics.Metric.MAJC_QUEUED;
++import static org.apache.accumulo.core.metrics.Metric.MAJC_RUNNING;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.TableDeletedException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
+import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
+import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
+import org.apache.accumulo.core.clientImpl.thrift.TInfo;
+import org.apache.accumulo.core.clientImpl.thrift.TableOperation;
+import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
+import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException;
+import org.apache.accumulo.core.compaction.thrift.CompactionCoordinatorService;
+import org.apache.accumulo.core.compaction.thrift.TCompactionState;
+import org.apache.accumulo.core.compaction.thrift.TCompactionStatusUpdate;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompaction;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompactionList;
+import org.apache.accumulo.core.compaction.thrift.TNextCompactionJob;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.NamespaceId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
+import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.FateId;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.FateKey;
+import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.iteratorsImpl.system.SystemIteratorUtil;
+import org.apache.accumulo.core.logging.TabletLogger;
+import org.apache.accumulo.core.manager.state.tables.TableState;
+import org.apache.accumulo.core.metadata.CompactableFileImpl;
+import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.schema.Ample;
+import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
+import org.apache.accumulo.core.metadata.schema.Ample.RejectionHandler;
+import org.apache.accumulo.core.metadata.schema.CompactionMetadata;
+import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.metadata.schema.filters.HasExternalCompactionsFilter;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
+import org.apache.accumulo.core.spi.compaction.CompactionJob;
+import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.spi.compaction.CompactionPlanner;
+import org.apache.accumulo.core.spi.compaction.CompactionServiceId;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
+import org.apache.accumulo.core.tabletserver.thrift.InputFile;
+import org.apache.accumulo.core.tabletserver.thrift.IteratorConfig;
+import org.apache.accumulo.core.tabletserver.thrift.TCompactionKind;
+import org.apache.accumulo.core.tabletserver.thrift.TCompactionStats;
+import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob;
+import org.apache.accumulo.core.util.Retry;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.core.util.cache.Caches.CacheName;
+import org.apache.accumulo.core.util.compaction.CompactionPlannerInitParams;
+import org.apache.accumulo.core.util.compaction.CompactionServicesConfig;
+import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil;
+import org.apache.accumulo.core.util.compaction.RunningCompaction;
+import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.core.util.threads.Threads;
+import org.apache.accumulo.core.util.time.SteadyTime;
+import org.apache.accumulo.core.volume.Volume;
+import org.apache.accumulo.manager.Manager;
+import org.apache.accumulo.manager.compaction.coordinator.commit.CommitCompaction;
+import org.apache.accumulo.manager.compaction.coordinator.commit.CompactionCommitData;
+import org.apache.accumulo.manager.compaction.coordinator.commit.RenameCompactionFile;
+import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue;
+import org.apache.accumulo.manager.compaction.queue.CompactionJobQueues;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.ServiceEnvironmentImpl;
+import org.apache.accumulo.server.compaction.CompactionConfigStorage;
+import org.apache.accumulo.server.compaction.CompactionPluginUtils;
+import org.apache.accumulo.server.security.SecurityOperation;
+import org.apache.accumulo.server.tablets.TabletNameGenerator;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.thrift.TException;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Weigher;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
+import com.google.common.net.HostAndPort;
+
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+
+public class CompactionCoordinator
+ implements CompactionCoordinatorService.Iface, Runnable, MetricsProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactionCoordinator.class);
+
+ public static final String RESTART_UPDATE_MSG =
+ "Coordinator restarted, compaction found in progress";
+
+ /*
+ * Map of compactionId to RunningCompactions. This is an informational cache of what external
+ * compactions may be running. Its possible it may contain external compactions that are not
+ * actually running. It may not contain compactions that are actually running. The metadata table
+ * is the most authoritative source of what external compactions are currently running, but it
+ * does not have the stats that this map has.
+ */
+ protected final Map<ExternalCompactionId,RunningCompaction> RUNNING_CACHE =
+ new ConcurrentHashMap<>();
+
+ /* Map of group name to last time compactor called to get a compaction job */
+ private final Map<CompactorGroupId,Long> TIME_COMPACTOR_LAST_CHECKED = new ConcurrentHashMap<>();
+
+ private final ServerContext ctx;
+ private final SecurityOperation security;
+ private final CompactionJobQueues jobQueues;
+ private final AtomicReference<Map<FateInstanceType,Fate<Manager>>> fateInstances;
+ // Exposed for tests
+ protected final CountDownLatch shutdown = new CountDownLatch(1);
+
+ private final ScheduledThreadPoolExecutor schedExecutor;
+
+ private final Cache<ExternalCompactionId,RunningCompaction> completed;
+ private final LoadingCache<FateId,CompactionConfig> compactionConfigCache;
+ private final Cache<Path,Integer> tabletDirCache;
+ private final DeadCompactionDetector deadCompactionDetector;
+
+ private final QueueMetrics queueMetrics;
+ private final Manager manager;
+
+ private final LoadingCache<String,Integer> compactorCounts;
+ private final int jobQueueInitialSize;
+
+ private volatile long coordinatorStartTime;
+
+ public CompactionCoordinator(ServerContext ctx, SecurityOperation security,
+ AtomicReference<Map<FateInstanceType,Fate<Manager>>> fateInstances, Manager manager) {
+ this.ctx = ctx;
+ this.schedExecutor = this.ctx.getScheduledExecutor();
+ this.security = security;
+ this.manager = Objects.requireNonNull(manager);
+
+ this.jobQueueInitialSize = ctx.getConfiguration()
+ .getCount(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_INITIAL_SIZE);
+
+ this.jobQueues = new CompactionJobQueues(jobQueueInitialSize);
+
+ this.queueMetrics = new QueueMetrics(jobQueues);
+
+ this.fateInstances = fateInstances;
+
+ completed = ctx.getCaches().createNewBuilder(CacheName.COMPACTIONS_COMPLETED, true)
+ .maximumSize(200).expireAfterWrite(10, TimeUnit.MINUTES).build();
+
+ CacheLoader<FateId,CompactionConfig> loader =
+ fateId -> CompactionConfigStorage.getConfig(ctx, fateId);
+
+ // Keep a small short lived cache of compaction config. Compaction config never changes, however
+ // when a compaction is canceled it is deleted which is why there is a time limit. It does not
+ // hurt to let a job that was canceled start, it will be canceled later. Caching this immutable
+ // config will help avoid reading the same data over and over.
+ compactionConfigCache = ctx.getCaches().createNewBuilder(CacheName.COMPACTION_CONFIGS, true)
+ .expireAfterWrite(30, SECONDS).maximumSize(100).build(loader);
+
+ Weigher<Path,Integer> weigher = (path, count) -> {
+ return path.toUri().toString().length();
+ };
+
+ tabletDirCache = ctx.getCaches().createNewBuilder(CacheName.COMPACTION_DIR_CACHE, true)
+ .maximumWeight(10485760L).weigher(weigher).build();
+
+ deadCompactionDetector =
+ new DeadCompactionDetector(this.ctx, this, schedExecutor, fateInstances);
+
+ compactorCounts = ctx.getCaches().createNewBuilder(CacheName.COMPACTOR_COUNTS, false)
+ .expireAfterWrite(30, TimeUnit.SECONDS).build(this::countCompactors);
+ // At this point the manager does not have its lock so no actions should be taken yet
+ }
+
+ protected int countCompactors(String groupName) {
+ return ExternalCompactionUtil.countCompactors(groupName, ctx);
+ }
+
+ private volatile Thread serviceThread = null;
+
+ public void start() {
+ serviceThread = Threads.createThread("CompactionCoordinator Thread", this);
+ serviceThread.start();
+ }
+
+ public void shutdown() {
+ shutdown.countDown();
+ var localThread = serviceThread;
+ if (localThread != null) {
+ try {
+ localThread.join();
+ } catch (InterruptedException e) {
+ LOG.error("Exception stopping compaction coordinator thread", e);
+ }
+ }
+ }
+
+ protected void startCompactorZKCleaner(ScheduledThreadPoolExecutor schedExecutor) {
+ ScheduledFuture<?> future = schedExecutor
+ .scheduleWithFixedDelay(this::cleanUpEmptyCompactorPathInZK, 0, 5, TimeUnit.MINUTES);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ }
+
+ protected void startInternalStateCleaner(ScheduledThreadPoolExecutor schedExecutor) {
+ ScheduledFuture<?> future =
+ schedExecutor.scheduleWithFixedDelay(this::cleanUpInternalState, 0, 5, TimeUnit.MINUTES);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ }
+
+ @Override
+ public void run() {
+
+ this.coordinatorStartTime = System.currentTimeMillis();
+ startCompactorZKCleaner(schedExecutor);
+
+ // On a re-start of the coordinator it's possible that external compactions are in-progress.
+ // Attempt to get the running compactions on the compactors and then resolve which tserver
+ // the external compaction came from to re-populate the RUNNING collection.
+ LOG.info("Checking for running external compactions");
+ // On re-start contact the running Compactors to try and seed the list of running compactions
+ List<RunningCompaction> running = getCompactionsRunningOnCompactors();
+ if (running.isEmpty()) {
+ LOG.info("No running external compactions found");
+ } else {
+ LOG.info("Found {} running external compactions", running.size());
+ running.forEach(rc -> {
+ TCompactionStatusUpdate update = new TCompactionStatusUpdate();
+ update.setState(TCompactionState.IN_PROGRESS);
+ update.setMessage(RESTART_UPDATE_MSG);
+ rc.addUpdate(System.currentTimeMillis(), update);
+ RUNNING_CACHE.put(ExternalCompactionId.of(rc.getJob().getExternalCompactionId()), rc);
+ });
+ }
+
+ startDeadCompactionDetector();
+ startInternalStateCleaner(schedExecutor);
+
+ try {
+ shutdown.await();
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted waiting for shutdown latch.", e);
+ }
+
+ LOG.info("Shutting down");
+ }
+
+ private Map<String,Set<HostAndPort>>
+ getIdleCompactors(Map<String,Set<HostAndPort>> runningCompactors) {
+
+ final Map<String,Set<HostAndPort>> allCompactors = new HashMap<>();
+ runningCompactors
+ .forEach((group, compactorList) -> allCompactors.put(group, new HashSet<>(compactorList)));
+
+ final Set<String> emptyQueues = new HashSet<>();
+
+ // Remove all of the compactors that are running a compaction
+ RUNNING_CACHE.values().forEach(rc -> {
+ Set<HostAndPort> busyCompactors = allCompactors.get(rc.getGroupName());
+ if (busyCompactors != null
+ && busyCompactors.remove(HostAndPort.fromString(rc.getCompactorAddress()))) {
+ if (busyCompactors.isEmpty()) {
+ emptyQueues.add(rc.getGroupName());
+ }
+ }
+ });
+ // Remove entries with empty queues
+ emptyQueues.forEach(e -> allCompactors.remove(e));
+ return allCompactors;
+ }
+
+ protected void startDeadCompactionDetector() {
+ deadCompactionDetector.start();
+ }
+
+ protected long getMissingCompactorWarningTime() {
+ return this.ctx.getConfiguration().getTimeInMillis(Property.COMPACTOR_MAX_JOB_WAIT_TIME) * 3;
+ }
+
+ protected long getTServerCheckInterval() {
+ return this.ctx.getConfiguration()
+ .getTimeInMillis(Property.COMPACTION_COORDINATOR_TSERVER_COMPACTION_CHECK_INTERVAL);
+ }
+
+ public long getNumRunningCompactions() {
+ return RUNNING_CACHE.size();
+ }
+
+ /**
+ * Return the next compaction job from the queue to a Compactor
+ *
+ * @param groupName group
+ * @param compactorAddress compactor address
+ * @throws ThriftSecurityException when permission error
+ * @return compaction job
+ */
+ @Override
+ public TNextCompactionJob getCompactionJob(TInfo tinfo, TCredentials credentials,
+ String groupName, String compactorAddress, String externalCompactionId)
+ throws ThriftSecurityException {
+
+ // do not expect users to call this directly, expect compactors to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ CompactorGroupId groupId = CompactorGroupId.of(groupName);
+ LOG.trace("getCompactionJob called for group {} by compactor {}", groupId, compactorAddress);
+ TIME_COMPACTOR_LAST_CHECKED.put(groupId, System.currentTimeMillis());
+
+ TExternalCompactionJob result = null;
+
+ CompactionJobQueues.MetaJob metaJob = jobQueues.poll(groupId);
+
+ while (metaJob != null) {
+
+ Optional<CompactionConfig> compactionConfig = getCompactionConfig(metaJob);
+
+ // this method may reread the metadata, do not use the metadata in metaJob for anything after
+ // this method
+ CompactionMetadata ecm = null;
+
+ var kind = metaJob.getJob().getKind();
+
+ // Only reserve user compactions when the config is present. When compactions are canceled the
+ // config is deleted.
+ var cid = ExternalCompactionId.from(externalCompactionId);
+ if (kind == CompactionKind.SYSTEM
+ || (kind == CompactionKind.USER && compactionConfig.isPresent())) {
+ ecm = reserveCompaction(metaJob, compactorAddress, cid);
+ }
+
+ if (ecm != null) {
+ result = createThriftJob(externalCompactionId, ecm, metaJob, compactionConfig);
+ // It is possible that by the time this added that the the compactor that made this request
+ // is dead. In this cases the compaction is not actually running.
+ RUNNING_CACHE.put(ExternalCompactionId.of(result.getExternalCompactionId()),
+ new RunningCompaction(result, compactorAddress, groupName));
+ TabletLogger.compacting(metaJob.getTabletMetadata(), cid, compactorAddress,
+ metaJob.getJob());
+ break;
+ } else {
+ LOG.debug(
+ "Unable to reserve compaction job for {}, pulling another off the queue for group {}",
+ metaJob.getTabletMetadata().getExtent(), groupName);
+ metaJob = jobQueues.poll(CompactorGroupId.of(groupName));
+ }
+ }
+
+ if (metaJob == null) {
+ LOG.debug("No jobs found in group {} ", groupName);
+ }
+
+ if (result == null) {
+ LOG.trace("No jobs found for group {}, returning empty job to compactor {}", groupName,
+ compactorAddress);
+ result = new TExternalCompactionJob();
+ }
+
+ return new TNextCompactionJob(result, compactorCounts.get(groupName));
+ }
+
+ @VisibleForTesting
+ public static boolean canReserveCompaction(TabletMetadata tablet, CompactionKind kind,
+ Set<StoredTabletFile> jobFiles, ServerContext ctx, SteadyTime steadyTime) {
+
+ if (tablet == null) {
+ // the tablet no longer exist
+ return false;
+ }
+
+ if (tablet.getOperationId() != null) {
+ return false;
+ }
+
+ if (ctx.getTableState(tablet.getTableId()) != TableState.ONLINE) {
+ return false;
+ }
+
+ if (!tablet.getFiles().containsAll(jobFiles)) {
+ return false;
+ }
+
+ var currentlyCompactingFiles = tablet.getExternalCompactions().values().stream()
+ .flatMap(ecm -> ecm.getJobFiles().stream()).collect(Collectors.toSet());
+
+ if (!Collections.disjoint(jobFiles, currentlyCompactingFiles)) {
+ return false;
+ }
+
+ switch (kind) {
+ case SYSTEM:
+ var userRequestedCompactions = tablet.getUserCompactionsRequested().size();
+ if (userRequestedCompactions > 0) {
+ LOG.debug(
+ "Unable to reserve {} for system compaction, tablet has {} pending requested user compactions",
+ tablet.getExtent(), userRequestedCompactions);
+ return false;
+ } else if (!Collections.disjoint(jobFiles,
+ getFilesReservedBySelection(tablet, steadyTime, ctx))) {
+ return false;
+ }
+ break;
+ case USER:
+ if (tablet.getSelectedFiles() == null
+ || !tablet.getSelectedFiles().getFiles().containsAll(jobFiles)) {
+ return false;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException("Not currently handling " + kind);
+ }
+
+ return true;
+ }
+
+ private void checkTabletDir(KeyExtent extent, Path path) {
+ try {
+ if (tabletDirCache.getIfPresent(path) == null) {
+ FileStatus[] files = null;
+ try {
+ files = ctx.getVolumeManager().listStatus(path);
+ } catch (FileNotFoundException ex) {
+ // ignored
+ }
+
+ if (files == null) {
+ LOG.debug("Tablet {} had no dir, creating {}", extent, path);
+
+ ctx.getVolumeManager().mkdirs(path);
+ }
+ tabletDirCache.put(path, 1);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ protected CompactionMetadata createExternalCompactionMetadata(CompactionJob job,
+ Set<StoredTabletFile> jobFiles, TabletMetadata tablet, String compactorAddress,
+ ExternalCompactionId externalCompactionId) {
+ boolean propDels;
+
+ FateId fateId = null;
+
+ switch (job.getKind()) {
+ case SYSTEM: {
+ boolean compactingAll = tablet.getFiles().equals(jobFiles);
+ propDels = !compactingAll;
+ }
+ break;
+ case USER: {
+ boolean compactingAll = tablet.getSelectedFiles().initiallySelectedAll()
+ && tablet.getSelectedFiles().getFiles().equals(jobFiles);
+ propDels = !compactingAll;
+ fateId = tablet.getSelectedFiles().getFateId();
+ }
+ break;
+ default:
+ throw new IllegalArgumentException();
+ }
+
+ Consumer<String> directoryCreator = dir -> checkTabletDir(tablet.getExtent(), new Path(dir));
+ ReferencedTabletFile newFile = TabletNameGenerator.getNextDataFilenameForMajc(propDels, ctx,
+ tablet, directoryCreator, externalCompactionId);
+
+ return new CompactionMetadata(jobFiles, newFile, compactorAddress, job.getKind(),
+ job.getPriority(), job.getGroup(), propDels, fateId);
+
+ }
+
+ protected CompactionMetadata reserveCompaction(CompactionJobQueues.MetaJob metaJob,
+ String compactorAddress, ExternalCompactionId externalCompactionId) {
+
+ Preconditions.checkArgument(metaJob.getJob().getKind() == CompactionKind.SYSTEM
+ || metaJob.getJob().getKind() == CompactionKind.USER);
+
+ var tabletMetadata = metaJob.getTabletMetadata();
+
+ var jobFiles = metaJob.getJob().getFiles().stream().map(CompactableFileImpl::toStoredTabletFile)
+ .collect(Collectors.toSet());
+
+ Retry retry = Retry.builder().maxRetries(5).retryAfter(Duration.ofMillis(100))
+ .incrementBy(Duration.ofMillis(100)).maxWait(Duration.ofSeconds(10)).backOffFactor(1.5)
+ .logInterval(Duration.ofMinutes(3)).createRetry();
+
+ while (retry.canRetry()) {
+ try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) {
+ var extent = metaJob.getTabletMetadata().getExtent();
+
+ if (!canReserveCompaction(tabletMetadata, metaJob.getJob().getKind(), jobFiles, ctx,
+ manager.getSteadyTime())) {
+ return null;
+ }
+
+ var ecm = createExternalCompactionMetadata(metaJob.getJob(), jobFiles, tabletMetadata,
+ compactorAddress, externalCompactionId);
+
+ // any data that is read from the tablet to make a decision about if it can compact or not
+ // must be included in the requireSame call
+ var tabletMutator = tabletsMutator.mutateTablet(extent).requireAbsentOperation()
+ .requireSame(tabletMetadata, FILES, SELECTED, ECOMP);
+
+ if (metaJob.getJob().getKind() == CompactionKind.SYSTEM) {
+ var selectedFiles = tabletMetadata.getSelectedFiles();
+ var reserved = getFilesReservedBySelection(tabletMetadata, manager.getSteadyTime(), ctx);
+
+ // If there is a selectedFiles column, and the reserved set is empty this means that
+ // either no user jobs were completed yet or the selection expiration time has passed
+ // so the column is eligible to be deleted so a system job can run instead
+ if (selectedFiles != null && reserved.isEmpty()
+ && !Collections.disjoint(jobFiles, selectedFiles.getFiles())) {
+ LOG.debug("Deleting user compaction selected files for {} {}", extent,
+ externalCompactionId);
+ tabletMutator.deleteSelectedFiles();
+ }
+ }
+
+ tabletMutator.putExternalCompaction(externalCompactionId, ecm);
+ tabletMutator.submit(tm -> tm.getExternalCompactions().containsKey(externalCompactionId));
+
+ var result = tabletsMutator.process().get(extent);
+
+ if (result.getStatus() == Ample.ConditionalResult.Status.ACCEPTED) {
+ return ecm;
+ } else {
+ tabletMetadata = result.readMetadata();
+ }
+ }
+
+ retry.useRetry();
+ try {
+ retry.waitForNextAttempt(LOG,
+ "Reserved compaction for " + metaJob.getTabletMetadata().getExtent());
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return null;
+ }
+
+ protected TExternalCompactionJob createThriftJob(String externalCompactionId,
+ CompactionMetadata ecm, CompactionJobQueues.MetaJob metaJob,
+ Optional<CompactionConfig> compactionConfig) {
+
+ Set<CompactableFile> selectedFiles;
+ if (metaJob.getJob().getKind() == CompactionKind.SYSTEM) {
+ selectedFiles = Set.of();
+ } else {
+ selectedFiles = metaJob.getTabletMetadata().getSelectedFiles().getFiles().stream()
+ .map(file -> new CompactableFileImpl(file,
+ metaJob.getTabletMetadata().getFilesMap().get(file)))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
+ Map<String,String> overrides = CompactionPluginUtils.computeOverrides(compactionConfig, ctx,
+ metaJob.getTabletMetadata().getExtent(), metaJob.getJob().getFiles(), selectedFiles);
+
+ IteratorConfig iteratorSettings = SystemIteratorUtil
+ .toIteratorConfig(compactionConfig.map(CompactionConfig::getIterators).orElse(List.of()));
+
+ var files = ecm.getJobFiles().stream().map(storedTabletFile -> {
+ var dfv = metaJob.getTabletMetadata().getFilesMap().get(storedTabletFile);
+ return new InputFile(storedTabletFile.getMetadata(), dfv.getSize(), dfv.getNumEntries(),
+ dfv.getTime());
+ }).collect(toList());
+
+ // The fateId here corresponds to the Fate transaction that is driving a user initiated
+ // compaction. A system initiated compaction has no Fate transaction driving it so its ok to set
+ // it to null. If anything tries to use the id for a system compaction and triggers a NPE it's
+ // probably a bug that needs to be fixed.
+ FateId fateId = null;
+ if (metaJob.getJob().getKind() == CompactionKind.USER) {
+ fateId = metaJob.getTabletMetadata().getSelectedFiles().getFateId();
+ }
+
+ return new TExternalCompactionJob(externalCompactionId,
+ metaJob.getTabletMetadata().getExtent().toThrift(), files, iteratorSettings,
+ ecm.getCompactTmpName().getNormalizedPathStr(), ecm.getPropagateDeletes(),
+ TCompactionKind.valueOf(ecm.getKind().name()), fateId == null ? null : fateId.toThrift(),
+ overrides);
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
- Gauge.builder(METRICS_MAJC_QUEUED, jobQueues, CompactionJobQueues::getQueuedJobCount)
++ Gauge.builder(MAJC_QUEUED.getName(), jobQueues, CompactionJobQueues::getQueuedJobCount)
+ .tag("subprocess", "compaction.coordinator")
+ .description("Number of queued major compactions").register(registry);
- Gauge.builder(METRICS_MAJC_RUNNING, this, CompactionCoordinator::getNumRunningCompactions)
++ Gauge.builder(MAJC_RUNNING.getName(), this, CompactionCoordinator::getNumRunningCompactions)
+ .tag("subprocess", "compaction.coordinator")
+ .description("Number of running major compactions").register(registry);
+
+ queueMetrics.registerMetrics(registry);
+ }
+
+ public void addJobs(TabletMetadata tabletMetadata, Collection<CompactionJob> jobs) {
+ jobQueues.add(tabletMetadata, jobs);
+ }
+
+ public CompactionCoordinatorService.Iface getThriftService() {
+ return this;
+ }
+
+ private Optional<CompactionConfig> getCompactionConfig(CompactionJobQueues.MetaJob metaJob) {
+ if (metaJob.getJob().getKind() == CompactionKind.USER
+ && metaJob.getTabletMetadata().getSelectedFiles() != null) {
+ var cconf =
+ compactionConfigCache.get(metaJob.getTabletMetadata().getSelectedFiles().getFateId());
+ return Optional.ofNullable(cconf);
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Compactors calls this method when they have finished a compaction. This method does the
+ * following.
+ *
+ * <ol>
+ * <li>Reads the tablets metadata and determines if the compaction can commit. Its possible that
+ * things changed while the compaction was running and it can no longer commit.</li>
+ * <li>Commit the compaction using a conditional mutation. If the tablets files or location
+ * changed since reading the tablets metadata, then conditional mutation will fail. When this
+ * happens it will reread the metadata and go back to step 1 conceptually. When committing a
+ * compaction the compacted files are removed and scan entries are added to the tablet in case the
+ * files are in use, this prevents GC from deleting the files between updating tablet metadata and
+ * refreshing the tablet. The scan entries are only added when a tablet has a location.</li>
+ * <li>After successful commit a refresh request is sent to the tablet if it has a location. This
+ * will cause the tablet to start using the newly compacted files for future scans. Also the
+ * tablet can delete the scan entries if there are no active scans using them.</li>
+ * </ol>
+ *
+ * <p>
+ * User compactions will be refreshed as part of the fate operation. The user compaction fate
+ * operation will see the compaction was committed after this code updates the tablet metadata,
+ * however if it were to rely on this code to do the refresh it would not be able to know when the
+ * refresh was actually done. Therefore, user compactions will refresh as part of the fate
+ * operation so that it's known to be done before the fate operation returns. Since the fate
+ * operation will do it, there is no need to do it here for user compactions.
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @param externalCompactionId compaction id
+ * @param textent tablet extent
+ * @param stats compaction stats
+ * @throws ThriftSecurityException when permission error
+ */
+ @Override
+ public void compactionCompleted(TInfo tinfo, TCredentials credentials,
+ String externalCompactionId, TKeyExtent textent, TCompactionStats stats)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+
+ // maybe fate has not started yet
+ var localFates = fateInstances.get();
+ while (localFates == null) {
+ UtilWaitThread.sleep(100);
+ if (shutdown.getCount() == 0) {
+ return;
+ }
+ localFates = fateInstances.get();
+ }
+
+ var extent = KeyExtent.fromThrift(textent);
+ var localFate = localFates.get(FateInstanceType.fromTableId(extent.tableId()));
+
+ LOG.info("Compaction completed, id: {}, stats: {}, extent: {}", externalCompactionId, stats,
+ extent);
+ final var ecid = ExternalCompactionId.of(externalCompactionId);
+
+ var tabletMeta =
+ ctx.getAmple().readTablet(extent, ECOMP, SELECTED, LOCATION, FILES, COMPACTED, OPID);
+
+ var tableState = manager.getContext().getTableState(extent.tableId());
+ if (tableState != TableState.ONLINE) {
+ // Its important this check is done after the compaction id is set in the metadata table to
+ // avoid race conditions with the client code that waits for tables to go offline. That code
+ // looks for compaction ids in the metadata table after setting the table state. When that
+ // client code sees nothing for a tablet its important that nothing will changes the tablets
+ // files after that point in time which this check ensure.
+ LOG.debug("Not committing compaction {} for {} because of table state {}", ecid, extent,
+ tableState);
+ // cleanup metadata table and files related to the compaction
+ compactionsFailed(Map.of(ecid, extent));
+ return;
+ }
+
+ if (!CommitCompaction.canCommitCompaction(ecid, tabletMeta)) {
+ return;
+ }
+
+ // Start a fate transaction to commit the compaction.
+ CompactionMetadata ecm = tabletMeta.getExternalCompactions().get(ecid);
+ var renameOp = new RenameCompactionFile(new CompactionCommitData(ecid, extent, ecm, stats));
+ var txid = localFate.seedTransaction("COMMIT_COMPACTION", FateKey.forCompactionCommit(ecid),
+ renameOp, true, "Commit compaction " + ecid);
+
+ txid.ifPresentOrElse(fateId -> LOG.debug("initiated compaction commit {} {}", ecid, fateId),
+ () -> LOG.debug("compaction commit already initiated for {}", ecid));
+ }
+
+ @Override
+ public void compactionFailed(TInfo tinfo, TCredentials credentials, String externalCompactionId,
+ TKeyExtent extent) throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ KeyExtent fromThriftExtent = KeyExtent.fromThrift(extent);
+ LOG.info("Compaction failed, id: {}, extent: {}", externalCompactionId, fromThriftExtent);
+ final var ecid = ExternalCompactionId.of(externalCompactionId);
+ compactionsFailed(Map.of(ecid, KeyExtent.fromThrift(extent)));
+ }
+
+ void compactionsFailed(Map<ExternalCompactionId,KeyExtent> compactions) {
+ // Need to process each level by itself because the conditional tablet mutator does not support
+ // mutating multiple data levels at the same time
+ compactions.entrySet().stream()
+ .collect(groupingBy(entry -> DataLevel.of(entry.getValue().tableId()),
+ Collectors.toMap(Entry::getKey, Entry::getValue)))
+ .forEach((level, compactionsByLevel) -> compactionFailedForLevel(compactionsByLevel));
+ }
+
+ void compactionFailedForLevel(Map<ExternalCompactionId,KeyExtent> compactions) {
+
+ try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) {
+ compactions.forEach((ecid, extent) -> {
+ try {
+ ctx.requireNotDeleted(extent.tableId());
+ tabletsMutator.mutateTablet(extent).requireAbsentOperation().requireCompaction(ecid)
+ .deleteExternalCompaction(ecid).submit(new RejectionHandler() {
+
+ @Override
+ public boolean callWhenTabletDoesNotExists() {
+ return true;
+ }
+
+ @Override
+ public boolean test(TabletMetadata tabletMetadata) {
+ return tabletMetadata == null
+ || !tabletMetadata.getExternalCompactions().containsKey(ecid);
+ }
+
+ });
+ } catch (TableDeletedException e) {
+ LOG.warn("Table {} was deleted, unable to update metadata for compaction failure.",
+ extent.tableId());
+ }
+ });
+
+ final List<ExternalCompactionId> ecidsForTablet = new ArrayList<>();
+ tabletsMutator.process().forEach((extent, result) -> {
+ if (result.getStatus() != Ample.ConditionalResult.Status.ACCEPTED) {
+
+ // this should try again later when the dead compaction detector runs, lets log it in case
+ // its a persistent problem
+ if (LOG.isDebugEnabled()) {
+ var ecid =
+ compactions.entrySet().stream().filter(entry -> entry.getValue().equals(extent))
+ .findFirst().map(Map.Entry::getKey).orElse(null);
+ LOG.debug("Unable to remove failed compaction {} {}", extent, ecid);
+ }
+ } else {
+ // compactionFailed is called from the Compactor when either a compaction fails or
+ // is cancelled and it's called from the DeadCompactionDetector. This block is
+ // entered when the conditional mutator above successfully deletes an ecid from
+ // the tablet metadata. Remove compaction tmp files from the tablet directory
+ // that have a corresponding ecid in the name.
+
+ ecidsForTablet.clear();
+ compactions.entrySet().stream().filter(e -> e.getValue().compareTo(extent) == 0)
+ .map(Entry::getKey).forEach(ecidsForTablet::add);
+
+ if (!ecidsForTablet.isEmpty()) {
+ final TabletMetadata tm = ctx.getAmple().readTablet(extent, ColumnType.DIR);
+ if (tm != null) {
+ final Collection<Volume> vols = ctx.getVolumeManager().getVolumes();
+ for (Volume vol : vols) {
+ try {
+ final String volPath =
+ vol.getBasePath() + Constants.HDFS_TABLES_DIR + Path.SEPARATOR
+ + extent.tableId().canonical() + Path.SEPARATOR + tm.getDirName();
+ final FileSystem fs = vol.getFileSystem();
+ for (ExternalCompactionId ecid : ecidsForTablet) {
+ final String fileSuffix = "_tmp_" + ecid.canonical();
+ FileStatus[] files = fs.listStatus(new Path(volPath), (path) -> {
+ return path.getName().endsWith(fileSuffix);
+ });
+ if (files.length > 0) {
+ for (FileStatus file : files) {
+ if (!fs.delete(file.getPath(), false)) {
+ LOG.warn("Unable to delete ecid tmp file: {}: ", file.getPath());
+ } else {
+ LOG.debug("Deleted ecid tmp file: {}", file.getPath());
+ }
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOG.error("Exception deleting compaction tmp files for tablet: {}", extent, e);
+ }
+ }
+ } else {
+ // TabletMetadata does not exist for the extent. This could be due to a merge or
+ // split operation. Use the utility to find tmp files at the table level
+ deadCompactionDetector.addTableId(extent.tableId());
+ }
+ }
+ }
+ });
+ }
+
+ compactions.forEach((k, v) -> recordCompletion(k));
+ }
+
+ /**
+ * Compactor calls to update the status of the assigned compaction
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @param externalCompactionId compaction id
+ * @param update compaction status update
+ * @param timestamp timestamp of the message
+ * @throws ThriftSecurityException when permission error
+ */
+ @Override
+ public void updateCompactionStatus(TInfo tinfo, TCredentials credentials,
+ String externalCompactionId, TCompactionStatusUpdate update, long timestamp)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ LOG.debug("Compaction status update, id: {}, timestamp: {}, update: {}", externalCompactionId,
+ timestamp, update);
+ final RunningCompaction rc = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId));
+ if (null != rc) {
+ rc.addUpdate(timestamp, update);
+ }
+ }
+
+ public void recordCompletion(ExternalCompactionId ecid) {
+ var rc = RUNNING_CACHE.remove(ecid);
+ if (rc != null) {
+ completed.put(ecid, rc);
+ }
+ }
+
+ protected Set<ExternalCompactionId> readExternalCompactionIds() {
+ try (TabletsMetadata tabletsMetadata =
+ this.ctx.getAmple().readTablets().forLevel(Ample.DataLevel.USER)
+ .filter(new HasExternalCompactionsFilter()).fetch(ECOMP).build()) {
+ return tabletsMetadata.stream().flatMap(tm -> tm.getExternalCompactions().keySet().stream())
+ .collect(Collectors.toSet());
+ }
+ }
+
+ /**
+ * Return information about running compactions
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @return map of ECID to TExternalCompaction objects
+ * @throws ThriftSecurityException permission error
+ */
+ @Override
+ public TExternalCompactionList getRunningCompactions(TInfo tinfo, TCredentials credentials)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+
+ final TExternalCompactionList result = new TExternalCompactionList();
+ RUNNING_CACHE.forEach((ecid, rc) -> {
+ TExternalCompaction trc = new TExternalCompaction();
+ trc.setGroupName(rc.getGroupName());
+ trc.setCompactor(rc.getCompactorAddress());
+ trc.setUpdates(rc.getUpdates());
+ trc.setJob(rc.getJob());
+ result.putToCompactions(ecid.canonical(), trc);
+ });
+ return result;
+ }
+
+ /**
+ * Return information about recently completed compactions
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @return map of ECID to TExternalCompaction objects
+ * @throws ThriftSecurityException permission error
+ */
+ @Override
+ public TExternalCompactionList getCompletedCompactions(TInfo tinfo, TCredentials credentials)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ final TExternalCompactionList result = new TExternalCompactionList();
+ completed.asMap().forEach((ecid, rc) -> {
+ TExternalCompaction trc = new TExternalCompaction();
+ trc.setGroupName(rc.getGroupName());
+ trc.setCompactor(rc.getCompactorAddress());
+ trc.setJob(rc.getJob());
+ trc.setUpdates(rc.getUpdates());
+ result.putToCompactions(ecid.canonical(), trc);
+ });
+ return result;
+ }
+
+ @Override
+ public void cancel(TInfo tinfo, TCredentials credentials, String externalCompactionId)
+ throws TException {
+ var runningCompaction = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId));
+ var extent = KeyExtent.fromThrift(runningCompaction.getJob().getExtent());
+ try {
+ NamespaceId nsId = this.ctx.getNamespaceId(extent.tableId());
+ if (!security.canCompact(credentials, extent.tableId(), nsId)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ } catch (TableNotFoundException e) {
+ throw new ThriftTableOperationException(extent.tableId().canonical(), null,
+ TableOperation.COMPACT_CANCEL, TableOperationExceptionType.NOTFOUND, e.getMessage());
+ }
+
+ cancelCompactionOnCompactor(runningCompaction.getCompactorAddress(), externalCompactionId);
+ }
+
+ /* Method exists to be called from test */
+ public CompactionJobQueues getJobQueues() {
+ return jobQueues;
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected List<RunningCompaction> getCompactionsRunningOnCompactors() {
+ return ExternalCompactionUtil.getCompactionsRunningOnCompactors(this.ctx);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected Map<String,Set<HostAndPort>> getRunningCompactors() {
+ return ExternalCompactionUtil.getCompactorAddrs(this.ctx);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected void cancelCompactionOnCompactor(String address, String externalCompactionId) {
+ HostAndPort hostPort = HostAndPort.fromString(address);
+ ExternalCompactionUtil.cancelCompaction(this.ctx, hostPort, externalCompactionId);
+ }
+
+ private void deleteEmpty(ZooReaderWriter zoorw, String path)
+ throws KeeperException, InterruptedException {
+ try {
+ LOG.debug("Deleting empty ZK node {}", path);
+ zoorw.delete(path);
+ } catch (KeeperException.NotEmptyException e) {
+ LOG.debug("Failed to delete {} its not empty, likely an expected race condition.", path);
+ }
+ }
+
+ private void cleanUpEmptyCompactorPathInZK() {
+ final String compactorQueuesPath = this.ctx.getZooKeeperRoot() + Constants.ZCOMPACTORS;
+
+ final var zoorw = this.ctx.getZooReaderWriter();
+ final double queueSizeFactor = ctx.getConfiguration()
+ .getFraction(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE_FACTOR);
+
+ try {
+ var groups = zoorw.getChildren(compactorQueuesPath);
+
+ for (String group : groups) {
+ final String qpath = compactorQueuesPath + "/" + group;
+ final CompactorGroupId cgid = CompactorGroupId.of(group);
+ final var compactors = zoorw.getChildren(qpath);
+
+ if (compactors.isEmpty()) {
+ deleteEmpty(zoorw, qpath);
+ // Group has no compactors, we can clear its
+ // associated priority queue of jobs
+ CompactionJobPriorityQueue queue = getJobQueues().getQueue(cgid);
+ if (queue != null) {
+ queue.clear();
+ queue.setMaxSize(this.jobQueueInitialSize);
+ }
+ } else {
+ int aliveCompactorsForGroup = 0;
+ for (String compactor : compactors) {
+ String cpath = compactorQueuesPath + "/" + group + "/" + compactor;
+ var lockNodes = zoorw.getChildren(compactorQueuesPath + "/" + group + "/" + compactor);
+ if (lockNodes.isEmpty()) {
+ deleteEmpty(zoorw, cpath);
+ } else {
+ aliveCompactorsForGroup++;
+ }
+ }
+ CompactionJobPriorityQueue queue = getJobQueues().getQueue(cgid);
+ if (queue != null) {
+ queue.setMaxSize(
+ Math.min((int) (aliveCompactorsForGroup * queueSizeFactor), Integer.MAX_VALUE));
+ }
+
+ }
+
+ }
+
+ } catch (KeeperException | RuntimeException e) {
+ LOG.warn("Failed to clean up compactors", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private Set<CompactorGroupId> getCompactionServicesConfigurationGroups()
+ throws ReflectiveOperationException, IllegalArgumentException, SecurityException {
+
+ Set<CompactorGroupId> groups = new HashSet<>();
+ AccumuloConfiguration config = ctx.getConfiguration();
+ CompactionServicesConfig servicesConfig = new CompactionServicesConfig(config);
+
+ for (var entry : servicesConfig.getPlanners().entrySet()) {
+ String serviceId = entry.getKey();
+ String plannerClassName = entry.getValue();
+
+ Class<? extends CompactionPlanner> plannerClass =
+ Class.forName(plannerClassName).asSubclass(CompactionPlanner.class);
+ CompactionPlanner planner = plannerClass.getDeclaredConstructor().newInstance();
+
+ var initParams = new CompactionPlannerInitParams(CompactionServiceId.of(serviceId),
+ servicesConfig.getPlannerPrefix(serviceId), servicesConfig.getOptions().get(serviceId),
+ new ServiceEnvironmentImpl(ctx));
+
+ planner.init(initParams);
+
+ groups.addAll(initParams.getRequestedGroups());
+ }
+ return groups;
+ }
+
+ public void cleanUpInternalState() {
+
+ // This method does the following:
+ //
+ // 1. Removes entries from RUNNING_CACHE that are not really running
+ // 2. Cancels running compactions for groups that are not in the current configuration
+ // 3. Remove groups not in configuration from TIME_COMPACTOR_LAST_CHECKED
+ // 4. Log groups with no compactors
+ // 5. Log compactors with no groups
+ // 6. Log groups with compactors and queued jos that have not checked in
+
+ // grab a snapshot of the ids in the set before reading the metadata table. This is done to
+ // avoid removing things that are added while reading the metadata.
+ final Set<ExternalCompactionId> idsSnapshot = Set.copyOf(RUNNING_CACHE.keySet());
+
+ // grab the ids that are listed as running in the metadata table. It important that this is done
+ // after getting the snapshot.
+ final Set<ExternalCompactionId> idsInMetadata = readExternalCompactionIds();
+
+ final Set<ExternalCompactionId> idsToRemove = Sets.difference(idsSnapshot, idsInMetadata);
+
+ // remove ids that are in the running set but not in the metadata table
+ idsToRemove.forEach(this::recordCompletion);
+ if (idsToRemove.size() > 0) {
+ LOG.debug("Removed stale entries from RUNNING_CACHE : {}", idsToRemove);
+ }
+
+ // Get the set of groups being referenced in the current configuration
+ Set<CompactorGroupId> groupsInConfiguration = null;
+ try {
+ groupsInConfiguration = getCompactionServicesConfigurationGroups();
+ } catch (RuntimeException | ReflectiveOperationException e) {
+ LOG.error(
+ "Error getting groups from the compaction services configuration. Unable to clean up internal state.",
+ e);
+ return;
+ }
+
+ // Compaction jobs are created in the TabletGroupWatcher and added to the Coordinator
+ // via the addJobs method which adds the job to the CompactionJobQueues object.
+ final Set<CompactorGroupId> groupsWithJobs = jobQueues.getQueueIds();
+
+ final Set<CompactorGroupId> jobGroupsNotInConfiguration =
+ Sets.difference(groupsWithJobs, groupsInConfiguration);
+
+ if (jobGroupsNotInConfiguration != null && !jobGroupsNotInConfiguration.isEmpty()) {
+ RUNNING_CACHE.values().forEach(rc -> {
+ if (jobGroupsNotInConfiguration.contains(CompactorGroupId.of(rc.getGroupName()))) {
+ LOG.warn(
+ "External compaction {} running in group {} on compactor {},"
+ + " but group not found in current configuration. Failing compaction...",
+ rc.getJob().getExternalCompactionId(), rc.getGroupName(), rc.getCompactorAddress());
+ cancelCompactionOnCompactor(rc.getCompactorAddress(),
+ rc.getJob().getExternalCompactionId());
+ }
+ });
+
+ final Set<CompactorGroupId> trackedGroups = Set.copyOf(TIME_COMPACTOR_LAST_CHECKED.keySet());
+ TIME_COMPACTOR_LAST_CHECKED.keySet().retainAll(groupsInConfiguration);
+ LOG.debug("No longer tracking compactor check-in times for groups: {}",
+ Sets.difference(trackedGroups, TIME_COMPACTOR_LAST_CHECKED.keySet()));
+ }
+
+ final Map<String,Set<HostAndPort>> runningCompactors = getRunningCompactors();
+
+ final Set<CompactorGroupId> runningCompactorGroups = new HashSet<>();
+ runningCompactors.keySet()
+ .forEach(group -> runningCompactorGroups.add(CompactorGroupId.of(group)));
+
+ final Set<CompactorGroupId> groupsWithNoCompactors =
+ Sets.difference(groupsInConfiguration, runningCompactorGroups);
+ if (groupsWithNoCompactors != null && !groupsWithNoCompactors.isEmpty()) {
+ for (CompactorGroupId group : groupsWithNoCompactors) {
+ long queuedJobCount = jobQueues.getQueuedJobs(group);
+ if (queuedJobCount > 0) {
+ LOG.warn("Compactor group {} has {} queued compactions but no running compactors", group,
+ queuedJobCount);
+ }
+ }
+ }
+
+ final Set<CompactorGroupId> compactorsWithNoGroups =
+ Sets.difference(runningCompactorGroups, groupsInConfiguration);
+ if (compactorsWithNoGroups != null && !compactorsWithNoGroups.isEmpty()) {
+ LOG.warn(
+ "The following groups have running compactors, but are not in the current configuration: {}",
+ compactorsWithNoGroups);
+ }
+
+ final long now = System.currentTimeMillis();
+ final long warningTime = getMissingCompactorWarningTime();
+ Map<String,Set<HostAndPort>> idleCompactors = getIdleCompactors(runningCompactors);
+ for (CompactorGroupId groupName : groupsInConfiguration) {
+ long lastCheckTime =
+ TIME_COMPACTOR_LAST_CHECKED.getOrDefault(groupName, coordinatorStartTime);
+ if ((now - lastCheckTime) > warningTime && jobQueues.getQueuedJobs(groupName) > 0
+ && idleCompactors.containsKey(groupName.canonical())) {
+ LOG.warn(
+ "The group {} has queued jobs and {} idle compactors, however none have checked in "
+ + "with coordinator for {}ms",
+ groupName, idleCompactors.get(groupName.canonical()).size(), warningTime);
+ }
+ }
+ }
+
+ private static Set<StoredTabletFile> getFilesReservedBySelection(TabletMetadata tabletMetadata,
+ SteadyTime steadyTime, ServerContext ctx) {
+ if (tabletMetadata.getSelectedFiles() == null) {
+ return Set.of();
+ }
+
+ if (tabletMetadata.getSelectedFiles().getCompletedJobs() > 0) {
+ return tabletMetadata.getSelectedFiles().getFiles();
+ }
+
+ long selectedExpirationDuration = ctx.getTableConfiguration(tabletMetadata.getTableId())
+ .getTimeInMillis(Property.TABLE_COMPACTION_SELECTION_EXPIRATION);
+
+ if (steadyTime.minus(tabletMetadata.getSelectedFiles().getSelectedTime()).toMillis()
+ < selectedExpirationDuration) {
+ return tabletMetadata.getSelectedFiles().getFiles();
+ }
+
+ return Set.of();
+ }
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/QueueMetrics.java
index da750e552b,0000000000..dbc4300f42
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/QueueMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/QueueMetrics.java
@@@ -1,155 -1,0 +1,162 @@@
+/*
+ * 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
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.manager.compaction.coordinator;
+
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUES;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH;
+import static org.apache.accumulo.core.metrics.MetricsUtil.formatString;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
+import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue;
+import org.apache.accumulo.manager.compaction.queue.CompactionJobQueues;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Sets;
+import com.google.common.collect.Sets.SetView;
+
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+
+public class QueueMetrics implements MetricsProducer {
+
+ private static class QueueMeters {
+ private final Gauge length;
+ private final Gauge jobsQueued;
+ private final Gauge jobsDequeued;
+ private final Gauge jobsRejected;
+ private final Gauge jobsLowestPriority;
+
+ public QueueMeters(MeterRegistry meterRegistry, CompactorGroupId cgid,
+ CompactionJobPriorityQueue queue) {
+ var queueId = formatString(cgid.canonical());
+
+ length =
- Gauge.builder(METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH, queue, q -> q.getMaxSize())
++ Gauge.builder(COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH.getName(), queue, q -> q.getMaxSize())
+ .description("Length of priority queues").tags(List.of(Tag.of("queue.id", queueId)))
+ .register(meterRegistry);
+
+ jobsQueued = Gauge
- .builder(METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED, queue, q -> q.getQueuedJobs())
++ .builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED.getName(), queue,
++ q -> q.getQueuedJobs())
+ .description("Count of queued jobs").tags(List.of(Tag.of("queue.id", queueId)))
+ .register(meterRegistry);
+
+ jobsDequeued = Gauge
- .builder(METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED, queue,
++ .builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED.getName(), queue,
+ q -> q.getDequeuedJobs())
+ .description("Count of jobs dequeued").tags(List.of(Tag.of("queue.id", queueId)))
+ .register(meterRegistry);
+
+ jobsRejected = Gauge
- .builder(METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED, queue,
++ .builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED.getName(), queue,
+ q -> q.getRejectedJobs())
+ .description("Count of rejected jobs").tags(List.of(Tag.of("queue.id", queueId)))
+ .register(meterRegistry);
+
+ jobsLowestPriority = Gauge
- .builder(METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY, queue,
++ .builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY.getName(), queue,
+ q -> q.getLowestPriority())
+ .description("Lowest priority queued job").tags(List.of(Tag.of("queue.id", queueId)))
+ .register(meterRegistry);
+ }
+
+ private void removeMeters(MeterRegistry registry) {
+ registry.remove(length);
+ registry.remove(jobsQueued);
+ registry.remove(jobsDequeued);
+ registry.remove(jobsRejected);
+ registry.remove(jobsLowestPriority);
+ }
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(QueueMetrics.class);
+ private static final long DEFAULT_MIN_REFRESH_DELAY = TimeUnit.SECONDS.toMillis(5);
+ private volatile MeterRegistry meterRegistry = null;
+ private final CompactionJobQueues compactionJobQueues;
+ private final Map<CompactorGroupId,QueueMeters> perQueueMetrics = new HashMap<>();
+ private Gauge queueCountMeter = null;
+
+ public QueueMetrics(CompactionJobQueues compactionJobQueues) {
+ this.compactionJobQueues = compactionJobQueues;
+ ScheduledExecutorService scheduler =
+ ThreadPools.getServerThreadPools().createScheduledExecutorService(1, "queueMetricsPoller");
+ Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdownNow));
+ ThreadPools.watchNonCriticalScheduledTask(scheduler.scheduleAtFixedRate(this::update,
+ DEFAULT_MIN_REFRESH_DELAY, DEFAULT_MIN_REFRESH_DELAY, TimeUnit.MILLISECONDS));
+ }
+
+ public void update() {
+
+ // read the volatile variable once so the rest of the method has consistent view
+ var localRegistry = meterRegistry;
+
+ if (queueCountMeter == null) {
+ queueCountMeter = Gauge
- .builder(METRICS_COMPACTOR_JOB_PRIORITY_QUEUES, compactionJobQueues,
++ .builder(COMPACTOR_JOB_PRIORITY_QUEUES.getName(), compactionJobQueues,
+ CompactionJobQueues::getQueueCount)
+ .description("Number of current Queues").register(localRegistry);
+ }
+ LOG.debug("update - cjq queues: {}", compactionJobQueues.getQueueIds());
+
+ Set<CompactorGroupId> definedQueues = compactionJobQueues.getQueueIds();
+ LOG.debug("update - defined queues: {}", definedQueues);
+
+ // Copy the keySet into a new Set so that changes to perQueueMetrics
+ // don't affect the collection
+ Set<CompactorGroupId> queuesWithMetrics = new HashSet<>(perQueueMetrics.keySet());
+ LOG.debug("update - queues with metrics: {}", queuesWithMetrics);
+
+ SetView<CompactorGroupId> queuesWithoutMetrics =
+ Sets.difference(definedQueues, queuesWithMetrics);
+ queuesWithoutMetrics.forEach(q -> {
+ LOG.debug("update - creating meters for queue: {}", q);
+ perQueueMetrics.put(q, new QueueMeters(localRegistry, q, compactionJobQueues.getQueue(q)));
+ });
+
+ SetView<CompactorGroupId> metricsWithoutQueues =
+ Sets.difference(queuesWithMetrics, definedQueues);
+ metricsWithoutQueues.forEach(q -> {
+ LOG.debug("update - removing meters for queue: {}", q);
+ perQueueMetrics.get(q).removeMeters(localRegistry);
+ perQueueMetrics.remove(q);
+ });
+
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ this.meterRegistry = registry;
+ }
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetrics.java
index a6c59f5328,02a30d4d22..40f8a6d262
--- a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/ManagerMetrics.java
@@@ -18,12 -18,8 +18,16 @@@
*/
package org.apache.accumulo.manager.metrics;
+import static java.util.Objects.requireNonNull;
++import static org.apache.accumulo.core.metrics.Metric.MANAGER_COMPACTION_SVC_ERRORS;
++import static org.apache.accumulo.core.metrics.Metric.MANAGER_META_TGW_ERRORS;
++import static org.apache.accumulo.core.metrics.Metric.MANAGER_ROOT_TGW_ERRORS;
++import static org.apache.accumulo.core.metrics.Metric.MANAGER_USER_TGW_ERRORS;
+
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
@@@ -31,69 -26,13 +35,69 @@@ import org.apache.accumulo.core.metadat
import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.manager.Manager;
import org.apache.accumulo.manager.metrics.fate.FateMetrics;
+import org.apache.accumulo.manager.metrics.fate.meta.MetaFateMetrics;
+import org.apache.accumulo.manager.metrics.fate.user.UserFateMetrics;
+
+import io.micrometer.core.instrument.MeterRegistry;
+
+public class ManagerMetrics implements MetricsProducer {
+
+ private final List<FateMetrics<?>> fateMetrics;
+
+ private final AtomicLong rootTGWErrorsGauge = new AtomicLong(0);
+ private final AtomicLong metadataTGWErrorsGauge = new AtomicLong(0);
+ private final AtomicLong userTGWErrorsGauge = new AtomicLong(0);
+ private final AtomicInteger compactionConfigurationError = new AtomicInteger(0);
+
+ public ManagerMetrics(final AccumuloConfiguration conf, final Manager manager) {
+ requireNonNull(conf, "AccumuloConfiguration must not be null");
+ requireNonNull(conf, "Manager must not be null");
+ fateMetrics = List.of(
+ new MetaFateMetrics(manager.getContext(),
+ conf.getTimeInMillis(Property.MANAGER_FATE_METRICS_MIN_UPDATE_INTERVAL)),
+ new UserFateMetrics(manager.getContext(),
+ conf.getTimeInMillis(Property.MANAGER_FATE_METRICS_MIN_UPDATE_INTERVAL)));
+ }
-public class ManagerMetrics {
+ public void incrementTabletGroupWatcherError(DataLevel level) {
+ switch (level) {
+ case METADATA:
+ metadataTGWErrorsGauge.incrementAndGet();
+ break;
+ case ROOT:
+ rootTGWErrorsGauge.incrementAndGet();
+ break;
+ case USER:
+ userTGWErrorsGauge.incrementAndGet();
+ break;
+ default:
+ throw new IllegalStateException("Unhandled DataLevel: " + level);
+ }
+ }
+
+ public void setCompactionServiceConfigurationError() {
+ this.compactionConfigurationError.set(1);
+ }
+
+ public void clearCompactionServiceConfigurationError() {
+ this.compactionConfigurationError.set(0);
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ fateMetrics.forEach(fm -> fm.registerMetrics(registry));
- registry.gauge(METRICS_MANAGER_ROOT_TGW_ERRORS, rootTGWErrorsGauge);
- registry.gauge(METRICS_MANAGER_META_TGW_ERRORS, metadataTGWErrorsGauge);
- registry.gauge(METRICS_MANAGER_USER_TGW_ERRORS, userTGWErrorsGauge);
- registry.gauge(METRICS_MANAGER_COMPACTION_SVC_ERRORS, compactionConfigurationError,
++ registry.gauge(MANAGER_ROOT_TGW_ERRORS.getName(), rootTGWErrorsGauge);
++ registry.gauge(MANAGER_META_TGW_ERRORS.getName(), metadataTGWErrorsGauge);
++ registry.gauge(MANAGER_USER_TGW_ERRORS.getName(), userTGWErrorsGauge);
++ registry.gauge(MANAGER_COMPACTION_SVC_ERRORS.getName(), compactionConfigurationError,
+ AtomicInteger::get);
+ }
- public static List<MetricsProducer> getProducers(AccumuloConfiguration conf, Manager manager) {
+ public List<MetricsProducer> getProducers(AccumuloConfiguration conf, Manager manager) {
ArrayList<MetricsProducer> producers = new ArrayList<>();
- producers.add(new FateMetrics(manager.getContext(),
- conf.getTimeInMillis(Property.MANAGER_FATE_METRICS_MIN_UPDATE_INTERVAL)));
+ producers.add(this);
+ producers.addAll(fateMetrics);
+ producers.add(manager.getCompactionCoordinator());
return producers;
}
}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/FateMetrics.java
index c9d704e374,ebbbec4316..be11b10584
--- a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/FateMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/FateMetrics.java
@@@ -18,9 -18,14 +18,13 @@@
*/
package org.apache.accumulo.manager.metrics.fate;
-import static org.apache.accumulo.core.metrics.Metric.FATE_ERRORS;
+ import static org.apache.accumulo.core.metrics.Metric.FATE_OPS;
-import static org.apache.accumulo.core.metrics.Metric.FATE_OPS_ACTIVITY;
+ import static org.apache.accumulo.core.metrics.Metric.FATE_TX;
+ import static org.apache.accumulo.core.metrics.Metric.FATE_TYPE_IN_PROGRESS;
+
import java.util.List;
import java.util.Map.Entry;
+import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@@ -104,56 -126,47 +108,56 @@@ public abstract class FateMetrics<T ext
log.warn("Unhandled status type: {}", vals.getKey());
}
}
-
metricValues.getOpTypeCounters().forEach((name, count) -> {
- Metrics.gauge(METRICS_FATE_TYPE_IN_PROGRESS, Tags.of(OP_TYPE_TAG, name), count);
+ Metrics.gauge(FATE_TYPE_IN_PROGRESS.getName(), Tags.of(OP_TYPE_TAG, name), count);
});
}
@Override
public void registerMetrics(final MeterRegistry registry) {
- totalCurrentOpsGauge = registry.gauge(FATE_OPS.getName(), new AtomicLong(0));
- totalOpsGauge = registry.gauge(FATE_OPS_ACTIVITY.getName(), new AtomicLong(0));
- fateErrorsGauge = registry.gauge(FATE_ERRORS.getName(),
- List.of(Tag.of("type", "zk.connection")), new AtomicLong(0));
- newTxGauge = registry.gauge(FATE_TX.getName(),
- List.of(Tag.of("state", ReadOnlyTStore.TStatus.NEW.name().toLowerCase())),
- new AtomicLong(0));
- submittedTxGauge = registry.gauge(FATE_TX.getName(),
- List.of(Tag.of("state", ReadOnlyTStore.TStatus.SUBMITTED.name().toLowerCase())),
- new AtomicLong(0));
- inProgressTxGauge = registry.gauge(FATE_TX.getName(),
- List.of(Tag.of("state", ReadOnlyTStore.TStatus.IN_PROGRESS.name().toLowerCase())),
- new AtomicLong(0));
- failedInProgressTxGauge = registry.gauge(FATE_TX.getName(),
- List.of(Tag.of("state", ReadOnlyTStore.TStatus.FAILED_IN_PROGRESS.name().toLowerCase())),
- new AtomicLong(0));
- failedTxGauge = registry.gauge(FATE_TX.getName(),
- List.of(Tag.of("state", ReadOnlyTStore.TStatus.FAILED.name().toLowerCase())),
- new AtomicLong(0));
- successfulTxGauge = registry.gauge(FATE_TX.getName(),
- List.of(Tag.of("state", ReadOnlyTStore.TStatus.SUCCESSFUL.name().toLowerCase())),
- new AtomicLong(0));
- unknownTxGauge = registry.gauge(FATE_TX.getName(),
- List.of(Tag.of("state", ReadOnlyTStore.TStatus.UNKNOWN.name().toLowerCase())),
- new AtomicLong(0));
-
- update();
+ var type = fateStore.type().name().toLowerCase();
+ var instanceTypeTag = Tag.of("instanceType", type);
+
- registry.gauge(METRICS_FATE_OPS, totalCurrentOpsGauge);
++ registry.gauge(FATE_OPS.getName(), totalCurrentOpsGauge);
+
- registry.gauge(METRICS_FATE_TX, List
++ registry.gauge(FATE_TX.getName(), List
+ .of(Tag.of("state", ReadOnlyFateStore.TStatus.NEW.name().toLowerCase()), instanceTypeTag),
+ newTxGauge);
- registry.gauge(METRICS_FATE_TX,
++ registry.gauge(FATE_TX.getName(),
+ List.of(Tag.of("state", ReadOnlyFateStore.TStatus.SUBMITTED.name().toLowerCase()),
+ instanceTypeTag),
+ submittedTxGauge);
- registry.gauge(METRICS_FATE_TX,
++ registry.gauge(FATE_TX.getName(),
+ List.of(Tag.of("state", ReadOnlyFateStore.TStatus.IN_PROGRESS.name().toLowerCase()),
+ instanceTypeTag),
+ inProgressTxGauge);
- registry.gauge(METRICS_FATE_TX,
++ registry.gauge(FATE_TX.getName(),
+ List.of(Tag.of("state", ReadOnlyFateStore.TStatus.FAILED_IN_PROGRESS.name().toLowerCase()),
+ instanceTypeTag),
+ failedInProgressTxGauge);
- registry.gauge(METRICS_FATE_TX,
++ registry.gauge(FATE_TX.getName(),
+ List.of(Tag.of("state", ReadOnlyFateStore.TStatus.FAILED.name().toLowerCase()),
+ instanceTypeTag),
+ failedTxGauge);
- registry.gauge(METRICS_FATE_TX,
++ registry.gauge(FATE_TX.getName(),
+ List.of(Tag.of("state", ReadOnlyFateStore.TStatus.SUCCESSFUL.name().toLowerCase()),
+ instanceTypeTag),
+ successfulTxGauge);
- registry.gauge(METRICS_FATE_TX,
++ registry.gauge(FATE_TX.getName(),
+ List.of(Tag.of("state", ReadOnlyFateStore.TStatus.UNKNOWN.name().toLowerCase()),
+ instanceTypeTag),
+ unknownTxGauge);
// get fate status is read only operation - no reason to be nice on shutdown.
- ScheduledExecutorService scheduler =
- ThreadPools.getServerThreadPools().createScheduledExecutorService(1, "fateMetricsPoller");
+ ScheduledExecutorService scheduler = ThreadPools.getServerThreadPools()
+ .createScheduledExecutorService(1, type + "FateMetricsPoller");
Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdownNow));
+ // Only update as part of the scheduler thread.
+ // We have to call update() in a new thread because this method to
+ // register metrics is called on start up in the Manager before it's finished
+ // initializing, so we can't scan the User fate store until after startup is done.
+ // If we called update() here in this method directly we would get stuck forever.
ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
try {
update();
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/meta/MetaFateMetrics.java
index 24570fa24b,0000000000..550ee7967e
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/meta/MetaFateMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/meta/MetaFateMetrics.java
@@@ -1,81 -1,0 +1,85 @@@
+/*
+ * 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
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.manager.metrics.fate.meta;
+
++import static org.apache.accumulo.core.metrics.Metric.FATE_ERRORS;
++import static org.apache.accumulo.core.metrics.Metric.FATE_OPS_ACTIVITY;
++
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.fate.MetaFateStore;
+import org.apache.accumulo.core.fate.ReadOnlyFateStore;
+import org.apache.accumulo.manager.metrics.fate.FateMetrics;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.zookeeper.KeeperException;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+
+public class MetaFateMetrics extends FateMetrics<MetaFateMetricValues> {
+
+ private final String fateRootPath;
+ private final AtomicLong totalOpsGauge = new AtomicLong(0);
+ private final AtomicLong fateErrorsGauge = new AtomicLong(0);
+
+ public MetaFateMetrics(ServerContext context, long minimumRefreshDelay) {
+ super(context, minimumRefreshDelay);
+ this.fateRootPath = getFateRootPath(context);
+ }
+
+ @Override
+ protected void update(MetaFateMetricValues metricValues) {
+ super.update(metricValues);
+ totalOpsGauge.set(metricValues.getZkFateChildOpsTotal());
+ fateErrorsGauge.set(metricValues.getZkConnectionErrors());
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ super.registerMetrics(registry);
- registry.gauge(METRICS_FATE_OPS_ACTIVITY, totalOpsGauge);
- registry.gauge(METRICS_FATE_ERRORS, List.of(Tag.of("type", "zk.connection")), fateErrorsGauge);
++ registry.gauge(FATE_OPS_ACTIVITY.getName(), totalOpsGauge);
++ registry.gauge(FATE_ERRORS.getName(), List.of(Tag.of("type", "zk.connection")),
++ fateErrorsGauge);
+ }
+
+ @Override
+ protected ReadOnlyFateStore<FateMetrics<MetaFateMetricValues>> buildStore(ServerContext context) {
+ try {
+ return new MetaFateStore<>(getFateRootPath(context), context.getZooReaderWriter());
+ } catch (KeeperException ex) {
+ throw new IllegalStateException(
+ "FATE Metrics - Failed to create zoo store - metrics unavailable", ex);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "FATE Metrics - Interrupt received while initializing zoo store");
+ }
+ }
+
+ @Override
+ protected MetaFateMetricValues getMetricValues() {
+ return MetaFateMetricValues.getMetaStoreMetrics(context, fateRootPath, fateStore);
+ }
+
+ private static String getFateRootPath(ServerContext context) {
+ return context.getZooKeeperRoot() + Constants.ZFATE;
+ }
+}
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java
index 847c39f572,26bbd11adb..76753b2c54
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetrics.java
@@@ -18,6 -18,25 +18,25 @@@
*/
package org.apache.accumulo.tserver.metrics;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_ENTRIES_READ;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_ENTRIES_WRITTEN;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_MAJC_STUCK;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_ENTRIES;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_HOLD;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_INGEST_BYTES;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_INGEST_MUTATIONS;
-import static org.apache.accumulo.core.metrics.Metric.TSERVER_MAJC_QUEUED;
-import static org.apache.accumulo.core.metrics.Metric.TSERVER_MAJC_RUNNING;
-import static org.apache.accumulo.core.metrics.Metric.TSERVER_MAJC_STUCK;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_MEM_ENTRIES;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_MINC_QUEUED;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_MINC_RUNNING;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_MINC_TOTAL;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_FILES;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_LONG_ASSIGNMENTS;
++import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_ONDEMAND_UNLOADED_FOR_MEM;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_ONLINE;
++import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_ONLINE_ONDEMAND;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_OPENING;
+ import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_UNOPENED;
+
import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.server.compaction.CompactionWatcher;
import org.apache.accumulo.server.compaction.FileCompactor;
@@@ -55,42 -74,42 +74,45 @@@ public class TabletServerMetrics implem
TabletServerMetrics::getTotalEntriesWritten)
.description("Number of entries written by all compactions that have run on this tserver")
.register(registry);
- LongTaskTimer timer = LongTaskTimer.builder(METRICS_COMPACTOR_MAJC_STUCK)
- LongTaskTimer timer = LongTaskTimer.builder(TSERVER_MAJC_STUCK.getName())
++ LongTaskTimer timer = LongTaskTimer.builder(COMPACTOR_MAJC_STUCK.getName())
.description("Number and duration of stuck major compactions").register(registry);
CompactionWatcher.setTimer(timer);
+
Gauge
- .builder(METRICS_TSERVER_TABLETS_LONG_ASSIGNMENTS, util,
+ .builder(TSERVER_TABLETS_LONG_ASSIGNMENTS.getName(), util,
TabletServerMetricsUtil::getLongTabletAssignments)
.description("Number of tablet assignments that are taking a long time").register(registry);
- Gauge.builder(METRICS_TSERVER_ENTRIES, util, TabletServerMetricsUtil::getEntries)
+ Gauge.builder(TSERVER_ENTRIES.getName(), util, TabletServerMetricsUtil::getEntries)
.description("Number of entries").register(registry);
- Gauge.builder(METRICS_TSERVER_MEM_ENTRIES, util, TabletServerMetricsUtil::getEntriesInMemory)
+ Gauge.builder(TSERVER_MEM_ENTRIES.getName(), util, TabletServerMetricsUtil::getEntriesInMemory)
.description("Number of entries in memory").register(registry);
- Gauge.builder(METRICS_TSERVER_MINC_RUNNING, util, TabletServerMetricsUtil::getMinorCompactions)
- Gauge
- .builder(TSERVER_MAJC_RUNNING.getName(), util, TabletServerMetricsUtil::getMajorCompactions)
- .description("Number of active major compactions").register(registry);
- Gauge
- .builder(TSERVER_MAJC_QUEUED.getName(), util,
- TabletServerMetricsUtil::getMajorCompactionsQueued)
- .description("Number of queued major compactions").register(registry);
+ Gauge
+ .builder(TSERVER_MINC_RUNNING.getName(), util, TabletServerMetricsUtil::getMinorCompactions)
.description("Number of active minor compactions").register(registry);
Gauge
- .builder(METRICS_TSERVER_MINC_QUEUED, util,
+ .builder(TSERVER_MINC_QUEUED.getName(), util,
TabletServerMetricsUtil::getMinorCompactionsQueued)
.description("Number of queued minor compactions").register(registry);
+ Gauge
- .builder(METRICS_TSERVER_TABLETS_ONLINE_ONDEMAND, util,
++ .builder(TSERVER_TABLETS_ONLINE_ONDEMAND.getName(), util,
+ TabletServerMetricsUtil::getOnDemandOnlineCount)
+ .description("Number of online on-demand tablets").register(registry);
+ Gauge
- .builder(METRICS_TSERVER_TABLETS_ONDEMAND_UNLOADED_FOR_MEM, util,
++ .builder(TSERVER_TABLETS_ONDEMAND_UNLOADED_FOR_MEM.getName(), util,
+ TabletServerMetricsUtil::getOnDemandUnloadedLowMem)
+ .description("Number of online on-demand tablets unloaded due to low memory")
+ .register(registry);
- Gauge.builder(METRICS_TSERVER_TABLETS_ONLINE, util, TabletServerMetricsUtil::getOnlineCount)
+ Gauge.builder(TSERVER_TABLETS_ONLINE.getName(), util, TabletServerMetricsUtil::getOnlineCount)
.description("Number of online tablets").register(registry);
- Gauge.builder(METRICS_TSERVER_TABLETS_OPENING, util, TabletServerMetricsUtil::getOpeningCount)
+ Gauge.builder(TSERVER_TABLETS_OPENING.getName(), util, TabletServerMetricsUtil::getOpeningCount)
.description("Number of opening tablets").register(registry);
- Gauge.builder(METRICS_TSERVER_TABLETS_UNOPENED, util, TabletServerMetricsUtil::getUnopenedCount)
+ Gauge
+ .builder(TSERVER_TABLETS_UNOPENED.getName(), util,
+ TabletServerMetricsUtil::getUnopenedCount)
.description("Number of unopened tablets").register(registry);
Gauge
- .builder(METRICS_TSERVER_MINC_TOTAL, util,
+ .builder(TSERVER_MINC_TOTAL.getName(), util,
TabletServerMetricsUtil::getTotalMinorCompactions)
.description("Total number of minor compactions performed").register(registry);
diff --cc test/src/main/java/org/apache/accumulo/test/compaction/BadCompactionServiceConfigIT.java
index 9aea90d60a,35813d9a93..b29b43aa77
--- a/test/src/main/java/org/apache/accumulo/test/compaction/BadCompactionServiceConfigIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/BadCompactionServiceConfigIT.java
@@@ -19,6 -19,6 +19,7 @@@
package org.apache.accumulo.test.compaction;
import static org.apache.accumulo.core.Constants.DEFAULT_COMPACTION_SERVICE_NAME;
++import static org.apache.accumulo.core.metrics.Metric.MANAGER_COMPACTION_SVC_ERRORS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.Collections;
@@@ -43,27 -42,14 +44,26 @@@ import org.apache.accumulo.core.data.Ke
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.Filter;
- import org.apache.accumulo.core.metrics.MetricsProducer;
-import org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner;
+import org.apache.accumulo.core.spi.compaction.RatioBasedCompactionPlanner;
+import org.apache.accumulo.core.spi.metrics.LoggingMeterRegistryFactory;
+import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.accumulo.test.util.Wait;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.google.common.collect.MoreCollectors;
@@@ -131,32 -87,6 +131,32 @@@ public class BadCompactionServiceConfig
@Test
public void testUsingMisconfiguredService() throws Exception {
+
+ final AtomicBoolean shutdownTailer = new AtomicBoolean(false);
+ final AtomicBoolean serviceMisconfigured = new AtomicBoolean(false);
+ final Thread thread = Threads.createThread("metric-tailer", () -> {
+ while (!shutdownTailer.get()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String s : statsDMetrics) {
+ if (shutdownTailer.get()) {
+ break;
+ }
- if (s.startsWith(MetricsProducer.METRICS_MANAGER_COMPACTION_SVC_ERRORS)) {
++ if (s.startsWith(MANAGER_COMPACTION_SVC_ERRORS.getName())) {
+ Metric m = TestStatsDSink.parseStatsDMetric(s);
+ Integer value = Integer.parseInt(m.getValue());
+ if (value == 0) {
+ serviceMisconfigured.set(false);
+ } else if (value == 1) {
+ serviceMisconfigured.set(true);
+ } else {
+ LOG.error("Invalid value received: " + m.getValue());
+ }
+ }
+ }
+ }
+ });
+ thread.start();
+
String table = getUniqueNames(1)[0];
// Create a table that is configured to use a compaction service with bad configuration.
@@@ -255,32 -173,6 +255,32 @@@
@Test
public void testUsingNonExistentService() throws Exception {
+
+ final AtomicBoolean shutdownTailer = new AtomicBoolean(false);
+ final AtomicBoolean serviceMisconfigured = new AtomicBoolean(false);
+ final Thread thread = Threads.createThread("metric-tailer", () -> {
+ while (!shutdownTailer.get()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String s : statsDMetrics) {
+ if (shutdownTailer.get()) {
+ break;
+ }
- if (s.startsWith(MetricsProducer.METRICS_MANAGER_COMPACTION_SVC_ERRORS)) {
++ if (s.startsWith(MANAGER_COMPACTION_SVC_ERRORS.getName())) {
+ Metric m = TestStatsDSink.parseStatsDMetric(s);
+ Integer value = Integer.parseInt(m.getValue());
+ if (value == 0) {
+ serviceMisconfigured.set(false);
+ } else if (value == 1) {
+ serviceMisconfigured.set(true);
+ } else {
+ LOG.error("Invalid value received: " + m.getValue());
+ }
+ }
+ }
+ }
+ });
+ thread.start();
+
String table = getUniqueNames(1)[0];
// Create a table that is configured to use a compaction service that does not exist
diff --cc test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
index c2a86fa6bc,0000000000..b976aab38e
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
@@@ -1,485 -1,0 +1,484 @@@
+/*
+ * 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
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.compaction;
+
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUES;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED;
++import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH;
+import static org.apache.accumulo.core.util.compaction.ExternalCompactionUtil.getCompactorAddrs;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.client.admin.TabletAvailability;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.file.FileOperations;
+import org.apache.accumulo.core.file.FileSKVWriter;
+import org.apache.accumulo.core.file.rfile.RFile;
+import org.apache.accumulo.core.metadata.UnreferencedTabletFile;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
- import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.metrics.MetricsUtil;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.spi.compaction.RatioBasedCompactionPlanner;
+import org.apache.accumulo.core.spi.compaction.SimpleCompactionDispatcher;
+import org.apache.accumulo.core.spi.crypto.NoCryptoServiceFactory;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.core.util.compaction.CompactionJobPrioritizer;
+import org.apache.accumulo.core.util.threads.Threads;
+import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.minicluster.MemoryUnit;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.test.functional.CompactionIT;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.util.Wait;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+// Use the shared Mini Cluster for all metric tests as Compaction Resources can be spun up and down per test.
+public class CompactionPriorityQueueMetricsIT extends SharedMiniClusterBase {
+
+ public static final Logger log = LoggerFactory.getLogger(CompactionPriorityQueueMetricsIT.class);
+
+ private static TestStatsDSink sink;
+ private String tableName;
+ private TableId tableId;
+ private AccumuloConfiguration aconf;
+ private FileSystem fs;
+ private String rootPath;
+
+ public static final String QUEUE1 = "METRICSQ1";
+ public static final String QUEUE1_METRIC_LABEL = MetricsUtil.formatString(QUEUE1);
+ public static final String QUEUE1_SERVICE = "Q1";
+ public static final int QUEUE1_SIZE = 6;
+
+ // Metrics collector Thread
+ final LinkedBlockingQueue<TestStatsDSink.Metric> queueMetrics = new LinkedBlockingQueue<>();
+ final AtomicBoolean shutdownTailer = new AtomicBoolean(false);
+ Thread metricsTailer;
+
+ @BeforeEach
+ public void setupMetricsTest() throws Exception {
+ getCluster().getClusterControl().stopAllServers(ServerType.COMPACTOR);
+ Wait.waitFor(() -> getCompactorAddrs(getCluster().getServerContext()).isEmpty());
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ tableName = getUniqueNames(1)[0];
+
+ Map<String,String> props =
+ Map.of("table.compaction.dispatcher", SimpleCompactionDispatcher.class.getName(),
+ "table.compaction.dispatcher.opts.service", QUEUE1_SERVICE);
+ NewTableConfiguration ntc = new NewTableConfiguration().setProperties(props)
+ .withInitialTabletAvailability(TabletAvailability.HOSTED);
+ c.tableOperations().create(tableName, ntc);
+
+ tableId = TableId.of(c.tableOperations().tableIdMap().get(tableName));
+ aconf = getCluster().getServerContext().getConfiguration();
+ fs = getCluster().getFileSystem();
+ rootPath = getCluster().getTemporaryPath().toString();
+ }
+ queueMetrics.clear();
+ shutdownTailer.set(false);
+ metricsTailer = Threads.createThread("metric-tailer", () -> {
+ while (!shutdownTailer.get()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String s : statsDMetrics) {
+ if (shutdownTailer.get()) {
+ break;
+ }
- if (s.startsWith(MetricsProducer.METRICS_COMPACTOR_PREFIX + "queue")) {
++ if (s.startsWith("accumulo.compactor.queue")) {
+ queueMetrics.add(TestStatsDSink.parseStatsDMetric(s));
+ }
+ }
+ }
+ });
+ metricsTailer.start();
+ }
+
+ @AfterEach
+ public void teardownMetricsTest() throws Exception {
+ shutdownTailer.set(true);
+ if (metricsTailer != null) {
+ metricsTailer.join();
+ }
+ }
+
+ private String getDir(String testName) throws Exception {
+ String dir = rootPath + testName + getUniqueNames(1)[0];
+ fs.delete(new Path(dir), true);
+ return dir;
+ }
+
+ @Override
+ protected Duration defaultTimeout() {
+ return Duration.ofMinutes(4);
+ }
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ sink = new TestStatsDSink();
+ SharedMiniClusterBase.startMiniClusterWithConfig(new CompactionPriorityQueueMetricsITConfig());
+ }
+
+ @AfterAll
+ public static void teardown() {
+ SharedMiniClusterBase.stopMiniCluster();
+ if (sink != null) {
+ sink.close();
+ }
+ }
+
+ public static class CompactionPriorityQueueMetricsITConfig
+ implements MiniClusterConfigurationCallback {
+ @Override
+ public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration conf) {
+ cfg.setMemory(ServerType.TABLET_SERVER, 512, MemoryUnit.MEGABYTE);
+ cfg.getClusterServerConfiguration().setNumDefaultCompactors(1);
+
+ // Create a new queue with zero compactors.
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + QUEUE1_SERVICE + ".planner",
+ RatioBasedCompactionPlanner.class.getName());
+ cfg.setProperty(
+ Property.COMPACTION_SERVICE_PREFIX.getKey() + QUEUE1_SERVICE + ".planner.opts.groups",
+ "[{'group':'" + QUEUE1 + "'}]");
+
+ cfg.setProperty(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_INITIAL_SIZE, "6");
+ cfg.getClusterServerConfiguration().addCompactorResourceGroup(QUEUE1, 0);
+
+ // This test waits for dead compactors to be absent in zookeeper. The following setting will
+ // make entries in ZK related to dead compactor processes expire sooner.
+ cfg.setProperty(Property.INSTANCE_ZK_TIMEOUT.getKey(), "10");
+
+ // use raw local file system
+ conf.set("fs.file.impl", RawLocalFileSystem.class.getName());
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY,
+ TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+ }
+ }
+
+ private String writeData(String file, AccumuloConfiguration aconf, int s, int e)
+ throws Exception {
+ FileSystem fs = getCluster().getFileSystem();
+ String filename = file + RFile.EXTENSION;
+ try (FileSKVWriter writer = FileOperations.getInstance().newWriterBuilder()
+ .forFile(UnreferencedTabletFile.of(fs, new Path(filename)), fs, fs.getConf(),
+ NoCryptoServiceFactory.NONE)
+ .withTableConfiguration(aconf).build()) {
+ writer.startDefaultLocalityGroup();
+ for (int i = s; i <= e; i++) {
+ writer.append(new Key(new Text(row(i))), new Value(Integer.toString(i)));
+ }
+ }
+
+ return hash(filename);
+ }
+
+ private void addSplits(AccumuloClient client, String tableName, String splitString)
+ throws Exception {
+ SortedSet<Text> splits = new TreeSet<>();
+ for (String split : splitString.split(" ")) {
+ splits.add(new Text(split));
+ }
+ client.tableOperations().addSplits(tableName, splits);
+ }
+
+ private void verifyData(AccumuloClient client, String table, int start, int end, boolean setTime)
+ throws Exception {
+ try (Scanner scanner = client.createScanner(table, Authorizations.EMPTY)) {
+
+ Iterator<Map.Entry<Key,Value>> iter = scanner.iterator();
+
+ for (int i = start; i <= end; i++) {
+ if (!iter.hasNext()) {
+ throw new Exception("row " + i + " not found");
+ }
+
+ Map.Entry<Key,Value> entry = iter.next();
+
+ String row = String.format("%04d", i);
+
+ if (!entry.getKey().getRow().equals(new Text(row))) {
+ throw new Exception("unexpected row " + entry.getKey() + " " + i);
+ }
+
+ if (Integer.parseInt(entry.getValue().toString()) != i) {
+ throw new Exception("unexpected value " + entry + " " + i);
+ }
+
+ if (setTime) {
+ assertEquals(1L, entry.getKey().getTimestamp());
+ }
+ }
+
+ if (iter.hasNext()) {
+ throw new Exception("found more than expected " + iter.next());
+ }
+ }
+ }
+
+ @SuppressFBWarnings(value = {"PATH_TRAVERSAL_IN", "WEAK_MESSAGE_DIGEST_SHA1"},
+ justification = "path provided by test; sha-1 is okay for test")
+ private String hash(String filename) {
+ try {
+ byte[] data = Files.readAllBytes(Paths.get(filename.replaceFirst("^file:", "")));
+ byte[] hash = MessageDigest.getInstance("SHA1").digest(data);
+ return new BigInteger(1, hash).toString(16);
+ } catch (IOException | NoSuchAlgorithmException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static String row(int r) {
+ return String.format("%04d", r);
+ }
+
+ @Test
+ public void testQueueMetrics() throws Exception {
+
+ long highestFileCount = 0L;
+ ServerContext context = getCluster().getServerContext();
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+
+ String dir = getDir("/testBulkFile-");
+ FileSystem fs = getCluster().getFileSystem();
+ fs.mkdirs(new Path(dir));
+
+ // Create splits so there are two groupings of tablets with similar file counts.
+ String splitString = "500 1000 1500 2000 3750 5500 7250 9000";
+ addSplits(c, tableName, splitString);
+
+ for (int i = 0; i < 100; i++) {
+ writeData(dir + "/f" + i + ".", aconf, i * 100, (i + 1) * 100 - 1);
+ }
+ c.tableOperations().importDirectory(dir).to(tableName).load();
+
+ IteratorSetting iterSetting = new IteratorSetting(100, CompactionIT.TestFilter.class);
+ iterSetting.addOption("expectedQ", QUEUE1);
+ iterSetting.addOption("modulus", 3 + "");
+ CompactionConfig config =
+ new CompactionConfig().setIterators(List.of(iterSetting)).setWait(false);
+ c.tableOperations().compact(tableName, config);
+
+ try (TabletsMetadata tm = context.getAmple().readTablets().forTable(tableId).build()) {
+ // Get each tablet's file sizes
+ for (TabletMetadata tablet : tm) {
+ long fileSize = tablet.getFiles().size();
+ log.info("Number of files in tablet {}: {}", tablet.getExtent().toString(), fileSize);
+ highestFileCount = Math.max(highestFileCount, fileSize);
+ }
+ }
+ verifyData(c, tableName, 0, 100 * 100 - 1, false);
+ }
+
+ boolean sawMetricsQ1 = false;
+ while (!sawMetricsQ1) {
+ while (!queueMetrics.isEmpty()) {
+ var qm = queueMetrics.take();
- if (qm.getName().contains(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED)
++ if (qm.getName().contains(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED.getName())
+ && qm.getTags().containsValue(QUEUE1_METRIC_LABEL)) {
+ if (Integer.parseInt(qm.getValue()) > 0) {
+ sawMetricsQ1 = true;
+ }
+ }
+ }
+ // If metrics are not found in the queue, sleep until the next poll.
+ UtilWaitThread.sleep(TestStatsDRegistryFactory.pollingFrequency.toMillis());
+ }
+
+ // Set lowest priority to the lowest possible system compaction priority
+ long lowestPriority = Short.MIN_VALUE;
+ long rejectedCount = 0L;
+ int queueSize = 0;
+
+ boolean sawQueues = false;
+ // An empty queue means that the last known value is the most recent.
+ while (!queueMetrics.isEmpty()) {
+ var metric = queueMetrics.take();
- if (metric.getName()
- .contains(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED)
++ if (metric.getName().contains(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED.getName())
+ && metric.getTags().containsValue(QUEUE1_METRIC_LABEL)) {
+ rejectedCount = Long.parseLong(metric.getValue());
- } else if (metric.getName()
- .contains(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY)
++ } else if (metric.getName().contains(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY.getName())
+ && metric.getTags().containsValue(QUEUE1_METRIC_LABEL)) {
+ lowestPriority = Math.max(lowestPriority, Long.parseLong(metric.getValue()));
- } else if (metric.getName()
- .contains(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH)
++ } else if (metric.getName().contains(COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH.getName())
+ && metric.getTags().containsValue(QUEUE1_METRIC_LABEL)) {
+ queueSize = Integer.parseInt(metric.getValue());
- } else if (metric.getName().contains(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUES)) {
++ } else if (metric.getName().contains(COMPACTOR_JOB_PRIORITY_QUEUES.getName())) {
+ sawQueues = true;
+ } else {
+ log.debug("{}", metric);
+ }
+ }
+
+ // Confirm metrics were generated and in some cases, validate contents.
+ assertTrue(rejectedCount > 0L);
+
+ // Priority is the file counts + number of compactions for that tablet.
+ // The lowestPriority job in the queue should have been
+ // at least 1 count higher than the highest file count.
+ short highestFileCountPrio = CompactionJobPrioritizer.createPriority(
+ getCluster().getServerContext().getTableId(tableName), CompactionKind.USER,
+ (int) highestFileCount, 0);
+ assertTrue(lowestPriority > highestFileCountPrio,
+ lowestPriority + " " + highestFileCount + " " + highestFileCountPrio);
+
+ // Multiple Queues have been created
+ assertTrue(sawQueues);
+
+ // Queue size matches the intended queue size
+ assertEquals(QUEUE1_SIZE, queueSize);
+
+ // Start Compactors
+ getCluster().getConfig().getClusterServerConfiguration().addCompactorResourceGroup(QUEUE1, 1);
+ getCluster().getClusterControl().start(ServerType.COMPACTOR);
+
+ boolean emptyQueue = false;
+
+ // Make sure that metrics added to the queue are recent
+ UtilWaitThread.sleep(TestStatsDRegistryFactory.pollingFrequency.toMillis());
+
+ while (!emptyQueue) {
+ while (!queueMetrics.isEmpty()) {
+ var metric = queueMetrics.take();
- if (metric.getName()
- .contains(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED)
++ if (metric.getName().contains(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED.getName())
+ && metric.getTags().containsValue(QUEUE1_METRIC_LABEL)) {
+ if (Integer.parseInt(metric.getValue()) == 0) {
+ emptyQueue = true;
+ }
+ }
+
+ // Check if the total number of queues is zero, if so then will not see metrics for the
+ // above queue.
- if (metric.getName().equals(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUES)) {
++ if (metric.getName().equals(COMPACTOR_JOB_PRIORITY_QUEUES.getName())) {
+ if (Integer.parseInt(metric.getValue()) == 0) {
+ emptyQueue = true;
+ }
+ }
+ }
+ UtilWaitThread.sleep(TestStatsDRegistryFactory.pollingFrequency.toMillis());
+ }
+ }
+
+ /**
+ * Test that the compaction queue is cleared when compactions no longer need to happen.
+ */
+ @Test
+ public void testCompactionQueueClearedWhenNotNeeded() throws Exception {
+ ServerContext context = getCluster().getServerContext();
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+
+ String dir = getDir("/testBulkFile-");
+ FileSystem fs = getCluster().getFileSystem();
+ fs.mkdirs(new Path(dir));
+
+ // Create splits so there are two groupings of tablets with similar file counts.
+ String splitString = "500 1000 1500 2000 3750 5500 7250 9000";
+ addSplits(c, tableName, splitString);
+
+ for (int i = 0; i < 100; i++) {
+ writeData(dir + "/f" + i + ".", aconf, i * 100, (i + 1) * 100 - 1);
+ }
+ c.tableOperations().importDirectory(dir).to(tableName).load();
+ verifyData(c, tableName, 0, 100 * 100 - 1, false);
+ }
+
+ final long sleepMillis = TestStatsDRegistryFactory.pollingFrequency.toMillis();
+
+ // wait for compaction jobs to be queued
+ Wait.waitFor(() -> getJobsQueued() > 0, 60_000, sleepMillis,
+ "Expected to see compaction jobs queued");
+
+ // change compactor settings so that compactions no longer need to run
+ context.tableOperations().setProperty(tableName, Property.TABLE_MAJC_RATIO.getKey(), "2000");
+ context.tableOperations().setProperty(tableName, Property.TABLE_FILE_MAX.getKey(), "2000");
+
+ // wait for queue to clear
+ Wait.waitFor(() -> getJobsQueued() == 0, 60_000, sleepMillis,
+ "Expected job queue to be cleared once compactions no longer need to happen");
+ }
+
+ /**
+ * @return the number of jobs queued in the compaction queue. Returns -1 if no metrics are found.
+ */
+ private int getJobsQueued() throws InterruptedException {
+ Integer jobsQueued = null;
+ while (!queueMetrics.isEmpty()) {
+ var metric = queueMetrics.take();
- if (metric.getName()
- .contains(MetricsProducer.METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED)
++ if (metric.getName().contains(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED.getName())
+ && metric.getTags().containsValue(QUEUE1_METRIC_LABEL)) {
+ jobsQueued = Integer.parseInt(metric.getValue());
+ }
+ }
+ if (jobsQueued == null) {
+ log.warn("No compaction job queue metrics found.");
+ return -1;
+ }
+ log.info("Jobs Queued: {}", jobsQueued);
+ return jobsQueued;
+ }
+
+}
diff --cc test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
index 26f6f90152,3ebd198b70..05219adfdf
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
@@@ -139,7 -127,7 +138,7 @@@ public class ExternalCompactionMetricsI
if (shutdownTailer.get()) {
break;
}
- if (s.startsWith(MetricsProducer.METRICS_COMPACTOR_PREFIX)) {
- if (s.startsWith(MAJC_QUEUED.getName())) {
++ if (s.startsWith("accumulo.tserver.compactions.")) {
queueMetrics.add(TestStatsDSink.parseStatsDMetric(s));
}
}
diff --cc test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java
index 515c67ec57,278c8a4e96..189ec85748
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java
@@@ -20,7 -20,9 +20,9 @@@ package org.apache.accumulo.test.compac
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_ENTRIES_READ;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_ENTRIES_WRITTEN;
-import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE1;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.GROUP1;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.compact;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.createTable;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.getRunningCompactions;
@@@ -52,8 -56,7 +54,7 @@@ import org.apache.accumulo.core.data.Ta
import org.apache.accumulo.core.iterators.IteratorUtil;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
- import org.apache.accumulo.core.metrics.MetricsProducer;
-import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil;
import org.apache.accumulo.core.util.compaction.RunningCompactionInfo;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.harness.AccumuloClusterHarness;
diff --cc test/src/main/java/org/apache/accumulo/test/functional/IdleProcessMetricsIT.java
index 8c04b63358,8ffa54fa5f..60896116bf
--- a/test/src/main/java/org/apache/accumulo/test/functional/IdleProcessMetricsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/IdleProcessMetricsIT.java
@@@ -18,8 -18,9 +18,9 @@@
*/
package org.apache.accumulo.test.functional;
+ import static org.apache.accumulo.core.metrics.Metric.SERVER_IDLE;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.GROUP1;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.MAX_DATA;
-import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE1;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.compact;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.createTable;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.verify;
@@@ -41,9 -43,7 +42,8 @@@ import org.apache.accumulo.core.client.
import org.apache.accumulo.core.client.ScannerBase;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.iterators.IteratorUtil;
- import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.spi.compaction.RatioBasedCompactionPlanner;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
import org.apache.accumulo.minicluster.ServerType;
@@@ -147,12 -126,10 +147,12 @@@ public class IdleProcessMetricsIT exten
AtomicBoolean sawCompactor = new AtomicBoolean(false);
AtomicBoolean sawSServer = new AtomicBoolean(false);
AtomicBoolean sawTServer = new AtomicBoolean(false);
+
Wait.waitFor(() -> {
List<String> statsDMetrics = sink.getLines();
- statsDMetrics.stream().filter(line -> line.startsWith(MetricsProducer.METRICS_SERVER_IDLE))
- .peek(log::info).map(TestStatsDSink::parseStatsDMetric)
+ statsDMetrics.stream().filter(line -> line.startsWith(SERVER_IDLE.getName())).peek(log::info)
- .map(TestStatsDSink::parseStatsDMetric).forEach(a -> {
++ .map(TestStatsDSink::parseStatsDMetric)
+ .filter(a -> a.getTags().get("resource.group").equals(IDLE_RESOURCE_GROUP)).forEach(a -> {
String processName = a.getTags().get("process.name");
int value = Integer.parseInt(a.getValue());
assertTrue(value == 0 || value == 1 || value == -1, "Unexpected value " + value);
diff --cc test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java
index 6bc1b5f109,eeb95017a0..8f9c38af3c
--- a/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java
@@@ -30,18 -29,15 +31,17 @@@ import java.util.Set
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.DoubleAdder;
+import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
-import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.admin.CompactionConfig;
import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.Property;
- import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
-import org.apache.accumulo.minicluster.MemoryUnit;
import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
@@@ -161,37 -137,25 +161,37 @@@ public class MemoryStarvedMajCIT extend
}
});
- int paused = MAJC_PAUSED.intValue();
- try (Scanner scanner = client.createScanner(table)) {
++ int paused = MAJC_PAUSED_COUNT.intValue();
+ assertEquals(0, paused);
- MemoryStarvedScanIT.consumeServerMemory(scanner);
+ // Calling getRunningCompaction on the MemoryConsumingCompactor
+ // will consume the free memory
+ LOG.info("Calling getRunningCompaction on {}", compactorAddr);
+ boolean success = false;
+ while (!success) {
+ try {
+ ExternalCompactionUtil.getRunningCompaction(compactorAddr, ctx);
+ success = true;
+ } catch (Exception e) {
+ UtilWaitThread.sleep(3000);
+ }
+ }
- int paused = MAJC_PAUSED_COUNT.intValue();
- assertEquals(0, paused);
+ ReadWriteIT.ingest(client, 100, 100, 100, 0, table);
+ compactionThread.start();
- waitFor(() -> MAJC_PAUSED.intValue() > 0);
- ReadWriteIT.ingest(client, 100, 100, 100, 0, table);
- compactionThread.start();
++ waitFor(() -> MAJC_PAUSED_COUNT.intValue() > 0);
- waitFor(() -> MAJC_PAUSED_COUNT.intValue() > 0);
+ // Calling cancel on the MemoryConsumingCompactor will free
+ // the consumed memory
+ LOG.info("Calling cancel on {}", compactorAddr);
+ ExternalCompactionUtil.cancelCompaction(ctx, compactorAddr, "fakeECID");
- MemoryStarvedScanIT.freeServerMemory(client);
- compactionThread.interrupt();
- compactionThread.join();
- assertNull(error.get());
- assertTrue(client.instanceOperations().getActiveCompactions().stream()
- .anyMatch(ac -> ac.getPausedCount() > 0));
- }
+ compactionThread.interrupt();
+ compactionThread.join();
+ assertNull(error.get());
+ assertTrue(client.instanceOperations().getActiveCompactions().stream()
+ .anyMatch(ac -> ac.getPausedCount() > 0));
}
}
diff --cc test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedScanIT.java
index 0c9b08e0ab,8e4da36bb9..3a12501b83
--- a/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedScanIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedScanIT.java
@@@ -48,14 -45,7 +50,13 @@@ import org.apache.accumulo.core.conf.Pr
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.iterators.WrappingIterator;
+import org.apache.accumulo.core.lock.ServiceLock;
+import org.apache.accumulo.core.lock.ServiceLockData;
+import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
- import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.rpc.ThriftUtil;
+import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.spi.metrics.LoggingMeterRegistryFactory;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
diff --cc test/src/main/java/org/apache/accumulo/test/functional/OnDemandTabletUnloadingIT.java
index b950c6b120,0000000000..b06da7b288
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/OnDemandTabletUnloadingIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/OnDemandTabletUnloadingIT.java
@@@ -1,275 -1,0 +1,275 @@@
+/*
+ * 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
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.functional;
+
++import static org.apache.accumulo.core.metrics.Metric.TSERVER_TABLETS_ONLINE_ONDEMAND;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.stream.IntStream;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.ScannerBase;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.ClientTabletCache;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
- import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.spi.scan.ConfigurableScanServerSelector;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.util.Wait;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class OnDemandTabletUnloadingIT extends SharedMiniClusterBase {
+
+ private static final int managerTabletGroupWatcherInterval = 5;
+ private static final int inactiveOnDemandTabletUnloaderInterval = 10;
+ private static TestStatsDSink sink;
+ private static Thread metricConsumer;
+ private static Long ONDEMAND_ONLINE_COUNT = 0L;
+
+ @BeforeAll
+ public static void beforeAll() throws Exception {
+
+ sink = new TestStatsDSink();
+ metricConsumer = new Thread(() -> {
+ while (!Thread.currentThread().isInterrupted()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String line : statsDMetrics) {
+ if (Thread.currentThread().isInterrupted()) {
+ break;
+ }
+ if (line.startsWith("accumulo")) {
+ TestStatsDSink.Metric metric = TestStatsDSink.parseStatsDMetric(line);
- if (MetricsProducer.METRICS_TSERVER_TABLETS_ONLINE_ONDEMAND.equals(metric.getName())) {
++ if (TSERVER_TABLETS_ONLINE_ONDEMAND.getName().equals(metric.getName())) {
+ Long val = Long.parseLong(metric.getValue());
+ System.out.println(val);
+ ONDEMAND_ONLINE_COUNT = val;
+ }
+ }
+ }
+ }
+ });
+ metricConsumer.start();
+
+ SharedMiniClusterBase.startMiniClusterWithConfig((cfg, core) -> {
+ cfg.getClusterServerConfiguration().setNumDefaultScanServers(1);
+ cfg.getClusterServerConfiguration().setNumDefaultTabletServers(1);
+ cfg.setProperty(Property.MANAGER_TABLET_GROUP_WATCHER_INTERVAL,
+ Integer.toString(managerTabletGroupWatcherInterval));
+ cfg.setProperty(Property.TSERV_ONDEMAND_UNLOADER_INTERVAL,
+ Integer.toString(inactiveOnDemandTabletUnloaderInterval));
+ cfg.setProperty("table.custom.ondemand.unloader.inactivity.threshold.seconds", "15");
+
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY,
+ TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+
+ });
+ }
+
+ @AfterAll
+ public static void after() throws Exception {
+ sink.close();
+ metricConsumer.interrupt();
+ metricConsumer.join();
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ // wait for tablets from previous test to be unloaded
+ Wait.waitFor(() -> ONDEMAND_ONLINE_COUNT == 0);
+ }
+
+ @Test
+ public void testTabletUnloader() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+
+ String tableName = super.getUniqueNames(1)[0];
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ ntc.withSplits(splits);
+ c.tableOperations().create(tableName, ntc);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+
+ // There should be no tablets online
+ List<TabletStats> stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ assertEquals(0, stats.size());
+ assertEquals(0, ClientTabletCache.getInstance((ClientContext) c, TableId.of(tableId))
+ .getTabletHostingRequestCount());
+ assertEquals(0, ONDEMAND_ONLINE_COUNT);
+
+ // loading data will cause tablets to be hosted
+ ManagerAssignmentIT.loadDataForScan(c, tableName);
+ verifyDataForScan(c, tableName);
+
+ // There should be four tablets online
+ stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ assertEquals(4, stats.size());
+ assertTrue(ClientTabletCache.getInstance((ClientContext) c, TableId.of(tableId))
+ .getTabletHostingRequestCount() > 0);
+ Wait.waitFor(() -> ONDEMAND_ONLINE_COUNT == 4);
+
+ // Waiting for tablets to be unloaded due to inactivity
+ Wait.waitFor(() -> ONDEMAND_ONLINE_COUNT == 0);
+ Wait.waitFor(() -> ManagerAssignmentIT.getTabletStats(c, tableId).size() == 0);
+ }
+ }
+
+ private static void verifyDataForScan(AccumuloClient c, String tableName)
+ throws TableNotFoundException, AccumuloSecurityException, AccumuloException {
+
+ Set<Integer> rows = new HashSet<>();
+ IntStream.range(97, 122).forEach(i -> rows.add(i));
+
+ try (Scanner s = c.createScanner(tableName)) {
+ s.iterator().forEachRemaining((e) -> rows.remove(e.getKey().getRow().charAt(0)));
+ }
+ assertEquals(0, rows.size());
+ }
+
+ public long countTabletsWithLocation(AccumuloClient client, String tableName) throws Exception {
+ var ctx = (ClientContext) client;
+ try (var tablets = ctx.getAmple().readTablets().forTable(ctx.getTableId(tableName)).build()) {
+ return tablets.stream().map(TabletMetadata::getLocation).filter(Objects::nonNull)
+ .filter(loc -> loc.getType() == TabletMetadata.LocationType.CURRENT).count();
+ }
+ }
+
+ public static final String SCAN_SERVER_SELECTOR_CONFIG =
+ "[{'isDefault':true,'maxBusyTimeout':'5m',"
+ + "'busyTimeoutMultiplier':8, 'scanTypeActivations':[], 'timeToWaitForScanServers':'120s',"
+ + "'attemptPlans':[{'servers':'3', 'busyTimeout':'33ms', 'salt':'one'},"
+ + "{'servers':'13', 'busyTimeout':'33ms', 'salt':'two'},"
+ + "{'servers':'100%', 'busyTimeout':'33ms'}]}]";
+
+ @Test
+ public void testScanHostedAndUnhosted() throws Exception {
+ String tableName = super.getUniqueNames(1)[0];
+
+ // create configuration that makes eventual scans wait for scan servers when there are none
+ // instead of falling back to the tserver.
+ var clientProps = new Properties();
+ clientProps.putAll(getClientProps());
+ String scanServerSelectorProfiles = "[{'isDefault':true,'maxBusyTimeout':'5m',"
+ + "'busyTimeoutMultiplier':8, 'scanTypeActivations':[], 'timeToWaitForScanServers':'120s',"
+ + "'attemptPlans':[{'servers':'3', 'busyTimeout':'1s'}]}]";
+ clientProps.put("scan.server.selector.impl", ConfigurableScanServerSelector.class.getName());
+ clientProps.put("scan.server.selector.opts.profiles",
+ scanServerSelectorProfiles.replace("'", "\""));
+
+ try (AccumuloClient c = Accumulo.newClient().from(clientProps).build()) {
+
+ SortedSet<Text> splits = new TreeSet<>(
+ List.of(new Text("005"), new Text("013"), new Text("027"), new Text("075")));
+ c.tableOperations().create(tableName, new NewTableConfiguration().withSplits(splits));
+ try (var writer = c.createBatchWriter(tableName)) {
+ IntStream.range(0, 100).mapToObj(i -> String.format("%03d", i)).forEach(row -> {
+ Mutation m = new Mutation(row);
+ m.put("", "", "");
+ try {
+ writer.addMutation(m);
+ } catch (MutationsRejectedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ c.tableOperations().flush(tableName, null, null, true);
+
+ // wait for all tablets to be unhosted
+ Wait.waitFor(() -> countTabletsWithLocation(c, tableName) == 0);
+
+ // scan a subset of the table causing some tablets to be hosted
+ try (var scanner = c.createScanner(tableName)) {
+ scanner.setRange(new Range("050", null));
+ assertEquals(50, scanner.stream().count());
+ }
+
+ // the above scan should only cause two tablets to be hosted so check this
+ assertTrue(countTabletsWithLocation(c, tableName) <= 2);
+
+ getCluster().getClusterControl().start(ServerType.SCAN_SERVER, "localhost");
+
+ // the cache will probably have locations for tablets, want to ensure cache can handle a mix
+ // of tablets with and without locations
+ c.tableOperations().clearLocatorCache(tableName);
+
+ // Scan should only use scan servers and should scan tablets with and without locations.
+ try (var scanner = c.createScanner(tableName)) {
+ scanner.setConsistencyLevel(ScannerBase.ConsistencyLevel.EVENTUAL);
+ assertEquals(100, scanner.stream().count());
+ }
+
+ try (var scanner = c.createBatchScanner(tableName)) {
+ scanner.setConsistencyLevel(ScannerBase.ConsistencyLevel.EVENTUAL);
+ scanner.setRanges(List.of(new Range()));
+ assertEquals(100, scanner.stream().count());
+ }
+
+ // ensure tablets without a location were not brought online by the eventual scan
+ assertTrue(countTabletsWithLocation(c, tableName) <= 2);
+
+ // esnure an immediate scans works when the cache contains tablets w/ and w/o locations
+ try (var scanner = c.createScanner(tableName)) {
+ scanner.setConsistencyLevel(ScannerBase.ConsistencyLevel.IMMEDIATE);
+ assertEquals(100, scanner.stream().count());
+ }
+
+ // delete table to unhost any tablets so they do not show up in counts for the next test
+ c.tableOperations().delete(tableName);
+ }
+ }
+}
diff --cc test/src/main/java/org/apache/accumulo/test/metrics/MetricsIT.java
index 529c116cdc,c77c41d049..42706273f6
--- a/test/src/main/java/org/apache/accumulo/test/metrics/MetricsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/metrics/MetricsIT.java
@@@ -18,7 -18,16 +18,17 @@@
*/
package org.apache.accumulo.test.metrics;
+ import static org.apache.accumulo.core.metrics.Metric.COMPACTOR_MAJC_STUCK;
+ import static org.apache.accumulo.core.metrics.Metric.FATE_TYPE_IN_PROGRESS;
+ import static org.apache.accumulo.core.metrics.Metric.MANAGER_BALANCER_MIGRATIONS_NEEDED;
+ import static org.apache.accumulo.core.metrics.Metric.SCAN_BUSY_TIMEOUT_COUNT;
+ import static org.apache.accumulo.core.metrics.Metric.SCAN_RESERVATION_CONFLICT_COUNTER;
+ import static org.apache.accumulo.core.metrics.Metric.SCAN_RESERVATION_TOTAL_TIMER;
+ import static org.apache.accumulo.core.metrics.Metric.SCAN_RESERVATION_WRITEOUT_TIMER;
+ import static org.apache.accumulo.core.metrics.Metric.SCAN_TABLET_METADATA_CACHE;
+ import static org.apache.accumulo.core.metrics.Metric.SCAN_YIELDS;
+ import static org.apache.accumulo.core.metrics.Metric.SERVER_IDLE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@@ -27,15 -35,13 +37,15 @@@ import static org.junit.jupiter.api.Ass
import static org.junit.jupiter.api.Assertions.fail;
import java.time.Duration;
+ import java.util.Arrays;
+import java.util.BitSet;
- import java.util.HashMap;
+ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicReference;
- import java.util.function.Predicate;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
@@@ -47,14 -52,11 +57,14 @@@ import org.apache.accumulo.core.client.
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.ReadOnlyFateStore.TStatus;
+ import org.apache.accumulo.core.metrics.Metric;
import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.core.spi.metrics.LoggingMeterRegistryFactory;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
import org.apache.accumulo.test.functional.ConfigurableMacBase;
+import org.apache.accumulo.test.functional.SlowIterator;
- import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.AfterAll;
@@@ -105,104 -106,59 +115,109 @@@ public class MetricsIT extends Configur
@Test
public void confirmMetricsPublished() throws Exception {
- doWorkToGenerateMetrics();
- cluster.stop();
// meter names sorted and formatting disabled to make it easier to diff changes
// @formatter:off
- Set<String> unexpectedMetrics =
- Set.of(METRICS_COMPACTOR_MAJC_STUCK,
- METRICS_SCAN_YIELDS);
+ Set<Metric> unexpectedMetrics = Set.of(
+ COMPACTOR_MAJC_STUCK,
+ SCAN_YIELDS
+ );
// add sserver as flaky until scan server included in mini tests.
- Set<String> flakyMetrics = Set.of(METRICS_FATE_TYPE_IN_PROGRESS,
- METRICS_MANAGER_BALANCER_MIGRATIONS_NEEDED,
- METRICS_SCAN_BUSY_TIMEOUT_COUNTER,
- METRICS_SCAN_RESERVATION_CONFLICT_COUNTER,
- METRICS_SCAN_RESERVATION_TOTAL_TIMER,
- METRICS_SCAN_RESERVATION_WRITEOUT_TIMER,
- METRICS_SCAN_TABLET_METADATA_CACHE,
- METRICS_SERVER_IDLE);
+ Set<Metric> flakyMetrics = Set.of(
+ FATE_TYPE_IN_PROGRESS,
+ MANAGER_BALANCER_MIGRATIONS_NEEDED,
+ SCAN_BUSY_TIMEOUT_COUNT,
+ SCAN_RESERVATION_CONFLICT_COUNTER,
+ SCAN_RESERVATION_TOTAL_TIMER,
+ SCAN_RESERVATION_WRITEOUT_TIMER,
+ SCAN_TABLET_METADATA_CACHE,
+ SERVER_IDLE
+ );
// @formatter:on
- Map<String,String> expectedMetricNames = this.getMetricFields();
- flakyMetrics.forEach(expectedMetricNames::remove); // might not see these
- unexpectedMetrics.forEach(expectedMetricNames::remove); // definitely shouldn't see these
- assertFalse(expectedMetricNames.isEmpty()); // make sure we didn't remove everything
+ Set<Metric> expectedMetrics = new HashSet<>(Arrays.asList(Metric.values()));
+ expectedMetrics.removeAll(flakyMetrics); // might not see these
+ expectedMetrics.removeAll(unexpectedMetrics); // definitely shouldn't see these
+ assertFalse(expectedMetrics.isEmpty()); // make sure we didn't remove everything
- Map<String,String> seenMetricNames = new HashMap<>();
+ Set<Metric> seenMetrics = new HashSet<>();
List<String> statsDMetrics;
+ final int compactionPriorityQueueLengthBit = 0;
+ final int compactionPriorityQueueQueuedBit = 1;
+ final int compactionPriorityQueueDequeuedBit = 2;
+ final int compactionPriorityQueueRejectedBit = 3;
+ final int compactionPriorityQueuePriorityBit = 4;
+
+ final BitSet trueSet = new BitSet(5);
+ trueSet.set(0, 4, true);
+
+ final BitSet queueMetricsSeen = new BitSet(5);
+
+ AtomicReference<Exception> error = new AtomicReference<>();
+ Thread workerThread = new Thread(() -> {
+ try {
+ doWorkToGenerateMetrics();
+ } catch (Exception e) {
+ error.set(e);
+ }
+ });
+ workerThread.start();
+
// loop until we run out of lines or until we see all expected metrics
- while (!(statsDMetrics = sink.getLines()).isEmpty() && !expectedMetricNames.isEmpty()
- while (!(statsDMetrics = sink.getLines()).isEmpty() && !expectedMetrics.isEmpty()) {
++ while (!(statsDMetrics = sink.getLines()).isEmpty() && !expectedMetrics.isEmpty()
+ && !queueMetricsSeen.intersects(trueSet)) {
// for each metric name not yet seen, check if it is expected, flaky, or unknown
statsDMetrics.stream().filter(line -> line.startsWith("accumulo"))
- .map(TestStatsDSink::parseStatsDMetric).map(Metric::getName)
- .filter(Predicate.not(seenMetricNames::containsKey)).forEach(name -> {
- if (expectedMetricNames.containsKey(name)) {
- // record expected metric names as seen, along with the value seen
- seenMetricNames.put(name, expectedMetricNames.remove(name));
- } else if (flakyMetrics.contains(name)) {
+ .map(TestStatsDSink::parseStatsDMetric).map(metric -> Metric.fromName(metric.getName()))
+ .filter(metric -> !seenMetrics.contains(metric)).forEach(metric -> {
+ if (expectedMetrics.contains(metric)) {
+ // record expected Metric as seen
+ seenMetrics.add(metric);
+ expectedMetrics.remove(metric);
+ } else if (flakyMetrics.contains(metric)) {
// ignore any flaky metric names seen
// these aren't always expected, but we shouldn't be surprised if we see them
- } else if (name.startsWith(METRICS_COMPACTOR_PREFIX)) {
++ } else if (metric.getName().startsWith("accumulo.compactor.")) {
+ // Compactor queue metrics are not guaranteed to be emitted
+ // during the call to doWorkToGenerateMetrics above. This will
+ // flip a bit in the BitSet when each metric is seen. The top-level
+ // loop will continue to iterate until all the metrics are seen.
- seenMetricNames.put(name, expectedMetricNames.remove(name));
- switch (name) {
- case METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH:
++ seenMetrics.add(metric);
++ expectedMetrics.remove(metric);
++ switch (metric) {
++ case COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH:
+ queueMetricsSeen.set(compactionPriorityQueueLengthBit, true);
+ break;
- case METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED:
++ case COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED:
+ queueMetricsSeen.set(compactionPriorityQueueQueuedBit, true);
+ break;
- case METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED:
++ case COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED:
+ queueMetricsSeen.set(compactionPriorityQueueDequeuedBit, true);
+ break;
- case METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED:
++ case COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED:
+ queueMetricsSeen.set(compactionPriorityQueueRejectedBit, true);
+ break;
- case METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY:
++ case COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY:
+ queueMetricsSeen.set(compactionPriorityQueuePriorityBit, true);
+ break;
+ }
} else {
// completely unexpected metric
- fail("Found accumulo metric not in expectedMetricNames or flakyMetricNames: " + name);
+ fail("Found accumulo metric not in expectedMetricNames or flakyMetricNames: "
+ + metric);
}
});
- log.debug("METRICS: metrics expected, but not seen so far: {}", expectedMetricNames);
++ log.debug("METRICS: metrics expected, but not seen so far: {}", expectedMetrics);
+ Thread.sleep(4_000);
}
- assertTrue(expectedMetricNames.isEmpty(),
- "Did not see all expected metric names, missing: " + expectedMetricNames.values());
+ assertTrue(expectedMetrics.isEmpty(),
+ "Did not see all expected metric names, missing: " + expectedMetrics);
+
+ workerThread.join();
+ assertNull(error.get());
+ cluster.stop();
-
}
private void doWorkToGenerateMetrics() throws Exception {