feat: Add pilot pilot logging (legacy and DiracX)

Author: Robin-Van-de-Merghel

Pilot/dirac-pilot.py

@@ -65,25 +65,34 @@
     # print the buffer, so we have a "classic' logger back in sync.
     sys.stdout.write(bufContent)
     # now the remote logger.
-    remote = pilotParams.pilotLogging and (pilotParams.loggerURL is not None)
-    if remote:
+    remote = pilotParams.pilotLogging and pilotParams.diracXServer
+    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 = ""
         if not sys.stdin.isatty():
             receivedContent = sys.stdin.read()
+        
         log = RemoteLogger(
-            pilotParams.loggerURL,
+            pilotParams.diracXServer,
             "Pilot",
             bufsize=pilotParams.loggerBufsize,
             pilotUUID=pilotParams.pilotUUID,
             debugFlag=pilotParams.debugFlag,
-            wnVO=pilotParams.wnVO,
+            jwt=pilotParams.jwt,
+            legacy_logging=pilotParams.isLegacyLogging,
+            clientID=pilotParams.clientID
         )
         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)
 
@@ -106,7 +115,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:
@@ -124,13 +133,16 @@
             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 pilotParams.jwt:
-        if not pilotParams.isLegacyPilot:
+        if remote:
+            log.buffer.flush(force=True)
+
+        if not pilotParams.isLegacyLogging:
             log.info("Revoking pilot token.")
             revokePilotToken(
                 pilotParams.diracXServer,

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,6 @@ def __init__(self, pilotParams):
     from shlex import quote
 except ImportError:
     from pipes import quote
-
 try:
     from Pilot.pilotTools import (
         CommandBase,
@@ -92,16 +90,20 @@ 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)})
+                self.log.error(str(exCode))
+                self.log.error(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
         except Exception as exc:
             # unexpected exit: document it and bail out.
             self.log.error(str(exc))
             self.log.error(traceback.format_exc())
+            self.log.buffer.flush(force=True)
             raise
         finally:
             self.log.buffer.cancelTimer()
@@ -132,7 +134,6 @@ 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())
@@ -1126,8 +1127,6 @@ def execute(self):
         self.__setInnerCEOpts()
         self.__startJobAgent()
 
-        sys.exit(0)
-
 
 class NagiosProbes(CommandBase):
     """Run one or more Nagios probe scripts that follow the Nagios Plugin API: