From cc17d5304cc9ca9910162f5dd83640d31175bbad Mon Sep 17 00:00:00 2001
From: John Chilton <jmchilton@gmail.com>
Date: Thu, 31 Mar 2016 08:30:52 -0400
Subject: [PATCH] Implement a Pulsar embedded job runner.

Closes #1378.
---
 config/job_conf.xml.sample_advanced | 20 +++++++++++
 lib/galaxy/jobs/runners/pulsar.py   | 51 +++++++++++++++++++++++++++++
 2 files changed, 71 insertions(+)

diff --git a/config/job_conf.xml.sample_advanced b/config/job_conf.xml.sample_advanced
index ea093ce8af5..c578a19cd1f 100644
--- a/config/job_conf.xml.sample_advanced
+++ b/config/job_conf.xml.sample_advanced
@@ -98,6 +98,26 @@
                deprecated and will disappear with a future release of Galaxy.
           -->
         </plugin>
+        <plugin id="pulsar_embedded" type="runner" load="galaxy.jobs.runners.pulsar:PulsarEmbeddedJobRunner">
+          <!-- The embedded Pulsar runner starts a Pulsar app
+               internal to Galaxy and communicates it directly.
+               This maybe be useful for instance when Pulsar
+               staging is important but a Pulsar server is
+               unneeded (most obviously for instance if compute
+               servers cannot mount Galaxy's files but Galaxy
+               can mount a scratch directory available on
+               compute). -->
+          <!-- Specify a complete description of the Pulsar app
+               to create. Currently this configuration (if set)
+               must create exactly on job manager. For more
+               information on configuring a Pulsar app see:
+
+               https://github.com/galaxyproject/pulsar/blob/master/app.yml.sample
+               http://pulsar.readthedocs.org/en/latest/configure.html
+          -->
+          <!-- <param id="pulsar_conf">path/to/pulsar/app.yml</param> -->
+        </plugin>
+
     </plugins>
     <handlers default="handlers">
         <!-- Additional job handlers - the id should match the name of a
diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py
index 27afad72303..66d28734c63 100644
--- a/lib/galaxy/jobs/runners/pulsar.py
+++ b/lib/galaxy/jobs/runners/pulsar.py
@@ -18,6 +18,10 @@ from pulsar.client import PulsarOutputs
 from pulsar.client import ClientOutputs
 from pulsar.client import PathMapper
 
+import pulsar.core
+
+import yaml
+
 from galaxy import model
 from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner
 from galaxy.jobs import ComputeEnvironment
@@ -35,6 +39,7 @@ __all__ = [
     'PulsarLegacyJobRunner',
     'PulsarRESTJobRunner',
     'PulsarMQJobRunner',
+    'PulsarEmbeddedJobRunner',
 ]
 
 NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE = "Pulsar misconfiguration - Pulsar client configured to set metadata remotely, but remote Pulsar isn't properly configured with a galaxy_home directory."
@@ -65,6 +70,10 @@ PULSAR_PARAM_SPECS = dict(
         map=specs.to_str_or_none,
         default=None,
     ),
+    pulsar_config=dict(
+        map=specs.to_str_or_none,
+        default=None,
+    ),
     manager=dict(
         map=specs.to_str_or_none,
         default=None,
@@ -144,6 +153,7 @@ class PulsarJobRunner( AsynchronousJobRunner ):
     """Base class for pulsar job runners."""
 
     runner_name = "PulsarJobRunner"
+    default_build_pulsar_app = False
 
     def __init__( self, app, nworkers, **kwds ):
         """Start the job runner."""
@@ -163,14 +173,40 @@ class PulsarJobRunner( AsynchronousJobRunner ):
         self._init_monitor_thread()
 
     def __init_client_manager( self ):
+        pulsar_conf = self.runner_params.get('pulsar_conf', None)
+        self.__init_pulsar_app(pulsar_conf)
+
         client_manager_kwargs = {}
         for kwd in 'manager', 'cache', 'transport', 'persistence_directory':
             client_manager_kwargs[ kwd ] = self.runner_params[ kwd ]
+        if self.pulsar_app is not None:
+            # TODO: Make this more generic and configurable - client_manager
+            # should define an app and client (destination) should reference
+            # a job manager.
+            job_manager = self.pulsar_app.only_manager
+            client_manager_kwargs[ "job_manager" ] = job_manager
+            # TODO: Hack remove this following line pulsar lib update
+            # that includes https://github.com/galaxyproject/pulsar/commit/ce0636a5b64fae52d165bcad77b2caa3f0e9c232
+            client_manager_kwargs[ "file_cache" ] = None
+
         for kwd in self.runner_params.keys():
             if kwd.startswith( 'amqp_' ):
                 client_manager_kwargs[ kwd ] = self.runner_params[ kwd ]
         self.client_manager = build_client_manager(**client_manager_kwargs)
 
+    def __init_pulsar_app( self, pulsar_conf_path ):
+        if pulsar_conf_path is None and not self.default_build_pulsar_app:
+            self.pulsar_app = None
+            return
+        conf = {}
+        if pulsar_conf_path is None:
+            log.info("Creating a Pulsar app with default configuration (no pulsar_conf specified).")
+        else:
+            log.info("Loading Pulsar app configuration from %s" % pulsar_conf_path)
+            with open(pulsar_conf_path, "r") as f:
+                conf.update(yaml.load(f) or {})
+        self.pulsar_app = pulsar.core.PulsarApp(**conf)
+
     def url_to_destination( self, url ):
         """Convert a legacy URL to a job destination."""
         return JobDestination( runner="pulsar", params=url_to_destination_params( url ) )
@@ -661,6 +697,21 @@ class PulsarRESTJobRunner( PulsarJobRunner ):
     )
 
 
+class PulsarEmbeddedJobRunner(PulsarJobRunner):
+    """Flavor of Puslar job runnner that runs Pulsar's server code directly within Galaxy.
+
+    This is an appropriate job runner for when the desire is to use Pulsar staging
+    but their is not need to run a remote service.
+    """
+
+    destination_defaults = dict(
+        default_file_action="copy",
+        rewrite_parameters="true",
+        dependency_resolution="remote",
+    )
+    default_build_pulsar_app = True
+
+
 class PulsarComputeEnvironment( ComputeEnvironment ):
 
     def __init__( self, pulsar_client, job_wrapper, remote_job_config ):