[refactor] use singularity instances

Instances in singularity are basically the same as being able to create
an empty container and then run something in it, so we can have the same
thing as with nodeos.

One main issue with the code as is: we should probably refactor the
container_manager code to modify the manifest on the fly according to
the options listed in it. It would make it easy to pass along stuff like
PWD, necessary binds and so on.

nrm/containers.py

@@ -112,7 +112,7 @@ class ContainerManager(object):
                                                                manifest)
         if creation_needed:
             logger.info("Creating container %s", container_name)
-            self.runtime.create(container)
+            self.runtime.create(container, self.downstream_event_uri)
             self.containers[container_name] = container
 
         # build context to execute
@@ -217,7 +217,7 @@ class ContainerRuntime(object):
     def __init__(self):
         pass
 
-    def create(self, container):
+    def create(self, container, downstream_uri):
         """Create the container defined by the container namedtuple on the
         system."""
         raise NotImplementedError
@@ -244,7 +244,7 @@ class NodeOSRuntime(ContainerRuntime):
         path/command."""
         self.client = NodeOSClient(argo_nodeos_config=path)
 
-    def create(self, container):
+    def create(self, container, downstream_uri):
         """Uses the container resource allocation to create a container."""
         self.client.create(container.uuid, container.resources)
 
@@ -257,29 +257,6 @@ class NodeOSRuntime(ContainerRuntime):
         self.client.delete(container_uuid, kill)
 
 
-class SingularityRootRuntime(ContainerRuntime):
-
-    """Implements the container runtime interface using the singularity
-    subprogram."""
-
-    def __init__(self, path="argo_singularity_config"):
-        """Creates the client for singularity, with an optional custom
-        path/command."""
-        self.client = SingularityClient(argo_singularity_config=path)
-
-    def create(self, container):
-        """Uses the container resource allocation to create a container."""
-        self.client.oci_start(container.uuid, container.resources)
-
-    def execute(self, container_uuid, args, environ):
-        """Launches a command in the container."""
-        return self.client.oci_execute(container_uuid, args, environ)
-
-    def delete(self, container_uuid, kill=False):
-        """Delete the container."""
-        self.client.oci_delete(container_uuid, kill)
-
-
 class SingularityUserRuntime(ContainerRuntime):
 
     """Implements the container runtime interface using the singularity
@@ -289,21 +266,20 @@ class SingularityUserRuntime(ContainerRuntime):
         """Creates the client for singularity, with an optional custom
         path/command."""
         self.client = SingularityClient(singularity_path=path)
-        self.containers = dict()
 
-    def create(self, container):
+    def create(self, container, downstream_uri):
         """Uses the container resource allocation to create a container."""
-        self.containers[container.uuid] = container
+        imageinfo = container.manifest.image
+        self.client.instance_start(container.uuid, imageinfo.path,
+                                   [downstream_uri])
 
     def execute(self, container_uuid, args, environ):
         """Launches a command in the container."""
-        imageinfo = self.containers[container_uuid].manifest.image
-        # not checking image because singularity supports all types
-        return self.client.execute(imageinfo.path, args, environ)
+        return self.client.execute(container_uuid, args, environ)
 
     def delete(self, container_uuid, kill=False):
         """Delete the container."""
-        del self.containers[container_uuid]
+        self.client.instance_stop(container_uuid, kill)
 
 
 class DummyRuntime(ContainerRuntime):
@@ -314,7 +290,7 @@ class DummyRuntime(ContainerRuntime):
     def __init__(self):
         pass
 
-    def create(self, container):
+    def create(self, container, downstream_uri):
         pass
 
     def execute(self, container_uuid, args, environ):

nrm/subprograms.py

@@ -149,27 +149,47 @@ class SingularityClient(object):
         """Load client configuration."""
         self.prefix = singularity_path
 
-    def execute(self, container_image, argv, environ):
+    def instance_start(self, instance_name, container_image, bind_list=[]):
+        """Start a named instance of a container image.
+
+        Note that singularity will also start the startscript if
+        defined in the container image, which might be an issue."""
+        args = [self.prefix]
+        args.extend(['instance', 'start'])
+        if bind_list:
+            args.extend(['--bind', ','.join(bind_list)])
+        args.extend([container_image, instance_name])
+        p = subprocess.Popen(args, stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+        stdout, stderr = p.communicate()
+        logpopen(p, args, stdout, stderr)
+
+    def execute(self, instance_name, argv, environ):
         """Execute argv inside container.
 
-        singularity exec --bind ... container.sif <command>"""
+        singularity exec instance://instance_name <command>"""
         args = [self.prefix]  # singularity
-        args.append('exec')
-        # if the monitoring is active, and is a file, we need to bind it into
-        # the container
-        uri = environ.get('ARGO_NRM_DOWNSTREAM_EVENT_URI')
-        if uri:
-            is_file = uri.startswith('ipc://')
-            if is_file:
-                bind_arg = uri[6:]
-                args.extend(['--bind', bind_arg])
-
-        args.append(container_image)
+        container_name = "instance://" + instance_name
+        args.extend(['exec', container_name])
         args.extend(argv)
         return process.Subprocess(args, env=environ,
                                   stdout=process.Subprocess.STREAM,
                                   stderr=process.Subprocess.STREAM,
-                                  close_fds=True)
+                                  close_fds=True,
+                                  cwd=environ['PWD'])
+
+    def instance_stop(self, instance_name, kill=False):
+        """Stop an instance and kill everything in it."""
+
+        args = [self.prefix]
+        args.extend(['instance', 'stop'])
+        if kill:
+            args.append("--force")
+        args.append(instance_name)
+        p = subprocess.Popen(args, stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+        stdout, stderr = p.communicate()
+        logpopen(p, args, stdout, stderr)
 
 
 class ChrtClient(object):