COMPLETED & recovery

ruff

doc

test group is called without errors

more tests

now it tests noplot correctness aswell,
Enabled offline, but if it fails it will mrked as xfail

changelog

Added the job_names calls, removed get_completed_files

Moved some tests/fixes in 4.2 && added setstatus/recovery

wip still missing something

wip

Author: dbeltrankyl

CHANGELOG.md

@@ -34,6 +34,8 @@ platform does not support wrappers anymore (it was used for testing).
 - Fixes an issue with multi-day applications dependencies bug #2631
 - Fixes an issue with all-filter #2565
 - Fixes an issue when setting a dependency to a different date or member # 2466 ( #2518 partially)
+- Fixes an issue with recovery not being able to cancel active jobs #2695
+- Fixes an issue with SQLALCHEMY not working correctly with the historical job_data.db #2695
 
 **Enhancements:**
 
@@ -60,6 +62,10 @@ platform does not support wrappers anymore (it was used for testing).
 - Added automated citation instructions to the website landing page #2480
 - Improving the error message handling for incorrect YAML syntax (too cryptic) #2651
 - Added a platform option to remove remote log files of the job's output after log recovery #2655
+- COMPLETED files are now fetch instead of downloaded. #2695, #2559
+- Improved recovery command perfomance by fetching all the COMPLETED files in a single call. related to #2695, #2570, #2563
+- Added --offline flag to recovery command to avoid not raise an error when the platform is not reachable. related to #2695
+- Improved setstatus #2695
 
 ### 4.1.15: Bug fixes, enhancements, and new features
 

autosubmit/autosubmit.py

