Ac slurmfix

CHANGELOG.md

@@ -1,3 +1,22 @@
+# Version 0.6.21
+
+## Fixes
+
+- Restore DRMAA (GridExecutor) as the primary cluster execution path; CLI-based
+  executors (SlurmExecutor etc.) are now a fallback for environments without DRMAA
+- Fix SlurmExecutor job ID parsing: sbatch outputs "Submitted batch job <id>" but
+  only the numeric ID was being extracted and validated correctly
+- Fix SlurmExecutor monitoring: use squeue to poll while the job is alive, then
+  sacct for final exit status, avoiding "Bad job/step specified" errors when sacct
+  is queried before the job is in the accounting database
+- Fix signal handler flood on pipeline exit: ruffus worker processes inherited the
+  SIGTERM handler via fork, causing every worker to log "Received signal 15" and
+  run cleanup when the process group exited normally
+- Fix get_caller() frame depth in run_workflow() so pipeline config files are
+  resolved relative to the calling pipeline module, not cgatcore itself
+- will_run_on_cluster() no longer raises when DRMAA is absent; returns False so
+  CLI executor fallback proceeds cleanly
+
 # Version 0.6.20
 
 - Fix pipeline parallelism: pass `multiprocess` to `ruffus.pipeline_run()` so multiple jobs run in parallel again

cgatcore/pipeline/execution.py

@@ -540,17 +540,14 @@ def will_run_on_cluster(options):
     # First check if we're explicitly running without cluster
     if options.get("without_cluster", False):
         return False
-        
+
     wants_cluster = options.get("to_cluster", True)
