Merge pull request #184 from PanDAWMS/next

3.11.1.15

Author: PalNilsson

.github/workflows/circular-imports.yml

@@ -0,0 +1,64 @@
+name: Circular import check
+
+on:
+  push:
+    branches: [master, next]
+    paths:
+      - "**.py"
+      - ".github/workflows/circular-imports.yml"
+  pull_request:
+    branches: [master, next]
+    paths:
+      - "**.py"
+      - ".github/workflows/circular-imports.yml"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  circular-imports:
+    name: circular-imports (py${{ matrix.python-version }})
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.9", "3.11", "3.12"]
+
+    concurrency:
+      group: circular-imports-${{ github.ref }}-py${{ matrix.python-version }}
+      cancel-in-progress: true
+
+    env:
+      # Pin or set to "" to use latest from PyPI
+      CIRC_DET_VERSION: "==1.0.16"
+      # If your code is in a subfolder, change this
+      PROJECT_PATH: "."
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+
+      - name: Install circular-import-detector
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install "circular-import-detector${CIRC_DET_VERSION}"
+
+      - name: Run detector (verbose) and save report
+        run: |
+          set -o pipefail
+          circular-import-detector "$PROJECT_PATH" --verbose --exit-code \
+            | tee "circular-import-report-py${{ matrix.python-version }}.txt"
+
+      - name: Upload report artifact
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: circular-import-report-py${{ matrix.python-version }}
+          path: circular-import-report-py${{ matrix.python-version }}.txt

.pre-commit-config.yaml

@@ -11,3 +11,15 @@ repos:
     rev: 6.1.0
     hooks:
     -   id: flake8
+-   repo: local
+    hooks:
+      - id: circular-imports
+        name: Check for circular imports (changed files)
+        entry: circular-import-precommit
+        language: python
+        additional_dependencies:
+          - circular-import-detector==1.0.18
+        pass_filenames: true
+        types_or: [python]
+        stages: [pre-commit]
+

PILOTVERSION

@@ -1 +1 @@
-3.11.0.29
+3.11.1.15

pilot.py

@@ -152,6 +152,14 @@ def main() -> int:  # noqa: C901
                 f"specified queue is NOT ACTIVE: {infosys.queuedata.name} -- aborting"
             )
             return errors.PANDAQUEUENOTACTIVE
+
+        # make sure the queue is online
+        if infosys.queuedata.status.lower() == "offline":
+            logger.critical(
+                f"specified queue is OFFLINE: {infosys.queuedata.name} -- aborting"
+            )
+            return errors.PANDAQUEUENOTONLINE
+
     except PilotException as error:
         logger.fatal(error)
         return error.get_error_code()

pilot/api/analytics.py

@@ -22,8 +22,7 @@
 """Functions for performing analytics including fitting of data."""
 
 import logging
-from typing import Any
-
+from typing import Any, Union
 from .services import Services
 from pilot.common.exception import NotDefined, NotSameLength, UnknownException
 from pilot.util.filehandling import get_table_from_file
@@ -51,11 +50,16 @@ def fit(self, x: list, y: list, model: str = "linear") -> Any:
 
         For a linear model: y(x) = slope * x + intersect
 
-        :param x: list of input data (list of floats or ints) (list)
-        :param y: list of input data (list of floats or ints) (list)
-        :param model: model name (str)
-        :raises UnknownException: in case Fit() fails
-        :return: fit (Any).
+        Args:
+            x (list[float] | list[int]): Input x data points.
+            y (list[float] | list[int]): Input y data points.
+            model (str): Model name. Defaults to "linear".
+
+        Raises:
+            UnknownException: If constructing the Fit object fails.
+
+        Returns:
+            Fit: Fit object containing the fitted parameters and diagnostics.
         """
         try:
             self._fit = Fit(x=x, y=y, model=model)
@@ -64,24 +68,29 @@ def fit(self, x: list, y: list, model: str = "linear") -> Any:
 
         return self._fit
 
-    def slope(self) -> float:
+    def slope(self) -> Union[float, None]:
         """
-        Return the slope of a linear fit, y(x) = slope * x + intersect.
+        Return the slope of a linear fit y(x) = slope * x + intersect.
 
-        :raises NotDefined: exception thrown if fit is not defined.
-        :return: slope (float).
+        Raises:
+            NotDefined: If the fit has not been defined.
+
+        Returns:
+            float | None: The slope value if available, otherwise None.
         """
         if not self._fit:
             raise NotDefined("Fit has not been defined")
 
         return self._fit.slope()
 
-    def intersect(self) -> float:
+    def intersect(self) -> Union[float, None]:
         """
-        Return the intersect of a linear fit, y(x) = slope * x + intersect.
+        Return the intersect of a linear fit y(x) = slope * x + intersect.
 
