Stage in out

README.md

@@ -33,7 +33,7 @@ Example JSON file:
 {
   "inputs": [
     {
-      "url": "https://cdn.kernel.org/pub/linux/kernel/v6.x/linux-6.5.5.tar.xz",
+      "url": "http://cdn.kernel.org/pub/linux/kernel/v6.x/linux-6.5.5.tar.xz",
       "path": "/data/kernel.tar.gz",
       "type": "FILE"
     }
@@ -47,7 +47,7 @@ Example JSON file:
       ],
       "workdir": "/data/",
       "stdout": "/tmp/stdout.log",
-      "stderr": "/tmp/stderr.log",
+      "stderr": "/tmp/stderr.log"
     }
   ]
 }

docker-compose.yaml

@@ -7,15 +7,19 @@ services:
       dockerfile: docker/tesp_api/Dockerfile
       target: development
     image: tesp-api
+    environment:
+      - CONTAINER_TYPE=singularity  # Set to "docker", "singularity", or "both"
     container_name: tesp-api
+    privileged: true
     ports:
       - "8080:8080"
     depends_on:
       - tesp-db
     volumes:
       - ./:/app
-    extra_hosts:
-      - "host.docker.internal:host-gateway"
+      - /opt/pysetup/files/staging/:/opt/pysetup/files/staging/
+      #    extra_hosts:
+      #      - "host.docker.internal:host-gateway"
 
   tesp-db:
     image: mongo:latest
@@ -103,3 +107,10 @@ services:
     volumes:
       - ./docker/ftpserver/data/:/tmp
       - ./docker/ftpserver/ftpserver.json:/app/ftpserver.json
+
+# Some cloud providers may require a lower MTU!
+networks:
+  default:
+    driver: bridge
+    driver_opts:
+      com.docker.network.driver.mtu: 1442

docker/.uuid

@@ -0,0 +1 @@
+496e1025-de54-47bf-8939-75bc2ccdfd7b

docker/ftpserver/.uuid

@@ -0,0 +1 @@
+d98140f5-e5fe-4201-acb7-7cf7c04d8d64

docker/minio/.uuid

@@ -0,0 +1 @@
+c25dba2d-8185-4503-869e-ac8dbb44da33

docker/minio/initial_data/.uuid

@@ -0,0 +1 @@
+b1e73b29-c1a9-447c-83cf-1124c37a1192

docker/pulsar_amqp/.uuid

@@ -0,0 +1 @@
+322072f7-5bbc-4ee4-88b2-1b8bcdee66fd

docker/pulsar_rest/.uuid

@@ -0,0 +1 @@
+1de80fb0-0436-4a19-9371-2e6048985315

docker/pulsar_rest/Dockerfile

@@ -37,4 +37,4 @@ WORKDIR $PYSETUP_PATH
 COPY startup.sh startup.sh
 RUN pulsar-config --host 0.0.0.0
 EXPOSE 8913
-CMD ["/bin/bash", "./startup.sh"]
+CMD ["/bin/bash", "./startup.sh"]

docker/rabbitmq/.uuid

@@ -0,0 +1 @@
+896b638d-1803-4d64-b134-ed43a1c7a649

docker/tesp_api/.uuid

@@ -0,0 +1 @@
+203b5739-380c-47c1-b0d8-c08fea4d83af

tes-runner/.uuid

@@ -0,0 +1 @@
+4a2ac601-8171-4f03-9a2a-0c3021d890ec

tes-runner/README.md

@@ -1,3 +1,10 @@
 # TES Runner prototype
 
