Merge branch 'master' into YAN-853

.travis.yml

@@ -48,11 +48,13 @@ matrix:
         - git config --global user.name $GITHUB_USERNAME
         - git config --global user.email "$GITHUB_USERNAME@gmail.com"
       script:
-        - python3 tools/xml2palette/xml2palette.py -i ./ -o $PROJECT_NAME-$TRAVIS_BRANCH.palette
+        - python3 tools/xml2palette/xml2palette.py -i ./ -t daliuge -o $PROJECT_NAME-$TRAVIS_BRANCH.palette
+        - python3 tools/xml2palette/xml2palette.py -i ./ -t template -o $PROJECT_NAME-$TRAVIS_BRANCH-template.palette
       after_success:
         - git clone https://$GITHUB_TOKEN@github.com/ICRAR/EAGLE_test_repo
         - mkdir -p EAGLE_test_repo/$PROJECT_NAME
         - mv $PROJECT_NAME-$TRAVIS_BRANCH.palette EAGLE_test_repo/$PROJECT_NAME/
+        - mv $PROJECT_NAME-$TRAVIS_BRANCH-template.palette EAGLE_test_repo/$PROJECT_NAME/
         - cd EAGLE_test_repo
         - git add *
         - git diff-index --quiet HEAD || git commit -m "Automatically generated DALiuGE palette (branch $TRAVIS_BRANCH, commit $PROJECT_VERSION)"

OpenAPI/tests/test.graph

@@ -204,7 +204,7 @@
                     "precious": false
                 }
             ],
-            "applicationParams": [],
+            "applicationArgs": [],
             "inputAppFields": [],
             "outputAppFields": [],
             "exitAppFields": [],
@@ -306,7 +306,7 @@
                     "precious": false
                 }
             ],
-            "applicationParams": [],
+            "applicationArgs": [],
             "inputAppFields": [],
             "outputAppFields": [],
             "exitAppFields": [],
@@ -388,7 +388,7 @@
                     "precious": false
                 }
             ],
-            "applicationParams": [],
+            "applicationArgs": [],
             "inputAppFields": [],
             "outputAppFields": [],
             "exitAppFields": [],

daliuge-common/dlg/common/__init__.py

@@ -39,6 +39,7 @@ class Categories:
     PLASMA = "Plasma"
     PLASMAFLIGHT = "PlasmaFlight"
     PARSET = "ParameterSet"
+    ENVIRONMENTVARS = "EnvironmentVars"
 
     MKN = "MKN"
     SCATTER = "Scatter"
@@ -72,7 +73,8 @@ class Categories:
     Categories.JSON,
     Categories.PLASMA,
     Categories.PLASMAFLIGHT,
