This is an automated email from the ASF dual-hosted git repository.
cshannon 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 d139704010 Add regex support for filtering listscans and
listcompactions (#5072)
d139704010 is described below
commit d139704010987b0dc6474d74db0fbabe52795b82
Author: Christopher L. Shannon <[email protected]>
AuthorDate: Sat Nov 23 13:09:10 2024 -0500
Add regex support for filtering listscans and listcompactions (#5072)
This commit adds new flags to filter listscans and listcompactions by
resource groups and/or servers using a regex pattern.
This closes #5069
---
.../core/client/admin/InstanceOperations.java | 15 +++
.../core/clientImpl/InstanceOperationsImpl.java | 8 ++
.../shell/commands/ActiveCompactionHelper.java | 35 +++++-
.../shell/commands/ListCompactionsCommand.java | 28 ++++-
.../accumulo/shell/commands/ListScansCommand.java | 53 +++++++--
.../shell/commands/ListCompactionsCommandTest.java | 36 ++++++
.../shell/commands/ListScansCommandTest.java | 129 +++++++++++++++++++++
.../apache/accumulo/test/shell/ShellServerIT.java | 90 +++++++++-----
8 files changed, 349 insertions(+), 45 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/admin/InstanceOperations.java
b/core/src/main/java/org/apache/accumulo/core/client/admin/InstanceOperations.java
index 30eb8e76ef..b325eaf52b 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/admin/InstanceOperations.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/admin/InstanceOperations.java
@@ -18,6 +18,7 @@
*/
package org.apache.accumulo.core.client.admin;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -28,6 +29,7 @@ import java.util.function.Predicate;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.admin.servers.ServerId;
+import org.apache.accumulo.core.client.admin.servers.ServerId.Type;
import org.apache.accumulo.core.data.InstanceId;
public interface InstanceOperations {
@@ -312,6 +314,19 @@ public interface InstanceOperations {
*/
List<ActiveCompaction> getActiveCompactions() throws AccumuloException,
AccumuloSecurityException;
+ /**
+ * List the active compaction running on a collection of TabletServers or
Compactors. The server
+ * addresses can be retrieved using {@link #getServers(Type)}. Use {@link
#getActiveCompactions()}
+ * to get a list of all compactions running on tservers and compactors.
+ *
+ * @param servers The collection of servers
+ * @return the list of active compactions
+ * @throws IllegalArgumentException if a type of server is not TABLET_SERVER
or COMPACTOR
+ * @since 4.0.0
+ */
+ List<ActiveCompaction> getActiveCompactions(Collection<ServerId> servers)
+ throws AccumuloException, AccumuloSecurityException;
+
/**
* Check to see if a server process at the host and port is up and
responding to RPC requests.
*
diff --git
a/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java
b/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java
index 9eb7123d9a..7886b9739b 100644
---
a/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java
+++
b/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java
@@ -28,6 +28,7 @@ import static
org.apache.accumulo.core.util.threads.ThreadPoolNames.INSTANCE_OPS
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.HashSet;
@@ -397,6 +398,13 @@ public class InstanceOperationsImpl implements
InstanceOperations {
compactionServers.addAll(getServers(ServerId.Type.COMPACTOR));
compactionServers.addAll(getServers(ServerId.Type.TABLET_SERVER));
+ return getActiveCompactions(compactionServers);
+ }
+
+ @Override
+ public List<ActiveCompaction> getActiveCompactions(Collection<ServerId>
compactionServers)
+ throws AccumuloException, AccumuloSecurityException {
+
int numThreads = Math.max(4, Math.min((compactionServers.size()) / 10,
256));
var executorService =
context.threadPools().getPoolBuilder(INSTANCE_OPS_COMPACTIONS_FINDER_POOL)
.numCoreThreads(numThreads).build();
diff --git
a/shell/src/main/java/org/apache/accumulo/shell/commands/ActiveCompactionHelper.java
b/shell/src/main/java/org/apache/accumulo/shell/commands/ActiveCompactionHelper.java
index 81d9e35cd1..732c68db57 100644
---
a/shell/src/main/java/org/apache/accumulo/shell/commands/ActiveCompactionHelper.java
+++
b/shell/src/main/java/org/apache/accumulo/shell/commands/ActiveCompactionHelper.java
@@ -21,8 +21,12 @@ package org.apache.accumulo.shell.commands;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
import java.util.stream.Stream;
import org.apache.accumulo.core.client.AccumuloException;
@@ -137,16 +141,37 @@ class ActiveCompactionHelper {
}
}
+ public static Stream<String> activeCompactions(InstanceOperations
instanceOps,
+ Predicate<String> resourceGroupPredicate, BiPredicate<String,Integer>
hostPortPredicate) {
+
+ try {
+ final Set<ServerId> compactionServers = new HashSet<>();
+ compactionServers.addAll(instanceOps.getServers(ServerId.Type.COMPACTOR,
+ resourceGroupPredicate, hostPortPredicate));
+
compactionServers.addAll(instanceOps.getServers(ServerId.Type.TABLET_SERVER,
+ resourceGroupPredicate, hostPortPredicate));
+
+ return
sortActiveCompactions(instanceOps.getActiveCompactions(compactionServers));
+ } catch (AccumuloException | AccumuloSecurityException e) {
+ LOG.debug("Failed to list active compactions with resource group and
server predicates", e);
+ return Stream.of("ERROR " + e.getMessage());
+ }
+ }
+
public static Stream<String> activeCompactions(InstanceOperations
instanceOps) {
- Comparator<ActiveCompaction> comparator =
- Comparator.comparing((ActiveCompaction ac) -> ac.getHost().getHost())
- .thenComparing(ac ->
ac.getHost().getPort()).thenComparing(COMPACTION_AGE_DESCENDING);
try {
- return instanceOps.getActiveCompactions().stream().sorted(comparator)
- .map(ActiveCompactionHelper::formatActiveCompactionLine);
+ return sortActiveCompactions(instanceOps.getActiveCompactions());
} catch (AccumuloException | AccumuloSecurityException e) {
return Stream.of("ERROR " + e.getMessage());
}
}
+ private static Stream<String> sortActiveCompactions(List<ActiveCompaction>
activeCompactions) {
+ Comparator<ActiveCompaction> comparator =
+ Comparator.comparing((ActiveCompaction ac) -> ac.getHost().getHost())
+ .thenComparing(ac ->
ac.getHost().getPort()).thenComparing(COMPACTION_AGE_DESCENDING);
+ return activeCompactions.stream().sorted(comparator)
+ .map(ActiveCompactionHelper::formatActiveCompactionLine);
+ }
+
}
diff --git
a/shell/src/main/java/org/apache/accumulo/shell/commands/ListCompactionsCommand.java
b/shell/src/main/java/org/apache/accumulo/shell/commands/ListCompactionsCommand.java
index 7c5f32cc48..2aec7dfce6 100644
---
a/shell/src/main/java/org/apache/accumulo/shell/commands/ListCompactionsCommand.java
+++
b/shell/src/main/java/org/apache/accumulo/shell/commands/ListCompactionsCommand.java
@@ -18,6 +18,10 @@
*/
package org.apache.accumulo.shell.commands;
+import static
org.apache.accumulo.shell.commands.ListScansCommand.getServerOptValue;
+import static
org.apache.accumulo.shell.commands.ListScansCommand.rgRegexPredicate;
+import static
org.apache.accumulo.shell.commands.ListScansCommand.serverRegexPredicate;
+
import java.util.regex.Pattern;
import java.util.stream.Stream;
@@ -30,7 +34,7 @@ import org.apache.commons.cli.Options;
public class ListCompactionsCommand extends Command {
- private Option tserverOption, disablePaginationOpt, filterOption;
+ private Option serverOpt, tserverOption, rgOpt, disablePaginationOpt,
filterOption;
@Override
public String description() {
@@ -51,9 +55,12 @@ public class ListCompactionsCommand extends Command {
Stream<String> activeCompactionStream;
- if (cl.hasOption(tserverOption.getOpt())) {
- activeCompactionStream = ActiveCompactionHelper
-
.activeCompactionsForServer(cl.getOptionValue(tserverOption.getOpt()),
instanceOps);
+ String serverValue = getServerOptValue(cl, serverOpt, tserverOption);
+ if (serverValue != null || cl.hasOption(rgOpt)) {
+ final var serverPredicate = serverRegexPredicate(serverValue);
+ final var rgPredicate = rgRegexPredicate(cl.getOptionValue(rgOpt));
+ activeCompactionStream =
+ ActiveCompactionHelper.activeCompactions(instanceOps, rgPredicate,
serverPredicate);
} else {
activeCompactionStream =
ActiveCompactionHelper.activeCompactions(instanceOps);
}
@@ -85,11 +92,22 @@ public class ListCompactionsCommand extends Command {
filterOption = new Option("f", "filter", true, "show only compactions that
match the regex");
opts.addOption(filterOption);
+ serverOpt = new Option("s", "server", true,
+ "tablet/compactor server regex to list compactions for. Regex will
match against strings like <host>:<port>");
+ serverOpt.setArgName("tablet/compactor server regex");
+ opts.addOption(serverOpt);
+
+ // Leaving here for backwards compatibility, same as serverOpt
tserverOption = new Option("ts", "tabletServer", true,
- "tablet server or compactor to list compactions for");
+ "tablet/compactor server regex to list compactions for");
tserverOption.setArgName("tablet server");
opts.addOption(tserverOption);
+ rgOpt = new Option("rg", "resourceGroup", true,
+ "tablet/compactor server resource group regex to list compactions
for");
+ rgOpt.setArgName("resource group");
+ opts.addOption(rgOpt);
+
disablePaginationOpt = new Option("np", "no-pagination", false, "disable
pagination of output");
opts.addOption(disablePaginationOpt);
diff --git
a/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java
b/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java
index 54ec60055b..c8283cfb22 100644
---
a/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java
+++
b/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java
@@ -19,7 +19,11 @@
package org.apache.accumulo.shell.commands;
import java.util.HashSet;
+import java.util.Optional;
import java.util.Set;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
import org.apache.accumulo.core.client.admin.InstanceOperations;
import org.apache.accumulo.core.client.admin.servers.ServerId;
@@ -29,11 +33,11 @@ import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
-import com.google.common.net.HostAndPort;
+import com.google.common.base.Preconditions;
public class ListScansCommand extends Command {
- private Option tserverOption, disablePaginationOpt;
+ private Option serverOpt, tserverOption, rgOpt, disablePaginationOpt;
@Override
public String description() {
@@ -49,13 +53,14 @@ public class ListScansCommand extends Command {
final boolean paginate = !cl.hasOption(disablePaginationOpt.getOpt());
final Set<ServerId> servers = new HashSet<>();
- if (cl.hasOption(tserverOption.getOpt())) {
- String serverAddress = cl.getOptionValue(tserverOption.getOpt());
- final HostAndPort hp = HostAndPort.fromString(serverAddress);
+ String serverValue = getServerOptValue(cl, serverOpt, tserverOption);
+ if (serverValue != null || cl.hasOption(rgOpt)) {
+ final var serverPredicate = serverRegexPredicate(serverValue);
+ final var rgPredicate = rgRegexPredicate(cl.getOptionValue(rgOpt));
servers
- .add(instanceOps.getServer(ServerId.Type.SCAN_SERVER, null,
hp.getHost(), hp.getPort()));
- servers.add(
- instanceOps.getServer(ServerId.Type.TABLET_SERVER, null,
hp.getHost(), hp.getPort()));
+ .addAll(instanceOps.getServers(ServerId.Type.SCAN_SERVER,
rgPredicate, serverPredicate));
+ servers.addAll(
+ instanceOps.getServers(ServerId.Type.TABLET_SERVER, rgPredicate,
serverPredicate));
} else {
servers.addAll(instanceOps.getServers(ServerId.Type.SCAN_SERVER));
servers.addAll(instanceOps.getServers(ServerId.Type.TABLET_SERVER));
@@ -75,14 +80,44 @@ public class ListScansCommand extends Command {
public Options getOptions() {
final Options opts = new Options();
- tserverOption = new Option("ts", "tabletServer", true, "tablet server to
list scans for");
+ serverOpt = new Option("s", "server", true,
+ "tablet/scan server regex to list scans for. Regex will match against
strings like <host>:<port>");
+ serverOpt.setArgName("tablet/scan server regex");
+ opts.addOption(serverOpt);
+
+ // Leaving here for backwards compatibility, same as serverOpt
+ tserverOption = new Option("ts", "tabletServer", true, "tablet/scan server
to list scans for");
tserverOption.setArgName("tablet server");
opts.addOption(tserverOption);
+ rgOpt = new Option("rg", "resourceGroup", true,
+ "tablet/scan server resource group regex to list scans for");
+ rgOpt.setArgName("resource group");
+ opts.addOption(rgOpt);
+
disablePaginationOpt = new Option("np", "no-pagination", false, "disable
pagination of output");
opts.addOption(disablePaginationOpt);
return opts;
}
+ static String getServerOptValue(CommandLine cl, Option serverOpt, Option
tserverOption) {
+ Preconditions.checkArgument(!(cl.hasOption(serverOpt) &&
cl.hasOption(tserverOption)),
+ "serverOpt and tserverOption may not be both set at the same time.");
+ return cl.hasOption(serverOpt) ? cl.getOptionValue(serverOpt)
+ : cl.getOptionValue(tserverOption);
+ }
+
+ static BiPredicate<String,Integer> serverRegexPredicate(String serverRegex) {
+ return Optional.ofNullable(serverRegex).map(regex ->
Pattern.compile(regex).asMatchPredicate())
+ .map(matcherPredicate -> (BiPredicate<String,
+ Integer>) (h, p) -> matcherPredicate.test(h + ":" + p))
+ .orElse((h, p) -> true);
+ }
+
+ static Predicate<String> rgRegexPredicate(String rgRegex) {
+ return Optional.ofNullable(rgRegex).map(regex ->
Pattern.compile(regex).asMatchPredicate())
+ .orElse(rg -> true);
+ }
+
}
diff --git
a/shell/src/test/java/org/apache/accumulo/shell/commands/ListCompactionsCommandTest.java
b/shell/src/test/java/org/apache/accumulo/shell/commands/ListCompactionsCommandTest.java
new file mode 100644
index 0000000000..71457f84b2
--- /dev/null
+++
b/shell/src/test/java/org/apache/accumulo/shell/commands/ListCompactionsCommandTest.java
@@ -0,0 +1,36 @@
+/*
+ * 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.shell.commands;
+
+import org.apache.commons.cli.ParseException;
+import org.junit.jupiter.api.Test;
+
+public class ListCompactionsCommandTest {
+
+ @Test
+ public void testServerRegexPredicate() throws ParseException {
+ ListScansCommandTest.testServerRegexPredicate(new
ListCompactionsCommand());
+ }
+
+ @Test
+ public void testResourceGroupRegexPredicate() throws ParseException {
+ ListScansCommandTest.testResourceGroupRegexPredicate(new
ListCompactionsCommand());
+ }
+
+}
diff --git
a/shell/src/test/java/org/apache/accumulo/shell/commands/ListScansCommandTest.java
b/shell/src/test/java/org/apache/accumulo/shell/commands/ListScansCommandTest.java
new file mode 100644
index 0000000000..d95a2af3b9
--- /dev/null
+++
b/shell/src/test/java/org/apache/accumulo/shell/commands/ListScansCommandTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.shell.commands;
+
+import static
org.apache.accumulo.shell.commands.ListScansCommand.getServerOptValue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
+
+import org.apache.accumulo.shell.Shell.Command;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.junit.jupiter.api.Test;
+
+public class ListScansCommandTest {
+
+ @Test
+ public void testTetServerOptValue() throws ParseException {
+ var cmd = new ListScansCommand();
+ CommandLineParser parser = new DefaultParser();
+ Options opts = cmd.getOptions();
+ Option serverOpt = opts.getOption("-s");
+ Option tserverOpt = opts.getOption("-ts");
+
+ assertThrows(IllegalArgumentException.class,
+ () -> getServerOptValue(
+ parser.parse(opts, new String[] {"-s", "server:1", "-ts",
"server:2"}), serverOpt,
+ tserverOpt));
+ assertEquals("server:1", getServerOptValue(parser.parse(opts, new String[]
{"-s", "server:1"}),
+ serverOpt, tserverOpt));
+ assertEquals("server:2", getServerOptValue(parser.parse(opts, new String[]
{"-ts", "server:2"}),
+ serverOpt, tserverOpt));
+ }
+
+ @Test
+ public void testServerRegexPredicate() throws ParseException {
+ testServerRegexPredicate(new ListScansCommand());
+ }
+
+ @Test
+ public void testResourceGroupRegexPredicate() throws ParseException {
+ testResourceGroupRegexPredicate(new ListScansCommand());
+ }
+
+ static void testServerRegexPredicate(Command cmd) throws ParseException {
+ Options opts = cmd.getOptions();
+ CommandLineParser parser = new DefaultParser();
+
+ List<String> matching =
+ List.of(".*:[0-9]*", "local.*:2000", "localhost:2000",
"l.*:200[0-9].*");
+ for (String serverRegex : matching) {
+ for (String opt : List.of("-s", "-ts")) {
+ var predicate = buildServerPredicate(opts, parser, opt, serverRegex);
+ assertTrue(predicate.test("localhost", 2000));
+ }
+ }
+
+ List<String> nonMatching = List.of(".*:[0-1]*", "local.*:2100",
"localhost:3000", "localhost");
+ for (String serverRegex : nonMatching) {
+ for (String opt : List.of("-s", "-ts")) {
+ var predicate = buildServerPredicate(opts, parser, opt, serverRegex);
+ assertFalse(predicate.test("localhost", 2000));
+ }
+ }
+
+ }
+
+ static void testResourceGroupRegexPredicate(Command cmd) throws
ParseException {
+ Options opts = cmd.getOptions();
+ CommandLineParser parser = new DefaultParser();
+
+ List<String> matching = List.of(".*", "test.*", ".*group", "testgroup");
+ for (String rgRegex : matching) {
+ var predicate = buildResourceGroupPredicate(opts, parser, rgRegex);
+ assertTrue(predicate.test("testgroup"));
+ }
+
+ List<String> nonMatching = List.of(".*gro", "test.*gr", "testgroup1",
"tg.*");
+ for (String rgRegex : nonMatching) {
+ var predicate = buildResourceGroupPredicate(opts, parser, rgRegex);
+ assertFalse(predicate.test("testgroup"));
+ }
+ }
+
+ static BiPredicate<String,Integer> buildServerPredicate(Options opts,
CommandLineParser parser,
+ String opt, String serverRegex) throws ParseException {
+
+ // Test flags for server regex
+ String[] args = {opt, serverRegex};
+ Option serverOpt = opts.getOption(opt);
+ CommandLine cli = parser.parse(opts, args);
+ return
ListScansCommand.serverRegexPredicate(cli.getOptionValue(serverOpt));
+ }
+
+ static Predicate<String> buildResourceGroupPredicate(Options opts,
CommandLineParser parser,
+ String rgRegex) throws ParseException {
+
+ // Test flag works for resource group regex
+ String[] args = {"-rg", rgRegex};
+ Option serverOpt = opts.getOption("-rg");
+ CommandLine cli = parser.parse(opts, args);
+ return ListScansCommand.rgRegexPredicate(cli.getOptionValue(serverOpt));
+ }
+}
diff --git
a/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java
b/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java
index dfbea21b04..791be6ff83 100644
--- a/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java
@@ -48,6 +48,7 @@ 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.SortedSet;
import java.util.regex.Pattern;
@@ -1528,12 +1529,23 @@ public class ShellServerIT extends
SharedMiniClusterBase {
ts.exec("insert d cf cq value", true);
ts.exec("flush -t " + table, true);
ts.exec("sleep 0.2", true);
- ts.exec("listcompactions", true, "default_tablet");
+ verifyListCompactions("listcompactions", "default_tablet");
+ // basic regex filtering test, more tests are in ListCompactionsCommandTest
+ verifyListCompactions("listcompactions -s .*:[0-9]*", "default_tablet");
+ verifyListCompactions("listcompactions -rg def.*", "default_tablet");
+ verifyListCompactions("listcompactions -s .*:[0-9]* -rg def.*",
"default_tablet");
+ // non matching
+ assertFalse(ts.exec("listcompactions -s bad.*",
true).contains("default_tablet"));
+ assertFalse(ts.exec("listcompactions -rg bad.*",
true).contains("default_tablet"));
+ ts.exec("deletetable -f " + table, true);
+ }
+
+ private void verifyListCompactions(String cmd, String expected) throws
IOException {
+ ts.exec(cmd, true, expected);
String[] lines = ts.output.get().split("\n");
String last = lines[lines.length - 1];
String[] parts = last.split("\\|");
assertEquals(13, parts.length);
- ts.exec("deletetable -f " + table, true);
}
@Test
@@ -1655,6 +1667,26 @@ public class ShellServerIT extends SharedMiniClusterBase
{
ts.exec("insert " + i + " cf cq value", true);
}
+ // Sanity checks that the regex will match
+ // Full regex tests are done in ListScansCommandTest
+ listscans(table, null, null, true);
+ listscans(table, ".*:[0-9]*", null, true);
+ listscans(table, null, "def.*", true);
+ listscans(table, ".*:[0-9]*", "def.*", true);
+
+ // check not matching
+ listscans(table, null, "bad.*", false);
+ listscans(table, "bad.*", null, false);
+
+ ts.exec("deletetable -f " + table, true);
+ }
+
+ private void listscans(String table, String serverRegex, String rgRegex,
boolean match)
+ throws Exception {
+ final StringBuilder cmd = new StringBuilder("listscans");
+ Optional.ofNullable(serverRegex).ifPresent(sr -> cmd.append(" -s
").append(sr));
+ Optional.ofNullable(rgRegex).ifPresent(rgr -> cmd.append(" -rg
").append(rgr));
+
try (AccumuloClient accumuloClient =
Accumulo.newClient().from(getClientProps()).build();
Scanner s = accumuloClient.createScanner(table, Authorizations.EMPTY))
{
IteratorSetting cfg = new IteratorSetting(30, SlowIterator.class);
@@ -1667,9 +1699,12 @@ public class ShellServerIT extends SharedMiniClusterBase
{
thread.start();
List<String> scans = new ArrayList<>();
- // Try to find the active scan for about 15seconds
- for (int i = 0; i < 50 && scans.isEmpty(); i++) {
- String currentScans = ts.exec("listscans", true);
+ // Try to find the active scan for about 15 seconds when should match
+ // else just 1 second to speed up test as the tests for the unmatching
case
+ // come after the matching so the scan should list quickly if they will
match
+ int attempts = match ? 50 : 3;
+ for (int i = 0; i < attempts && scans.isEmpty(); i++) {
+ String currentScans = ts.exec(cmd.toString(), true);
log.info("Got output from listscans:\n{}", currentScans);
String[] lines = currentScans.split("\n");
for (int scanOffset = 2; scanOffset < lines.length; scanOffset++) {
@@ -1685,31 +1720,34 @@ public class ShellServerIT extends
SharedMiniClusterBase {
}
thread.join();
- assertFalse(scans.isEmpty(), "Could not find any active scans over table
" + table);
+ if (match) {
+ assertFalse(scans.isEmpty(), "Could not find any active scans over
table " + table);
- for (String scan : scans) {
- if (!scan.contains("RUNNING")) {
- log.info("Ignoring scan because it doesn't contain 'RUNNING': {}",
scan);
- continue;
+ for (String scan : scans) {
+ if (!scan.contains("RUNNING")) {
+ log.info("Ignoring scan because it doesn't contain 'RUNNING': {}",
scan);
+ continue;
+ }
+ String[] parts = scan.split("\\|");
+ assertEquals(15, parts.length, "Expected 15 colums, but found " +
parts.length
+ + " instead for '" + Arrays.toString(parts) + "'");
+ String tserver = parts[1].trim();
+ // TODO: any way to tell if the client address is accurate? could be
local IP, host,
+ // loopback...?
+ String hostPortPattern = ".+:\\d+";
+ assertMatches(tserver, hostPortPattern);
+
assertTrue(accumuloClient.instanceOperations().getServers(ServerId.Type.TABLET_SERVER)
+ .stream().anyMatch((srv) ->
srv.toHostPortString().equals(tserver)));
+ String client = parts[1].trim();
+ assertMatches(client, hostPortPattern);
+ // Scan ID should be a long (throwing an exception if it fails to
parse)
+ Long r = Long.parseLong(parts[12].trim());
+ assertNotNull(r);
}
- String[] parts = scan.split("\\|");
- assertEquals(15, parts.length, "Expected 15 colums, but found " +
parts.length
- + " instead for '" + Arrays.toString(parts) + "'");
- String tserver = parts[1].trim();
- // TODO: any way to tell if the client address is accurate? could be
local IP, host,
- // loopback...?
- String hostPortPattern = ".+:\\d+";
- assertMatches(tserver, hostPortPattern);
-
assertTrue(accumuloClient.instanceOperations().getServers(ServerId.Type.TABLET_SERVER)
- .stream().anyMatch((srv) ->
srv.toHostPortString().equals(tserver)));
- String client = parts[1].trim();
- assertMatches(client, hostPortPattern);
- // Scan ID should be a long (throwing an exception if it fails to
parse)
- Long r = Long.parseLong(parts[12].trim());
- assertNotNull(r);
+ } else {
+ assertTrue(scans.isEmpty(), "Should not find any active scans over
table " + table);
}
}
- ts.exec("deletetable -f " + table, true);
}
@Test