-**This directory contains a prototype of a TES runner for the Galaxy project. The code is taken from the [Pull Request (https://github.com/galaxyproject/galaxy/pull/14462)](https://github.com/galaxyproject/galaxy/pull/14462) and modified.**
+**This directory contains a prototype of a TES runner for the Galaxy project. The code is taken from the [Pull Request (https://github.com/galaxyproject/galaxy/pull/14462)](https://github.com/galaxyproject/galaxy/pull/14462) and modified.**
+
+This directory contains two files:
+- `tes.py` - TES Runner for Galaxy
+- `job_config.yaml` - Sample config for the TES Runner
+
+Runners are located in `<galaxy>/lib/galaxy/jobs/runners/`.  
+Job config is located in `<galaxy>/config/`.

tes-runner/job_config.yaml

@@ -0,0 +1,16 @@
+runners:
+  local_runner:
+    load: galaxy.jobs.runners.local:LocalJobRunner
+    workers: 1
+  tes_runner:
+    load: galaxy.jobs.runners.tes:TESJobRunner
+
+execution:
+  default: tesp_env
+  environments:
+    local_env:
+      runner: local_runner
+    tesp_env:
+      runner: tes_runner
+      tes_master_addr: http://<tes-server-address>:8080
+      default_docker_image: hub.cerit.io/josef_handl/tesp-test-image # prototype - for testing purpose only

tes-runner/tes.py

@@ -143,14 +143,18 @@ def file_creation_executor(self, docker_image: str, work_dir: str):
         }
         return file_executor
 
-    def job_executor(self, remote_image: str, command_line: str, env: dict):
+    def job_executor(self, remote_image: str, command_line: str, env: dict, work_dir: str):
         """
         Returns the executor for executing jobs
         """
+        command_list = shlex.split(command_line)
+        replacements = {'../outputs/tool_stdout': 'tool_stdout',
+                        '../outputs/tool_stderr;': 'tool_stderr;'}
+        command_list = [replacements.get(item, item) for item in command_list]
         job_executor = {
-            "workdir": "/",
+            "workdir": work_dir,
             "image": remote_image,
-            "command": shlex.split(command_line),
+            "command": command_list,
             "env": env
         }
         return job_executor
@@ -166,18 +170,15 @@ def file_staging_out_executor(self, docker_image: str, command: list):
         }
         return staging_out_executor
 
-    def base_job_script(self, mounted_dir: list, work_dir: str, output_files: list, description: str):
+    def base_job_script(self, mounted_dir: list, work_dir: str, description: str):
         """
         Retruns the basic structure for job-script
         """
         execution_script = {
             "name": "Galaxy Job Execution",
             "description": description,
-            "inputs": [
-                {
-                    "path": os.path.join(work_dir, 'createfiles.sh'),
-                    "content": self.output_file_gen_script(output_files),
-                }],
+            "inputs": [],
+            "outputs": [],
             "executors": [],
             "volumes": mounted_dir
         }
@@ -197,17 +198,6 @@ def output_file_gen_script(self, output_files: list):
         script[-1] = '\n'
         return ' '.join(script)
 
-    def staging_out_command(self, script_path: str, output_files: list, api_url: str, work_dir: str):
-        """
-            Generates command for staging out of files
-        """
-        command = ["python", script_path, api_url, work_dir]
-
-        for file in output_files:
-            command.append(file)
-
-        return command
-
     def env_variables(self, job_wrapper: JobWrapper):
         """
             Get environment variables from job_wrapper
@@ -223,26 +213,41 @@ def env_variables(self, job_wrapper: JobWrapper):
         env_vars["_GALAXY_JOB_HOME_DIR"] = self.container_workdir
         return env_vars
 
-    def input_url(self, api_url: str, path: str):
+    def inout_url(self, api_url: str, path: str):
         """
             Get URL for path
         """
         file_link = f"{api_url}&path={path}"
         return file_link
 
-    def input_descriptors(self, api_url: str, input_paths: list, type: str = "FILE"):
+    def in_descriptors(self, api_url: str, in_paths: list, type: str = "FILE"):
         """
-            Get Input Descriptor for Jobfile
+            Get Input / Output Descriptor for Jobfile
         """
-        input_description = []
+        in_description = []
 
-        for path in input_paths:
-            input_description.append({
-                "url": self.input_url(api_url, path),
+        for path in in_paths:
+            in_description.append({
+                "url": self.inout_url(api_url, path),
                 "path": path,
                 "type": type
             })
-        return input_description
+        return in_description
+
+    def out_descriptors(self, api_url: str, out_paths: list[dict], type: str = "FILE"):
+        """
+            Get Input / Output Descriptor for Jobfile
+        """
+        out_description = []
+
+        for path in out_paths:
+            out_description.append({
+                "url": self.inout_url(api_url, path['return_path']),
+                "path": path['tes_path'],
+                "type": type
+            })
+        return out_description
+
 
     def get_job_directory_files(self, work_dir: str):
         """
@@ -292,10 +297,9 @@ def build_script(self, job_wrapper: JobWrapper, client_args: dict):
         tool_dir = job_wrapper.tool.tool_dir
         work_dir = job_wrapper.working_directory
         object_store_path = job_wrapper.object_store.file_path
-        script_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "util/file-staging.py")
 
         input_files = self.__get_inputs(job_wrapper)
-        output_files = self.get_output_files(job_wrapper)
+        output_files_dict = self.get_output_files(job_wrapper)
         extra_files = job_wrapper.extra_filenames
         tool_files = pulsar.client.staging.up.JobInputs(job_wrapper.command_line, extra_files).find_referenced_subfiles(tool_dir)
 
@@ -307,7 +311,7 @@ def build_script(self, job_wrapper: JobWrapper, client_args: dict):
             include_metadata=False,
             create_tool_working_directory=False,
             include_work_dir_outputs=False,
-            remote_job_directory=job_wrapper.working_directory
+            remote_job_directory=work_dir
         )
 
         env_var = self.env_variables(job_wrapper)
@@ -322,18 +326,16 @@ def build_script(self, job_wrapper: JobWrapper, client_args: dict):
         if (staging_out_url is None):
             staging_out_url = client_args['files_endpoint']
 
-        staging_out_command = self.staging_out_command(script_path, output_files, staging_out_url, job_wrapper.working_directory)
 
