Repository: ExaWorks/psij-python (language: python)

Issue:
Remove SAGA
Worked as a test case, but likely not needed anymore. 


Discussion / hints:

Fix patch:
diff --git a/docs/programming.rst b/docs/programming.rst
index a4113a06..20749321 100644
--- a/docs/programming.rst
+++ b/docs/programming.rst
@@ -54,7 +54,7 @@ Executors
 
 Executors are concrete implementations of mechanisms that execute jobs.
 To get an instance of a specific executor, call
-:meth:`JobExecutor.get_instance(name) <psij.job_executor.JobExecutor.get_instance>`, 
+:meth:`JobExecutor.get_instance(name) <psij.job_executor.JobExecutor.get_instance>`,
 with ``name`` being one of the installed executor names. Executors can be
 installed from multiple sources, so the precise list of executors
 avaiable to a specific installation of the PSI/J Python library can vary.
@@ -70,8 +70,8 @@ The executor implementations provided by this version of PSI/J Python
 are:
 
 :class:`cobalt <psij.executors.batch.cobalt.CobaltJobExecutor>`
-    A job executor that can interact with the 
-    `Cobalt HPC Job Scheduler <https://xgitlab.cels.anl.gov/aig-public/cobalt>`_, 
+    A job executor that can interact with the
+    `Cobalt HPC Job Scheduler <https://xgitlab.cels.anl.gov/aig-public/cobalt>`_,
     which is used by `Argonne's <www.anl.gov>`_ `ALCF <www.alcf.anl.gov>`_ systems.
 
 :class:`flux <psij.executors.flux.FluxJobExecutor>`
@@ -81,20 +81,15 @@ are:
     A job executor that runs jobs locally by forking a subprocess.
 
 :class:`lsf <psij.executors.batch.lsf.LsfJobExecutor>`
-    An implementation of a job executor for the 
+    An implementation of a job executor for the
     `IBM Spectrum LSF workload manager <https://www.ibm.com/docs/en/spectrum-lsf>`_.
 
 :class:`rp <psij.executors.rp.RPJobExecutor>`
     Job executor for the `RADICAL Pilot system
     <https://radical-cybertools.github.io/radical-pilot/>`_.
 
-:class:`saga <psij.executors.saga.SagaExecutor>`
-    An executor that delegates job execution to the 
-    `RADICAL-SAGA <https://radical-cybertools.github.io/saga-python/index.html>`_
-    library.
-
 :class:`slurm <psij.executors.batch.slurm.SlurmJobExecutor>`
-    An executor for the 
+    An executor for the
     `Slurm Workload Manager <https://slurm.schedmd.com/overview.html>`_.
 
 
@@ -108,7 +103,7 @@ once a set of nodes has been allocated for the job. In essence, launchers
 are wrappers around the job executable which can provide additional
 features, such as setting up an MPI environment, starting a copy of the
 job executable on each allocated node, etc. To get a launcher instance,
-call :meth:`Launcher.get_instance(name) <psij.launcher.Launcher.get_instance>` 
+call :meth:`Launcher.get_instance(name) <psij.launcher.Launcher.get_instance>`
 with ``name`` being the name of a launcher. Like job executors, above,
 launchers are plugins and can come from various places. To obtain a list
 of launchers, you can run:
@@ -120,16 +115,16 @@ of launchers, you can run:
 The PSI/J Python library comes with a core set of launchers, which are:
 
 :class:`aprun <psij.launchers.aprun.AprunLauncher>`
-    Launches jobs using 
+    Launches jobs using
     `Cobalt's <https://xgitlab.cels.anl.gov/aig-public/cobalt>`_ ``aprun``.
 
 :class:`jsrun <psij.launchers.jsrun.JsrunLauncher>`
-    Starts jobs using ``jsrun`` provided by the 
+    Starts jobs using ``jsrun`` provided by the
     `IBM Spectrum LSF workload manager <https://www.ibm.com/docs/en/spectrum-lsf>`_.
 
 :class:`mpirun <psij.launchers.mpirun.MPILauncher>`
