Merge branch 'zones'

Author: lens161

modules/dcache/src/main/java/diskCacheV111/poolManager/PoolMonitorV5.java

@@ -252,18 +252,45 @@ public SelectedPool selectWritePool(long preallocated)
                             .toLowerCase()));
             }
 
+            if (_zone.isPresent()) {
+                SelectedPool pool = filterWritePool(levels, true, preallocated);
+                if (pool != null) {
+                    return pool;
+                }
+            }
+
+            SelectedPool pool = filterWritePool(levels, false, preallocated);
+            if (pool != null) {
+                return pool;
+            }
+
+            throw new CacheException(CacheException.NO_POOL_ONLINE,
+                  noOnlinePoolsErrorMessage(DirectionType.WRITE.name()
+                        .toLowerCase()));
+        }
+
+       @Nullable
+        private SelectedPool filterWritePool(PoolPreferenceLevel[] levels,
+                                           boolean filterByZone,
+                                           long preallocated) throws CacheException{
             CostException fallback = null;
+
             for (PoolPreferenceLevel level : levels) {
                 List<PoolInfo> pools =
-                      level.getPoolList().stream()
-                            .map(_costModule::getPoolInfo)
-                            .filter(Objects::nonNull)
-                            .collect(toList());
+                        level.getPoolList().stream()
+                                .map(_costModule::getPoolInfo)
+                                .filter(Objects::nonNull)
+                                .collect(toList());
+
+                if (filterByZone){
+                    pools = filterByZone(pools);
+                }
+
                 if (!pools.isEmpty()) {
                     Partition partition = _partitionManager.getPartition(level.getTag());
                     try {
                         return partition.selectWritePool(_costModule, pools, _fileAttributes,
-                              preallocated);
+                                preallocated);
                     } catch (CostException e) {
                         if (!e.shouldFallBack()) {
                             throw e;
@@ -273,16 +300,11 @@ public SelectedPool selectWritePool(long preallocated)
                 }
             }
 
-            /* We were asked to fall back, but all available links were
-             * exhausted. Let the caller deal with it.
-             */
-            if (fallback != null) {
+            if (!filterByZone && fallback != null) {
                 throw fallback;
             }
 
-            throw new CacheException(CacheException.NO_POOL_ONLINE,
-                  noOnlinePoolsErrorMessage(DirectionType.WRITE.name()
-                        .toLowerCase()));
+            return null;
         }
 
         @Override
@@ -316,12 +338,37 @@ public SelectedPool selectReadPool()
                             .toLowerCase()));
             }
 
+
+            if (_zone.isPresent()) {
+                SelectedPool pool = filterReadPool(level, onlinePoolsWithFile, true);
+                if (pool != null) {
+                    return pool;
+                }
+            }
+
+            SelectedPool pool = filterReadPool(level, onlinePoolsWithFile, false);
+            if (pool != null) {
+                return pool;
+            }
+
+            /* None of the pools we were allowed to read from were
+             * online or had the file.
+             */
+            throw new PermissionDeniedCacheException(
+                  "File is online, but not in read-allowed pool");
+        }
+
+        @Nullable
+        private SelectedPool filterReadPool(PoolPreferenceLevel[] level,
+                                  Map<String, PoolInfo> onlinePoolsWithFile,
+                                  boolean filterByZone) throws CacheException {
+
             CostException costException = null;
 
             for (int prio = 0; prio < level.length; prio++) {
                 List<String> poolsInCurrentLevel = level[prio].getPoolList();
                 LOGGER.debug("[read] Allowed pools at level {}: {}",
-                      prio, poolsInCurrentLevel);
+                        prio, poolsInCurrentLevel);
 
                 if (poolsInCurrentLevel.isEmpty()) {
                     // No pools in this level....skip it.
@@ -339,7 +386,11 @@ public SelectedPool selectReadPool()
                     }
                 }
                 LOGGER.debug("[read] Available pools at level {}: {}",
-                      prio, pools);
+                        prio, pools);
+
+                if (filterByZone) {
+                    pools = filterByZone(pools);
+                }
 
                 /* If allowed, fallback to next link if current link doesn't point
                  * to any pool holding the file.
@@ -356,14 +407,14 @@ public SelectedPool selectReadPool()
                  * to select a pool.
                  */
                 _partition =
-                      _partitionManager.getPartition(level[prio].getTag());
+                        _partitionManager.getPartition(level[prio].getTag());
 
                 /* The actual pool selection is delegated to the
                  * Partition.
                  */
                 try {
                     return _partition.selectReadPool(_costModule, pools,
-                          _fileAttributes);
+                            _fileAttributes);
                 } catch (CostException e) {
                     costException = e;
                     if (!e.shouldFallBack()) {
@@ -372,25 +423,24 @@ public SelectedPool selectReadPool()
                 }
             }
 