-    Categories.PARSET
+    Categories.PARSET,
+    Categories.ENVIRONMENTVARS
 }
 APP_DROP_TYPES = [
     Categories.COMPONENT,

daliuge-common/docker/Dockerfile.dev

@@ -7,18 +7,13 @@ FROM ubuntu:20.04
 ARG BUILD_ID
 LABEL stage=builder
 LABEL build=$BUILD_ID
-RUN apt-get update && apt-get install -y gcc python3 python3.8-venv && apt-get clean
+RUN apt-get update && apt-get install -y gcc python3 python3.8-venv python3-pip python3-distutils libmetis-dev curl && apt-get clean
 
 COPY / /daliuge
 
-RUN cd && python3 -m venv dlg && cd /daliuge && \
-    . ${HOME}/dlg/bin/activate && \
-    pip install numpy && \
-    pip install . && \
-    apt-get remove -y gcc && \
-    apt-get autoremove -y
+RUN cd / && python3 -m venv dlg && cd /daliuge && \
+    . /dlg/bin/activate && \
+    pip install wheel numpy && \
+    pip install .
 
-
-FROM ubuntu:20.04
-RUN apt-get update && apt-get install -y bash
-COPY --from=0 /root/dlg /root/dlg
+# we don't clean this up, will be done in the derived images

daliuge-engine/dlg/apps/archiving.py

@@ -81,6 +81,7 @@ def store(self, inputDrop):
 # @details Takes an input and archives it in an NGAS server.
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/appclass Application class/dlg.apps.archiving.NgasArchivingApp/String/readonly/
 #     \~English Application class
 # @param[in] param/ngasSrv NGAS Server URL/localhost/String/readwrite/

daliuge-engine/dlg/apps/bash_shell_app.py

@@ -157,17 +157,28 @@ class BashShellBase(object):
     Common class for BashShell apps. It simply requires a command to be
     specified.
     """
-
+    #TODO: use the shlex module for most of the construction of the
+    # command line to get a proper and safe shell syntax
     command = dlg_string_param("Bash command", None)
 
     def initialize(self, **kwargs):
         super(BashShellBase, self).initialize(**kwargs)
 
         self.proc = None
+        self._inputRedirect = self._getArg(kwargs, "input_redirection", "")
+        self._outputRedirect = self._getArg(kwargs, "output_redirection", "")
+        self._cmdLineArgs = self._getArg(kwargs, "command_line_arguments", "")
+        self._applicationArgs = self._getArg(kwargs, "applicationArgs", {})
+        self._argumentPrefix = self._getArg(kwargs, "argumentPrefix", "--")
+        self._paramValueSeparator = self._getArg(kwargs, \
+            "paramValueSeparator", " ")
+
         if not self.command:
-            raise InvalidDropException(
-                self, "No command specified, cannot create BashShellApp"
-            )
+            self.command = self._getArg(kwargs, "command", None)
+            if not self.command:
+                raise InvalidDropException(
+                    self, "No command specified, cannot create BashShellApp"
+                )
 
     def _run_bash(self, inputs, outputs, stdin=None, stdout=subprocess.PIPE):
         """
@@ -186,7 +197,16 @@ def _run_bash(self, inputs, outputs, stdin=None, stdout=subprocess.PIPE):
         session_id = (
             self._dlg_session.sessionId if self._dlg_session is not None else ""
         )
-        cmd = self.command
+        argumentString = droputils.serialize_applicationArgs(self._applicationArgs, \
+            self._argumentPrefix, self._paramValueSeparator)
+        # complete command including all additional parameters and optional redirects
+        cmd = f"{self.command} {argumentString} {self._cmdLineArgs} "
+        if self._outputRedirect:
+            cmd = f"{cmd} > {self._outputRedirect}"
+        if self._inputRedirect:
+            cmd = f"cat {self._inputRedirect} > {cmd}"
+        cmd = cmd.strip()
+
         app_uid = self.uid
         # self.run_bash(self._command, self.uid, session_id, *args, **kwargs)
 
@@ -210,7 +230,7 @@ def _run_bash(self, inputs, outputs, stdin=None, stdout=subprocess.PIPE):
 
         # Wrap everything inside bash
         cmd = ("/bin/bash", "-c", cmd)
-        logger.debug("Command after user creation and wrapping is: %s", cmd)
+        logger.debug("Command after wrapping is: %s", cmd)
 
         start = time.time()
 
@@ -305,6 +325,21 @@ def execute(self, data):
 # * input-only stream
 # * full-stream
 #
+##
+# @brief BashShellApp
+# @details An application component able to run an arbitrary command within the Bash Shell
+# @par EAGLE_START
+# @param category BashShellApp
+# @param tag template
+# @param[in] param/command Command//String/readwrite/
+#     \~English The command to be executed
+# @param[in] param/input_redirection Input Redirection//String/readwrite/
+#     \~English The command line argument that specifies the input into this application
+# @param[in] param/output_redirection Output Redirection//String/readwrite/
+#     \~English The command line argument that specifies the output from this application
+# @param[in] param/command_line_arguments Command Line Arguments//String/readwrite/
+#     \~English Additional command line arguments to be added to the command line to be executed
+# @par EAGLE_END
 class BashShellApp(BashShellBase, BarrierAppDROP):
     """
     An app that runs a bash command in batch mode; that is, it waits until all

daliuge-engine/dlg/apps/crc.py

@@ -78,6 +78,7 @@ def run(self):
 # i.e. A "streamingConsumer" of its predecessor in the graph
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/appclass Application Class/dlg.apps.crc.CRCStreamApp/String/readonly/
 #     \~English Application class
 # @param[out] port/data Data/String/

daliuge-engine/dlg/apps/dockerapp.py

@@ -67,6 +67,31 @@ def waitForIp(self, timeout=None):
         return self._uid, self._containerIp
 
 
+##
+# @brief Docker
+# @details
+# @par EAGLE_START
+# @param category Docker
+# @param tag template
+# @param[in] param/image Image//String/readwrite/
+#     \~English The name of the docker image to be used for this application
+# @param[in] param/tag Tag/1.0/String/readwrite/
+#     \~English The tag of the docker image to be used for this application
+# @param[in] param/digest Digest//String/readwrite/
+#     \~English The hexadecimal hash (long version) of the docker image to be used for this application
+# @param[in] param/command Command//String/readwrite/
+#     \~English The command line to run within the docker instance. The specified command will be executed in a bash shell. That means that images will need a bash shell.
+# @param[in] param/user User//String/readwrite/
+#     \~English Username of the user who will run the application within the docker image
+# @param[in] param/ensureUserAndSwitch Ensure User And Switch/False/Boolean/readwrite/
+#     \~English Make sure the user specified in the User parameter exists and then run the docker container as that user
+# @param[in] param/removeContainer Remove Container/True/Boolean/readwrite/
+#     \~English Instruct Docker engine to delete the container after execution is complete
+# @param[in] param/additionalBindings Additional Bindings//String/readwrite/
+#     \~English Directories which will be visible inside the container during run-time. Format is srcdir_on_host:trgtdir_on_container. Multiple entries can be separated by commas.
+# @param[in] param/portMappings Port Mappings//String/readwrite/
+#     \~English Port mappings on the host machine
+# @par EAGLE_END
 class DockerApp(BarrierAppDROP):
     """
     A BarrierAppDROP that represents a process running in a container
@@ -204,17 +229,28 @@ def initialize(self, **kwargs):
             )
 
         self._command = self._getArg(kwargs, "command", None)