-    Launches jobs using ``mpirun``, which is a tool provided by 
-    `MPI <https://www.mpi-forum.org/>`_ implementations, such as 
+    Launches jobs using ``mpirun``, which is a tool provided by
+    `MPI <https://www.mpi-forum.org/>`_ implementations, such as
     `Open MPI <https://www.open-mpi.org/>`_.
 
 :class:`multiple <psij.launchers.multiple.MultipleLauncher>`
@@ -141,7 +136,7 @@ The PSI/J Python library comes with a core set of launchers, which are:
     default launcher, which is used if nothing is explicitly specified.
 
 :class:`srun <psij.launchers.srun.SrunLauncher>`
-    Starts job using ``srun``, which is part of the 
+    Starts job using ``srun``, which is part of the
     `Slurm Workload Manager <https://slurm.schedmd.com/overview.html>`_.
 
 
diff --git a/requirements-connector-saga.txt b/requirements-connector-saga.txt
deleted file mode 100644
index dbf595fa..00000000
--- a/requirements-connector-saga.txt
+++ /dev/null
@@ -1,3 +0,0 @@
-# For SAGA executor
-radical.saga >= 1.8.0
-radical.utils
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 3aa03f3f..f64a450f 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -2,7 +2,6 @@
 -r requirements-docs.txt
 
 # Executors
--r requirements-connector-saga.txt
 -r requirements-connector-radical.txt
 
 # Testing / QA
diff --git a/requirements-docs.txt b/requirements-docs.txt
index c408246d..47db5498 100644
--- a/requirements-docs.txt
+++ b/requirements-docs.txt
@@ -2,7 +2,6 @@
 -r requirements.txt
 
 # Executors
--r requirements-connector-saga.txt
 -r requirements-connector-radical.txt
 
 
diff --git a/scripts/psijcli.py b/scripts/psijcli.py
index 62fc0572..e69bf179 100644
--- a/scripts/psijcli.py
+++ b/scripts/psijcli.py
@@ -23,12 +23,11 @@
                                         "flux",
                                         "lsf",
                                         "rp",
-                                        "saga",
                                         "slurm"
                                         ],
                             )
 execute_parser.add_argument("file", help="JobSpec file")
