pool: update sweeper purge command to accept option storage class

Motivation:
Sometimes admin want to purge only one particular storage class, but
`sweep purge` command doesn't accept a filter.

Modification:
Update `sweeper purge` command to accept optional `-storageClass` option.

Result:
More flexible file purging in pool

Acked-by: Dmitry Litvintsev
Target: master
Require-book: no
Require-notes: yes

modules/dcache/src/main/java/org/dcache/pool/classic/SpaceSweeper2.java

@@ -26,6 +26,8 @@
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.concurrent.Callable;
+import java.util.function.Predicate;
+
 import org.dcache.namespace.FileAttribute;
 import org.dcache.pool.PoolDataBeanProvider;
 import org.dcache.pool.classic.json.SweeperData;
@@ -224,19 +226,25 @@ public String call() {
                 "even if it has been marked for removal.")
     public class SweeperPurgeCommand implements Callable<String> {
 
+
+        @Option(name="storageClass", usage = "Purge only files of the specified storage class.", metaVar = "storage-class")
+        String storageClass;
+
         @Override
         public String call() {
             new Thread("sweeper-purge") {
                 @Override
                 public void run() {
                     try {
-                        long bytes = reclaim(Long.MAX_VALUE, "'sweeper purge' command");
+                        Predicate<CacheEntry> filter = storageClass == null ? e -> true
+                                : e -> e.getFileAttributes().getStorageClass().equals(storageClass);
+                        long bytes = reclaim(Long.MAX_VALUE, "'sweeper purge' command", filter);
                         LOGGER.info("'sweeper purge' reclaimed {} bytes.", bytes);
                     } catch (InterruptedException e) {
                     }
                 }
             }.start();
-            return "Purging all removable files from pool.";
+            return "Purging "  + (storageClass == null ? "all" : storageClass) + " removable files from pool.";
         }
     }
 
@@ -254,7 +262,7 @@ public String call() {
                 @Override
                 public void run() {
                     try {
-                        long bytes = reclaim(bytesToFree, "'sweeper free' command");
+                        long bytes = reclaim(bytesToFree, "'sweeper free' command", e -> true);
                         LOGGER.info("'sweeper free {}' reclaimed {} bytes.", bytesToFree, bytes);
                     } catch (InterruptedException e) {
                     }
@@ -402,7 +410,15 @@ public String call() {
         }
     }
 
-    private long reclaim(long amount, String why)
+    /**
+     * Reclaims space by removing removable files from the pool.
+     * @param amount the amount of space to reclaim.
+     * @param why the reason for reclaiming space.
+     * @param filter remove only files that match the filter
+     * @return the amount of space reclaimed.
+     * @throws InterruptedException
+     */
+    private long reclaim(long amount, String why, Predicate<CacheEntry> filter)
           throws InterruptedException {
         LOGGER.debug("Sweeper tries to reclaim {} bytes.", amount);
 
@@ -429,6 +445,11 @@ private long reclaim(long amount, String why)
                     continue;
                 }
 
+                if (!filter.test(entry)) {
+                    LOGGER.debug("File skipped by sweeper (filter): {}", entry);
+                    continue;
+                }
+
                 long size = entry.getReplicaSize();
                 LOGGER.debug("Sweeper removes {}.", id);
                 _repository.setState(id, ReplicaState.REMOVED, why);
@@ -475,7 +496,7 @@ public long waitForRequests()
     public void run() {
         try {
             while (true) {
-                if (reclaim(waitForRequests(), "sweeper making space for new data") == 0) {
+                if (reclaim(waitForRequests(), "sweeper making space for new data", e -> true) == 0) {
                     /* The list maintained by the sweeper is imperfect
                      * in the sense that it can contain locked entries
                      * or entries in use. Thus we could be caught in a