pool: implememt hot file replication

Motivation:
-----------

Thousands of simultaneous accesses to the same set of files (so called popular files)
result in hot pools that dCache previously mitigated by hot pool replication that
triggered p2p of any files accessed on the hot pool when certain, configurable,
cost parameters were met. Proper tuning of this feature is non-trivial and often hot pool
replication only further aggravates the situation.

Modification:
-------------

Introduce hot file replication mechanism on pools, where the pool itself tiggers replication of
hot (popular) files. When the number of in-flight transfers for a file reaches certain configurable
threshold, the pool invokes the migration module to create a pre-defined configurable number of
replicas on other pools in the same pool group.

Result:
-------

Hot spots are eliminated by spreading the hot (popular) files across multiple pools, thus spreading
the load across multiple pools.

Ticket: -
Acked-by: Dmitry Litvintsev, Tigran Mkrtchyan
Patch: https://rb.dcache.org/r/14583/diff/raw/
Commit:
Target: trunk
Request: -
Require-book: yes
Require-notes: yes

Author: greenc-FNAL

docs/TheBook/src/main/markdown/rf-cc-pool.md

@@ -485,15 +485,21 @@ The task completed successfully
 
 ## migration ls
 
-migration ls - Lists all migration jobs.
+migration ls - List migration jobs.
 
 ### synopsis
 
 migration ls
 
 ### Description
 
-Lists all migration jobs.
+Lists migration jobs.
+
+The list of migration jobs returned includes:
+
+- All non-hot file migration jobs
+- Running hot file migration jobs
+- A limited number of finished hot file migration jobs (sorted by creation time).
 
 ## migration move
 
@@ -531,4 +537,58 @@ migration resume job
 
 Resumes a suspended migration job.
 
+## hotfile set replicas
+
+hotfile set replicas - Set the number of replicas to create for hot files.
+
+### synopsis
+
+hotfile set replicas <replicas>
+
+<replicas>
+The number of replicas to maintain for files identified as hot.
+
+### Description
+
+Sets the target number of replicas for files that exceed the hot file threshold. The system will attempt to create this many copies of the file on different pools (and preferably different hosts) to distribute the load.
+
+## hotfile get replicas
+
+hotfile get replicas - Get the current hot file replication factor.
+
+### synopsis
+
+hotfile get replicas
+
+### Description
+
+Returns the currently configured number of replicas for hot files.
+
+## hotfile set threshold
+
+hotfile set threshold - Set the access frequency threshold for identifying hot files.
+
+### synopsis
+
+hotfile set threshold <threshold>
+
+<threshold>
+The threshold value (e.g., accesses per minute) that triggers hot file replication
+
+### Description
+
+Sets the threshold for classifying a file as "hot". Files accessed more frequently than this threshold will be candidates for automatic replication.
+
+## hotfile get threshold
+
+hotfile get threshold - Get the current threshold for triggering hot file replication
+
+### synopsis
+
+hotfile get threshold
+
+### Description
+
+Returns the current threshold for triggering hot file replication
+
   [???]: #cf-pm-classic-space

modules/dcache-resilience/src/main/java/org/dcache/resilience/util/CopyLocationExtractor.java

@@ -62,6 +62,7 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import java.util.Collection;
 import java.util.Map;
 import org.dcache.resilience.data.PoolInfoMap;
+import org.dcache.util.pool.AbstractLocationExtractor;
 
 /**
  * <p>Implementation of the {@link AbstractLocationExtractor}

modules/dcache-resilience/src/main/java/org/dcache/resilience/util/CostModuleLocationExtractor.java

@@ -59,29 +59,23 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
  */
 package org.dcache.resilience.util;
 
-import com.google.common.collect.ImmutableMap;
 import diskCacheV111.poolManager.CostModule;
 import java.util.Collection;
-import java.util.Map;
-import org.dcache.poolmanager.PoolInfo;
+import org.dcache.util.pool.CostModuleTagProvider;
+import org.dcache.util.pool.PoolTagBasedExtractor;
 
 /**
- * <p>Implementation of the {@link AbstractLocationExtractor} which uses
- * {@link CostModule} to extract tags.</p>
+ * <p>Implementation of the {@link PoolTagBasedExtractor} which uses
+ * {@link CostModule} to extract tags for resilience operations.</p>
  */
