Scitags: fallback mechanism (#7775)

This brings back the fallback mechanism to generate fireflies based on the subject's virtual organisation,
even if there are no transfer tags indicated in the protocols. 

It introduces a default mapping via the configuration variable pool.firefly.vo-mapping

Adds fallback mechanism to determine network tags from the Subject's
virtual organisation. A configurable mapping is used to map VO names
to the corresponding experiment IDs.

Signed-off-by: Marian Babik <[email protected]>

Author: marian-babik

modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java

@@ -34,8 +34,12 @@
 import java.time.Instant;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.HashMap;
 import javax.annotation.Nonnull;
 import javax.security.auth.Subject;
+import org.dcache.auth.Subjects;
 import org.dcache.net.FlowMarker.FlowMarkerBuilder;
 import org.dcache.util.IPMatcher;
 import org.slf4j.Logger;
@@ -60,6 +64,8 @@ public class TransferLifeCycle {
 
     private boolean enabled;
 
+    private Map<String, Integer> voToExpId = new HashMap<>();
+
     /**
      * Mark transfer start.
      * @param src remote client endpoint
@@ -82,9 +88,14 @@ public void onStart(InetSocketAddress src, InetSocketAddress dst, ProtocolInfo p
             return;
         }
 
+        var optionalExpId = getExperimentId(protocolInfo, subject);
+        if (optionalExpId.isEmpty()) {
+            return;
+        }
+
         var data = new FlowMarkerBuilder()
               .withStartedAt(Instant.now())
-              .withExperimentId(getExperimentId(protocolInfo))
+              .withExperimentId(optionalExpId.getAsInt())
               .withActivityId(getActivity(protocolInfo))
               .wittApplication(getApplication(protocolInfo))
               .withProtocol("tcp")
@@ -118,10 +129,15 @@ public void onEnd(InetSocketAddress src, InetSocketAddress dst, ProtocolInfo pro
             return;
         }
 
+        var optionalExpId = getExperimentId(protocolInfo, subject);
+        if (optionalExpId.isEmpty()) {
+            return;
+        }
+
         var data = new FlowMarkerBuilder()
               .withStartedAt(Instant.now())
               .withFinishedAt(Instant.now())
-              .withExperimentId(getExperimentId(protocolInfo))
+              .withExperimentId(optionalExpId.getAsInt())
               .withActivityId(getActivity(protocolInfo))
               .wittApplication(getApplication(protocolInfo))
               .withProtocol("tcp")
@@ -153,6 +169,29 @@ public void setEnabled(boolean isEnabled) {
         enabled = isEnabled;
     }
 
+    /**
+     * Configures VO (Virtual Organization) to Experiment ID mapping.
+     *       
+     * @param voMap A comma-separated string of VO mapping entries in the format
+     *              "voName:expId".
+     */
+    public void setVoMapping(String voMap) {
+        voToExpId.clear();
+
+        for (String entry : voMap.split(",")) {
+            String[] parts = entry.split(":");
+            if (parts.length == 2) {
+                try {
+                    voToExpId.put(parts[0].trim().toLowerCase(), Integer.parseInt(parts[1].trim()));
+                } catch (NumberFormatException e) {
+                    LOGGER.warn("Invalid VO mapping entry: {}", entry);
+                }        
+            } else {
+                LOGGER.warn("Invalid VO mapping entry: {}", entry);
+            }
+        }
+    }
+
     /**
      * Send flow marker.
      *
@@ -175,21 +214,6 @@ private void send(InetSocketAddress dst, @Nonnull String payload)
 
     private boolean needMarker(ProtocolInfo protocolInfo) {
 
-        if (protocolInfo.getTransferTag().isEmpty()) {
-            return false;
-        }
-
-        try {
-            int transferTag = Integer.parseInt(protocolInfo.getTransferTag());
-            if (transferTag <= 64 || transferTag >= 65536) {
-                LOGGER.warn("Invalid integer range for transfer tag: {}", protocolInfo.getTransferTag());
-                return false;
-            }
-        } catch (NumberFormatException e) {
-            LOGGER.warn("Invalid transfer tag: {}", protocolInfo.getTransferTag());
-            return false;
-        }
-
         switch (protocolInfo.getProtocol().toLowerCase()) {
             case "xrootd":
             case "http":
@@ -204,9 +228,38 @@ private String getApplication(ProtocolInfo protocolInfo) {
         return protocolInfo.getProtocol().toLowerCase();
     }
 
-    private int getExperimentId(ProtocolInfo protocolInfo) {
-        // scitag = exp_id << 6 | act_id
-        return Integer.parseInt(protocolInfo.getTransferTag()) >> 6;
+    /**
+     * Determine experiment ID, initially from the ProtocolInfo (xroot/http), 
+     * if that fails then fallback to the Subject's primary FQAN. 
+     *
+     * @param protocolInfo the ProtocolInfo object containing transfer-related metadata
+     * @param subject the Subject representing the user or entity associated with the transfer
+     * @return the experiment ID, or -1 if it cannot be determined
+     */
+    private OptionalInt getExperimentId(ProtocolInfo protocolInfo, Subject subject) {
+        if (protocolInfo.getTransferTag() != null && !protocolInfo.getTransferTag().isEmpty()) {
+            try {
+                int transferTag = Integer.parseInt(protocolInfo.getTransferTag());
+                if (transferTag <= 64 || transferTag >= 65536) {
+                    LOGGER.warn("Invalid integer range for transfer tag: {}", protocolInfo.getTransferTag());
+                    return OptionalInt.empty();
+                }
+                // scitag = exp_id << 6 | act_id
+                return OptionalInt.of(transferTag >> 6);
+            } catch (NumberFormatException e) {
+                LOGGER.warn("Invalid transfer tag: {}", protocolInfo.getTransferTag());
+                return OptionalInt.empty();
+            }
+        }
+
+        var vo = Subjects.getPrimaryFqan(subject);
+        if (vo == null) {
+            return OptionalInt.empty();
+        }
+
+        return voToExpId.containsKey(vo.getGroup().toLowerCase()) 
+                ? OptionalInt.of(voToExpId.get(vo.getGroup().toLowerCase())) 
+                : OptionalInt.empty();
     }
 
     private boolean isLocalTransfer(InetSocketAddress dst) {
@@ -215,8 +268,13 @@ private boolean isLocalTransfer(InetSocketAddress dst) {
     }
 
     private int getActivity(ProtocolInfo protocolInfo) {
-        // scitag = exp_id << 6 | act_id
-        return Integer.parseInt(protocolInfo.getTransferTag()) & 0x3F;
+        if (!protocolInfo.getTransferTag().isEmpty()) {
+            // scitag = exp_id << 6 | act_id
+            return Integer.parseInt(protocolInfo.getTransferTag()) & 0x3F;
+        } else {
+            // default activity id = 1
+            return 1;
+        }
     }
 
     private String toAFI(InetSocketAddress dst) {

modules/dcache/src/main/resources/org/dcache/pool/classic/pool.xml

@@ -350,6 +350,7 @@
     <property name="enabled" value="${pool.enable.firefly}" />
     <property name="fireflyDestination" value="${pool.firefly.destination}" />
     <property name="excludes" value="${pool.firefly.excludes}" />
+    <property name="voMapping" value="${pool.firefly.vo-mapping}" />
   </bean>
 
   <bean id="default-transfer-service" class="org.dcache.pool.classic.MoverMapTransferService"

skel/share/defaults/pool.properties

@@ -940,4 +940,12 @@ pool.firefly.destination=
 #
 # example: a.b.c.d/16, aa:bb:cc:dd:ee:ff/64, v.x.w.z
 #
-pool.firefly.excludes=
+pool.firefly.excludes=
+
+#
+# A comma separated list of virtual organisations and their mappings to Scitags experiment IDs 
+# Experiment IDs must be based on https://scitags.docs.cern.ch/api.json 
+#
+# example: atlas:2, cms:3,etc.
+#
+pool.firefly.vo-mapping=atlas:2, cms:3, lhcb:4, alice:5, belleii:6, ska:7, dune:8, lsst:9, ilc:10, auger:11, juno:12, nova:13, xenon:14