@@ -3041,6 +3041,7 @@ def recovery(expid, noplot, save, all_jobs, hide, group_by=None, expand=list(),
         """
         Method to check all active jobs. If COMPLETED file is found, job status will be changed to COMPLETED,
         otherwise it will be set to WAITING. It will also update the jobs list.
+
         :param expid: experiment identifier
         :param noplot: if True, no plot will be generated
         :param save: if True, changes will be saved to the job list

autosubmit/history/database_managers/experiment_history_db_manager.py

@@ -494,7 +494,7 @@ class SqlAlchemyExperimentHistoryDbManager:
     def __init__(self, options: dict):
         default_file = None if BasicConfig.DATABASE_BACKEND == "postgres" else f"job_data_{options.get('expid',options.get('schema',None))}.db"
         job_data_file = options.get('jobdata_file', default_file)
-        default_dir =  BasicConfig.DATABASE_CONN_URL if BasicConfig.DATABASE_BACKEND == "postgres" else BasicConfig.JOBDATA_DIR
+        default_dir = BasicConfig.DATABASE_CONN_URL if BasicConfig.DATABASE_BACKEND == "postgres" else BasicConfig.JOBDATA_DIR
         jobdata_path = Path(options.get('jobdata_path', default_dir))
         if job_data_file:
             jobdata_path = jobdata_path / job_data_file
@@ -852,8 +852,7 @@ def get_jobs_data_last_row(self, job_names) -> dict[str, Any]:
                 jobs_data_by_name[job['job_name']] = job
         return jobs_data_by_name
 
-
-    def select_jobs_data(self, table, job_names) -> List[tuple[str, Any]]:
+    def select_jobs_data(self, table, job_names) -> list[tuple[str, Any]]:
         query = select(table).where(
             and_(
                 table.c.last == 1,
@@ -865,7 +864,6 @@ def select_jobs_data(self, table, job_names) -> List[tuple[str, Any]]:
         columns = table.c.keys()
         return [tuple(zip(columns, row)) for row in rows]
 
-
 def create_experiment_history_db_manager(db_engine: str, **options: Any) -> ExperimentHistoryDatabaseManager:
     if db_engine == 'postgres':
         return cast(ExperimentHistoryDatabaseManager, SqlAlchemyExperimentHistoryDbManager(options))

autosubmit/history/experiment_history.py

@@ -40,7 +40,7 @@ def __init__(self, expid, jobdata_dir_path=DEFAULT_JOBDATA_DIR, historiclog_dir_
         BasicConfig.read()
         self._log = Logging(expid, BasicConfig.HISTORICAL_LOG_DIR)
         self._job_data_dir_path = BasicConfig.JOBDATA_DIR
-        self._job_data_file = f"job_data_{expid}.db" if BasicConfig.DATABASE_BACKEND == "sqlite" else None
+        self._job_data_file = f"job_data_{expid}.db" if BasicConfig.DATABASE_BACKEND == "sqlite" else ""
         self._historiclog_dir_path = BasicConfig.HISTORICAL_LOG_DIR
         self.force_sql_alchemy = force_sql_alchemy
         self.manager: Optional[ExperimentHistoryDatabaseManager] = None

autosubmit/job/job_list.py

@@ -4130,8 +4130,10 @@ def recover_last_data(self):
                     job.local_logs = jobs_data[job.name]["out"]
                     job.remote_logs = jobs_data[job.name]["err"]
                     job.ready_date = datetime.datetime.fromtimestamp(jobs_data[job.name]["start"]).strftime('%Y%m%d%H%M%S')
-                    # TODO: (new feature, another PR) If not last, update_log must be False so it can be recovered on the next autosubmit run ( check the run_id of all jobs somehow)
-                    job.updated_log = True
+                    if job.status in [Status.COMPLETED, Status.FAILED]:
+                        job.updated_log = True
+                    else:
+                        job.updated_log = False
 
     def _get_job_names(self, status: Optional[list[int]] = None, platform: Platform = None) -> List[str]:
         """Get a list of all job names in the job list.

autosubmit/platforms/paramiko_platform.py

@@ -510,7 +510,7 @@ def get_logs_files(self, exp_id: str, remote_logs: tuple[str, str]) -> None:
 
     def get_list_of_files(self):
         return self._ftpChannel.get(self.get_files_path)
-    
+
     def _chunked_md5(self, file_buffer: BufferedReader) -> str:
         """Calculate the MD5 checksum of a file in chunks to avoid high memory usage.
 
@@ -522,7 +522,7 @@ def _chunked_md5(self, file_buffer: BufferedReader) -> str:
         for chunk in iter(lambda: file_buffer.read(CHUNK_SIZE), b""):
             md5_hash.update(chunk)
         return md5_hash.hexdigest()
-    
+
     def _checksum_validation(self, local_path: str, remote_path: str) -> bool:
         """
         Validates that the checksum of the local file matches the checksum of the remote file.
@@ -758,44 +758,47 @@ def job_is_over_wallclock(self, job, job_status, cancel=False):
                     Log.debug(f"Error cancelling job {job.id}: {str(e)}")
         return job_status
 
-    def get_completed_job_names(self, job_names: Optional[list[str]] = None) -> list[str]:
-        """
-        Retrieve the names of all files ending with '_COMPLETED' from the remote log directory using SSH.
+    def get_completed_job_names(self, job_names_provided: Optional[list[str]] = None) -> list[str]:
+        """Retrieve the names of all files ending with '_COMPLETED' from the remote log directory using SSH.
 
-        :param job_names: If provided, filters the results to include only these job names.
-        :type job_names: Optional[List[str]]
+        :param job_names_provided: If provided, filters the results to include only these job names.
+        :type job_names_provided: Optional[List[str]]
         :return: List of job names with COMPLETED files.
         :rtype: List[str]
         """
-        if not job_names:
-            cmd = f"find {self.remote_log_dir} -maxdepth 1 -name '*_COMPLETED' -type f"
-        else:
-            patterns = ' -o '.join([f"-name '{name}_COMPLETED'" for name in job_names])
-            cmd = f"find {self.remote_log_dir} -maxdepth 1 \\( {patterns} \\) -type f"
-        self.send_command(cmd)
-        output = self.get_ssh_output()
-        completed_files = output.strip().split('\n') if output else []
-        job_names = [Path(file).name.replace('_COMPLETED', '') for file in completed_files]
+        job_names = []
+        if self.expid in str(self.remote_log_dir):  # Ensure we are in the right experiment
+            if not job_names_provided:
+                cmd = f"find {self.remote_log_dir} -maxdepth 1 -name '*_COMPLETED' -type f"
+            else:
+                patterns = ' -o '.join([f"-name '{name}_COMPLETED'" for name in job_names_provided])
+                cmd = f"find {self.remote_log_dir} -maxdepth 1 \\( {patterns} \\) -type f"
+            self.send_command(cmd)
+            output = self.get_ssh_output()
+            completed_files = output.strip().split('\n') if output else []
+            job_names = [Path(file).name.replace('_COMPLETED', '') for file in completed_files]
         return job_names
 
-    def get_failed_job_names(self, job_names: Optional[list[str]] = None) -> list[str]:
+    def get_failed_job_names(self, job_names_provided: Optional[list[str]] = None) -> list[str]:
         """
         Retrieve the names of all files ending with '_COMPLETED' from the remote log directory using SSH.
 
-        :param job_names: If provided, filters the results to include only these job names.
-        :type job_names: Optional[List[str]]
+        :param job_names_provided: If provided, filters the results to include only these job names.
+        :type job_names_provided: Optional[List[str]]
         :return: List of job names with COMPLETED files.
         :rtype: List[str]
         """
-        if not job_names:
-            cmd = f"find {self.remote_log_dir} -maxdepth 1 -name '*_FAILED' -type f"
-        else:
-            patterns = ' -o '.join([f"-name '{name}_FAILED'" for name in job_names])
-            cmd = f"find {self.remote_log_dir} -maxdepth 1 \\( {patterns} \\) -type f"
-        self.send_command(cmd)
-        output = self.get_ssh_output()
-        completed_files = output.strip().split('\n') if output else []
-        job_names = [Path(file).name.replace('_FAILED', '') for file in completed_files]
+        job_names = []
+        if self.expid in str(self.remote_log_dir):  # Ensure we are in the right experiment
+            if not job_names_provided:
+                cmd = f"find {self.remote_log_dir} -maxdepth 1 -name '*_FAILED' -type f"
+            else:
+                patterns = ' -o '.join([f"-name '{name}_FAILED'" for name in job_names])
+                cmd = f"find {self.remote_log_dir} -maxdepth 1 \\( {patterns} \\) -type f"
+            self.send_command(cmd)
+            output = self.get_ssh_output()
+            completed_files = output.strip().split('\n') if output else []
+            job_names = [Path(file).name.replace('_FAILED', '') for file in completed_files]
         return job_names
 
     def delete_failed_and_completed_names(self, job_names: list[str]) -> None:

docs/source/userguide/modifying_workflow/index.rst

@@ -8,8 +8,6 @@ How to recover an experiment
 
 We utilize the recovery command to restart Autosubmit when it is killed in an ungraceful way. 
 
-.. warning:: You can only recover when the workflow is not running (i.e. there no QUEUING, SUBMITTED, or RUNNING tasks) 
-
 The recovery command will change the state of all the tasks that are (or can be) in READY status to COMPLETED if a completed file for that task is found. 
 
 To recover a running workflow, we must issue the recovery command with -f, so Autosubmit kills the tasks in remote before resetting their status.
@@ -24,15 +22,29 @@ Examples:
 
 ::
 
-    # check locally for completion files, do not kill running tasks, and do not commit changes
+    # check for completion files of the current active jobs ( SUBBMITED, QUEUING, RUNNING )
     autosubmit recovery <EXPID>
-    # commit to the previous states
+
+    # check for completion files of all jobs
+    autosubmit recovery <EXPID> --all
+
+    # Save changes
     autosubmit recovery <EXPID> -s
-    # check locally for completion file AND issue kill commands in remote
-    autosubmit recovery <EXPID> -f -s
+
+    # If there are active jobs, kill them in remote.
+    autosubmit recovery <EXPID> -f
+
+    # If some platform is not reachable, force the recovery without checking remote
+    autosubmit recovery <EXPID> --offline -f --all
+
     # check in remote for completion files
     autosubmit recovery <EXPID> --all -s
 
+    # Resume the experiment:
+    autosubmit create <EXPID> -np
+    autosubmit recovery <EXPID> -all -s [--offline] [-f]
+    autosubmit run <EXPID>
+
 Options:
 
 .. runcmd:: autosubmit recovery -h