-        :raises NotDefined: exception thrown if fit is not defined
-        :return: intersect (float).
+        Raises:
+            NotDefined: If the fit has not been defined.
+
+        Returns:
         """
         if not self._fit:
             raise NotDefined("Fit has not been defined")
@@ -397,9 +406,9 @@ def set_intersect(self):
         """
         Calculate and set the intersect of the linear fit.
         """
-        if self._ym and self._slope and self._xm:
+        if self._ym is not None and self._slope is not None and self._xm is not None:
             self._intersect = self._ym - self._slope * self._xm
-            logger.info("-- intersect: %s", self._intersect)
+            logger.debug(f"{self._ym} - {self._slope} * {self._xm} = {self._intersect}")
         else:
             self._intersect = None
             logger.info("could not calculate intersect")

pilot/api/data.py

@@ -22,7 +22,71 @@
 # - Tobias Wegner, tobias.wegner@cern.ch, 2017-2018
 # - Alexey Anisenkov, anisyonk@cern.ch, 2018-2024
 
-"""API for data transfers."""
+"""
+API for data transfers.
+
+This module provides a high-level API for managing data transfers (stage-in and stage-out)
+within the Pilot framework. It serves as an abstraction layer over various underlying transfer
+protocols and tools, collectively known as "copytools." The primary goal is to provide a
+unified interface for staging data, regardless of the specific technology used for the transfer.
+
+Core Classes:
+- `StagingClient`: This is the base class that provides the common framework for data staging.
+  It handles the dynamic selection of copytools based on site configuration and the type of
+  activity (e.g., 'read_lan', 'write_wan'). It includes methods for resolving file replicas
+  from catalogs like Rucio, sorting them based on priority (e.g., LAN vs. WAN), and
+  orchestrating the transfer process through the `transfer` method. It also manages tracing
+  and logging for transfers.
+
+- `StageInClient`: This class inherits from `StagingClient` and specializes in handling the
+  stage-in of input files. It contains logic to resolve the best replica for input files,
+  considering factors like direct access modes (LAN/WAN), allowed schemas (e.g., 'root', 'https'),
+  and site-specific storage configurations. It is also responsible for checking available
+  disk space and verifying that input file sizes are within configured limits.
+
+- `StageOutClient`: This class, also inheriting from `StagingClient`, is responsible for
+  staging out output files. Its key functionality includes preparing destinations by resolving
+  the correct output storage element (RSE) based on the activity. It constructs the final
+  destination SURL (Storage URL) for the output files, calculates checksums for verification,
+  and ensures that output files exist and have a non-zero size before initiating the transfer.
+
+Key Concepts:
+- Copytools: The actual file transfers are delegated to specific "copytool" modules, which
+  are located in the `pilot/copytool/` directory (e.g., `rucio`, `xrdcp`, `gfal`). The
+  `StagingClient` dynamically imports and uses the appropriate copytool based on the
+  `acopytools` configuration for a given activity. This design makes the system extensible
+  to new transfer protocols.
+
+- Replica Resolution: For stage-in, the client queries a catalog (like Rucio) to find all
+  available replicas (copies) of a file. These replicas are then sorted by priority
+  (e.g., network proximity, site preference) to select the most efficient source for the
+  transfer.
+
+- Protocol and Destination Resolution: For stage-out, the client determines the correct
+  destination storage and the protocol to use for the transfer based on site and experiment
+  configurations stored in `ddmconf` and `astorages`.
+
+- Direct Access: The `StageInClient` supports a "direct access" or "remote I/O" mode, where
+  files are not physically copied to the worker node but are accessed remotely by the payload.
+  The client identifies when this mode is applicable and sets the file status accordingly,
+  providing the payload with the correct remote TURL (Transport URL).
+
+Workflow:
+1. A `StageInClient` or `StageOutClient` is instantiated with site-specific information.
+2. The `transfer` method is called with a list of `FileSpec` objects that represent the
+   files to be transferred.
+3. The client determines the appropriate copytool(s) for the given activity.
+4. For each copytool, the client prepares the files:
+   - Stage-in: Resolves replicas, selects the best source URL, and checks for direct
+     access possibilities.
+   - Stage-out: Resolves the destination URL and prepares the source file by checking for
+     its existence and calculating its checksum.
+5. The client calls the `copy_in` or `copy_out` function of the selected copytool module,
+   passing the list of files to be transferred.
+6. The copytool executes the transfer.
+7. The client updates the status of the `FileSpec` objects and handles any errors. If one
+   copytool fails, it can try the next one in the configured list.
+"""
 
 import os
 import hashlib

pilot/common/errorcodes.py

@@ -187,6 +187,7 @@ class ErrorCodes:
     STAGEOUTAUTHENTICATIONFAILURE = 1383
     QUEUENOTSETUPFORCONTAINERS = 1384
     NOJOBSINPANDA = 1385  # internally used code
+    PANDAQUEUENOTONLINE = 1386
 
     _error_messages = {
         GENERALERROR: "General pilot error, consult batch log",
@@ -335,6 +336,8 @@ class ErrorCodes:
         STAGEOUTAUTHENTICATIONFAILURE: "Authentication failure during stage-out",
         QUEUENOTSETUPFORCONTAINERS: "Queue is not set up for containers",
         NOJOBSINPANDA: "No jobs in PanDA",
+        PANDAQUEUENOTONLINE: "PanDA queue is not online",
+
     }
 
     put_error_codes = [1135, 1136, 1137, 1141, 1152, 1181]

pilot/control/data.py

@@ -23,7 +23,33 @@
 # - Wen Guan, wen.guan@cern.ch, 2018
 # - Alexey Anisenkov, anisyonk@cern.ch, 2018
 
-"""Control interface to data API."""
+"""
+Control interface to data API.
+
+This module manages all data transfer operations for pilot jobs, including stage-in of input files
+and stage-out of output files and logs. It operates as a control layer, orchestrating data transfers
+through a system of "copytools."
+
+The core of this module is a set of threads that continuously monitor queues for jobs that require
+data transfers:
+- `copytool_in`: This thread handles the stage-in of input files for jobs. It retrieves jobs from
+  the `data_in` queue, determines the appropriate copytool to use, and initiates the transfer.
+- `copytool_out`: This thread manages the stage-out of output files. Once a job has finished
+  execution, it is placed in the `data_out` queue, and this thread handles the upload of
+  output files to the appropriate storage elements.
+- `queue_monitoring`: This thread monitors the status of the data transfer queues and handles
+  jobs that have either completed or failed the transfer process.
+
+The actual data transfer logic is abstracted away into "copytools," which are individual modules
+located in the `pilot/copytool/` directory. Each copytool is responsible for a specific transfer
+protocol or technology, such as Rucio, xrdcp, etc. This module interacts with the copytools
+through the `StageInClient` and `StageOutClient` classes from the `pilot.api.data` module, which
+provide a unified interface for initiating transfers regardless of the underlying copytool used.
+
+The selection of which copytool to use is determined by the site configuration and the protocols
+supported by the storage elements involved in the transfer. This allows for a flexible and
+extensible data transfer framework that can be adapted to different computing environments.
+"""
 
 import logging
 import os
@@ -890,6 +916,39 @@ def _do_stageout(job: JobData, args: object, xdata: list, activity: list, title:
     """
     Use the `StageOutClient` in the Data API to perform stage-out.
 