+
         if not self._command:
             logger.warning(
                 "No command specified. Assume that a default command is executed in the container"
             )
+            # The above also means that we can't pass applicationArgs
             # raise InvalidDropException(
             #     self, "No command specified, cannot create DockerApp")
+        else:
+            self._applicationArgs = self._getArg(kwargs, "applicationArgs", {})
+
+            # construct the actual command line from all application parameters
+            argumentPrefix = self._getArg(kwargs, "argumentPrefix", "--")
+            argumentString = droputils.serialize_applicationArgs(self._applicationArgs, \
+                argumentPrefix)
+            self._command = f"{self._command} {argumentString}"
 
         # The user used to run the process in the docker container
         # By default docker containers run as root, but we don't want to run
         # a process using a different user because otherwise anything that that
         # process writes to the filesystem
+        # TODO: User switching should be changed to be transparent
         self._user = self._getArg(kwargs, "user", None)
 
         # In some cases we want to make sure the command in the container runs

daliuge-engine/dlg/apps/dynlib.py

@@ -348,6 +348,15 @@ def addStreamingInput(self, streamingInputDrop, back=True):
         self._c_app.n_streaming_inputs += 1
 
 
+##
+# @brief DynlibApp
+# @details An application component run from a dynamic library
+# @par EAGLE_START
+# @param category DynlibApp
+# @param tag template
+# @param[in] param/libpath Library Path//String/readwrite/
+#     \~English The location of the shared object/DLL that implements this application
+# @par EAGLE_END
 class DynlibApp(DynlibAppBase, BarrierAppDROP):
     """Loads a dynamic library into the current process and runs it"""
 

daliuge-engine/dlg/apps/mpi.py

@@ -32,7 +32,15 @@
 
 logger = logging.getLogger(__name__)
 
-
+##
+# @brief MPI
+# @details An application component using the Message Passing Interface (MPI)
+# @par EAGLE_START
+# @param category Mpi
+# @param tag template
+# @param[in] param/num_of_procs Num procs//Integer/readwrite/
+#     \~English Number of processes used for this application
+# @par EAGLE_END
 class MPIApp(BarrierAppDROP):
     """
     An application drop representing an MPI job.

daliuge-engine/dlg/apps/plasma.py

@@ -48,6 +48,7 @@
 # via Plasma.
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/plasma_path Plasma Path//String/readwrite/
 #     \~English Path to plasma store.
 # @param[in] param/appclass Application class/dlg.apps.plasma.MSStreamingPlasmaConsumer/String/readonly/
@@ -135,6 +136,7 @@ def dropCompleted(self, uid, drop_state):
 # via Plasma.
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/plasma_path Plasma Path//String/readwrite/
 #     \~English Path to plasma store
 # @param[in] param/appclass Application class/dlg.apps.plasma.MSStreamingPlasmaProducer/String/readonly/
@@ -203,6 +205,7 @@ def run(self):
 # @details Batch read entire Measurement Set from Plasma.
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/appclass Application class/dlg.apps.plasma.MSPlasmaReader/String/readonly/
 #     \~English Application class
 # @param[in] port/plasma_ms_input Plasma MS Input/Measurement Set/
@@ -265,6 +268,7 @@ def run(self, **kwargs):
 # @details Batch write entire Measurement Set to Plasma.
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/appclass Application class/dlg.apps.plasma.MSPlasmaWriter/String/readonly/
 #     \~English Application class
 # @param[in] port/input_ms Input MS/Measurement Set/

daliuge-engine/dlg/apps/pyfunc.py

@@ -107,6 +107,7 @@ def import_using_code(code):
 # being written to its corresponding output.
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/appclass Application Class/dlg.apps.pyfunc.PyFuncApp/String/readonly/
 #     \~English Application class
 # @param[in] param/func_name Function Name//String/readwrite/

daliuge-engine/dlg/apps/scp.py

@@ -45,6 +45,7 @@
 # single output via SSH's scp protocol.
 # @par EAGLE_START
 # @param category PythonApp
+# @param tag daliuge
 # @param[in] param/appclass Application Class/dlg.apps.scp.ScpApp/String/readonly/
 #     \~English Application class
 # @param[in] param/remoteUser Remote User//String/readwrite/