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 d68cb7d2a5 Compute compaction related suggestions in Monitor (#6239)
d68cb7d2a5 is described below
commit d68cb7d2a5e49c6d9d28aece1cbafb0a3dc08924
Author: Dave Marion <[email protected]>
AuthorDate: Tue Mar 24 17:40:44 2026 -0400
Compute compaction related suggestions in Monitor (#6239)
Modified SystemInformation.finish to compute compaction
related problems after all of the metrics have been
gathered. Additionally centralized some duplicated
code related to processing the value of the FMetric.
---
.../coordinator/CompactionCoordinator.java | 46 ------------------
.../compaction/CompactionCoordinatorTest.java | 4 --
.../accumulo/monitor/next/SystemInformation.java | 56 ++++++++++++++++++----
.../next/serializers/FMetricSerializer.java | 8 +++-
.../next/serializers/MetricResponseSerializer.java | 11 +----
.../monitor/next/sservers/ScanServerView.java | 15 +-----
6 files changed, 56 insertions(+), 84 deletions(-)
diff --git
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
index aee9679cc0..135369fe7a 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
@@ -207,9 +207,6 @@ public class CompactionCoordinator
protected final Map<ExternalCompactionId,TExternalCompaction> RUNNING_CACHE =
new ConcurrentHashMap<>();
- /* Map of group name to last time compactor called to get a compaction job */
- private final Map<ResourceGroupId,Long> TIME_COMPACTOR_LAST_CHECKED = new
ConcurrentHashMap<>();
-
private final ServerContext ctx;
private final AuditedSecurityOperation security;
private final CompactionJobQueues jobQueues;
@@ -382,30 +379,6 @@ public class CompactionCoordinator
LOG.info("Shutting down");
}
- private Map<String,Set<HostAndPort>> getIdleCompactors(Set<ServerId>
runningCompactors) {
-
- final Map<String,Set<HostAndPort>> allCompactors = new HashMap<>();
- runningCompactors.forEach((csi) -> allCompactors
- .computeIfAbsent(csi.getResourceGroup().canonical(), (k) -> new
HashSet<>())
- .add(HostAndPort.fromParts(csi.getHost(), csi.getPort())));
-
- final Set<String> emptyQueues = new HashSet<>();
-
- // Remove all of the compactors that are running a compaction
- RUNNING_CACHE.values().forEach(tec -> {
- Set<HostAndPort> busyCompactors = allCompactors.get(tec.getGroupName());
- if (busyCompactors != null
- &&
busyCompactors.remove(HostAndPort.fromString(tec.getCompactor()))) {
- if (busyCompactors.isEmpty()) {
- emptyQueues.add(tec.getGroupName());
- }
- }
- });
- // Remove entries with empty queues
- emptyQueues.forEach(e -> allCompactors.remove(e));
- return allCompactors;
- }
-
protected void startDeadCompactionDetector() {
deadCompactionDetector.start();
}
@@ -438,7 +411,6 @@ public class CompactionCoordinator
}
ResourceGroupId groupId = ResourceGroupId.of(groupName);
LOG.trace("getCompactionJob called for group {} by compactor {}", groupId,
compactorAddress);
- TIME_COMPACTOR_LAST_CHECKED.put(groupId, System.currentTimeMillis());
TExternalCompactionJob result = null;
@@ -1270,10 +1242,6 @@ public class CompactionCoordinator
}
});
- final Set<ResourceGroupId> 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 Set<ServerId> runningCompactors = getRunningCompactors();
@@ -1302,19 +1270,5 @@ public class CompactionCoordinator
compactorsWithNoGroups);
}
- final long now = System.currentTimeMillis();
- final long warningTime = getMissingCompactorWarningTime();
- Map<String,Set<HostAndPort>> idleCompactors =
getIdleCompactors(runningCompactors);
- for (ResourceGroupId 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);
- }
- }
}
}
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java
b/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java
index a11e345a99..881cfa51ce 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java
@@ -39,7 +39,6 @@ import java.util.UUID;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.admin.CompactionConfig;
import org.apache.accumulo.core.client.admin.servers.ServerId;
import org.apache.accumulo.core.clientImpl.thrift.TInfo;
@@ -370,11 +369,8 @@ public class CompactionCoordinatorTest {
@Test
public void testCleanUpRunning() throws Exception {
TExternalCompaction ext1 = createMock(TExternalCompaction.class);
-
expect(ext1.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce();
TExternalCompaction ext2 = createMock(TExternalCompaction.class);
-
expect(ext2.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce();
TExternalCompaction ext3 = createMock(TExternalCompaction.class);
-
expect(ext3.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce();
replay(ext1, ext2, ext3);
TestCoordinator coordinator = new TestCoordinator(manager, new
ArrayList<>());
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 36bf500a99..27b22d1700 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
@@ -27,6 +27,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -50,6 +51,7 @@ import org.apache.accumulo.core.dataImpl.KeyExtent;
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.flatbuffers.FMetric;
import org.apache.accumulo.core.metrics.flatbuffers.FTag;
import org.apache.accumulo.core.process.thrift.MetricResponse;
@@ -463,24 +465,18 @@ public class SystemInformation {
break;
}
}
- double value = fm.dvalue();
- if (value == 0.0) {
- value = fm.ivalue();
- if (value == 0.0) {
- value = fm.lvalue();
- }
- }
+ Number value = getMetricValue(fm);
final Id id = new Id(name,
(statisticTag == null) ? Tags.empty() : Tags.of(statisticTag.key(),
statisticTag.value()),
null, null, Type.valueOf(fm.type()));
total
.computeIfAbsent(id,
(k) -> new CumulativeDistributionSummary(id, Clock.SYSTEM, DSC,
1.0, false))
- .record(value);
+ .record(value.doubleValue());
rgMetrics
.computeIfAbsent(id,
(k) -> new CumulativeDistributionSummary(id, Clock.SYSTEM, DSC,
1.0, false))
- .record(value);
+ .record(value.doubleValue());
});
}
@@ -587,6 +583,36 @@ public class SystemInformation {
+ " group " + balancerRG + ", but there are no TabletServers.");
}
}
+ for (String rg : getResourceGroups()) {
+ Set<ServerId> rgCompactors = getCompactorResourceGroupServers(rg);
+ List<FMetric> metrics = queueMetrics.get(rg);
+ Optional<FMetric> queued = metrics.stream()
+ .filter(fm ->
fm.name().equals(Metric.COMPACTOR_JOB_PRIORITY_QUEUE_JOBS_QUEUED.getName()))
+ .findFirst();
+ if (queued.isPresent()) {
+ Number numQueued = getMetricValue(queued.orElseThrow());
+ if (numQueued.longValue() > 0) {
+ if (rgCompactors == null || rgCompactors.size() == 0) {
+ suggestions.add("Compactor group " + rg + " has " +
numQueued.longValue()
+ + " queued compactions but no running compactors");
+ } else {
+ // Check for idle compactors.
+ Map<Id,CumulativeDistributionSummary> rgMetrics =
+ getCompactorResourceGroupMetricSummary(rg);
+ Optional<Entry<Id,CumulativeDistributionSummary>> idleMetric =
rgMetrics.entrySet()
+ .stream().filter(e ->
e.getKey().getName().equals(Metric.SERVER_IDLE.getName()))
+ .findFirst();
+ if (idleMetric.isPresent()) {
+ var metric = idleMetric.orElseThrow().getValue();
+ if (metric.max() == 1.0D) {
+ suggestions.add("Compactor group " + rg + " has queued jobs
and idle compactors.");
+ }
+ }
+
+ }
+ }
+ }
+ }
Set<ServerId> scanServers = new HashSet<>();
sservers.values().forEach(scanServers::addAll);
int problemScanServerCount = (int) problemHosts.stream()
@@ -692,4 +718,16 @@ public class SystemInformation {
return this.scanServerView;
}
+ public static Number getMetricValue(FMetric metric) {
+ if (metric.ivalue() != 0) {
+ return metric.ivalue();
+ }
+ if (metric.lvalue() != 0L) {
+ return metric.lvalue();
+ }
+ if (metric.dvalue() != 0.0d) {
+ return metric.dvalue();
+ }
+ return 0;
+ }
}
diff --git
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/FMetricSerializer.java
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/FMetricSerializer.java
index 51823415d7..01219713dd 100644
---
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/FMetricSerializer.java
+++
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/FMetricSerializer.java
@@ -47,6 +47,12 @@ public class FMetricSerializer extends
JsonSerializer<FMetric> {
gen.writeEndObject();
}
gen.writeEndArray();
+ serializeValueField(gen, fm);
+ gen.writeEndObject();
+
+ }
+
+ public static void serializeValueField(JsonGenerator gen, FMetric fm) throws
IOException {
// Write the number as the value (preserve negatives)
if (fm.ivalue() != 0) {
gen.writeNumberField("value", fm.ivalue());
@@ -57,8 +63,6 @@ public class FMetricSerializer extends
JsonSerializer<FMetric> {
} else {
gen.writeNumberField("value", 0);
}
- gen.writeEndObject();
-
}
}
diff --git
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/MetricResponseSerializer.java
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/MetricResponseSerializer.java
index a058e054f7..54ce5f8ade 100644
---
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/MetricResponseSerializer.java
+++
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/serializers/MetricResponseSerializer.java
@@ -54,16 +54,7 @@ public class MetricResponseSerializer extends
JsonSerializer<MetricResponse> {
gen.writeEndObject();
}
gen.writeEndArray();
- // Write the number as the value (preserve negatives)
- if (fm.ivalue() != 0) {
- gen.writeNumberField("value", fm.ivalue());
- } else if (fm.lvalue() != 0L) {
- gen.writeNumberField("value", fm.lvalue());
- } else if (fm.dvalue() != 0.0d) {
- gen.writeNumberField("value", fm.dvalue());
- } else {
- gen.writeNumberField("value", 0);
- }
+ FMetricSerializer.serializeValueField(gen, fm);
gen.writeEndObject();
}
}
diff --git
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/sservers/ScanServerView.java
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/sservers/ScanServerView.java
index a212424877..05ee018d82 100644
---
a/server/monitor/src/main/java/org/apache/accumulo/monitor/next/sservers/ScanServerView.java
+++
b/server/monitor/src/main/java/org/apache/accumulo/monitor/next/sservers/ScanServerView.java
@@ -31,6 +31,7 @@ import org.apache.accumulo.core.metrics.Metric;
import org.apache.accumulo.core.metrics.flatbuffers.FMetric;
import org.apache.accumulo.core.metrics.flatbuffers.FTag;
import org.apache.accumulo.core.process.thrift.MetricResponse;
+import org.apache.accumulo.monitor.next.SystemInformation;
import org.apache.accumulo.server.metrics.MetricResponseWrapper;
/**
@@ -145,7 +146,7 @@ public record ScanServerView(long lastUpdate, List<Row>
servers, Status status)
var metricStatistic = extractStatistic(metric);
if (metricStatistic == null || metricStatistic.equals("value")
|| metricStatistic.equals("count")) {
- values.putIfAbsent(metric.name(), metricNumericValue(metric));
+ values.putIfAbsent(metric.name(),
SystemInformation.getMetricValue(metric));
}
}
return values;
@@ -168,16 +169,4 @@ public record ScanServerView(long lastUpdate, List<Row>
servers, Status status)
return statistic.toLowerCase();
}
- private static Number metricNumericValue(FMetric metric) {
- if (metric.ivalue() != 0) {
- return metric.ivalue();
- }
- if (metric.lvalue() != 0L) {
- return metric.lvalue();
- }
- if (metric.dvalue() != 0.0d) {
- return metric.dvalue();
- }
- return 0;
- }
}