-
-    # Only raise error if DRMAA is required but not available
-    if wants_cluster and not HAS_DRMAA:
-        raise ValueError(
-            "Cluster execution requested (to_cluster=True) but DRMAA library is not available. "
-            "Please install drmaa package in your environment to enable cluster execution.")
-
-    return (wants_cluster 
-            and HAS_DRMAA 
+
+    # Return True only when DRMAA is present and a session is active.
+    # When DRMAA is absent the CLI-based executors in get_executor() are used
+    # as a fallback, so we do not raise here.
+    return (wants_cluster
+            and HAS_DRMAA
             and GLOBAL_SESSION is not None)
 
 
@@ -622,8 +619,6 @@ def __init__(self, **kwargs):
         if self.monitor_interval_running is None:
             self.monitor_interval_running = get_params()["cluster"].get(
                 'monitor_interval_running_default', GEVENT_TIMEOUT_WAIT)
-        # Set up signal handlers for clean-up on interruption
-        self.setup_signal_handlers()
 
     def __enter__(self):
         return self
@@ -926,9 +921,21 @@ def cleanup_all_jobs(self):
         self.active_jobs.clear()  # Clear the list after cleanup
 
     def setup_signal_handlers(self):
-        """Set up signal handlers to clean up jobs on SIGINT and SIGTERM."""
+        """Set up signal handlers to clean up jobs on SIGINT and SIGTERM.
+
+        Call this once from the main control loop only.  The handler guards
+        against firing in ruffus worker subprocesses: ruffus forks N workers
+        (multiprocess=N) after the handler is installed, so each worker
+        inherits it and would otherwise run cleanup when the process group
+        receives SIGTERM on normal exit.
+        """
+        import multiprocessing
 
         def signal_handler(signum, frame):
+            # Workers inherit this handler via fork.  Only the main process
+            # should perform clean-up.
+            if multiprocessing.current_process().name != "MainProcess":
+                return
             self.logger.info(f"Received signal {signum}. Starting clean-up.")
             self.cleanup_all_jobs()
             exit(1)
@@ -1522,8 +1529,16 @@ def run(statement, **kwargs):
             logger.info("Dry-run: {}".format(statement))
         return []
 
-    # Use get_executor to get the appropriate executor
-    executor = get_executor(options)  # Updated to use get_executor
+    # Prefer DRMAA (GridExecutor) when available - it is the stable, battle-tested
+    # path.  Fall back to the CLI-based executors only when DRMAA is absent.
+    wants_cluster = (
+        options.get("to_cluster", True)
+        and not options.get("without_cluster", False)
+    )
+    if wants_cluster and HAS_DRMAA and GLOBAL_SESSION is not None:
+        executor = GridExecutor(**options)
+    else:
+        executor = get_executor(options)
 
     # Execute statement list within the context of the executor
     with executor as e:

cgatcore/pipeline/executors.py

@@ -118,7 +118,13 @@ def run(self, statement_list):
                 self.logger.error(f"Slurm job submission failed: {process.stderr}")
                 raise RuntimeError(f"Slurm job submission failed: {process.stderr}")
 
-            job_id = process.stdout.strip()
+            # sbatch outputs "Submitted batch job <id>" - extract the numeric ID
+            stdout = process.stdout.strip()
+            try:
+                job_id = stdout.split()[-1]
+                int(job_id)  # validate it's numeric
+            except (IndexError, ValueError):
+                raise RuntimeError(f"Could not parse job ID from sbatch output: {stdout!r}")
             self.logger.info(f"Slurm job submitted with ID: {job_id}")
 
             # Monitor job completion
@@ -135,34 +141,62 @@ def build_job_script(self, statement):
     def monitor_job_completion(self, job_id):
         """Monitor the completion of a Slurm job.
 
+        Uses squeue to poll while the job is alive (pending/running), then
+        falls back to sacct to retrieve the final exit status.  This avoids
+        querying the accounting database before the job has been recorded.
+
         Args:
             job_id (str): The Slurm job ID to monitor.
 
         Raises:
             RuntimeError: If the job fails or times out.
         """
+        TERMINAL_STATES = {"FAILED", "TIMEOUT", "CANCELLED", "NODE_FAIL", "OUT_OF_MEMORY"}
+
+        # Poll with squeue while the job is still in the queue
         while True:
-            # Use sacct to get job status
-            cmd = f"sacct -j {job_id} --format=State --noheader --parsable2"
+            cmd = f"squeue -j {job_id} -h -o %T"
             process = subprocess.run(cmd, shell=True, capture_output=True, text=True)
-
-            if process.returncode != 0:
-                self.logger.error(f"Failed to get job status: {process.stderr}")
-                raise RuntimeError(f"Failed to get job status: {process.stderr}")
+            state = process.stdout.strip()
 
-            status = process.stdout.strip()
-
-            # Check job status
-            if status in ["COMPLETED", "COMPLETED+"]:
-                self.logger.info(f"Job {job_id} completed successfully")
+            if not state:
+                # Job has left the queue - check accounting for final status
                 break
-            elif status in ["FAILED", "TIMEOUT", "CANCELLED", "NODE_FAIL"]:
-                self.logger.error(f"Job {job_id} failed with status: {status}")
-                raise RuntimeError(f"Job {job_id} failed with status: {status}")
-
-            # Wait before checking again
+
+            self.logger.info(f"Job {job_id} state: {state}")
+
+            if state in TERMINAL_STATES:
+                raise RuntimeError(f"Job {job_id} failed with status: {state}")
+
             time.sleep(10)
 
+        # Job is no longer in squeue; query sacct for final status.
+        # Retry a few times because sacct may lag slightly behind job completion.
+        for attempt in range(10):
+            cmd = f"sacct -j {job_id} --format=State,ExitCode --noheader --parsable2"
+            process = subprocess.run(cmd, shell=True, capture_output=True, text=True)
+
+            if process.returncode != 0:
+                raise RuntimeError(f"sacct query failed for job {job_id}: {process.stderr}")
+
+            # sacct returns one line per job step; examine the batch step (first line)
+            lines = [line.strip() for line in process.stdout.splitlines() if line.strip()]
+            if lines:
+                # format is "STATE|EXITCODE"
+                state = lines[0].split("|")[0].split()[0]  # handle "CANCELLED by X"
+                self.logger.info(f"Job {job_id} final state: {state}")
+
+                if state == "COMPLETED":
+                    self.logger.info(f"Job {job_id} completed successfully")
+                    return
+                elif any(state.startswith(s) for s in TERMINAL_STATES):
+                    raise RuntimeError(f"Job {job_id} failed with status: {state}")
+                # Any other state (e.g. COMPLETING) - wait and retry
+
+            time.sleep(5)
+
+        raise RuntimeError(f"Job {job_id} did not reach a terminal state after sacct retries")
+
     def collect_benchmark_data(self, statements, resource_usage=None):
         """Collect benchmark data for Slurm jobs.
 

docs/getting_started/installation.md

@@ -1,118 +1,76 @@
 # Installation
 
-The following sections describe how to install the `cgatcore` framework.
+## Requirements
 
-## Conda installation {#conda-installation}
+- Python 3.8 or higher
+- Linux or macOS
+- Conda (recommended) or pip
 
-The preferred method of installation is using Conda. If you do not have Conda installed, you can install it using [Miniconda](https://conda.io/miniconda.html) or [Anaconda](https://www.anaconda.com/download/#macos).
+## Conda (recommended)
 
-`cgatcore` is installed via the Bioconda channel, and the recipe can be found on [GitHub](https://github.com/bioconda/bioconda-recipes/tree/b1a943da5a73b4c3fad93fdf281915b397401908/recipes/cgat-core). To install `cgatcore`, run the following command:
+cgatcore is available on the Bioconda channel:
 
 ```bash
 conda install -c conda-forge -c bioconda cgatcore
 ```
 
-### Prerequisites {#prerequisites}
-
-Before installing `cgatcore`, ensure that you have the following prerequisites:
-
-- **Operating System**: Linux or macOS
-- **Python**: Version 3.6 or higher
-- **Conda**: Recommended for dependency management
-
-### Troubleshooting {#troubleshooting}
-
-- **Conda Issues**: If you encounter issues with Conda, ensure that the Bioconda and Conda-Forge channels are added and prioritized correctly.
-- **Pip Dependencies**: When using pip, manually install any missing dependencies listed in the error messages.
-- **Script Errors**: If the installation script fails, check the script's output for error messages and ensure all prerequisites are met.
-
-### Verification {#verification}
-
-After installation, verify the installation by running:
-
-```bash
-python
-```
-
-```python
-import cgatcore
-print(cgatcore.__version__)
-```
-
-This should display the installed version of `cgatcore`.
-
-## Pip installation {#pip-installation}
-
-We recommend installation through Conda because it manages dependencies automatically. However, `cgatcore` is generally lightweight and can also be installed using the `pip` package manager. Note that you may need to manually install other dependencies as needed:
+## pip
 
 ```bash
 pip install cgatcore
 ```
 
-## Automated installation {#automated-installation}
-
-The preferred method to install `cgatcore` is using Conda. However, we have also created a Bash installation script, which uses [Conda](https://conda.io/docs/) under the hood.
+## Development installation
 
-Here are the steps:
-
-```bash
-# Download the installation script:
-curl -O https://raw.githubusercontent.com/cgat-developers/cgat-core/master/install.sh
-
-# See help:
-bash install.sh
-
-# Install the development version (recommended, as there is no production version yet):
-bash install.sh --devel [--location </full/path/to/folder/without/trailing/slash>]
-
-# To download the code in Git format instead of the default zip format, use:
---git # for an HTTPS clone
---git-ssh # for an SSH clone (you need to be a cgat-developer contributor on GitHub to do this)
-
-# Enable the Conda environment as instructed by the installation script
-# Note: you might want to automate this by adding the following instructions to your .bashrc
-source </full/path/to/folder/without/trailing/slash>/conda-install/etc/profile.d/conda.sh
-conda activate base
-conda activate cgat-c
-```
-
-The installation script will place everything under the specified location. The aim of the script is to provide a portable installation that does not interfere with existing software environments. As a result, you will have a dedicated Conda environment that can be activated as needed to work with `cgatcore`.
-
-## Manual installation {#manual-installation}
-
-To obtain the latest code, check it out from the public Git repository and activate it:
+To install from source for development:
 
 ```bash
 git clone https://github.com/cgat-developers/cgat-core.git
 cd cgat-core
-python setup.py develop
+pip install -e .
 ```
 
-To update to the latest version, simply pull the latest changes:
+To update:
 
 ```bash
 git pull
+pip install -e .
 ```
 
-## Installing additional software {#installing-additional-software}
-
-When building your own workflows, we recommend using Conda to install software into your environment where possible. This ensures compatibility and ease of installation.
-
-To search for and install a package using Conda:
+## Verify installation
 
-```bash
-conda search <package>
-conda install <package>
+```python
+import cgatcore
+print(cgatcore.__version__)
 ```
 
-## Accessing the libdrmaa shared library {#accessing-libdrmaa}
+## DRMAA support (for HPC clusters)
 
-You may also need access to the `libdrmaa.so.1.0` C library, which can often be installed as part of the `libdrmaa-dev` package on most Unix systems. Once installed, you may need to specify the location of the DRMAA library if it is not in a default library path. Set the `DRMAA_LIBRARY_PATH` environment variable to point to the library location.
+To submit jobs to a cluster via DRMAA, install the Python DRMAA bindings and
+point the environment variable at your cluster's DRMAA shared library:
 
-To set this variable permanently, add the following line to your `.bashrc` file (adjusting the path as necessary):
+```bash
+pip install drmaa
+# or, for Slurm specifically:
+conda install -c conda-forge slurm-drmaa
+```
 
 ```bash
 export DRMAA_LIBRARY_PATH=/usr/lib/libdrmaa.so.1.0
 ```
 
-[Conda documentation](https://conda.io)
+Add the `export` line to your `~/.bashrc` (or your conda environment's
+activation script) so it is set every session.
+
+If DRMAA is not available, cgatcore falls back to direct CLI job submission
+(`sbatch`, `qsub`, etc.) — see [Cluster configuration](../pipeline_modules/cluster.md).
+
+## Troubleshooting
+
+- **Conda channel priority**: ensure `conda-forge` and `bioconda` are listed in
+  your `.condarc` and that `conda-forge` is first.
+- **Missing dependencies**: with a bare `pip install`, you may need to install
+  optional dependencies (e.g. `drmaa`, `gevent`, `ruffus`) manually.
+- **`DRMAA_LIBRARY_PATH` not set**: jobs will silently fall back to local
+  execution unless the CLI fallback is configured
+  (`cluster.queue_manager` in `.cgat.yml`).