From 85090e0be2bb397acf78690e1ce4cef023d985d5 Mon Sep 17 00:00:00 2001 From: arbaazkhan1 Date: Mon, 3 Aug 2026 14:04:42 -0400 Subject: [PATCH] moved Zk cleaner threads into utility class --- .../apache/accumulo/core/conf/Property.java | 7 +- .../core/util/ExpectedProcessCounts.java | 126 ++++++++++ .../accumulo/server/util/CleanZooKeeper.java | 216 ++++++++++++++++++ .../accumulo/server/util/ProcessStatus.java | 177 ++++++++++++++ .../org/apache/accumulo/manager/Manager.java | 45 ---- .../coordinator/CompactionCoordinator.java | 61 ----- .../compaction/CompactionCoordinatorTest.java | 3 - .../monitor/next/SystemInformation.java | 38 +++ .../accumulo/test/start/KeywordStartIT.java | 4 + 9 files changed, 567 insertions(+), 110 deletions(-) create mode 100644 core/src/main/java/org/apache/accumulo/core/util/ExpectedProcessCounts.java create mode 100644 server/base/src/main/java/org/apache/accumulo/server/util/CleanZooKeeper.java create mode 100644 server/base/src/main/java/org/apache/accumulo/server/util/ProcessStatus.java diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java index b148c1a60e5..aac26a444d6 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java @@ -1423,7 +1423,12 @@ start with the category prefix, followed by a scope (minc, majc, scan, \ "The interval at which to check for dead compactors.", "2.1.0"), GENERAL_AMPLE_CONDITIONAL_WRITER_THREADS_MAX("general.ample.conditional.writer.threads.max", "8", PropertyType.COUNT, - "The maximum number of threads for the shared ConditionalWriter used by Ample.", "4.0.0"); + "The maximum number of threads for the shared ConditionalWriter used by Ample.", "4.0.0"), + GENERAL_EXPECTED_PROCESS_COUNTS("general.expected.process.counts", "", PropertyType.STRING, + "Declare the expected number of server processes per type and resource group. Used by the process-status utility to identify compactor and scan servers that are down." + + "Format: comma-separated list of .= entries. valid types are 'compactor' and 'sserver'." + + "Example: compactor.default=2, compactor.CTEST=3, sserver.default=1 ", + "4.0.0"); private final String key; private final String defaultValue; diff --git a/core/src/main/java/org/apache/accumulo/core/util/ExpectedProcessCounts.java b/core/src/main/java/org/apache/accumulo/core/util/ExpectedProcessCounts.java new file mode 100644 index 00000000000..6aebd80d65e --- /dev/null +++ b/core/src/main/java/org/apache/accumulo/core/util/ExpectedProcessCounts.java @@ -0,0 +1,126 @@ +/* + * 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.util; + +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.Map; +import java.util.OptionalInt; + +import org.apache.accumulo.core.client.admin.servers.ServerId; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ExpectedProcessCounts { + + private static final Logger log = LoggerFactory.getLogger(ExpectedProcessCounts.class); + + private static final Map SUPPORTED_TYPES = + Map.of("compactor", ServerId.Type.COMPACTOR, "sserver", ServerId.Type.SCAN_SERVER); + + private final Map> counts; + + private ExpectedProcessCounts(Map> counts) { + this.counts = counts; + } + + public static ExpectedProcessCounts parse(String propertyValue) { + Map> result = new EnumMap<>(ServerId.Type.class); + + if (propertyValue == null || propertyValue.isBlank()) { + return new ExpectedProcessCounts(result); + } + + for (String entry : propertyValue.split(",")) { + entry = entry.trim(); + if (entry.isEmpty()) { + continue; + } + + int eqIdx = entry.lastIndexOf('='); + if (eqIdx < 0) { + log.warn("Ignoring malformed entry in {} (missing '='): {}", + "general.expected.process.counts", entry); + continue; + } + + String key = entry.substring(0, eqIdx).trim(); + String valueStr = entry.substring(eqIdx + 1).trim(); + + int dotIdx = key.indexOf('.'); + if (dotIdx < 0) { + log.warn("Ignoring malformed key in {} (expected '.'): {}", + "general.expected.process.counts", key); + continue; + } + + String typeName = key.substring(0, dotIdx).trim().toLowerCase(); + String groupName = key.substring(dotIdx + 1).trim(); + + ServerId.Type serverType = SUPPORTED_TYPES.get(typeName); + if (serverType == null) { + log.warn("Ignoring unknown server type '{}' in general.expected.process.counts." + + " Supported types: {}", typeName, SUPPORTED_TYPES.keySet()); + continue; + } + + if (groupName.isEmpty()) { + log.warn("Ignoring entry with empty resource group name in" + + " general.expected.process.counts: {}", entry); + continue; + } + + int count; + try { + count = Integer.parseInt(valueStr); + if (count < 0) { + throw new NumberFormatException("count must be non-negative"); + } + } catch (NumberFormatException e) { + log.warn("Ignoring entry with invalid count '{}' in general.expected.process.counts: {}", + valueStr, entry); + continue; + } + + result.computeIfAbsent(serverType, t -> new HashMap<>()).put(ResourceGroupId.of(groupName), + count); + } + + return new ExpectedProcessCounts(Collections.unmodifiableMap(result)); + } + + public OptionalInt getExpectedCount(ServerId.Type type, ResourceGroupId group) { + var groupMap = counts.get(type); + if (groupMap == null) { + return OptionalInt.empty(); + } + Integer count = groupMap.get(group); + return count == null ? OptionalInt.empty() : OptionalInt.of(count); + } + + public Map> all() { + return counts; + } + + public boolean isEmpty() { + return counts.isEmpty(); + } +} diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/CleanZooKeeper.java b/server/base/src/main/java/org/apache/accumulo/server/util/CleanZooKeeper.java new file mode 100644 index 00000000000..669f119d714 --- /dev/null +++ b/server/base/src/main/java/org/apache/accumulo/server/util/CleanZooKeeper.java @@ -0,0 +1,216 @@ +/* + * 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.server.util; + +import java.util.Arrays; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.cli.ServerOpts; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter; +import org.apache.accumulo.core.lock.ServiceLock; +import org.apache.accumulo.core.lock.ServiceLockData; +import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector; +import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate; +import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath; +import org.apache.accumulo.core.zookeeper.ZcStat; +import org.apache.accumulo.server.ServerContext; +import org.apache.accumulo.server.util.CleanZooKeeper.CleanOpts; +import org.apache.accumulo.start.spi.CommandGroup; +import org.apache.accumulo.start.spi.CommandGroups; +import org.apache.accumulo.start.spi.KeywordExecutable; +import org.apache.zookeeper.KeeperException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.google.auto.service.AutoService; + +@AutoService(KeywordExecutable.class) +public class CleanZooKeeper extends ServerKeywordExecutable { + + private static final Logger log = LoggerFactory.getLogger(CleanZooKeeper.class); + + public CleanZooKeeper() { + super(new CleanOpts()); + } + + static class CleanOpts extends ServerOpts { + @Parameter(names = "--compactors", + description = "Remove ZooKeeper paths for compactors that are no longer running") + boolean cleanCompactors = false; + + @Parameter(names = "--sservers", + description = "Remove ZooKeeper paths for scan servers that are no longer running") + boolean cleanScanServers = false; + + @Parameter(names = "--include-groups", + description = "Comma-separated list of resource groups to include (default: all)") + String includeGroups; + + @Parameter(names = "--dry-run", + description = "Print paths that would be removed without making any changes") + boolean dryRun = false; + + @Parameter(names = "--verbose", description = "Print progress messages") + boolean verbose = false; + } + + @Override + public String keyword() { + return "clean-zk"; + } + + @Override + public String description() { + return "Removes ZooKeeper paths for compactors and scan servers that are no longer running." + + " Run only when permanently changing deployment topology, not during normal operations."; + } + + @Override + public CommandGroup commandGroup() { + return CommandGroups.ZOOKEEPER; + } + + @Override + public void execute(JCommander cl, CleanOpts opts) throws Exception { + if (!opts.cleanCompactors && !opts.cleanScanServers) { + new JCommander(opts).usage(); + return; + } + + final ResourceGroupPredicate rgp; + if (opts.includeGroups != null) { + Set groups = Arrays.stream(opts.includeGroups.split(",")).map(String::trim) + .map(ResourceGroupId::of).collect(Collectors.toSet()); + rgp = groups::contains; + } else { + rgp = ResourceGroupPredicate.ANY; + } + + var context = getServerContext(); + var zrw = context.getZooSession().asReaderWriter(); + + if (opts.cleanCompactors) { + cleanCompactors(context, zrw, rgp, opts); + } + if (opts.cleanScanServers) { + cleanScanServers(context, zrw, rgp, opts); + } + } + + private void cleanCompactors(ServerContext context, ZooReaderWriter zrw, + ResourceGroupPredicate rgp, CleanOpts opts) { + + // Remove individual compactor nodes with no active lock + Set compactorPaths = + context.getServerPaths().getCompactor(rgp, AddressSelector.all(), false); + + for (ServiceLockPath path : compactorPaths) { + ZcStat stat = new ZcStat(); + Optional lockData = + ServiceLock.getLockData(context.getZooCache(), path, stat); + if (lockData.isEmpty()) { + message("Removing empty compactor ZK path: " + path, opts); + if (!opts.dryRun) { + try { + zrw.delete(path.toString()); + } catch (KeeperException.NotEmptyException e) { + log.debug("Failed to delete compactor ZK node {} because it is not empty," + + " likely an expected race condition.", path); + } catch (KeeperException | InterruptedException e) { + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + log.warn("Failed to delete compactor ZK node {}", path, e); + } + } + } + } + + // Remove empty resource group parent nodes under ZCOMPACTORS + try { + var groups = zrw.getChildren(Constants.ZCOMPACTORS); + for (String group : groups) { + ResourceGroupId rgid = ResourceGroupId.of(group); + if (!rgp.test(rgid)) { + continue; + } + String groupPath = Constants.ZCOMPACTORS + "/" + group; + var children = zrw.getChildren(groupPath); + if (children.isEmpty()) { + message("Removing empty compactor group ZK path: " + groupPath, opts); + if (!opts.dryRun) { + try { + zrw.delete(groupPath); + } catch (KeeperException.NotEmptyException e) { + log.debug("Failed to delete compactor group ZK node {} because it is not empty.", + groupPath); + } + } + } + } + } catch (KeeperException | InterruptedException e) { + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + log.warn("Failed to clean up empty compactor group paths", e); + } + } + + private void cleanScanServers(org.apache.accumulo.server.ServerContext context, + ZooReaderWriter zrw, ResourceGroupPredicate rgp, CleanOpts opts) { + try { + Set scanServerPaths = + context.getServerPaths().getScanServer(rgp, AddressSelector.all(), false); + + for (ServiceLockPath path : scanServerPaths) { + ZcStat stat = new ZcStat(); + Optional lockData = + ServiceLock.getLockData(context.getZooCache(), path, stat); + if (lockData.isEmpty()) { + message("Removing empty scan server ZK path: " + path, opts); + if (!opts.dryRun) { + try { + zrw.delete(path.toString()); + } catch (KeeperException.NotEmptyException e) { + log.debug("Failed to delete scan server ZK node {} because it is not empty," + + " likely an expected race condition.", path); + } + } + } + } + } catch (KeeperException e) { + log.error("Exception trying to delete empty scan server ZK paths", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.error("Interrupted trying to delete empty scan server ZK paths", e); + } + } + + private static void message(String msg, CleanOpts opts) { + if (opts.verbose || opts.dryRun) { + System.out.println(msg); + } + } +} diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/ProcessStatus.java b/server/base/src/main/java/org/apache/accumulo/server/util/ProcessStatus.java new file mode 100644 index 00000000000..eeccfe24800 --- /dev/null +++ b/server/base/src/main/java/org/apache/accumulo/server/util/ProcessStatus.java @@ -0,0 +1,177 @@ +/* + * 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.server.util; + +import java.util.Arrays; +import java.util.Optional; +import java.util.OptionalInt; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.accumulo.core.cli.ServerOpts; +import org.apache.accumulo.core.client.admin.servers.ServerId; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.apache.accumulo.core.lock.ServiceLock; +import org.apache.accumulo.core.lock.ServiceLockData; +import org.apache.accumulo.core.lock.ServiceLockPaths; +import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector; +import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate; +import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath; +import org.apache.accumulo.core.util.ExpectedProcessCounts; +import org.apache.accumulo.core.zookeeper.ZcStat; +import org.apache.accumulo.server.ServerContext; +import org.apache.accumulo.start.spi.CommandGroup; +import org.apache.accumulo.start.spi.CommandGroups; +import org.apache.accumulo.start.spi.KeywordExecutable; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.google.auto.service.AutoService; + +@AutoService(KeywordExecutable.class) +public class ProcessStatus extends ServerKeywordExecutable { + public ProcessStatus() { + super(new ProcessStatusOpts()); + } + + static class ProcessStatusOpts extends ServerOpts { + @Parameter(names = "--include-groups", + description = "Comma-separated list of resource groups to check (default: all declared groups)") + String includeGroups; + } + + @Override + public String keyword() { + return "process-status"; + } + + @Override + public String description() { + return "Compares declared expected process counts against running compactors and scan servers," + + " reporting groups that are below capacity. Requires " + + Property.GENERAL_EXPECTED_PROCESS_COUNTS.getKey() + " to be configured."; + } + + @Override + public CommandGroup commandGroup() { + return CommandGroups.PROCESS; + } + + @Override + public void execute(JCommander cl, ProcessStatusOpts opts) throws Exception { + ServerContext context = getServerContext(); + + String rawProperty = context.getConfiguration().get(Property.GENERAL_EXPECTED_PROCESS_COUNTS); + ExpectedProcessCounts expected = ExpectedProcessCounts.parse(rawProperty); + + if (expected.isEmpty()) { + System.out.println("No expected process counts declared in '" + + Property.GENERAL_EXPECTED_PROCESS_COUNTS.getKey() + "'."); + System.out.println("Set this property to enable process health checking."); + System.out.println("Example value: compactor.default=2,compactor.CTEST=3,sserver.default=1"); + return; + } + + final ResourceGroupPredicate rgp; + if (opts.includeGroups != null) { + Set groups = Arrays.stream(opts.includeGroups.split(",")).map(String::trim) + .map(ResourceGroupId::of).collect(Collectors.toSet()); + rgp = groups::contains; + } else { + rgp = ResourceGroupPredicate.ANY; + } + + System.out.printf("%-12s %-20s %-10s %-10s %-6s%n", "Type", "Group", "Expected", "Running", + "Down"); + System.out.println("-".repeat(63)); + + boolean anyDegraded = false; + + for (var typeEntry : expected.all().entrySet()) { + ServerId.Type serverType = typeEntry.getKey(); + for (var groupEntry : typeEntry.getValue().entrySet()) { + ResourceGroupId group = groupEntry.getKey(); + OptionalInt maybeExpected = expected.getExpectedCount(serverType, group); + if (maybeExpected.isEmpty()) { + continue; // defensive — should not happen since we're iterating expected.all() + } + int expectedCount = maybeExpected.getAsInt(); + if (!rgp.test(group)) { + continue; + } + + int runningCount = countRunning(context, serverType, group); + int downCount = Math.max(0, expectedCount - runningCount); + boolean degraded = downCount > 0; + if (degraded) { + anyDegraded = true; + } + + System.out.printf("%-12s %-20s %-10d %-10d %-6d%s%n", typeName(serverType), + group.canonical(), expectedCount, runningCount, downCount, + degraded ? " ** DEGRADED **" : ""); + } + } + + System.out.println(); + if (anyDegraded) { + System.out.println("WARNING: One or more server groups are running below expected capacity."); + } else { + System.out.println("All declared server groups are running at expected capacity."); + } + } + + // Counts servers of the given type and resource group that currently hold a ZooKeeper lock. + private int countRunning(ServerContext context, ServerId.Type serverType, ResourceGroupId group) { + ServiceLockPaths.ResourceGroupPredicate exactGroup = + ServiceLockPaths.ResourceGroupPredicate.exact(group); + Set paths; + + switch (serverType) { + case COMPACTOR: + paths = context.getServerPaths().getCompactor(exactGroup, AddressSelector.all(), false); + break; + case SCAN_SERVER: + paths = context.getServerPaths().getScanServer(exactGroup, AddressSelector.all(), false); + break; + default: + return 0; + } + + int count = 0; + for (ServiceLockPath path : paths) { + ZcStat stat = new ZcStat(); + Optional lockData = + ServiceLock.getLockData(context.getZooCache(), path, stat); + if (lockData.isPresent()) { + count++; + } + } + return count; + } + + private static String typeName(ServerId.Type type) { + return switch (type) { + case COMPACTOR -> "compactor"; + case SCAN_SERVER -> "sserver"; + default -> type.name().toLowerCase(); + }; + } +} diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java index ed20d3e0ffc..86ad76ebda1 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java @@ -40,7 +40,6 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; -import java.util.Optional; import java.util.Set; import java.util.SortedMap; import java.util.UUID; @@ -90,7 +89,6 @@ import org.apache.accumulo.core.lock.ServiceLockData.ThriftService; import org.apache.accumulo.core.lock.ServiceLockPaths; import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector; -import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate; import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath; import org.apache.accumulo.core.lock.ServiceLockSupport; import org.apache.accumulo.core.lock.ServiceLockSupport.HAServiceLockWatcher; @@ -113,7 +111,6 @@ 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.zookeeper.ZcStat; import org.apache.accumulo.manager.compaction.coordinator.CompactionCoordinator; import org.apache.accumulo.manager.fate.FateManager; import org.apache.accumulo.manager.fate.FateNotifier; @@ -575,47 +572,6 @@ public void hostOndemand(List extents) { } } - private class ScanServerZKCleaner implements Runnable { - - @Override - public void run() { - - final ZooReaderWriter zrw = getContext().getZooSession().asReaderWriter(); - - while (stillManager()) { - try { - Set scanServerPaths = getContext().getServerPaths() - .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false); - for (ServiceLockPath path : scanServerPaths) { - - ZcStat stat = new ZcStat(); - Optional lockData = - ServiceLock.getLockData(getContext().getZooCache(), path, stat); - - if (lockData.isEmpty()) { - try { - log.debug("Deleting empty ScanServer ZK node {}", path); - zrw.delete(path.toString()); - } catch (KeeperException.NotEmptyException e) { - log.debug( - "Failed to delete ScanServer ZK node {} its not empty, likely an expected race condition.", - path); - } - } - } - } catch (KeeperException e) { - log.error("Exception trying to delete empty scan server ZNodes, will retry", e); - } catch (InterruptedException e) { - log.error("Interrupted trying to delete empty scan server ZNodes, will retry", e); - } finally { - // sleep for 5 mins - sleepUninterruptibly(CLEANUP_INTERVAL_MINUTES, MINUTES); - } - } - } - - } - boolean canBalance(DataLevel dataLevel, TServerStatus tServerStatus) { Set serversToShutdown; if (!badServers.isEmpty()) { @@ -1154,7 +1110,6 @@ boolean canSuspendTablets() { } balanceManager.startBackGroundTask(); - Threads.createCriticalThread("ScanServer Cleanup Thread", new ScanServerZKCleaner()).start(); // Don't call start the CompactionCoordinator until we have tservers and upgrade is complete. compactionCoordinator.start(); 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 bbe9dd22ea1..1833f9d6184 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 @@ -31,7 +31,6 @@ 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; @@ -56,7 +55,6 @@ import java.util.function.Supplier; 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.admin.CompactionConfig; @@ -77,7 +75,6 @@ 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.ConditionalLogger.ConditionalLogAction; import org.apache.accumulo.core.logging.TabletLogger; @@ -110,7 +107,6 @@ 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.manager.compaction.queue.ResolvedCompactionJob; import org.apache.accumulo.manager.tableOps.FateEnv; @@ -122,7 +118,6 @@ import org.apache.accumulo.server.util.FindCompactionTmpFiles; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.Path; -import org.apache.zookeeper.KeeperException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -274,12 +269,6 @@ public void shutdown() { } } - 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::resizeThreadPools, 0, 5, TimeUnit.MINUTES); @@ -301,7 +290,6 @@ private void checkForConfigChanges() { @Override public void run() { startConfigMonitor(ctx.getScheduledExecutor()); - startCompactorZKCleaner(ctx.getScheduledExecutor()); startDeadCompactionDetector(); startFailureSummaryLogging(); @@ -820,55 +808,6 @@ public CompactionJobQueues getJobQueues() { return jobQueues; } - 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 var zoorw = this.ctx.getZooSession().asReaderWriter(); - - try { - var groups = zoorw.getChildren(Constants.ZCOMPACTORS); - - for (String group : groups) { - final String qpath = Constants.ZCOMPACTORS + "/" + group; - final ResourceGroupId cgid = ResourceGroupId.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.clearIfInactive(Duration.ofMinutes(10)); - } - } else { - for (String compactor : compactors) { - String cpath = Constants.ZCOMPACTORS + "/" + group + "/" + compactor; - var lockNodes = - zoorw.getChildren(Constants.ZCOMPACTORS + "/" + group + "/" + compactor); - if (lockNodes.isEmpty()) { - deleteEmpty(zoorw, cpath); - } - } - } - } - } catch (KeeperException | RuntimeException e) { - LOG.warn("Failed to clean up compactors", e); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IllegalStateException(e); - } - } - public void resizeThreadPools() { var config = ctx.getConfiguration(); ThreadPools.resizePool(reservationPools.get(DataLevel.ROOT), config, 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 01e44585049..f1248aea46e 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 @@ -110,9 +110,6 @@ protected void startFailureSummaryLogging() {} @Override protected void startDeadCompactionDetector() {} - @Override - protected void startCompactorZKCleaner(ScheduledThreadPoolExecutor schedExecutor) {} - @Override protected void startInternalStateCleaner(ScheduledThreadPoolExecutor schedExecutor) { // This is called from CompactionCoordinator.run(). Counting down 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 70dfe828cf6..ad40ce04242 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 @@ -41,6 +41,7 @@ import java.util.Map.Entry; import java.util.Objects; import java.util.Optional; +import java.util.OptionalInt; import java.util.Set; import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; @@ -84,6 +85,7 @@ import org.apache.accumulo.core.process.thrift.MetricResponse; import org.apache.accumulo.core.spi.balancer.TableLoadBalancer; import org.apache.accumulo.core.tabletscan.thrift.ActiveScan; +import org.apache.accumulo.core.util.ExpectedProcessCounts; import org.apache.accumulo.core.util.compaction.RunningCompactionInfo; import org.apache.accumulo.monitor.next.InformationFetcher.MetricFetcher; import org.apache.accumulo.monitor.next.InformationFetcher.TableInformationFetcher; @@ -1406,6 +1408,42 @@ private void computeAlerts(final List failures, } } + ExpectedProcessCounts expectedCounts = ExpectedProcessCounts + .parse(ctx.getConfiguration().get(Property.GENERAL_EXPECTED_PROCESS_COUNTS)); + + if (!expectedCounts.isEmpty()) { + for (var typeEntry : expectedCounts.all().entrySet()) { + ServerId.Type serverType = typeEntry.getKey(); + for (var groupEntry : typeEntry.getValue().entrySet()) { + ResourceGroupId group = groupEntry.getKey(); + OptionalInt expected = expectedCounts.getExpectedCount(serverType, group); + if (expected.isEmpty()) { + continue; + } + + int running = switch (serverType) { + case COMPACTOR -> { + Set s = compactors.get(group.canonical()); + yield s == null ? 0 : s.size(); + } + case SCAN_SERVER -> { + Set s = sservers.get(group.canonical()); + yield s == null ? 0 : s.size(); + } + default -> expected.getAsInt(); + }; + + int down = expected.getAsInt() - running; + if (down > 0) { + String typeName = serverType == ServerId.Type.COMPACTOR ? "Compactor" : "ScanServer"; + addAlert(High, Resource, + typeName + " group " + group.canonical() + " is degraded: expected " + + expected.getAsInt() + ", running " + running + ", down " + down); + } + } + } + } + } public void finish(final List failures, final List cancelled, diff --git a/test/src/main/java/org/apache/accumulo/test/start/KeywordStartIT.java b/test/src/main/java/org/apache/accumulo/test/start/KeywordStartIT.java index f277baeeee5..49c023cd72a 100644 --- a/test/src/main/java/org/apache/accumulo/test/start/KeywordStartIT.java +++ b/test/src/main/java/org/apache/accumulo/test/start/KeywordStartIT.java @@ -64,6 +64,7 @@ import org.apache.accumulo.server.conf.util.ZooPropEditor; import org.apache.accumulo.server.init.Initialize; import org.apache.accumulo.server.util.CancelCompaction; +import org.apache.accumulo.server.util.CleanZooKeeper; import org.apache.accumulo.server.util.DumpZookeeper; import org.apache.accumulo.server.util.FindCompactionTmpFiles; import org.apache.accumulo.server.util.FindOfflineTablets; @@ -72,6 +73,7 @@ import org.apache.accumulo.server.util.ListCompactors; import org.apache.accumulo.server.util.ListOnlineOnDemandTablets; import org.apache.accumulo.server.util.LoginProperties; +import org.apache.accumulo.server.util.ProcessStatus; import org.apache.accumulo.server.util.RemoveEntriesForMissingFiles; import org.apache.accumulo.server.util.ScanServerMetadataEntries; import org.apache.accumulo.server.util.UpgradeUtil; @@ -254,6 +256,8 @@ public void testExpectedClasses() { expectSet.add(new CommandInfo(CommandGroups.ZOOKEEPER, "prop-editor", ZooPropEditor.class)); expectSet.add(new CommandInfo(CommandGroups.ZOOKEEPER, "zap", ZooZap.class)); expectSet.add(new CommandInfo(CommandGroups.ZOOKEEPER, "cli", ZooKeeperMain.class)); + expectSet.add(new CommandInfo(CommandGroups.ZOOKEEPER, "clean-zk", CleanZooKeeper.class)); + expectSet.add(new CommandInfo(CommandGroups.PROCESS, "process-status", ProcessStatus.class)); Map> actualExecutables = getKeywordExecutables(); SortedSet actualSet = new TreeSet<>();