-            /* If we have a CostException where a pool was selected and
-             * shouldTryAlternatives not set then we return that pool anyway.
-             * REVISIT: consider updating partitions so they don't throw
-             * an exception in this case.
-             */
+            if (filterByZone){
+                return null;
+            }
             if (costException != null) {
                 if (costException.getPool() != null
-                      && !costException.shouldTryAlternatives()) {
+                        && !costException.shouldTryAlternatives()) {
                     return costException.getPool();
                 }
-
                 throw costException;
             }
+            return null;
+        }
 
-            /* None of the pools we were allowed to read from were
-             * online or had the file.
-             */
-            throw new PermissionDeniedCacheException(
-                  "File is online, but not in read-allowed pool");
+        private List<PoolInfo> filterByZone(List<PoolInfo> pools) {
+                pools = pools.stream()
+                        .filter(p -> Objects.equals(p.getTags().get("zone"), _zone.get()))
+                        .toList();
+            return pools;
         }
 
         @Nullable
@@ -459,6 +509,7 @@ public Partition.P2pPair selectPool2Pool(String poolGroup,
                           Lists.newArrayList(sources.values()),
                           pools,
                           _fileAttributes,
+                          _zone,
                           force);
                 }
             }
@@ -470,25 +521,48 @@ public Partition.P2pPair selectPool2Pool(String poolGroup,
 
         @Override
         public SelectedPool selectStagePool(Optional<PoolInfo> previous)
-              throws CacheException {
+                throws CacheException {
             Collection<String> locations = filteredFileLocations();
             LOGGER.debug("[stage] Existing locations of the file: {}", locations);
 
+            if (_zone.isPresent()) {
+                SelectedPool pool = filterStagePool(locations, previous, true);
+                if (pool != null) {
+                    return pool;
+                }
+            }
+
+            SelectedPool pool = filterStagePool(locations, previous, false);
+            if (pool != null) {
+                return pool;
+            }
+
+            throw new CacheException(149, "No pool candidates available/configured/left for stage");
+        }
+
+        public SelectedPool filterStagePool(Collection<String> locations,
+                                            Optional<PoolInfo> previous,
+                                            boolean filterByZone) throws CacheException{
+
             CostException costException = null;
             for (PoolPreferenceLevel level : match(DirectionType.CACHE)) {
                 try {
                     List<PoolInfo> pools =
-                          level.getPoolList().stream()
-                                .filter(pool -> !locations.contains(pool))
-                                .map(_costModule::getPoolInfo)
-                                .filter(Objects::nonNull)
-                                .collect(toList());
+                            level.getPoolList().stream()
+                                    .filter(pool -> !locations.contains(pool))
+                                    .map(_costModule::getPoolInfo)
+                                    .filter(Objects::nonNull)
+                                    .collect(toList());
+                    if (filterByZone) {
+                        pools = filterByZone(pools);
+                    }
+
                     if (!pools.isEmpty()) {
                         LOGGER.debug("[stage] Online stage candidates: {}", pools);
                         Partition partition =
-                              _partitionManager.getPartition(level.getTag());
+                                _partitionManager.getPartition(level.getTag());
                         return partition.selectStagePool(_costModule, pools,
-                              previous, _fileAttributes);
+                                previous, _fileAttributes);
                     }
                 } catch (CostException e) {
                     costException = e;
@@ -498,15 +572,15 @@ public SelectedPool selectStagePool(Optional<PoolInfo> previous)
                 }
             }
 
-            if (costException != null) {
+            if (costException != null && !filterByZone) {
                 SelectedPool pool = costException.getPool();
                 if (pool != null) {
                     return pool;
                 }
                 throw costException;
             }
 
-            throw new CacheException(149, "No pool candidates available/configured/left for stage");
+            return null;
         }
 
         // FIXME: There is a fair amount of overlap between this method

modules/dcache/src/main/java/org/dcache/poolmanager/BufferPartition.java

@@ -145,7 +145,8 @@ public SelectedPool selectReadPool(CostModule cm, List<PoolInfo> pools,
 
     @Override
     public P2pPair selectPool2Pool(CostModule cm, List<PoolInfo> src, List<PoolInfo> dst,
-          FileAttributes attributes, boolean force) throws CacheException {
+          FileAttributes attributes, Optional<String> zone, boolean force) throws CacheException {
+        dst = filterByZone(dst, zone);
         return new P2pPair(selectPool(src), selectPool(dst, attributes.getSize()));
     }
 

modules/dcache/src/main/java/org/dcache/poolmanager/LruPartition.java

@@ -133,8 +133,10 @@ public P2pPair selectPool2Pool(CostModule cm,
           List<PoolInfo> src,
           List<PoolInfo> dst,
           FileAttributes attributes,
+          Optional<String> zone,
           boolean force)
           throws CacheException {
+        dst = filterByZone(dst, zone);
         return new P2pPair(new SelectedPool(select(src, _lastRead)),
               selectWritePool(cm, dst, attributes, attributes.getSize()));
     }

modules/dcache/src/main/java/org/dcache/poolmanager/Partition.java

@@ -13,8 +13,11 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
+import java.util.Objects;
+
 import org.dcache.vehicles.FileAttributes;
 
 /**
@@ -335,6 +338,7 @@ public abstract P2pPair selectPool2Pool(CostModule cm,
           List<PoolInfo> src,
           List<PoolInfo> dst,
           FileAttributes attributes,
+          Optional<String> zone,
           boolean force)
           throws CacheException;
 
@@ -350,6 +354,15 @@ public abstract SelectedPool selectStagePool(CostModule cm,
           FileAttributes attributes)
           throws CacheException;
 
+    static List<PoolInfo> filterByZone(List<PoolInfo> pools, Optional<String> zone) {
+        if (zone.isEmpty()) return pools;
+        List<PoolInfo> poolsUnfiltered = pools;
+        pools = pools.stream()
+                .filter(p -> Objects.equals(p.getTags().get("zone"), zone.get()))
+                .toList();
+        return (pools.isEmpty()) ? poolsUnfiltered : pools;
+    }
+
     /**
      * Immutable helper class to represent a source and destination pair for pool to pool
      * selection.

modules/dcache/src/main/java/org/dcache/poolmanager/RandomPartition.java

@@ -82,8 +82,10 @@ public SelectedPool selectReadPool(CostModule cm,
           List<PoolInfo> src,
           List<PoolInfo> dst,
           FileAttributes attributes,
+          Optional<String> zone,
           boolean force)
           throws CacheException {
+        dst = filterByZone(dst, zone);
         return new P2pPair(new SelectedPool(select(src)),
               selectWritePool(cm, dst, attributes, attributes.getSize()));
     }

modules/dcache/src/main/java/org/dcache/poolmanager/WRandomPartition.java

@@ -51,7 +51,7 @@ public String getType() {
     }
 
     @Override
-    public P2pPair selectPool2Pool(CostModule cm, List<PoolInfo> src, List<PoolInfo> dst, FileAttributes attributes, boolean force) throws CacheException {
+    public P2pPair selectPool2Pool(CostModule cm, List<PoolInfo> src, List<PoolInfo> dst, FileAttributes attributes, Optional<String> zone, boolean force) throws CacheException {
 
         Collections.shuffle(src);
 
@@ -62,6 +62,8 @@ public P2pPair selectPool2Pool(CostModule cm, List<PoolInfo> src, List<PoolInfo>
             List<PoolInfo> tryList = Lists.newArrayList(
                     filter(dst, new DifferentHost(srcPoolInfo.getHostName())));
 
+            tryList = filterByZone(tryList, zone);
+
             if (!tryList.isEmpty()) {
                 SelectedPool destPoolInfo = selectWritePool(cm, tryList, attributes, attributes.getSize());
                 return new P2pPair(new SelectedPool(srcPoolInfo), destPoolInfo);

modules/dcache/src/main/java/org/dcache/poolmanager/WassPartition.java

@@ -83,6 +83,7 @@ public P2pPair selectPool2Pool(CostModule cm,
           List<PoolInfo> src,
           List<PoolInfo> dst,
           FileAttributes attributes,
+          Optional<String> zone,
           boolean force)
           throws CacheException {
         checkState(!src.isEmpty());
@@ -153,6 +154,8 @@ public P2pPair selectPool2Pool(CostModule cm,
                           .collect(toList());
                 }
 
+                destinations = filterByZone(destinations, zone);
+
                 PoolInfo destination =
                       wass.selectByAvailableSpace(destinations, filesize, PoolInfo::getCostInfo);
                 if (destination != null) {
@@ -169,6 +172,8 @@ public P2pPair selectPool2Pool(CostModule cm,
             }
         }
 
+        dst = filterByZone(dst, zone);
+
         PoolInfo destination = wass.selectByAvailableSpace(dst, filesize, PoolInfo::getCostInfo);
         if (destination == null) {
             throw new DestinationCostException("All pools are full");

modules/dcache/src/test/java/org/dcache/mock/SimplePartition.java

@@ -66,7 +66,7 @@ public SelectedPool selectReadPool(CostModule cm, List<PoolInfo> pools,
 
     @Override
     public P2pPair selectPool2Pool(CostModule cm, List<PoolInfo> src,
-          List<PoolInfo> dst, FileAttributes attributes, boolean force)
+          List<PoolInfo> dst, FileAttributes attributes, Optional<String> zone, boolean force)
           throws CacheException {
         throw new UnsupportedOperationException("Not supported yet.");
     }