feat: Add pilot auth, pilot logging and refresh

Author: Robin-Van-de-Merghel

.github/workflows/basic.yml

@@ -23,7 +23,7 @@ jobs:
     strategy:
       matrix:
         python: 
-          - 2.7.18
+          # - 2.7.18
           - 3.6.15
           - 3.9.17
 
@@ -53,7 +53,7 @@ jobs:
     strategy:
       matrix:
         python: 
-          - 2.7.18
+          # - 2.7.18
           - 3.6.15
           - 3.9.17
 
@@ -77,7 +77,7 @@ jobs:
     strategy:
       matrix:
         python: 
-          - 2.7.18
+          # - 2.7.18
           - 3.6.15
           - 3.9.17
 

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__":
@@ -63,25 +69,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)
 
@@ -104,7 +119,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 +137,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,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())
@@ -549,7 +550,13 @@ def __init__(self, pilotParams):
 
     @logFinalizer
     def execute(self):
-        """Calls dirac-admin-add-pilot"""
+        """Calls dirac-admin-add-pilot
+        
+        Deprecated with DIRAC V9 and DiracX."""
+
+        if self.pp.jwt:
+            self.log.warn("Skipping module, normally it is already done via DiracX secret-exchange.")
+            return
 
         if not self.pp.pilotReference:
             self.log.warn("Skipping module, no pilot reference found")
@@ -1113,8 +1120,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:
@@ -1231,4 +1236,4 @@ def _runNagiosProbes(self):
     def execute(self):
         """Standard entry point to a pilot command"""
         self._setNagiosOptions()
-        self._runNagiosProbes()
+        self._runNagiosProbes()