This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new bfbda8b9ef Created constant for queue.id tag key (#6337)
bfbda8b9ef is described below
commit bfbda8b9efe3afad57e1fbe0927a67b2194867ee
Author: Dave Marion <[email protected]>
AuthorDate: Fri Apr 24 14:21:51 2026 -0400
Created constant for queue.id tag key (#6337)
Closes #6335
---
.../java/org/apache/accumulo/core/metrics/Metric.java | 6 ++++--
.../org/apache/accumulo/core/metrics/MetricsInfo.java | 5 +----
.../server/metrics/MetricResponseWrapper.java | 17 +++++++++++++++--
.../java/org/apache/accumulo/compactor/Compactor.java | 18 +++++++++---------
.../manager/compaction/coordinator/QueueMetrics.java | 19 ++++++++++---------
.../accumulo/monitor/next/SystemInformation.java | 3 ++-
.../test/compaction/ExternalCompactionMetricsIT.java | 5 +++--
7 files changed, 44 insertions(+), 29 deletions(-)
diff --git a/core/src/main/java/org/apache/accumulo/core/metrics/Metric.java
b/core/src/main/java/org/apache/accumulo/core/metrics/Metric.java
index c00c1fb16e..ce55be21d9 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
@@ -493,8 +493,10 @@ public enum Metric {
GENERAL_SERVER("General Server Metrics", "Metrics that are generated
across all server types."),
COMPACTION("Compaction Metrics",
"Metrics specific to compactions, both minor and major. Metrics for
major compactions"
- + " will likely have a 'queue.id' tag. The CompactionCoordinator
component in the Manager creates a queue for each CompactionService"
- + " in the configuration. The 'queue.id' tag may map directly to
the name of a Compactor resource group."),
+ + " will likely have a '" + MetricsInfo.QUEUE_TAG_KEY
+ + "' tag. The CompactionCoordinator component in the Manager
creates a queue for each CompactionService"
+ + " in the configuration. The '" + MetricsInfo.QUEUE_TAG_KEY
+ + "' tag may map directly to the name of a Compactor resource
group."),
COMPACTOR("Compactor Metrics", "Metrics that are generated by the
Compactor processes."),
FATE("Fate Metrics",
"Metrics that are generated by the Fate component in the Manager
process."),
diff --git
a/core/src/main/java/org/apache/accumulo/core/metrics/MetricsInfo.java
b/core/src/main/java/org/apache/accumulo/core/metrics/MetricsInfo.java
index 0b689b8e43..8045ca50c7 100644
--- a/core/src/main/java/org/apache/accumulo/core/metrics/MetricsInfo.java
+++ b/core/src/main/java/org/apache/accumulo/core/metrics/MetricsInfo.java
@@ -23,7 +23,6 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
-import java.util.Set;
import org.apache.accumulo.core.data.ResourceGroupId;
@@ -38,9 +37,7 @@ public interface MetricsInfo {
String RESOURCE_GROUP_TAG_KEY = "resource.group";
String HOST_TAG_KEY = "host";
String PORT_TAG_KEY = "port";
-
- Set<String> allTags = Set.of(INSTANCE_NAME_TAG_KEY, PROCESS_NAME_TAG_KEY,
RESOURCE_GROUP_TAG_KEY,
- HOST_TAG_KEY, PORT_TAG_KEY);
+ String QUEUE_TAG_KEY = "queue.id";
/**
* Convenience method to create tag name / value pair for the instance name
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricResponseWrapper.java
b/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricResponseWrapper.java
index 2c7f8e8bc6..4eff86464d 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricResponseWrapper.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/metrics/MetricResponseWrapper.java
@@ -18,14 +18,20 @@
*/
package org.apache.accumulo.server.metrics;
+import static org.apache.accumulo.core.metrics.MetricsInfo.HOST_TAG_KEY;
+import static
org.apache.accumulo.core.metrics.MetricsInfo.INSTANCE_NAME_TAG_KEY;
+import static org.apache.accumulo.core.metrics.MetricsInfo.PORT_TAG_KEY;
+import static
org.apache.accumulo.core.metrics.MetricsInfo.PROCESS_NAME_TAG_KEY;
+import static
org.apache.accumulo.core.metrics.MetricsInfo.RESOURCE_GROUP_TAG_KEY;
+
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import org.apache.accumulo.core.metrics.MetricsInfo;
import org.apache.accumulo.core.metrics.flatbuffers.FMetric;
import org.apache.accumulo.core.metrics.flatbuffers.FTag;
import org.apache.accumulo.core.process.thrift.MetricResponse;
@@ -48,6 +54,13 @@ import
io.micrometer.core.instrument.distribution.ValueAtPercentile;
*/
public class MetricResponseWrapper extends MetricResponse {
+ /**
+ * Set of tags that may be found on the metrics. These tags can be removed
because this
+ * information is already set on the MetricResponse object.
+ */
+ private static final Set<String> DUPLICATE_TAGS =
Set.of(INSTANCE_NAME_TAG_KEY,
+ PROCESS_NAME_TAG_KEY, RESOURCE_GROUP_TAG_KEY, HOST_TAG_KEY,
PORT_TAG_KEY);
+
private static class CommonRefs {
int nameRef;
int typeRef;
@@ -77,7 +90,7 @@ public class MetricResponseWrapper extends MetricResponse {
*/
private List<Tag> reduceTags(List<Tag> tags, List<Tag> extraTags) {
return Stream.concat(tags.stream(), extraTags.stream())
- .filter(t ->
!MetricsInfo.allTags.contains(t.getKey())).collect(Collectors.toList());
+ .filter(t ->
!DUPLICATE_TAGS.contains(t.getKey())).collect(Collectors.toList());
}
private void parseAndCreateCommonInfo(Meter.Id id, List<Tag> extraTags) {
diff --git
a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
index 4a8df5602b..8799e25090 100644
---
a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
+++
b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
@@ -276,35 +276,35 @@ public class Compactor extends AbstractServer implements
MetricsProducer, Compac
final String rgName =
MetricsUtil.formatString(getResourceGroup().canonical());
FunctionCounter.builder(COMPACTOR_ENTRIES_READ.getName(), this,
Compactor::getTotalEntriesRead)
.description(COMPACTOR_ENTRIES_READ.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
FunctionCounter
.builder(COMPACTOR_ENTRIES_WRITTEN.getName(), this,
Compactor::getTotalEntriesWritten)
.description(COMPACTOR_ENTRIES_WRITTEN.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
Gauge.builder(COMPACTOR_MAJC_IN_PROGRESS.getName(), this,
Compactor::compactionInProgress)
.description(COMPACTOR_MAJC_IN_PROGRESS.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
LongTaskTimer timer = LongTaskTimer.builder(COMPACTOR_MAJC_STUCK.getName())
.description(COMPACTOR_MAJC_STUCK.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
FunctionCounter.builder(COMPACTOR_MAJC_CANCELLED.getName(), this,
Compactor::getCancellations)
.description(COMPACTOR_MAJC_CANCELLED.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
FunctionCounter.builder(COMPACTOR_MAJC_COMPLETED.getName(), this,
Compactor::getCompletions)
.description(COMPACTOR_MAJC_COMPLETED.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
FunctionCounter.builder(COMPACTOR_MAJC_FAILED.getName(), this,
Compactor::getFailures)
.description(COMPACTOR_MAJC_FAILED.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
FunctionCounter
.builder(COMPACTOR_MAJC_FAILURES_TERMINATION.getName(), this,
Compactor::getTerminated)
.description(COMPACTOR_MAJC_FAILURES_TERMINATION.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
Gauge
.builder(COMPACTOR_MAJC_FAILURES_CONSECUTIVE.getName(), this,
Compactor::getConsecutiveFailures)
.description(COMPACTOR_MAJC_FAILURES_CONSECUTIVE.getDescription())
- .tags(List.of(Tag.of("queue.id", rgName))).register(registry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
rgName))).register(registry);
CompactionWatcher.setTimer(timer);
}
diff --git
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
index a6f0e34a09..756dff5d1c 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
@@ -39,6 +39,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.data.ResourceGroupId;
+import org.apache.accumulo.core.metrics.MetricsInfo;
import org.apache.accumulo.core.metrics.MetricsProducer;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue;
@@ -75,53 +76,53 @@ public class QueueMetrics implements MetricsProducer {
.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED.getName(), queue,
q -> q.getQueuedJobs())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsQueuedSize = Gauge
.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_SIZE.getName(), queue,
q -> q.getQueuedJobsSize())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_SIZE.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsDequeued = Gauge
.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED.getName(), queue,
q -> q.getDequeuedJobs())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_DEQUEUED.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsRejected = Gauge
.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED.getName(), queue,
q -> q.getRejectedJobs())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_REJECTED.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsLowestPriority = Gauge
.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY.getName(), queue,
q -> q.getLowestPriority())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_PRIORITY.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsMinAge = Gauge
.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_MIN_AGE.getName(), queue,
q -> q.getJobQueueStats().getMinAge().toSeconds())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_MIN_AGE.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsMaxAge = Gauge
.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_MAX_AGE.getName(), queue,
q -> q.getJobQueueStats().getMaxAge().toSeconds())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_MAX_AGE.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsAvgAge =
Gauge.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_AVG_AGE.getName(), queue,
// Divide by 1000.0 instead of using toSeconds() so we get a double
q -> q.getJobQueueStats().getAvgAge().toMillis() / 1000.0)
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_AVG_AGE.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
jobsQueueTimer =
Timer.builder(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_POLL_TIMER.getName())
.description(COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_POLL_TIMER.getDescription())
- .tags(List.of(Tag.of("queue.id", queueId))).register(meterRegistry);
+ .tags(List.of(Tag.of(MetricsInfo.QUEUE_TAG_KEY,
queueId))).register(meterRegistry);
queue.setJobQueueTimerCallback(jobsQueueTimer);
}
diff --git
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/SystemInformation.java
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/SystemInformation.java
index db9e077f9e..2bd4444e42 100644
---
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/SystemInformation.java
+++
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/SystemInformation.java
@@ -57,6 +57,7 @@ import org.apache.accumulo.core.dataImpl.TabletIdImpl;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.metadata.TabletState;
import org.apache.accumulo.core.metrics.Metric;
+import org.apache.accumulo.core.metrics.MetricsInfo;
import org.apache.accumulo.core.metrics.flatbuffers.FMetric;
import org.apache.accumulo.core.metrics.flatbuffers.FTag;
import org.apache.accumulo.core.process.thrift.MetricResponse;
@@ -495,7 +496,7 @@ public class SystemInformation {
FMetric fm = FMetric.getRootAsFMetric(binary);
for (int i = 0; i < fm.tagsLength(); i++) {
FTag t = fm.tags(i);
- if (t.key().equals("queue.id")) {
+ if (t.key().equals(MetricsInfo.QUEUE_TAG_KEY)) {
queueMetrics
.computeIfAbsent(t.value(), (k) ->
Collections.synchronizedList(new ArrayList<>()))
.add(fm);
diff --git
a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
index 5373eb5be8..a83950041b 100644
---
a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionMetricsIT.java
@@ -49,6 +49,7 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.metrics.MetricsInfo;
import org.apache.accumulo.core.spi.metrics.LoggingMeterRegistryFactory;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.util.threads.Threads;
@@ -232,7 +233,7 @@ public class ExternalCompactionMetricsIT extends
SharedMiniClusterBase {
private static boolean match(Metric input, String queue, String value) {
if (input.getTags() != null) {
- String id = input.getTags().get("queue.id");
+ String id = input.getTags().get(MetricsInfo.QUEUE_TAG_KEY);
if (id != null && id.equals(queue) && input.getValue().equals(value)) {
return true;
}
@@ -243,7 +244,7 @@ public class ExternalCompactionMetricsIT extends
SharedMiniClusterBase {
private static boolean assertMetric(Metric input, String queue, String name,
DoublePredicate valuePredicate) {
if (input.getTags() != null) {
- String id = input.getTags().get("queue.id");
+ String id = input.getTags().get(MetricsInfo.QUEUE_TAG_KEY);
if (id != null && id.equals(queue) && input.getName().equals(name)
&& valuePredicate.test(Double.parseDouble(input.getValue()))) {
return true;