-        job_script = self.base_job_script([work_dir, object_store_path], work_dir, output_files, job_wrapper.tool.description)
+        job_script = self.base_job_script([work_dir, object_store_path], work_dir, job_wrapper.tool.description)
 
-        job_script["inputs"].extend(self.input_descriptors(client_args['files_endpoint'], [script_path]))
-        job_script["inputs"].extend(self.input_descriptors(client_args['files_endpoint'], tool_files))
-        job_script["inputs"].extend(self.input_descriptors(client_args['files_endpoint'], self.get_job_directory_files(work_dir)))
-        job_script["inputs"].extend(self.input_descriptors(client_args['files_endpoint'], input_files))
+        job_script["inputs"].extend(self.in_descriptors(client_args['files_endpoint'], tool_files))
+        job_script["inputs"].extend(self.in_descriptors(client_args['files_endpoint'], self.get_job_directory_files(work_dir)))
+        job_script["inputs"].extend(self.in_descriptors(client_args['files_endpoint'], input_files))
 
-        job_script["executors"].append(self.file_creation_executor(staging_out_image, work_dir))
-        job_script["executors"].append(self.job_executor(remote_image, command_line, env_var))
-        job_script["executors"].append(self.file_staging_out_executor(staging_out_image, staging_out_command))
+        job_script["outputs"].extend(self.out_descriptors(client_args['files_endpoint'], output_files_dict))
+
+        job_script["executors"].append(self.job_executor(remote_image, command_line, env_var, work_dir))
 
         return job_script
 
@@ -384,8 +386,23 @@ def get_output_files(self, job_wrapper: JobWrapper):
         """
         Utility for getting list of Output Files
         """
-        output_paths = job_wrapper.job_io.get_output_fnames()
-        return [str(o) for o in output_paths]
+        out_dicts = []
+        work_dir = job_wrapper.working_directory
+        existing_fpaths = []
+        for pair in self.get_work_dir_outputs(job_wrapper):
+            out_dicts.append({
+                'tes_path': work_dir + "/" + os.path.basename(pair[0]),
+                'return_path': pair[1]
+            })
+            existing_fpaths.append(pair[1])
+        for output in job_wrapper.job_io.get_output_fnames():
+            if str(output) not in existing_fpaths:
+                out_dicts.append({
+                    'tes_path': str(output),
+                    'return_path': str(output)
+                })
+
+        return out_dicts
 
     def __finish_job(self, data: dict, job_wrapper: JobWrapper):
         """
@@ -499,11 +516,11 @@ def recover(self, job: model.Job, job_wrapper: JobWrapper):
         if job_id is None:
             self.put(job_wrapper)
             return
-        job_state = TESJobState(job_wrapper=job_wrapper, files_dir=self.app.config.cluster_files_directory)
+        job_state = TESJobState(job_wrapper=job_wrapper, files_dir=self.app.config.job_working_directory)
         job_state.job_id = str(job_id)
         job_state.job_wrapper = job_wrapper
         job_state.job_destination = job_wrapper.job_destination
-        job_state.user_log = os.path.join(self.app.config.cluster_files_directory, 'galaxy_%s.tes.log' % galaxy_id_tag)
+        job_state.user_log = os.path.join(self.app.config.job_working_directory, 'galaxy_%s.tes.log' % galaxy_id_tag)
         job_state.register_cleanup_file_attribute('user_log')
         if job.state == model.Job.states.RUNNING:
             log.debug("(%s/%s) is still in running state, adding to the DRM queue" % (job.id, job.job_runner_external_id))

tesp_api/.uuid

@@ -0,0 +1 @@
+f3737a9d-3a5b-4a48-a6dd-3a946ba323e0

tesp_api/api/.uuid

@@ -0,0 +1 @@
+d7ddef8d-e6ce-4527-87c3-5c9184d2a581

tesp_api/api/endpoints/.uuid

@@ -0,0 +1 @@
+4af13e94-e4f1-4a64-bf8e-9ba24f2a1aa8

tesp_api/api/endpoints/task_endpoints.py

@@ -33,6 +33,9 @@
              response_model=TesCreateTaskResponseModel,
              description=descriptions["tasks-create"])
 async def create_task(tes_task: TesTask = Body(...)) -> Response:
+    print("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
+    print("TES TASK JSON")
+    print(tes_task.json())
     task_to_create = RegisteredTesTask(
         **tes_task.dict(),
         state=TesTaskState.QUEUED,

tesp_api/api/model/.uuid

@@ -0,0 +1 @@
+277ef504-a166-41c4-965d-ef782b11a716

tesp_api/config/.uuid

@@ -0,0 +1 @@
+fa24e399-e81b-4782-9431-6f6dce6b8eec

tesp_api/repository/.uuid

@@ -0,0 +1 @@
+2bfe5a17-3edb-48ac-8822-41567a217155

tesp_api/repository/model/.uuid

@@ -0,0 +1 @@
+a6006489-8472-4031-a60e-dd4a74cb4768