add support for eainfo streams

Author: kazeborja

Collectors/DetailedCollector.py

@@ -42,6 +42,12 @@ def __init__(self, *args, **kw):
         self._exchange_tpc = self.config.get('AMQP', 'exchange_tpc')
         self._wlcg_exchange_tpc = self.config.get('AMQP', 'wlcg_exchange_tpc')
 
+        if 'SCITAGS' in self.config:
+            self._scitags_mapping_file = self.config.get('SCITAGS', 'mapping_file_path')
+            with open(self._scitags_mapping_file) as scitag_file:
+                self._scitags_mapping = json.load(scitag_file)
+        else:
+            self._scitags_mapping = None
 
         self.last_flush = time.time()
         self.seq_data = {}
@@ -131,6 +137,7 @@ def addRecord(self, sid, userID, fileClose, timestamp, addr, openTime, fileToClo
             u = self._users[sid][userInfo].get('userinfo', None)
             auth = self._users[sid][userInfo].get('authinfo', None)
             appinfo = self._users[sid][userInfo].get('appinfo', None)
+            eainfo = self._users[sid][userInfo].get('eainfo', None)
 
             if u is not None:
                 hostname = u.host.decode('idna')
@@ -151,6 +158,9 @@ def addRecord(self, sid, userID, fileClose, timestamp, addr, openTime, fileToClo
                     rec['vo'] = auth.on.decode('utf-8')
             if appinfo is not None:
                 rec['appinfo'] = appinfo
+            if eainfo is not None:
+                rec['activity'] = eainfo.activity
+                rec['experiment'] = eainfo.experiment
 
         except KeyError:
             self.logger.exception("File close record from unknown UserID=%i, SID=%s", userID, sid)
@@ -582,6 +592,16 @@ def process(self, data, addr, port):
             elif stream_type == "P":
                 self.process_gstream_tpc(decoded_gstream, addr,sid)
 
+        elif header.code == b'U': #eainfo
+            if self._scitags_mapping:
+                eainfo = decoding.eaInfo(rest, self._scitags_mapping)
+
+                try:
+                    self.logger.debug("Adding new eainfo: %s.", eainfo)
+                    self._users[sid][userInfo]['eainfo'] = eainfo
+                except KeyError:
+                    self.logger.warning("Received eainfo for a server or user not seen yet")
+
         else:
             infolen = len(data) - 4
             mm = decoding.mapheader._make(struct.unpack("!I" + str(infolen) + "s", data))

Collectors/decoding.py

@@ -15,6 +15,7 @@
 prginfo  = namedtuple("prginfo", ["xfn", "tod", "sz", "at", "ct", "mt", "fn"])
 xfrinfo  = namedtuple("xfrinfo", ["lfn", "tod", "sz", "tm", "op", "rc", "pd"])
 pathinfo = namedtuple("pathinfo", ["userinfo", "path"])
+eainfo   = namedtuple("eainfo", ["udid", "experiment", "activity"])
 
 fileOpen  = namedtuple("fileOpen",  ["rectype", "recFlag", "recSize", "fileID", "fileSize", "userID", "fileName"])
 fileXfr   = namedtuple("fileXfr",   ["rectype", "recFlag", "recSize", "fileID", "read", "readv", "write"])
@@ -143,6 +144,21 @@ def xfrInfo(message):
         pd = b''
     return xfrinfo([lfn, tod, sz, tm, op, rc, pd])
 
+
+def eaInfo(message, scitags_mapping):
+    if isinstance(message, str):
+        message = message.encode('utf-8')
+    r = message.split(b'&')
+    
+    udid = r[1].split(b'=')[1]
+    expc = int(r[2].split(b'=')[1]) - 1 
+    actc = int(r[3].split(b'=')[1]) - 1
+
+    experiment = scitags_maping['experiments'][expc]['expName']
+    activity = scitags_maping['experiments'][expc]['activities'][actc]['activityName']
+    return eainfo(udid, experiment, activity)   
+
+
 def gStream(message):
     # int, int, int64, null terminated string
 

Collectors/wlcg_converter.py

@@ -100,11 +100,17 @@ def Convert(source_record):
         to_return["user_protocol"] = source_record["protocol"]
 
     # vo
-    if "vo" in source_record:
+    if "experiment" in source_record:
+        to_return["vo"] = source_record["experiment"]
+    elif "vo" in source_record:
         to_return["vo"] = source_record["vo"]
 
     # write_bytes
     to_return["write_bytes"] = source_record['write']
+
+    # activity
+    to_return["activity"] = source_record.get('activity', 'unknown')
+
     # remote_access - boolean
     # is_transfer (optional)
     # user_fqan (optional)