-execute_parser.add_argument("-n", 
+execute_parser.add_argument("-n",
                             "--number-of-jobs",
                             dest = "jobs",
                             type = int,
@@ -48,7 +47,7 @@
                     help="print debug information")
 
 
-# parser.print_help()    
+# parser.print_help()
 
 args = parser.parse_args()
 
@@ -58,47 +57,47 @@
     if args.verbose:
         print("Validating " + args.file)
     job_spec = i.load(args.file)
-    
+
     if job_spec and  isinstance(job_spec, JobSpec):
         print("File ok")
     else:
         sys.exit("Not a valid file, could not import " + args.file)
 elif args.command == "run":
-    
+
     if not args.executor:
         sys.exit("Missing argument executor")
-    
+
     if args.verbose:
         print("Importing " + args.file)
     job_spec = i.load(args.file)
     if not (job_spec and  isinstance(job_spec, JobSpec)):
         sys.exit("Something wrong with JobSpec")
-    
-    # Get job executor    
-    if args.verbose:    
+
+    # Get job executor
+    if args.verbose:
         print("Initializing job executor")
-    
+
     jex = None
-    
-    try:    
+
+    try:
         jex = psij.JobExecutor.get_instance(args.executor)
     except ValueError as err:
-        sys.exit(f"Panic, {err}") 
+        sys.exit(f"Panic, {err}")
     except BaseException as err:
         sys.exit(f"Unexpected: {err}, {type(err)}")
 
-    # Submit jobs      
+    # Submit jobs
     number_of_jobs = args.jobs
     if args.verbose:
         print("Submitting " + str(number_of_jobs) + " job(s)")
-    
+
     jobs = [] # list of created jobs
     for i in range(number_of_jobs):
         job = psij.Job()
         job.spec = job_spec
         jobs.append(job)
         jex.submit(job)
-    
+
     if args.verbose:
         print("Waiting for jobs to finish")
     for i in range(number_of_jobs):
@@ -106,4 +105,4 @@
 else:
     # Should never be here
     sys.stderr.write("Missig command. Use --help for more information.\n")
-    parser.print_help(sys.stderr)
\ No newline at end of file
+    parser.print_help(sys.stderr)
diff --git a/src/psij-descriptors/saga_descriptor.py b/src/psij-descriptors/saga_descriptor.py
deleted file mode 100644
index 70cb751b..00000000
--- a/src/psij-descriptors/saga_descriptor.py
+++ /dev/null
@@ -1,7 +0,0 @@
-from distutils.version import StrictVersion
-
-from psij.descriptor import Descriptor
-
-
-__PSI_J_EXECUTORS__ = [Descriptor(name='saga', version=StrictVersion('0.0.1'),
-                                  cls='psij.executors.saga.SagaExecutor')]
diff --git a/src/psij/executors/saga.py b/src/psij/executors/saga.py
deleted file mode 100644
index ea7ba53a..00000000
--- a/src/psij/executors/saga.py
+++ /dev/null
@@ -1,244 +0,0 @@
-import atexit
-import logging
-from threading import RLock
-from typing import Optional, Dict, List
-
-import radical.saga as rs
-import radical.saga.job.constants as ct
-import time
-
-from psij import JobExecutor, JobExecutorConfig, JobState, JobStatus, Job, \
-    SubmitException, InvalidJobException
-
-logger = logging.getLogger(__name__)
-
-
-# map SAGA states to jpsi states.
-_STATE_MAP = {
-    rs.job.NEW: JobState.NEW,
-    rs.job.PENDING: JobState.QUEUED,
-    rs.job.RUNNING: JobState.ACTIVE,
-    rs.job.DONE: JobState.COMPLETED,
-    rs.job.FAILED: JobState.FAILED,
-    rs.job.CANCELED: JobState.CANCELED,
-}
-
-
-class _JobMapping:
-    def __init__(self, jpsi_job: Job, saga_job: Optional[rs.job.Job] = None):
-        self.jpsi_job = jpsi_job
-        self.saga_job = saga_job
-
-
-class SagaExecutor(JobExecutor):
-    """
-    An executor that use SAGA to run jobs.
-
-    For details about SAGA, see https://en.wikipedia.org/wiki/Simple_API_for_Grid_Applications
-    and related links.
-    """
-
-    def __init__(self, url: Optional[str] = None,
-                 config: Optional[JobExecutorConfig] = None) -> None:
-        """
-        Initializes a `SagaExecutor`.
-
-        Parameters
-        ----------
-        url
-            An URL representing a SAGA resource manager to connect to.
-        config
-            Not used.
-        """
-        self._jobs = dict()  # type: Dict[str, _JobMapping]
-        self._js = None
-
-        if not config:
-            config = JobExecutorConfig()
-
-        super().__init__(url=url, config=config)
-
-        self._lock = RLock()
-        self._js = rs.job.Service(self.url)
-        self._cb = None
-
-        atexit.register(self._close)
-
-    def _close(self) -> None:
-        if self._js:
-            self._js.close()
-            self._js = None
-
-    def __del__(self) -> None:
-        """
-        Removes process shutdown hooks added earlier.
-
-        The hooks are used to close the SAGA service used by this executor.
-        """
-        atexit.unregister(self._close)
-        self._close()
-
-    def _state_cb(self, saga_job: rs.job.Job, metric: object, ctx: object) -> bool:
-        jpsi_ids = list()
-        with self._lock:
-            for job_mapping in self._jobs.values():
-                if job_mapping.saga_job and job_mapping.saga_job.id == saga_job.id:
-                    jpsi_ids.append(job_mapping.jpsi_job.id)
-
-        for jpsi_id in jpsi_ids:
-            jpsi_job = self._get_jpsi_job(jpsi_id)
-            ec = None
-
-            print('%s: %s - %s' % (jpsi_job.id, saga_job.state, saga_job.id))
-            if not jpsi_job._native_id:
-                jpsi_job._native_id = saga_job.id
-
-            if saga_job.state in [rs.FAILED, rs.DONE]:
-                ec = saga_job.exit_code
-
-            logger.debug('%s --> %s', jpsi_id, saga_job.state)
-
-            job_status = JobStatus(_STATE_MAP[saga_job.state], time.time(), exit_code=ec)
-            self._set_job_status(jpsi_job, job_status)
-
-        return True
-
-    def submit(self, job: Job) -> None:
-        """
-        Submits the specified :class:`~psij.Job` to be run using SAGA.
-
-        Parameters
-        ----------
-        job
-            The :class:`~psij.Job` to be submitted.
-        """
-        assert(self._js)
-        # derive SAGA job description and submit it
-        job_mapping = _JobMapping(job)
-        with self._lock:
-            self._jobs[job.id] = job_mapping
-        job.executor = self
-
-        jd = self._job_2_descr(job)
-        saga_job = self._js.create_job(jd)
-
-        job_mapping.saga_job = saga_job
-        saga_job.add_callback(rs.STATE, self._state_cb)
-
-        # TODO: the backend attribute is not standard; why do we need it?
-        self._set_job_status(job, JobStatus(JobState.QUEUED, metadata={'backend': self.url}))
-
-        saga_job.run()
-
-    def _get_saga_job(self, uid: str) -> rs.job.Job:
-        with self._lock:
-            return self._jobs[uid].saga_job
-
-    def _get_jpsi_job(self, uid: str) -> Job:
-        with self._lock:
-            return self._jobs[uid].jpsi_job
-
-    def list(self) -> List[str]:
-        """
-        Retrieves a list of native IDs corresponding to jobs known to this executor.
-
-        This method does not query the SAGA backend or any of the backends behind SAGA for jobs
-        not submitted through this executor. Furthermore, it simply returns the PSI-J job IDs of
-        the jobs rather than native IDs.
-        Returns
-        -------
-        A list of job IDs.
-        """
-        with self._lock:
-            return list(self._jobs.keys())
-
-    def attach(self, job: Job, native_id: str) -> None:
-        """
-        Attaches a job to a backend job.
-
-        This method does not quite do what it is meant to do at this time.
-
-        Parameters
-        ----------
-        job
-            The job to attach.
-        native_id
-            A PSI-J job ID
-
-        """
-        saga_job = None
-        job.executor = self
-        with self._lock:
-            # try to find job in known jobs
-            for job_mapping in self._jobs.values():
-                if job_mapping.jpsi_job.status.metadata:
-                    job_native_id = job_mapping.jpsi_job.status.metadata.get('native_id')
-                    if not job_native_id:
-                        continue
-                    if job_native_id == native_id:
-                        saga_job = job_mapping.saga_job
-                        if not saga_job:
-                            continue
-                        # found the native saga job, attach it to this job
-                        job_mapping = _JobMapping(job, saga_job)
-                        self._jobs[job.id] = job_mapping
-                        break
-
-            # saga job is not known to this executor - re-connect via backend:
-            try:
-                assert(self._js)
-                saga_job = self._js.get_job(native_id)
-                job_mapping = _JobMapping(job, saga_job)
-                self._jobs[job.id] = job_mapping
-            except Exception:
-                pass
-
-        if not saga_job:
-            # TODO: There is a discussion on the suitability of cancel()
-            #       throwing a SubmitException - same applies here
-            raise SubmitException('Unknown native id: "{}'.format(native_id))
-
-        # we found the job and reconnected to it - update the job state
-        job._native_id = native_id
-        state = _STATE_MAP[saga_job.state]
-        if state.final:
-            self._set_job_status(job, JobStatus(state, exit_code=saga_job.exit_code))
-        else:
-            self._set_job_status(job, JobStatus(state))
-
-    def _job_2_descr(self, job: Job) -> rs.job.Description:
-        spec = job.spec
-        if not spec:
-            raise InvalidJobException('Missing specification.')
-
-        jd = rs.job.Description()
-        if spec.executable:
-            jd.set_attribute(ct.EXECUTABLE, spec.executable)
-        if spec.arguments:
-            jd.set_attribute(ct.ARGUMENTS, spec.arguments)
-        if spec.environment:
-            jd.set_attribute(ct.ENVIRONMENT, spec.environment)
-        if spec.stdin_path:
-            jd.set_attribute(ct.INPUT, spec.stdin_path)
-        if spec.stdout_path:
-            jd.set_attribute(ct.OUTPUT, spec.stdout_path)
-        if spec.stderr_path:
-            jd.set_attribute(ct.ERROR, spec.stderr_path)
-        if spec.directory:
-            jd.set_attribute(ct.WORKING_DIRECTORY, spec.directory)
-
-        return jd
-
-    def cancel(self, job: Job) -> None:
-        """
-        Cancels a job.
-
-        Parameters
-        ----------
-        job
-            The job to cancel.
-        """
-        saga_job = self._get_saga_job(job.id)
-
-        if saga_job.state not in rs.FINAL:
-            saga_job.cancel()


Test patch:
diff --git a/requirements-tests.txt b/requirements-tests.txt
index 73da73d6..1226c027 100644
--- a/requirements-tests.txt
+++ b/requirements-tests.txt
@@ -1,3 +1,2 @@
 -r requirements.txt
--r requirements-connector-saga.txt
 -r requirements-connector-radical.txt
diff --git a/testing.conf b/testing.conf
index 16ee8db1..452d00ff 100644
--- a/testing.conf
+++ b/testing.conf
@@ -3,7 +3,7 @@
 # are public. The options are as follows:
 #
 # hostname:
-#     The host name of the machine on which the tests are invoked. This is 
+#     The host name of the machine on which the tests are invoked. This is
 #     useful in identifying the machine such that the community is aware of the
 #     status of PSI/J on it.
 #
@@ -26,28 +26,28 @@ id = hostname
 #
 # maintainer_email = <string>
 
-maintainer_email = 
+maintainer_email =
 
 
-# 
-# The scope selects which branches get tested. This selection can potentially 
+#
+# The scope selects which branches get tested. This selection can potentially
 # impact security. The options are:
-# 
-# stable: 
-#     Only test stable branches (currently the main branch). The stable 
-#     branches are branches in the core repository which are considered stable 
-#     and have been extensively reviewed and tested on other machines. They 
+#
+# stable:
+#     Only test stable branches (currently the main branch). The stable
+#     branches are branches in the core repository which are considered stable
+#     and have been extensively reviewed and tested on other machines. They
 #     carry the lowest risks.
 #
 # core:
-#     Test all core branches. This tests all stable branches, as above, and 
-#     all branches that are in the core namespace and are part of pull 
+#     Test all core branches. This tests all stable branches, as above, and
+#     all branches that are in the core namespace and are part of pull
 #     requests. Since only core developers can create branches in the core
-#     namespace, this limits the tested code to code written by core 
+#     namespace, this limits the tested code to code written by core
 #     developers.
 # tagged:
 #     Test all core branches and all branches that are part of pull requests
-#     that have been explicitly tagged by core developers for testing. The 
+#     that have been explicitly tagged by core developers for testing. The
 #     branches can exist in private namespaces, so, technically, anybody can
 #     submit such pull requests. However, a core developer needs to explicitly
 #     approve testing on such pull requests. This option carries the most
@@ -59,20 +59,19 @@ scope = core
 
 
 #
-# Which executors to test. This is a comma separated list of 
-# <executor_name>[:<launcher_name>[:<url>]] items. The "auto_q" executor means 
-# that the testing infrastructure will attempt to determine which queuing system 
+# Which executors to test. This is a comma separated list of
+# <executor_name>[:<launcher_name>[:<url>]] items. The "auto_q" executor means
+# that the testing infrastructure will attempt to determine which queuing system
 # is present on the test machine and use the corresponding executor for testing.
-# Similar to "auto_q", "auto_l" means that the testing infrastructure will 
+# Similar to "auto_q", "auto_l" means that the testing infrastructure will
 # determine an appropriate launcher for the system. The launcher name can be
-# empty if a particular executor does not use launchers but uses URLs 
+# empty if a particular executor does not use launchers but uses URLs
 # (e.g., saga::localhost). The special value "auto" (without quotes) means that
-# the executors to be tested should be determined automatically based on a 
+# the executors to be tested should be determined automatically based on a
 # survey of the runtime environment.
 #
 # executors = local:single, local:multiple, \
 #            batch-test:single, batch-test:multiple, \
-#            saga::fork://localhost/, \
 #            auto_q:single, auto_q:multiple, auto_q:auto_l
 
 executors = auto
@@ -107,14 +106,14 @@ server_url = https://psij.testing.exaworks.org
 # A key authenticates a test ID such that it reduces the odds that two machines
 # compete to overwrite each other's results. The machines are identified by their
 # IDs. However, without additional steps, any other test deployment can set its
-# ID to an existing one. The "key" setting allows a secret to be used to 
+# ID to an existing one. The "key" setting allows a secret to be used to
 # authenticate requests. There are two valid values:
-# 
+#
 # random:
 #     Generates a random key on the first upload attempt. The server will then
 #     only accept subsequent requests from the same ID if the key matches. There
-#     is a 7 day expiration of a pairing which allows for a new key to be used 
-#     if an old key has not been used for 7 or more days. Random keys are 
+#     is a 7 day expiration of a pairing which allows for a new key to be used
+#     if an old key has not been used for 7 or more days. Random keys are
 #     persisted locally in ~/.psij
 #
 # "<string>":
@@ -126,12 +125,12 @@ key = random
 
 
 #
-# The maximum age, in hours, for saved test results to be kept on disk. When 
-# tests are run with the "--save-results", test results are saved on disk. 
-# They can be later uploaded to the aggregation server using 
-# "make upload-results". In order to prevent a very large number of test 
-# results from occupying disk space indefinitely, the various tools involved 
-# purge old results from disk. This parameter sets a limit on the age of 
+# The maximum age, in hours, for saved test results to be kept on disk. When
+# tests are run with the "--save-results", test results are saved on disk.
+# They can be later uploaded to the aggregation server using
+# "make upload-results". In order to prevent a very large number of test
+# results from occupying disk space indefinitely, the various tools involved
+# purge old results from disk. This parameter sets a limit on the age of
 # results that should be kept on disk.
 #
 # max_age = 48
diff --git a/tests/conftest.py b/tests/conftest.py
index 805e0c18..6ef84fe1 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -96,10 +96,6 @@ def _get_executors(config: Dict[str, str]) -> List[str]:
 def _translate_executor(config: Dict[str, str], executor: str) -> List[str]:
     if executor == 'auto':
         execs = ['local:single', 'local:multiple', 'batch-test:single', 'batch-test:multiple']
-        if config.option.environment.get('has_saga'):
-            execs.append('saga::fork://localhost/')
-            if config.option.environment.get('can_ssh_to_localhost'):
-                execs.append('saga::ssh://localhost/')
         queue_execs = _translate_executor(config, 'auto_q')
         assert len(queue_execs) in [0, 1]
         queue_exec = None
@@ -303,17 +299,6 @@ def _get_run_id(config):
     return run_id
 
 
-def _has_saga():
-    # try to instantiate SAGA executor
-    try:
-        from psij.executors.saga import SagaExecutor
-        SagaExecutor()
-        return True
-    except Exception as ex:
-        logger.info(ex)
-        return False
-
-
 def _has_flux():
     try:
         import flux
@@ -382,7 +367,6 @@ def _discover_environment(config):
         env['has_slurm'] = shutil.which('sbatch') is not None
         env['has_mpirun'] = shutil.which('mpirun') is not None
         env['has_flux'] = _has_flux()
-        env['has_saga'] = _has_saga()
         env['can_ssh_to_localhost'] = _try_run_command(['ssh', '-oBatchMode=yes',
                                                         '-oStrictHostKeyChecking=no', 'localhost',
                                                         'true'], timeout=5)
diff --git a/tests/test_executor.py b/tests/test_executor.py
index 19791f39..e0f3db62 100644
--- a/tests/test_executor.py
+++ b/tests/test_executor.py
@@ -8,7 +8,7 @@
 from executor_test_params import ExecutorTestParams
 
 
-_QUICK_EXECUTORS = set(['local', 'batch-test', 'saga'])
+_QUICK_EXECUTORS = set(['local', 'batch-test'])
 
 
 def _get_timeout(execparams: ExecutorTestParams) -> Optional[timedelta]: