Scitags: Firefly usage and stats (#7778)

* Introduces usage and storage statistics in fireflies (FlowMaker), which
are used to track the data transfer(s) direction and estimate storage
performance wrt. network usage.

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

Author: marian-babik

modules/dcache-vehicles/src/main/java/diskCacheV111/vehicles/MoverInfoMessage.java

@@ -22,6 +22,8 @@ public class MoverInfoMessage extends PnfsFileInfoMessage {
     private Duration _readIdle;
     private Duration _writeActive;
     private Duration _writeIdle;
+    private double bytesRead = Double.NaN;
+    private double bytesWritten = Double.NaN;
 
     private static final long serialVersionUID = -7013160118909496211L;
     private String _transferPath;
@@ -140,6 +142,23 @@ public void setLocalEndpoint(InetSocketAddress endpoint) {
     public Optional<InetSocketAddress> getLocalEndpoint() {
         return Optional.ofNullable(_localEndpoint);
     }
+
+    public double getBytesRead() {
+        return bytesRead;
+    }
+
+    public void setBytesRead(double bytesRead) {
+        this.bytesRead = bytesRead;
+    }
+
+    public double getBytesWritten() {
+        return bytesWritten;
+    }
+
+    public void setBytesWritten(double bytesWritten) {
+        this.bytesWritten = bytesWritten;
+    }
+
     @Override
     public String toString() {
         return "MoverInfoMessage{" +

modules/dcache/src/main/java/org/dcache/net/FlowMarker.java

@@ -18,9 +18,13 @@
 package org.dcache.net;
 
 import com.google.common.net.InetAddresses;
+
+import diskCacheV111.vehicles.MoverInfoMessage;
+
 import java.net.InetSocketAddress;
 import java.time.Instant;
 import java.time.format.DateTimeFormatter;
+
 import org.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,6 +44,8 @@ public static class FlowMarkerBuilder {
         private final JSONObject lifecycle = new JSONObject();
         private final JSONObject context = new JSONObject();
         private final JSONObject flow = new JSONObject();
+        private final JSONObject usage = new JSONObject();
+        private final JSONObject stats = new JSONObject();
 
         public FlowMarkerBuilder withStartedAt(Instant startTime) {
             lifecycle.put("start-time", DateTimeFormatter.ISO_INSTANT.format(startTime));
@@ -104,6 +110,31 @@ public FlowMarkerBuilder withProtocol(String proto) {
             return this;
         }
 
+        public FlowMarkerBuilder withUsage(double bytesRead, double bytesWritten) {
+            // approx. - assumes bytesWritten to disk are bytes "received" over the network
+            // and bytesRead from disk are bytes "sent" to the network
+            usage.put("received", Double.isFinite(bytesWritten) ? (long) bytesWritten : 0L);
+            usage.put("sent", Double.isFinite(bytesRead) ? (long) bytesRead : 0L);
+            return this;
+        }
+
+        public FlowMarkerBuilder withStats(MoverInfoMessage message) {
+            stats.put("bytes-transferred", message.getDataTransferred());
+            stats.put("connection-time", message.getConnectionTime());
+            stats.put("read-bw", Double.isFinite(message.getMeanReadBandwidth()) ? (long) message.getMeanReadBandwidth() : 0L);
+            stats.put("write-bw", Double.isFinite(message.getMeanWriteBandwidth()) ? (long) message.getMeanWriteBandwidth() : 0L);
+            stats.put("read-active", message.getReadActive().isPresent() ? message.getReadActive().get().toString() : "0");
+            stats.put("read-idle", message.getReadIdle().isPresent() ? message.getReadIdle().get().toString() : "0");  
+            stats.put("write-active", message.getWriteActive().isPresent() ? message.getWriteActive().get().toString() : "0");  
+            stats.put("write-idle", message.getWriteIdle().isPresent() ? message.getWriteIdle().get().toString() : "0");
+            return this;
+        }
+
+        public FlowMarkerBuilder withFlowId(String id) {
+            flow.put("flow-id", id);
+            return this;
+        }
+
         public String build(String state) {
 
             switch (state) {
@@ -119,6 +150,9 @@ public String build(String state) {
             payload.put("flow-lifecycle", lifecycle);
             payload.put("context", context);
             payload.put("flow-id", flow);
+            payload.put("usage", usage);
+            if(!stats.isEmpty())
+                payload.put("storage-stats", stats);
 
             lifecycle.put("state", state);
             lifecycle.put("current-time", DateTimeFormatter.ISO_INSTANT.format(Instant.now()));

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

@@ -208,12 +208,14 @@ private static MoverInfoMessage updateIoStatistics(MoverInfoMessage info,
         SnapshotStatistics writeStats = writes.statistics();
 
         if (readStats.requestedBytes().getN() > 0) {
+            info.setBytesRead(readStats.transferredBytes().getSum());
             info.setMeanReadBandwidth(readStats.instantaneousBandwidth().getMean());
             info.setReadActive(reads.active());
             info.setReadIdle(reads.idle());
         }
 
         if (writeStats.requestedBytes().getN() > 0) {
+            info.setBytesWritten(writeStats.transferredBytes().getSum());
             info.setMeanWriteBandwidth(writeStats.instantaneousBandwidth().getMean());
             info.setWriteActive(writes.active());
             info.setWriteIdle(writes.idle());
@@ -240,8 +242,7 @@ public void sendFinished(Mover<?> mover, MoverInfoMessage moverInfoMessage) {
         mover.getLocalEndpoint().ifPresent(e ->
                 transferLifeCycle.onEnd(((IpProtocolInfo) mover.getProtocolInfo()).getSocketAddress(),
                         e,
-                        mover.getProtocolInfo(),
-                        mover.getSubject()));
+                        moverInfoMessage));
 
         _door.notify(mover.getPathToDoor(), finished);
     }

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

@@ -22,6 +22,8 @@
 
 import com.google.common.base.Strings;
 import com.google.common.net.HostAndPort;
+
+import diskCacheV111.vehicles.MoverInfoMessage;
 import diskCacheV111.vehicles.ProtocolInfo;
 import java.io.IOException;
 import java.net.DatagramPacket;
@@ -63,6 +65,7 @@ public class TransferLifeCycle {
     private Predicate<InetAddress> localSubnet = a -> false;
 
     private boolean enabled;
+    private boolean storageStatisticsEnabled;
 
     private Map<String, Integer> voToExpId = new HashMap<>();
 
@@ -75,7 +78,7 @@ public class TransferLifeCycle {
      */
     public void onStart(InetSocketAddress src, InetSocketAddress dst, ProtocolInfo protocolInfo,
           Subject subject) {
-
+        
         if (!enabled) {
             return;
         }
@@ -114,9 +117,11 @@ public void onStart(InetSocketAddress src, InetSocketAddress dst, ProtocolInfo p
      * @param protocolInfo access protocol information
      * @param subject associated with the transfer
      */
-    public void onEnd(InetSocketAddress src, InetSocketAddress dst, ProtocolInfo protocolInfo,
-          Subject subject) {
+    public void onEnd(InetSocketAddress src, InetSocketAddress dst, MoverInfoMessage mover) {
+        ProtocolInfo protocolInfo = mover.getProtocolInfo();
+        Subject subject = mover.getSubject();
 
+        
         if (!enabled) {
             return;
         }
@@ -140,13 +145,17 @@ public void onEnd(InetSocketAddress src, InetSocketAddress dst, ProtocolInfo pro
               .withExperimentId(optionalExpId.getAsInt())
               .withActivityId(getActivity(protocolInfo))
               .wittApplication(getApplication(protocolInfo))
+              .withUsage(mover.getBytesRead(), mover.getBytesWritten())
               .withProtocol("tcp")
               .withAFI(toAFI(dst))
               .withDestination(dst)
-              .withSource(src)
-              .build("end");
+              .withSource(src);
+        if (storageStatisticsEnabled) {
+            data.withStats(mover);
+        }
+        var firefly = data.build("end");
 
-        send(toFireflyDestination.apply(src), data);
+        send(toFireflyDestination.apply(src), firefly);
     }
 
     public void setFireflyDestination(String addr) {
@@ -169,6 +178,10 @@ public void setEnabled(boolean isEnabled) {
         enabled = isEnabled;
     }
 
+    public void setStorageStatisticsEnabled(boolean isEnabled) {
+        storageStatisticsEnabled = isEnabled;
+    }
+
     /**
      * Configures VO (Virtual Organization) to Experiment ID mapping.
      *       

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

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

skel/share/defaults/pool.properties

@@ -948,4 +948,9 @@ pool.firefly.excludes=
 #
 # 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
+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
+
+#
+# Enable sending storage statistics as part of the firefly packets
+#
+pool.firefly.storage-statistics=false