feat (diracx): add initial future client for DiracX job monitoring

Author: chaen

docs/source/AdministratorGuide/ServerInstallations/environment_variable_configuration.rst

@@ -18,6 +18,9 @@ DIRAC_DEPRECATED_FAIL
   If set, the use of functions or objects that are marked ``@deprecated`` will fail. Useful for example in continuous
   integration tests against future versions of DIRAC
 
+DIRAC_ENABLE_DIRACX_JOB_MONITORING
+  If set, calls the diracx job monitoring service. Off by default.
+
 DIRAC_FEWER_CFG_LOCKS
   If ``true`` or ``yes`` or ``on`` or ``1`` or ``y`` or ``t``, DIRAC will reduce the number of locks used when accessing the CS for better performance (default, ``no``).
 

src/DIRAC/WorkloadManagementSystem/Client/JobMonitoringClient.py

@@ -1,5 +1,6 @@
 """ Class that contains client access to the job monitoring handler. """
 
+import os
 from DIRAC.Core.Base.Client import Client, createClient
 from DIRAC.Core.Utilities.DEncode import ignoreEncodeWarning
 from DIRAC.Core.Utilities.JEncode import strToIntDict
@@ -11,6 +12,13 @@ def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.setServer("WorkloadManagement/JobMonitoring")
 
+    if os.getenv("DIRAC_ENABLE_DIRACX_JOB_MONITORING", "No").lower() in ("yes", "true"):
+        from DIRAC.WorkloadManagementSystem.FutureClient.JobMonitoringClient import (
+            JobMonitoringClient as futureJobMonitoringClient,
+        )
+
+        httpsClient = futureJobMonitoringClient
+
     @ignoreEncodeWarning
     def getJobsStatus(self, jobIDs):
         res = self._getRPC().getJobsStatus(jobIDs)

src/DIRAC/WorkloadManagementSystem/FutureClient/JobMonitoringClient.py

@@ -0,0 +1,36 @@
+# pylint: disable=import-error
+import os
+
+if os.getenv("DIRAC_ENABLE_DIRACX_JOB_MONITORING", "No").lower() in ("yes", "true"):
+    from diracx.client import Dirac
+    from diracx.client.models import JobSearchParams
+
+    from diracx.cli.utils import get_auth_headers
+    from diracx.core.preferences import DiracxPreferences
+
+    from DIRAC.Core.Utilities.ReturnValues import convertToReturnValue
+
+    class JobMonitoringClient:
+        def __init__(self, *args, **kwargs):
+            self.endpoint = DiracxPreferences().url
+
+        def fetch(self, parameters, jobIDs):
+            with Dirac(endpoint=self.endpoint) as api:
+                jobs = api.jobs.search(
+                    parameters=["JobID"] + parameters,
+                    search=[{"parameter": "JobID", "operator": "in", "values": jobIDs}],
+                    headers=get_auth_headers(),
+                )
+                return {j["JobID"]: {param: j[param] for param in parameters} for j in jobs}
+
+        @convertToReturnValue
+        def getJobsMinorStatus(self, jobIDs):
+            return self.fetch(["MinorStatus"], jobIDs)
+
+        @convertToReturnValue
+        def getJobsStates(self, jobIDs):
+            return self.fetch(["Status", "MinorStatus", "ApplicationStatus"], jobIDs)
+
+        @convertToReturnValue
+        def getJobsSites(self, jobIDs):
+            return self.fetch(["Site"], jobIDs)