+    This function orchestrates the stage-out process, including a fallback mechanism known as
+    "alternative stage-out." This allows the pilot to attempt transferring output files to a
+    secondary storage element (RSE) if the primary one fails.
+
+    The alternative stage-out logic is as follows:
+    1.  **Initial Transfer Attempt**: The function first calls `client.transfer()` to attempt
+        uploading all files to their primary destination RSEs. Crucially, if alternative
+        stage-out is enabled for the job (`job.allow_altstageout()` is True), the `transfer`
+        method is called with `raise_exception=False`. This ensures that if a transfer fails,
+        the function does not immediately exit but continues execution, allowing for a retry.
+
+    2.  **Failure Detection**: After the initial attempt, the code checks for any files that
+        were not successfully transferred by creating a `remain_files` list.
+
+    3.  **Conditions for Retry**: An alternative stage-out is attempted only if all of the
+        following conditions are met:
+        a. The job is configured to allow alternative stage-out (`altstageout` is True).
+        b. There are files remaining that failed the first transfer attempt (`remain_files` is not empty).
+        c. Every file that failed has an alternative destination defined (`has_altstorage` is True).
+           The alternative destination is stored in the `ddmendpoint_alt` attribute of the file spec.
+
+    4.  **Executing the Alternative Transfer**: If the conditions are met, the function iterates
+        through the list of failed files. For each file, it swaps the primary destination
+        (`entry.ddmendpoint`) with the alternative one (`entry.ddmendpoint_alt`).
+
+    5.  **Second Transfer Attempt**: The `client.transfer()` method is called a second time.
+        The `StageOutClient` will now attempt to transfer the previously failed files to their
+        newly assigned alternative destinations. Files that were successfully transferred in the
+        first attempt are ignored.
+
+    The function returns `True` only if all files are successfully transferred after these
+    attempts.
+
     The rucio host is internally set by Rucio via the client config file. This can be set directly as a pilot option
     --rucio-host.