stage: reapply changes; filter stageing pool by zone, test

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