[9.0] [POC] Pilot legacy logging

.github/workflows/integration.yml

@@ -37,6 +37,8 @@ jobs:
             ARGS: DIRAC_USE_JSON_ENCODE=NO MYSQL_VER=mysql:8.0.40
           - TEST_NAME: "Backward Compatibility"
             ARGS: CLIENT_INSTALLATION_BRANCH=rel-v8r0 PILOT_INSTALLATION_BRANCH=rel-v8r0
+          - TEST_NAME: "Test DiracX latest"
+            ARGS: TEST_DIRACX=Yes --diracx-dist-dir $GITHUB_WORKSPACE/diracx-dist
 
     steps:
     - uses: actions/checkout@v4
@@ -55,7 +57,22 @@ jobs:
             packaging \
             pyyaml \
             requests \
-            typer
+            typer \
+            build
+    - name: Building wheels
+      run: |
+        # Clone diracx
+        git clone --single-branch --branch robin-pilot-legacy-logging https://github.com/Robin-Van-de-Merghel/diracx.git $GITHUB_WORKSPACE/diracx
+
+        # Create dist dir
+        mkdir -p $GITHUB_WORKSPACE/diracx-dist
+
+        # Building diracx
+        for pkg_dir in $GITHUB_WORKSPACE/diracx/diracx-* $GITHUB_WORKSPACE/diracx; do
+          echo "Building $pkg_dir"
+          python -m build --outdir "$GITHUB_WORKSPACE/diracx-dist" $pkg_dir
+        done
+
     - name: Prepare environment
       run: ./integration_tests.py prepare-environment ${{ matrix.ARGS }}
     - name: Install server
@@ -79,3 +96,8 @@ jobs:
         if [ -f client-tests-failed ]; then has_error=1; echo "Client tests failed"; fi
         if [ -f pilot-tests-failed ]; then has_error=1; echo "pilot tests failed"; fi
         if [ ${has_error} = 1 ]; then exit 1; fi
+    - name: DiracX filtered logs
+      if: ${{ always() && contains(matrix.ARGS, 'TEST_DIRACX=Yes') }}
+      run: |
+        # Used to debug, and see which requests were done.
+        docker logs diracx | grep -Fv 'GET /.well-known/openid-configuration HTTP/1.1" 200 OK'

docs/source/DeveloperGuide/CodeTesting/index.rst

@@ -309,10 +309,8 @@ Running the above might take a while. Supposing you are interested in running on
     ./integration_tests.py prepare-environment [FLAGS]
     ./integration_tests.py install-server
 
