poolmanager: filter staging pools by zone

Motivation:
When zones are enabled and configured, staged files should land in a pool in the same zone as the client that requested it.

Modification:
refactor selectStagePool with new filterStagePool helper method in PoolMonitorV5.

Result:
Best effort ("soft") filtering of stage pool lists by zone. Fallback to full unfiltered list, if no eligible pool in correct zone was found.
if configured, p2p should trigger and copy to a pool in clients zone when selcted stage pool is not a read pool (no extra modification needed).

Target: master
Require-book: no

Author: lens161

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

@@ -521,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;
@@ -549,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/test/java/org/dcache/tests/poolmanager/PoolMonitorTest.java

@@ -223,6 +223,39 @@ public void testWritePoolFallBackAllPoolsInZoneFull() throws Exception {
 
         assertEquals("pool2", selector.selectWritePool(0).name());
     }
+    @Test
+    public void testStagePoolZonePreference() throws Exception {
+        _partitionManager.setProperties(null, Map.of("fallback-onspace", "yes"));
+        prepareCostModule(false, true, false, false);
+
+        FileAttributes attributes = FileAttributes.of()
+                .pnfsId(_pnfsId)
+                .locations(Collections.emptyList())
+                .build();
+        StorageInfos.injectInto(_storageInfo, attributes);
+
+        PoolSelector selector = _poolMonitor.getPoolSelector(attributes, _protocolInfo,
+                null, Optional.of("1"), Collections.EMPTY_SET);
+
+        assertEquals("pool1", selector.selectStagePool(Optional.empty()).name());
+    }
+
+    @Test
+    public void testStagePoolZoneFallback() throws Exception {
+        _partitionManager.setProperties(null, Map.of("fallback-onspace", "yes"));
+        prepareCostModule(false, true, false, true);
+
+        FileAttributes attributes = FileAttributes.of()
+                .pnfsId(_pnfsId)
+                .locations(Collections.emptyList())
+                .build();
+        StorageInfos.injectInto(_storageInfo, attributes);
+
+        PoolSelector selector = _poolMonitor.getPoolSelector(attributes, _protocolInfo,
+                null, Optional.of("1"), Collections.EMPTY_SET);
+
+        assertEquals("pool2", selector.selectStagePool(Optional.empty()).name());
+    }
 
     private void prepareCostModule(boolean linkPerPool) throws Exception {
         prepareCostModule(linkPerPool, false);
@@ -255,7 +288,7 @@ private void prepareCostModule(boolean linkPerPool, boolean withZones, boolean p
         }
 
         PoolManagerPoolUpMessage pool1UpMessage = new PoolManagerPoolUpMessage("pool1",
-              serialId, poolMode, poolCost1);
+                serialId, poolMode, poolCost1);
         pool1UpMessage.setHostName(_localhost);
 
         if (withZones) {
@@ -266,21 +299,25 @@ private void prepareCostModule(boolean linkPerPool, boolean withZones, boolean p
         envelope1.addSourceAddress(new CellAddressCore("pool1"));
         _costModule.messageArrived(envelope1, pool1UpMessage);
 
+        _selectionUnit.getPool("pool1").setHsmInstances(Set.of("osm"));
+
         if (!pool2Offline) {
             PoolCostInfo poolCost2 = new PoolCostInfo("pool2", IoQueueManager.DEFAULT_QUEUE);
             poolCost2.setSpaceUsage(100, 20, 30, 50);
             PoolManagerPoolUpMessage pool2UpMessage = new PoolManagerPoolUpMessage("pool2",
-                  serialId, poolMode, poolCost2);
+                    serialId, poolMode, poolCost2);
             pool2UpMessage.setHostName(_localhost);
             if (withZones) {
                 pool2UpMessage.setTagMap(Map.of("zone", "2"));
             }
             CellMessage envelope2 = new CellMessage(new CellAddressCore("PoolManager"), null);
             envelope2.addSourceAddress(new CellAddressCore("pool2"));
             _costModule.messageArrived(envelope2, pool2UpMessage);
+            _selectionUnit.getPool("pool2").setHsmInstances(Set.of("osm"));
         }
     }
 
+
     private PoolSelector prepareHostExclusion() throws Exception {
         /*
          * Emulate the poolup message arrival on the PoolManager