admin-shell: bulk send commands to zone

Motivation:
There was no way to scope a bulk admin-shell command to a zones.

Modification:
Add a '#<zone>' suffix to '\s' command patterns that filters targets
to cells whose domain is registered under that zone in ZooKeeper.
Domains register themselves on startup when dcache.zone is set.

Result:
Admins can now run e.g. '\s pool*#<zone> rep ls' to target a zone.

Target: master
Require-book: yes

Author: lens161

docs/TheBook/src/main/markdown/config-admin.md

@@ -181,7 +181,7 @@ send supported cell commands to other cells. Once logged in you are prompted to
 \h [command]...  # display help for cell commands
 \l [cell[@domain]|pool/poolgroup]...  # list cells
 \q # quit
-\s [OPTIONS] (cell[@domain]|pool/poolgroup)[,(cell[@domain]|pool/poolgroup)]... command...  # send command
+\s [OPTIONS] (cell[@domain]|pool[#zone][/poolgroup])[,(cell[@domain]|pool[#zone][/poolgroup])]... command...  # send command
 \sl [options] pnfsid|path command...  # send to locations
 \sn [options] command...  # send pnfsmanager command
 \sp [options] command...  # send poolmanager command
@@ -191,7 +191,7 @@ send supported cell commands to other cells. Once logged in you are prompted to
 ```
 
 Shell commands are always available at command prompt, whereas in order to execute cell commands you have to either connect to the cell
-using `\c cell[@domain]` and execute command or send command to the cell using `\s [OPTIONS] (cell[@domain]|pool/poolgroup)[,(cell[@domain]|pool/poolgroup)]... command...`. For instance:
+using `\c cell[@domain]` and execute command or send command to the cell using `\s [OPTIONS] (cell[@domain]|pool[#zone][/poolgroup])[,...] command...`. For instance:
 
 ```
 [headnode] (local) enstore > \? \c
@@ -355,6 +355,41 @@ pool_2={Tag={{hostname=example.org}};size=0;SC=2.7939677238464355E-4;CC=0.0;}
 
 While the first line for each pool gives the information stored in the cache of the cost module, the second line gives the    costs (SC: [space cost](rf-glossary.md#space-cost), CC: [performance cost](rf-glossary.md#performance-cost)) calculated for a (hypothetical) file of zero size. For details on how these are calculated and their meaning, see [the section called “Classic Partitions”](#config-poolmanager.md#classic-partitions).
 
+## Zone-scoped bulk commands
+
+When dCache is deployed with zones (`dcache.zone` property), the `\s` command supports a `#<zone>` suffix to restrict a bulk command to cells that belong to a specific zone.
+
+The general syntax is:
+
+```
+\s [OPTIONS] <cellPattern>#<zone>[/<poolgroup>] command...
+```
+
+For example, to send `rep ls` to all pools in zone `zone`:
+
+```
+(local) admin > \s pool*#zone rep ls
+```
+
+To further restrict to a pool group within the zone:
+
+```
+(local) admin > \s pool*#zone/poolGroup rep ls
+```
+
+The zone filter is resolved through ZooKeeper: each domain registers itself under
+`/dcache/zones/<zone>/<domainName>` at startup when `dcache.zone` is configured.
+The admin shell looks up the children of that ZooKeeper path and keeps only those
+cells whose domain is listed there.
+
+You can combine the zone filter with glob patterns as usual:
+
+```
+(local) admin > \s *#zone st set max-active 10
+```
+
+This sends the command to every well-known cell in the selected zone.
+
 ## Creating a new user
 
 To create a new user, <new-user> and set a new password for the user `\c` from the local prompt `((local) admin >)` to the acm, the access control manager, and run following command sequence:

modules/dcache/src/main/java/diskCacheV111/admin/UserAdminShell.java

@@ -25,6 +25,7 @@
 import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
 import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Range;
 import com.google.common.util.concurrent.Futures;
@@ -80,6 +81,8 @@
 import java.util.stream.StreamSupport;
 import jline.console.completer.Completer;
 import jline.console.completer.StringsCompleter;
+import org.apache.curator.framework.CuratorFramework;
+import org.checkerframework.checker.nullness.qual.NonNull;
 import org.dcache.auth.Subjects;
 import org.dcache.auth.attributes.Restrictions;
 import org.dcache.cells.CellStub;
@@ -157,6 +160,8 @@ public class UserAdminShell
      */
     private Completer _pnfsManagerCompleter;
 
+    private CuratorFramework _curator;
+
     /**
      * Client handler for listing directories in the dCache name space.
      */
@@ -238,6 +243,10 @@ public void setPnfsManager(CellStub stub) {
         _pnfsManagerCompleter = createRemoteCompleter(_pnfsManager.getDestinationPath());
     }
 
+    public void setCurator(CuratorFramework curator){
+        _curator = curator;
+    }
+
     public void setListHandler(ListDirectoryHandler list) {
         _list = list;
     }
@@ -390,8 +399,7 @@ private ListenableFuture<List<String>> getPoolsInGroups(Predicate<String> predic
      * The result is sorted lexicographically and case insensitive, but the order of the input
      * patterns is preserved (ie. the output contains matching cells in the same order).
      */
-    private List<String> expandCellPatterns(List<String> patterns)
-          throws CacheException, InterruptedException, ExecutionException, NoRouteToCellException {
+    private List<String> expandCellPatterns(List<String> patterns) throws CommandException, CacheException, NoRouteToCellException, InterruptedException, ExecutionException {
         /* Query domains and well-known cells on demand. */
         Supplier<Future<Map<String, Collection<String>>>> domains =
               Suppliers.memoize(() -> transform(_cellStub.send(new CellPath("RoutingMgr"),
@@ -414,32 +422,66 @@ private List<String> expandCellPatterns(List<String> patterns)
                 continue;
             }
 
-            i = pattern.indexOf('/');
-            if (i >= 0) {
-                Predicate<String> matchesPool = toGlobPredicate(pattern.substring(0, i));
+            List<ListenableFuture<List<String>>> zoneCellFutures;
+            List<String> zoneCells;
+
+            int zoneIdx = pattern.indexOf('#');
+            int pGroupIdx = pattern.indexOf('/');
+            if (zoneIdx >= 0) {
+                String zone = pGroupIdx < 0 ? pattern.substring(zoneIdx + 1) : pattern.substring(zoneIdx + 1, pGroupIdx);
+                try {
+                    List<String> zoneDomains = _curator.getChildren().forPath("/dcache/zones/" + zone);
+                    Predicate<String> matchesCellName = toGlobPredicate(pattern.substring(0, zoneIdx));
+                    zoneCellFutures = zoneDomains.stream()
+                            .sorted(CASE_INSENSITIVE_ORDER)
+                            .map(domain -> getCells(domain, matchesCellName))
+                            .toList();
+
+                    zoneCells = Objects.requireNonNull(allAsList(zoneCellFutures).get()).stream()
+                            .flatMap(List::stream)
+                            .map(addr -> addr.contains("@") ? addr.substring(0, addr.indexOf('@')) : addr)
+                            .toList();
+
+                } catch (Exception e) {
+                    throw new CommandException("Can not send to zone: " + zone, e.getCause());
+                }
+
+            } else {
+                zoneCells = null;
+                zoneCellFutures = null;
+            }
 
-                if (i + 1 == pattern.length()) {
+            if (pGroupIdx >= 0) {
+                int endPoolName = (zoneIdx >= 0) ? zoneIdx : pGroupIdx;
+                Predicate<String> matchesPool = toGlobPredicate(pattern.substring(0, endPoolName));
+                Predicate<String> isInZone = cell -> zoneCells == null || zoneCells.contains(cell);
+                if (pGroupIdx + 1 == pattern.length()) {
                     /* Special case when no pool group is specified - matches over all pools, even those
                      * not in a pool group.
                      */
                     futures.add(transform(getPools(matchesPool),
                           (List<String> pools) ->
                                 pools.stream()
-                                      .sorted(CASE_INSENSITIVE_ORDER)
-                                      .collect(toList()), MoreExecutors.directExecutor()));
+                                        .filter(isInZone)
+                                        .sorted(CASE_INSENSITIVE_ORDER)
+                                        .collect(toList()), MoreExecutors.directExecutor()));
                 } else {
                     /* Find the pools of each matching pool group.
                      */
-                    Predicate<String> matchesPoolGroup = toGlobPredicate(pattern.substring(i + 1));
+                    Predicate<String> matchesPoolGroup = toGlobPredicate(pattern.substring(pGroupIdx + 1));
                     futures.add(
                           transform(getPoolsInGroups(matchesPoolGroup),
                                 (List<String> pools) ->
                                       pools.stream()
-                                            .filter(matchesPool)
-                                            .sorted(CASE_INSENSITIVE_ORDER)
-                                            .collect(toList()), MoreExecutors.directExecutor()));
+                                              .filter(matchesPool)
+                                              .filter(isInZone)
+                                              .sorted(CASE_INSENSITIVE_ORDER)
+                                              .collect(toList()), MoreExecutors.directExecutor()));
                 }
                 continue;
+            } else if (zoneCellFutures != null) {
+                futures.addAll(zoneCellFutures);
+                continue;
             }
 
             Predicate<String> matchesCellName = toGlobPredicate(pattern);
@@ -463,7 +505,7 @@ private List<String> expandCellPatterns(List<String> patterns)
      */
     private static boolean isExpandable(String s) {
         return !s.contains(":") && (s.startsWith("@") || s.endsWith("@") || Glob.isGlob(s)
-              || s.indexOf('/') > -1);
+              || s.indexOf('/') > -1 || s.indexOf('#') > -1);
     }
 
     /**
@@ -734,7 +776,7 @@ class PoolManagerCommand implements Callable<Serializable> {
 
         @Override
         public Serializable call()
-              throws InterruptedException, NoRouteToCellException, CommandException, AclException {
+                throws InterruptedException, NoRouteToCellException, CommandException, AclException {
             return sendObject(_poolManager.getDestinationPath(), args.toString());
         }
     }
@@ -756,8 +798,7 @@ class SendCommand implements Callable<Serializable> {
 
         @Override
         public Serializable call()
-              throws InterruptedException, ExecutionException, CacheException, AclException,
-              CommandException, NoRouteToCellException {
+                throws Exception {
             args.shift();
             AuthorizedString command = new AuthorizedString(_user, args.toString());
 
@@ -768,18 +809,24 @@ public Serializable call()
                 return sendObject(destination, command);
             }
 
-            /* Expand wildcards.
-             */
             Map<Boolean, List<String>> expandable =
-                  StreamSupport.stream(Glob.expandList(destination).spliterator(), false)
-                        .collect(partitioningBy(UserAdminShell::isExpandable));
+                    StreamSupport.stream(Glob.expandList(destination).spliterator(), false)
+                            .collect(partitioningBy(UserAdminShell::isExpandable));
             Iterable<String> destinations = concat(expandable.get(false),
-                  expandCellPatterns(expandable.get(true)));
+                    expandCellPatterns(expandable.get(true)));
 
             return sendToMany(destinations, command);
         }
     }
 
+    private @NonNull Iterable<String> expandWildcards(String destination) throws CacheException, InterruptedException, ExecutionException, NoRouteToCellException, CommandException {Map<Boolean, List<String>> expandable =
+            StreamSupport.stream(Glob.expandList(destination).spliterator(), false)
+                    .collect(partitioningBy(UserAdminShell::isExpandable));
+
+        return concat(expandable.get(false), expandCellPatterns(expandable.get(true)));
+    }
+
+
     @Command(name = "\\sl", hint = "send to locations",
           description = "Sends COMMAND to all pools hosting a copy of the given file. If the " +
                 "string $1 occurs in the command, the string is replaced by the PNFS ID " +
@@ -884,6 +931,9 @@ private int completeConnectCommand(String buffer, int cursor, List<CharSequence>
             return -1;
         } catch (InterruptedException e) {
             return -1;
+        } catch (CommandException e) {
+            LOGGER.warn(e.getMessage());
+            return -1;
         }
     }
 
@@ -923,6 +973,9 @@ private int completeCellWildcard(String buffer, int cursor, List<CharSequence> c
             return -1;
         } catch (InterruptedException e) {
             return -1;
+        } catch (CommandException e) {
+            LOGGER.warn(e.getMessage());
+            return -1;
         }
     }
 

modules/dcache/src/main/java/org/dcache/boot/Domain.java

@@ -1,6 +1,7 @@
 package org.dcache.boot;
 
 import static com.google.common.base.Strings.emptyToNull;
+
 import static org.dcache.boot.Properties.PATH_DELIMITER;
 import static org.dcache.boot.Properties.PROPERTY_DOMAIN_NAME;
 import static org.dcache.boot.Properties.PROPERTY_DOMAIN_PRELOAD;
@@ -26,7 +27,6 @@
 import static org.dcache.boot.Properties.PROPERTY_ZOOKEEPER_PEM_KEY;
 import static org.dcache.boot.Properties.PROPERTY_CRL_MODE;
 import static org.dcache.boot.Properties.PROPERTY_OCSP_MODE;
-
 import static org.dcache.util.Exceptions.genericCheck;
 
 import ch.qos.logback.classic.LoggerContext;
@@ -41,6 +41,7 @@
 import dmg.cells.nucleus.CellShell;
 import dmg.cells.nucleus.SerializationHandler;
 import dmg.cells.nucleus.SystemCell;
+import dmg.cells.zookeeper.LmPersistentNode;
 import dmg.util.CommandException;
 import java.io.File;
 import java.io.IOException;
@@ -56,13 +57,15 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 
 import eu.emi.security.authn.x509.CrlCheckingMode;
 import eu.emi.security.authn.x509.OCSPCheckingMode;
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.client.ZKClientConfig;
 import org.dcache.ssl.CanlContextFactory;
 import org.dcache.util.Args;
@@ -165,8 +168,9 @@ public void start() throws Exception {
 
         SerializationHandler.Serializer cellSerializer = SerializationHandler.enumFromConfigString(
               _properties.getValue(PROPERTY_MSG_PAYLOAD_SERIALIZER));
+        CuratorFramework curator = createCuratorFramework();
         SystemCell systemCell = SystemCell.create(domainName,
-              createCuratorFramework(), zone, cellSerializer);
+              curator, zone, cellSerializer);
         try {
             systemCell.start().get();
         } catch (ExecutionException e) {
@@ -176,6 +180,20 @@ public void start() throws Exception {
         }
         LOGGER.info("Starting {}", domainName);
 
+        if(zone.isPresent()){
+            try{
+                LmPersistentNode.createOrUpdate(
+                        curator,
+                        "/dcache/zones/" + zone.get() + "/" + domainName,
+                        new byte[0],
+                        Function.identity(),
+                        null
+                );
+            } catch (Exception e) {
+                LOGGER.warn("Failed to add zone {} in ZooKeeper", zone.get(), e);
+            }
+        }
+
         executePreload(systemCell);
         for (ConfigurationProperties serviceConfig : _services) {
             executeService(systemCell, serviceConfig);

modules/dcache/src/main/java/org/dcache/boot/Properties.java

@@ -24,6 +24,7 @@ public class Properties {
     public static final String PROPERTY_DOMAIN_PRELOAD = "dcache.domain.preload";
     public static final String PROPERTY_LOG_CONFIG = "dcache.log.configuration";
     public static final String PROPERTY_ZONE = "dcache.zone";
+    public static final String PROPERTY_ZONE_IS_MAIN = "dcache.zone.main";
     public static final String PROPERTY_MSG_PAYLOAD_SERIALIZER = "dcache.broker.channel.msg-payload-serializer";
 
     public static final String PROPERTY_ZOOKEPER_CONNECTION = "dcache.zookeeper.connection";

modules/dcache/src/main/java/org/dcache/services/ssh2/AdminShellFactory.java

@@ -4,14 +4,18 @@
 import dmg.cells.nucleus.CellEndpoint;
 import dmg.cells.nucleus.CellMessageSender;
 import java.io.File;
+
+import org.apache.curator.framework.CuratorFramework;
 import org.apache.sshd.server.channel.ChannelSession;
 import org.apache.sshd.server.command.Command;
 import org.apache.sshd.server.shell.ShellFactory;
 import org.dcache.cells.CellStub;
+import org.dcache.cells.CuratorFrameworkAware;
+import org.dcache.http.CustomResponseHeadersHandler;
 import org.dcache.util.list.ListDirectoryHandler;
 import org.springframework.beans.factory.annotation.Required;
 
-public class AdminShellFactory implements ShellFactory, CellMessageSender {
+public class AdminShellFactory implements ShellFactory, CellMessageSender, CuratorFrameworkAware {
 
     private CellEndpoint _endpoint;
     private File _historyFile;
@@ -20,6 +24,7 @@ public class AdminShellFactory implements ShellFactory, CellMessageSender {
     private CellStub _pnfsManager;
     private CellStub _poolManager;
     private CellStub _acm;
+    private CuratorFramework _curator;
     private String _prompt;
     private ListDirectoryHandler _list;
 
@@ -68,6 +73,10 @@ public void setCellEndpoint(CellEndpoint endpoint) {
         _endpoint = endpoint;
     }
 
+    public void setCuratorFramework(CuratorFramework curator) {
+        _curator = curator;
+    }
+
     @Override
     public Command createShell(ChannelSession channelSession) {
         return new ShellCommand(_historyFile, _historySize, _useColor, createAdminShell());
@@ -78,8 +87,10 @@ private UserAdminShell createAdminShell() {
         shell.setCellEndpoint(_endpoint);
         shell.setPnfsManager(_pnfsManager);
         shell.setPoolManager(_poolManager);
+        shell.setCurator(_curator);
         shell.setAcm(_acm);
         shell.setListHandler(_list);
         return shell;
     }
+
 }