Robin add pilot logging

Pilot/dirac-pilot.py

@@ -49,6 +49,12 @@
         getCommand,
         pythonPathCheck,
     )
+
+try:
+    from Pilot.proxyTools import revokePilotToken
+except ImportError:
+    from proxyTools import revokePilotToken
+
 ############################
 
 if __name__ == "__main__":
@@ -64,7 +70,7 @@
     sys.stdout.write(bufContent)
     # now the remote logger.
     remote = pilotParams.pilotLogging and (pilotParams.loggerURL is not None)
-    if remote:
+    if remote and pilotParams.jwt:
         # In a remote logger enabled Dirac version we would have some classic logger content from a wrapper,
         # which we passed in:
         receivedContent = ""
@@ -76,12 +82,18 @@
             bufsize=pilotParams.loggerBufsize,
             pilotUUID=pilotParams.pilotUUID,
             debugFlag=pilotParams.debugFlag,
-            wnVO=pilotParams.wnVO,
+            jwt=pilotParams.jwt
         )
         log.info("Remote logger activated")
-        log.buffer.write(receivedContent)
+        log.buffer.write(log.format_to_json(
+            "INFO",
+            receivedContent,
+        ))
         log.buffer.flush()
-        log.buffer.write(bufContent)
+        log.buffer.write(log.format_to_json(
+            "INFO",
+            bufContent,
+        ))
     else:
         log = Logger("Pilot", debugFlag=pilotParams.debugFlag)
 
@@ -104,7 +116,7 @@
 
     log.info("Executing commands: %s" % str(pilotParams.commands))
 
-    if remote:
+    if remote and pilotParams.jwt:
         # It's safer to cancel the timer here. Each command has got its own logger object with a timer cancelled by the
         # finaliser. No need for a timer in the "else" code segment below.
         try:
@@ -122,5 +134,22 @@
             log.error("Command %s could not be instantiated" % commandName)
             # send the last message and abandon ship.
             if remote:
-                log.buffer.flush()
+                log.buffer.flush(force=True)
             sys.exit(-1)
+
+    log.info("Pilot tasks finished.")
+
+    if not remote:
+        log.buffer.flush()
+
+    if pilotParams.jwt:
+        if remote:
+            log.buffer.flush(force=True)
+
+        log.info("Revoking pilot token.")
+        revokePilotToken(
+            pilotParams.diracXServer,
+            pilotParams.pilotUUID,
+            pilotParams.jwt,
+            pilotParams.clientID
+        )

Pilot/pilotCommands.py

@@ -28,7 +28,6 @@ def __init__(self, pilotParams):
 import sys
 import time
 import traceback
-import subprocess
 from collections import Counter
 
 ############################
@@ -44,7 +43,7 @@ def __init__(self, pilotParams):
     from shlex import quote
 except ImportError:
     from pipes import quote
-
+    
 try:
     from Pilot.pilotTools import (
         CommandBase,
@@ -61,6 +60,16 @@ def __init__(self, pilotParams):
         safe_listdir,
         sendMessage,
     )
+
+try:
+    from Pilot.proxyTools import BaseRequest, refreshTokenLoop
+except ImportError:
+    from proxyTools import BaseRequest, refreshTokenLoop
+
+try:
+    from urllib.error import HTTPError, URLError
+except ImportError:
+    from urllib2 import HTTPError, URLError
 ############################
 
 
@@ -92,16 +101,37 @@ def wrapper(self):
             self.log.info(
                 "Flushing the remote logger buffer for pilot on sys.exit(): %s (exit code:%s)" % (pRef, str(exCode))
             )
-            self.log.buffer.flush()  # flush the buffer unconditionally (on sys.exit()).
-            try:
-                sendMessage(self.log.url, self.log.pilotUUID, self.log.wnVO, "finaliseLogs", {"retCode": str(exCode)})
-            except Exception as exc:
-                self.log.error("Remote logger couldn't be finalised %s " % str(exc))
+            if self.pp.jwt:
+                try:
+                    sendMessage(self.log.url, self.log.pilotUUID,  self.pp.jwt, [
+                        {
+                            "severity": "ERROR",
+                            "message": str(exCode)
+                        },
+                        {
+                            "severity": "ERROR",
+                            "message": traceback.format_exc()
+                        }
+                    ])
+
+                    self.log.buffer.flush(force=True)
+                except Exception as exc:
+                    self.log.error("Remote logger couldn't be finalised %s " % str(exc))
+                raise
+
+                # No force here because there's no remote logger if we're here
+                self.log.buffer.flush()
             raise
         except Exception as exc:
             # unexpected exit: document it and bail out.
             self.log.error(str(exc))
             self.log.error(traceback.format_exc())
+
+            if self.pp.jwt:
+                # Force flush if it's a remote logger
+                self.log.buffer.flush(force=True)
+            else:
+                self.log.buffer.flush()
             raise
         finally:
             self.log.buffer.cancelTimer()
@@ -132,7 +162,7 @@ def __init__(self, pilotParams):
     @logFinalizer
     def execute(self):
         """Get host and local user info, and other basic checks, e.g. space available"""
-
+        
         self.log.info("Uname      = %s" % " ".join(os.uname()))
         self.log.info("Host Name  = %s" % socket.gethostname())
         self.log.info("Host FQDN  = %s" % socket.getfqdn())
@@ -1232,3 +1262,5 @@ def execute(self):
         """Standard entry point to a pilot command"""
         self._setNagiosOptions()
         self._runNagiosProbes()
+
+