-public final class CostModuleLocationExtractor extends AbstractLocationExtractor {
-
-    private CostModule module;
+public final class CostModuleLocationExtractor extends PoolTagBasedExtractor {
 
     public CostModuleLocationExtractor(Collection<String> onlyOneCopyPer,
-          CostModule module) {
-        super(onlyOneCopyPer);
-        this.module = module;
+          CostModule costModule) {
+        super(onlyOneCopyPer, createTagProvider(costModule));
     }
 
-    @Override
-    protected Map<String, String> getPoolTagsFor(String location) {
-        PoolInfo poolinfo = module.getPoolInfo(location);
-        return poolinfo == null ? ImmutableMap.of() : poolinfo.getTags();
+    private static CostModuleTagProvider createTagProvider(CostModule costModule) {
+        return new CostModuleTagProvider(costModule);
     }
 }

modules/dcache-resilience/src/main/java/org/dcache/resilience/util/RemoveLocationExtractor.java

@@ -69,23 +69,19 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.dcache.resilience.data.PoolInfoMap;
+import org.dcache.util.pool.AbstractLocationExtractor;
 
 /**
- * <p> Implementation of the {@link PoolTagConstraintDiscriminator} which returns
- * a list containing the pools or locations with maximum weight.  The weights are computed in terms
- * of the sum of the size of the value partitions to which they belong.  For instance, if pool1 has
- * tag1 with value A and tag2 with value X, and A is shared by 3 other pools but X by 2, then pool1
- * has a weight of 5.</p>
+ * <p> Implementation of the {@link AbstractLocationExtractor}
+ * which uses {@link PoolInfoMap} to get the pool tags.</p>
  *
- * <p>The use case envisaged here is as follows:  a pnfsId/file has a set
- * of locations which exceeds the maximum by some number K.  The caller responsible for removal will
- * run getCandidateLocations() on this list up to K times, each time picking a location from the
- * returned list using some selection algorithm, and removing the selected location from the
- * original list, which is then passed in to the i+1 iteration. The set of weights is recomputed at
- * each iteration based on the remaining locations.</p>
+ * <p>A RemoveLocationExtractor can receive notification that one or more locations are no longer
+ * hosting or should no longer host a given replica.  These locations are added to the excluded set,
+ * and then subtracted from the candidate locations returned by the call to
+ * getCandidateLocations().</p>
  */
 public final class RemoveLocationExtractor
-      extends PoolTagConstraintDiscriminator {
+      extends AbstractLocationExtractor {
 
     class WeightedLocation {
 

modules/dcache/pom.xml

@@ -262,6 +262,13 @@
         <artifactId>spring-tx</artifactId>
     </dependency>
 
+    <dependency>
+        <groupId>org.springframework</groupId>
+        <artifactId>spring-test</artifactId>
+        <version>${version.spring}</version>
+        <scope>test</scope>
+    </dependency>
+
     <dependency>
         <groupId>org.aspectj</groupId>
         <artifactId>aspectjrt</artifactId>
@@ -382,6 +389,13 @@
         <artifactId>everit-json-schema</artifactId>
         <scope>test</scope>
       </dependency>
+      <!-- Mockito for unit testing -->
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>3.2.4</version>
+      <!-- <scope>test</scope> Fails if this is uncommented; no idea why -->
+    </dependency>
   </dependencies>
 
   <build>

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

@@ -0,0 +1,18 @@
+package org.dcache.pool.classic;
+
+import diskCacheV111.util.PnfsId;
+
+/**
+ * Abstract interface for monitoring file requests in the pool.
+ */
+public interface FileRequestMonitor {
+
+    /**
+     * Report a file request for the given pnfsId and number of requests.
+     *
+     * @param pnfsId           the file identifier
+     * @param numberOfRequests the number of requests for this file
+     */
+    void reportFileRequest(PnfsId pnfsId, long numberOfRequests);
+}
+

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

@@ -5,6 +5,7 @@
 import static java.util.stream.Collectors.joining;
 
 import diskCacheV111.util.CacheException;
+import diskCacheV111.util.PnfsId;
 import diskCacheV111.vehicles.IoJobInfo;
 import dmg.cells.nucleus.CellCommandListener;
 import dmg.cells.nucleus.CellSetupProvider;
@@ -213,12 +214,16 @@ private static void toMoverString(MoverRequestScheduler.PrioritizedRequest j,
         sb.append(j.getId()).append(" : ").append(j).append('\n');
         if (displaySubject) {
             sb.append(
-                  j.getMover().getSubject().getPrincipals().stream().map(Objects::toString).collect(
-                        Collectors.joining(",", "    <", ">")))
-            .append("\n");
+                        j.getMover().getSubject().getPrincipals().stream().map(Objects::toString).collect(
+                              Collectors.joining(",", "    <", ">")))
+                  .append("\n");
         }
     }
 
+    public long numberOfRequestsFor(PnfsId pnfsId) {
+        return queues().stream().mapToLong((q) -> q.numberOfRequestsFor(pnfsId)).sum();
+    }
+
     @AffectsSetup
     @Command(name = "mover set max active",
           hint = "set the maximum number of active client transfers",

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

@@ -12,6 +12,7 @@
 import diskCacheV111.pools.PoolCostInfo.NamedPoolQueueInfo;
 import diskCacheV111.util.CacheException;
 import diskCacheV111.util.DiskErrorCacheException;
+import diskCacheV111.util.PnfsId;
 import diskCacheV111.vehicles.IoJobInfo;
 import diskCacheV111.vehicles.JobInfo;
 import diskCacheV111.vehicles.ProtocolInfo;
@@ -56,6 +57,11 @@ public class MoverRequestScheduler {
     private static final long DEFAULT_LAST_ACCESSED = 0;
     private static final long DEFAULT_TOTAL = 0;
 
+    public long numberOfRequestsFor(PnfsId pnfsId) {
+        return _jobs.values().stream()
+              .filter((pr) -> pr.getMover().getFileAttributes().getPnfsId().equals(pnfsId)).count();
+    }
+
     /**
      * A RuntimeException that wraps a CacheException.
      */
@@ -184,8 +190,9 @@ public synchronized void setOrder(Order order) {
     }
 
     /**
-     * Get mover id for given door request. If there is no mover associated with {@code
-     * doorUniqueueRequest} a new mover will be created by using provided {@code moverSupplier}.
+     * Get mover id for given door request. If there is no mover associated with
+     * {@code doorUniqueueRequest} a new mover will be created by using provided
+     * {@code moverSupplier}.
      * <p>
      * The returned mover id generated with following encoding: | 31- queue id -24|23- job id -0|
      *
@@ -517,16 +524,17 @@ public void failed(Throwable exc, Void attachment) {
                               FaultAction faultAction = null;
                               //TODO this is done because the FileStoreState is in another module
                               // to be improved
-                              switch (((DiskErrorCacheException) exc).checkStatus(exc.getMessage())){
+                              switch (((DiskErrorCacheException) exc).checkStatus(
+                                    exc.getMessage())) {
                                   case READ_ONLY:
                                       faultAction = FaultAction.READONLY;
-                                  break;
+                                      break;
                                   default:
                                       faultAction = FaultAction.DISABLED;
-                                  break;
+                                      break;
                               }
                               FaultEvent faultEvent = new FaultEvent("transfer",
-                                      faultAction, exc.getMessage(), exc);
+                                    faultAction, exc.getMessage(), exc);
                               _faultListeners.forEach(l -> l.faultOccurred(faultEvent));
                           } else if (exc instanceof OutOfDiskException) {
                               FaultEvent faultEvent = new FaultEvent(
@@ -552,7 +560,8 @@ public void completed(Void result, Void attachment) {
                                         public void failed(Throwable exc, Void attachment) {
                                             if (exc instanceof DiskErrorCacheException) {
                                                 FaultAction faultAction = null;
-                                                switch (((DiskErrorCacheException) exc).checkStatus(exc.getMessage())){
+                                                switch (((DiskErrorCacheException) exc).checkStatus(
+                                                      exc.getMessage())) {
                                                     case READ_ONLY:
                                                         faultAction = FaultAction.READONLY;
                                                         break;
@@ -562,7 +571,7 @@ public void failed(Throwable exc, Void attachment) {
                                                 }
                                                 FaultEvent faultEvent = new FaultEvent(
                                                       "post-processing",
-                                                        faultAction,
+                                                      faultAction,
                                                       exc.getMessage(), exc);
                                                 _faultListeners.forEach(
                                                       l -> l.faultOccurred(faultEvent));

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

@@ -214,12 +214,31 @@ public class PoolV4
 
     private ThreadFactory _threadFactory;
 
+    // Hot file monitoring
+    private FileRequestMonitor _fileRequestMonitor;
+    private boolean _hotFileReplicationEnabled = true;
+
+    public void setFileRequestMonitor(FileRequestMonitor fileRequestMonitor) {
+        _fileRequestMonitor = fileRequestMonitor;
+    }
+
+    public FileRequestMonitor getFileRequestMonitor() {
+        return _fileRequestMonitor;
+    }
+
+    @Required
+    public void setHotFileReplicationEnabled(boolean hotFileReplicationEnabled) {
+        _hotFileReplicationEnabled = hotFileReplicationEnabled;
+    }
+
+    public boolean getHotFileReplicationEnabled() {
+        return _hotFileReplicationEnabled;
+    }
 
     protected void assertNotRunning(String error) {
         checkState(!_running, error);
     }
 
-
     @Autowired(required = false)
     @Qualifier("remove")
     public void setKafkaTemplate(KafkaTemplate kafkaTemplate) {
@@ -581,7 +600,8 @@ public void stateChanged(StateChangeEvent event) {
                 try {
                     _kafkaSender.accept(msg);
                 } catch (KafkaException | org.apache.kafka.common.KafkaException e) {
-                    LOGGER.warn("Failed to send message to kafka: {} ", Throwables.getRootCause(e).getMessage());
+                    LOGGER.warn("Failed to send message to kafka: {} ",
+                          Throwables.getRootCause(e).getMessage());
                 }
             }
         }
@@ -732,6 +752,12 @@ private int queueIoRequest(CellMessage envelope, PoolIoFileMessage message)
     private void ioFile(CellMessage envelope, PoolIoFileMessage message) {
         try {
             message.setMoverId(queueIoRequest(envelope, message));
+            LOGGER.debug("moverId {} received request for pnfsId {}", message.getMoverId(),
+                  message.getPnfsId());
+            if (_hotFileReplicationEnabled) {
+                _fileRequestMonitor.reportFileRequest(message.getPnfsId(),
+                      _ioQueue.numberOfRequestsFor(message.getPnfsId()));
+            }
             message.setSucceeded();
         } catch (OutOfDateCacheException e) {
             if (_pingLimiter.tryAcquire()) {