-which (in some minutes) will give you a fully dockerized server setup
-(`docker container ls` will list the created container, and you can see what's going on inside with the standard `docker exec -it server /bin/bash`.
-Now, suppose that you want to run `WorkloadManagementSystem/Test_JobDB.py`,
-the first thing to do is that you should first login in the docker container, by doing:
+which (in some minutes) will give you a fully dockerized server setup. `docker container ls` will list the created container, and you can see what's going on inside with the standard `docker exec -it server /bin/bash`.
+Now, suppose that you want to run `WorkloadManagementSystem/Test_JobDB.py`, the first thing to do is that you should first login in the docker container, by doing:
 
 .. code-block:: bash
 
@@ -326,7 +324,7 @@ Now you can run the test with:
 
     pytest --no-check-dirac-environment LocalRepo/ALTERNATIVE_MODULES/DIRAC/tests/Integration/WorkloadManagementSystem/Test_JobDB.py
 
-You can find the logs of the services in `/home/dirac/ServerInstallDIR/diracos/runit/`
+You can find the logs of the services in `/home/dirac/ServerInstallDIR/diracos/runit/`.
 
 You can also login in client and mysql with:
 
@@ -335,7 +333,45 @@ You can also login in client and mysql with:
     ./integration_tests.py exec-client
     ./integration_tests.py exec-mysql
 
+To restart a service, you can go into the server docker, and run `runsvctrl`:
 
+.. code-block:: bash
+
+    ./integration_tests.py exec-server
+    ...
+    runsvctrl t /home/dirac/ServerInstallDIR/diracos/runit/WorkloadManagement/JobMonitor/
+
+And you can also restart all services (it can take some time):
+
+.. code-block:: bash
+
+    ./integration_tests.py exec-server
+    ...
+    runsvctrl t /home/dirac/ServerInstallDIR/diracos/runit/*/*
+
+
+You can also test DiracX in integration tests. To do that, you have to provide in the `prepare-environment` command the following flag: `TEST_DIRACX=Yes`. It will run DiracX alongside DIRAC, and use the available and activated legacy adapted services.
+
+To deactivate a service from being used with DiracX, you can add it in `integration_tests.py` in the `DIRACX_DISABLED_SERVICES` list:
+
+.. code-block:: python
+
+    DIRACX_DISABLED_SERVICES = [
+      "WorkloadManagement/JobMonitoring",
+    ]
+
+By setting `TEST_DIRACX=Yes` only, it will take the last version of DiracX by default. If you want to provide your own, you have to build your DiracX project, and provide the `dist` folder path when calling `prepare-client`. This path has to be absolute.
+
+.. code-block:: bash
+
+    ./integration-tests.py prepare-client TEST_DIRACX=Yes --diracx-dist-dir my-dist-folder/
+
+It will then mount your dist folder into DIRAC and DiracX (in `/diracx_sources`) to install the right dependencies.
+
+For MacOS, there are two bugs that can be fixed.
+
+- The first one is about `docker compose` not being recognized. To fix that, you can set in your environment variables `DOCKER_COMPOSE_CMD="docker-compose"`.
+- The second one, is for macs with M1 (or more recent ones) chips which are not supported by the `opensearch` docker image. By setting the `ES_PLATFORM` flag to `linux/arm64` you will be able to start `opensearch` without issue.
 
 Validation and System tests
 ---------------------------

integration_tests.py

@@ -25,6 +25,9 @@
 DEFAULT_HOST_OS = "el9"
 DEFAULT_MYSQL_VER = "mysql:8.4.4"
 DEFAULT_ES_VER = "opensearchproject/opensearch:2.18.0"
+# In MacOSX with Arm (MX), there's an issue with opensearch
+# You *must* set the ES_PLATFORM flag to `linux/arm64` to make it work.
+DEFAULT_ES_PLATFORM = "linux/amd64"
 DEFAULT_IAM_VER = "indigoiam/iam-login-service:v1.10.2"
 FEATURE_VARIABLES = {
     "DIRACOSVER": "master",
@@ -36,8 +39,12 @@
     "INSTALLATION_BRANCH": "",
     "DEBUG": "Yes",
 }
-DIRACX_OPTIONS = ()
 DEFAULT_MODULES = {"DIRAC": Path(__file__).parent.absolute()}
+# All services that have a FutureClient, but we *explicitly* deactivate
+# (for example if we did not finish to develop it)
+DIRACX_DISABLED_SERVICES = [
+    "WorkloadManagement/JobMonitoring",
+]
 
 # Static configuration
 DB_USER = "Dirac"
@@ -71,6 +78,10 @@
 }
 LOG_PATTERN = re.compile(r"^[\d\-]{10} [\d:]{8} UTC [^\s]+ ([A-Z]+):")
 
+# In niche cases where we use MacOSX with Orbstack, some commands may not work with docker compose
+# If you're in that case, set in your environment `export DOCKER_COMPOSE_CMD="docker-compose"`
+DOCKER_COMPOSE_CMD = shlex.split(os.environ.get("DOCKER_COMPOSE_CMD", "docker compose"))
+
 
 class NaturalOrderGroup(typer.core.TyperGroup):
     """Group for showing subcommands in the correct order"""
@@ -140,6 +151,19 @@ def list_commands(self, ctx):
 After restarting your terminal you command completion is available using:
 
   typer ./integration_tests.py run ...
+
+## DiracX
+
+If you want to activate DiracX, you have to set the flag TEST_DIRACX to "Yes".
+It will search for legacy adapted services (services with a future client activated)
+and do the necessary to make DIRAC work alongside DiracX.
+
+To deactivate a legacy adapted service (to pass CI for example), you have to add it in
+the `DIRACX_DISABLED_SERVICES` list. If you don't, the program will set this service to be used
+with DiracX, and if it is badly adapted, errors will be raised.
+
+> Note that you can provide a DiracX project (repository, branch) by building it and providing
+the dist folder to the prepare-environment command.
 """,
 )
 
@@ -193,8 +217,8 @@ def destroy():
     typer.secho("Shutting down and removing containers", err=True, fg=c.GREEN)
     with _gen_docker_compose(DEFAULT_MODULES) as docker_compose_fn:
         os.execvpe(
-            "docker",
-            ["docker", "compose", "-f", docker_compose_fn, "down", "--remove-orphans", "-t", "0", "--volumes"],
+            DOCKER_COMPOSE_CMD[0],
+            [*DOCKER_COMPOSE_CMD, "-f", docker_compose_fn, "down", "--remove-orphans", "-t", "0", "--volumes"],
             _make_env({}),
         )
 
@@ -253,7 +277,7 @@ def prepare_environment(
     typer.secho("Running docker compose to create containers", fg=c.GREEN)
     with _gen_docker_compose(modules, diracx_dist_dir=diracx_dist_dir) as docker_compose_fn:
         subprocess.run(
-            ["docker", "compose", "-f", docker_compose_fn, "up", "-d", "dirac-server", "dirac-client", "dirac-pilot"]
+            [*DOCKER_COMPOSE_CMD, "-f", docker_compose_fn, "up", "-d", "dirac-server", "dirac-client", "dirac-pilot"]
             + extra_services,
             check=True,
             env=docker_compose_env,
@@ -360,7 +384,7 @@ def prepare_environment(
         subStderr = open(docker_compose_fn_final / "stderr", "w")
 
         subprocess.Popen(
-            ["docker", "compose", "-f", docker_compose_fn_final / "docker-compose.yml", "up", "-d", "diracx"],
+            [*DOCKER_COMPOSE_CMD, "-f", docker_compose_fn_final / "docker-compose.yml", "up", "-d", "diracx"],
             env=docker_compose_env,
             stdin=None,
             stdout=subStdout,
@@ -569,7 +593,7 @@ def _gen_docker_compose(modules, *, diracx_dist_dir=None):
     # Load the docker compose configuration and mount the necessary volumes
     input_fn = Path(__file__).parent / "tests/CI/docker-compose.yml"
     docker_compose = yaml.safe_load(input_fn.read_text())
-    # diracx-wait-for-db needs the volume to be able to run the witing script
+    # diracx-wait-for-db needs the volume to be able to run the waiting script
     for ctn in ("dirac-server", "dirac-client", "dirac-pilot", "diracx-wait-for-db"):
         if "volumes" not in docker_compose["services"][ctn]:
             docker_compose["services"][ctn]["volumes"] = []
@@ -619,7 +643,7 @@ def _gen_docker_compose(modules, *, diracx_dist_dir=None):
 def _check_containers_running(*, is_up=True):
     with _gen_docker_compose(DEFAULT_MODULES) as docker_compose_fn:
         running_containers = subprocess.run(
-            ["docker", "compose", "-f", docker_compose_fn, "ps", "-q", "-a"],
+            [*DOCKER_COMPOSE_CMD, "-f", docker_compose_fn, "ps", "-q", "-a"],
             stdout=subprocess.PIPE,
             env=_make_env({}),
             # docker compose ps has a non-zero exit code when no containers are running
@@ -701,6 +725,7 @@ def _make_env(flags):
     else:
         env["MYSQL_ADMIN_COMMAND"] = "mysqladmin"
     env["ES_VER"] = flags.pop("ES_VER", DEFAULT_ES_VER)
+    env["ES_PLATFORM"] = flags.pop("ES_PLATFORM", DEFAULT_ES_PLATFORM)
     env["IAM_VER"] = flags.pop("IAM_VER", DEFAULT_IAM_VER)
     if "CVMFS_DIR" not in env or not Path(env["CVMFS_DIR"]).is_dir():
         typer.secho(f"CVMFS_DIR environment value: {env.get('CVMFS_DIR', 'NOT SET')}", fg=c.YELLOW)
@@ -1163,10 +1188,16 @@ def _make_config(modules, flags, release_var, editable):
             typer.secho(f"Required feature variable {key!r} is missing", err=True, fg=c.RED)
             raise typer.Exit(code=1)
 
-    # If we test DiracX, enable all the options
+    # If we test DiracX, add specific config
     if config["TEST_DIRACX"].lower() in ("yes", "true"):
-        for key in DIRACX_OPTIONS:
-            config[key] = "Yes"
+        if DIRACX_DISABLED_SERVICES:
+            # We link all disabled services
+            # config["DIRACX_DISABLED_SERVICES"] = "Service1 Service2 Service3 ..."
+            diracx_disabled_services = " ".join(DIRACX_DISABLED_SERVICES)
+
+            typer.secho(f"The following services won't be legacy adapted: {diracx_disabled_services}", fg="yellow")
+
+            config["DIRACX_DISABLED_SERVICES"] = diracx_disabled_services
 
     config["TESTREPO"] = [f"/home/dirac/LocalRepo/TestCode/{name}" for name in modules]
     config["ALTERNATIVE_MODULES"] = [f"/home/dirac/LocalRepo/ALTERNATIVE_MODULES/{name}" for name in modules]

src/DIRAC/ConfigurationSystem/Client/Helpers/Registry.py

@@ -457,7 +457,7 @@ def getVOForGroup(group):
 
     :return: str
     """
-    return getVO() or gConfig.getValue(f"{gBaseRegistrySection}/Groups/{group}/VO", "")
+    return gConfig.getValue(f"{gBaseRegistrySection}/Groups/{group}/VO", "") or getVO()
 
 
 def getIdPForGroup(group):

src/DIRAC/Core/Security/DiracX.py

@@ -47,7 +47,7 @@
 def addTokenToPEM(pemPath, group):
     from DIRAC.Core.Base.Client import Client
 
-    vo = Registry.getVOMSVOForGroup(group)
+    vo = Registry.getVOForGroup(group)
     if not vo:
         gLogger.error(f"ERROR: Could not find VO for group {group}, DiracX will not work!")
     disabledVOs = gConfig.getValue("/DiracX/DisabledVOs", [])

src/DIRAC/Core/Tornado/Client/ClientSelector.py

@@ -17,7 +17,6 @@
 from DIRAC.Core.DISET.TransferClient import TransferClient
 from DIRAC.Core.Tornado.Client.TornadoClient import TornadoClient
 
-
 sLog = gLogger.getSubLogger(__name__)
 
 
@@ -82,6 +81,10 @@ def ClientSelector(disetClient, *args, **kwargs):  # We use same interface as RP
             rpc = tornadoClient(*args, **kwargs)
         else:
             rpc = disetClient(*args, **kwargs)
+    except NotImplementedError as e:
+        # We catch explicitly NotImplementedError to avoid just printing "there's an error"
+        # If we mis-configured the CS for legacy adapted services, we MUST have an error.
+        raise e
     except Exception as e:  # pylint: disable=broad-except
         # If anything went wrong in the resolution, we return default RPCClient
         # So the behaviour is exactly the same as before implementation of Tornado

src/DIRAC/Core/Utilities/Extensions.py

@@ -73,6 +73,15 @@ def findServices(modules):
     return findModules(modules, "Service", "*Handler")
 
 
+def findFutureServices(modules):
+    """Find the legacy adapted services for one or more DIRAC extension(s)
+
+    :param list/str/module module: One or more Python modules or Python module names
+    :returns: list of tuples of the form (SystemName, ServiceName)
+    """
+    return findModules(modules, "FutureClient")
+
+
 @iterateThenSort
 def findDatabases(module):
     """Find the DB SQL schema defintions for one or more DIRAC extension(s)
@@ -182,7 +191,7 @@ def parseArgs():
     parser = argparse.ArgumentParser()
     subparsers = parser.add_subparsers(required=True, dest="function")
     defaultExtensions = extensionsByPriority()
-    for func in [findSystems, findAgents, findExecutors, findServices, findDatabases]:
+    for func in [findSystems, findAgents, findExecutors, findServices, findDatabases, findFutureServices]:
         subparser = subparsers.add_parser(func.__name__)
         subparser.add_argument("--extensions", nargs="+", default=defaultExtensions)
         subparser.set_defaults(func=func)

src/DIRAC/WorkloadManagementSystem/Client/JobMonitoringClient.py

@@ -14,12 +14,14 @@
 
 @createClient("WorkloadManagement/JobMonitoring")
 class JobMonitoringClient(Client):
+    # Set to None to raise an error if this service is set as "legacy adapted"
+    # See ClientSelector
+    diracxClient = None
+
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.setServer("WorkloadManagement/JobMonitoring")
 
-    diracxClient = futureJobMonitoringClient
-
     @ignoreEncodeWarning
     def getJobsStatus(self, jobIDs):
         res = self._getRPC().getJobsStatus(jobIDs)

src/DIRAC/WorkloadManagementSystem/Client/PilotManagerClient.py

@@ -4,10 +4,17 @@
 from DIRAC.Core.Base.Client import Client, createClient
 
 
+from DIRAC.WorkloadManagementSystem.FutureClient.PilotManagerClient import (
+    PilotManagerClient as futurePilotManagerClient,
+)
+
+
 @createClient("WorkloadManagement/PilotManager")
 class PilotManagerClient(Client):
     """PilotManagerClient sets url for the PilotManagerHandler."""
 
+    diracxClient = futurePilotManagerClient
+
     def __init__(self, url=None, **kwargs):
         """
         Sets URL for PilotManager handler

src/DIRAC/WorkloadManagementSystem/FutureClient/JobStateUpdateClient.py

@@ -77,11 +77,11 @@ def setJobApplicationStatus(self, jobID: str | int, appStatus: str, source: str
     def setJobAttribute(self, jobID: str | int, attribute: str, value: str):
         with DiracXClient() as api:
             if attribute == "Status":
-                api.jobs.set_job_statuses(
+                return api.jobs.set_job_statuses(
                     {jobID: {datetime.now(tz=timezone.utc): {"Status": value}}},
                 )
             else:
-                api.jobs.patch_metadata({jobID: {attribute: value}})
+                return api.jobs.patch_metadata({jobID: {attribute: value}})
 
     @stripValueIfOK
     @convertToReturnValue