Either one is totally possible. @daniel just described how to do the former

We'll probably ship a solid for this in one of our libraries in the near future, but for now it's not hard to write yourself

Hey @sashank! I wanted to follow up on this question. I developed a solid similar to what you described (it was surprisingly easy btw), but i have a situation: When the external container is called as a solid, I use the

docker.containers.run()

command to sync up the execution of the solid displayed in dagit with the execution of the container. If I enable the

detach

option in that function, Dagit will only register the startup period as the solid execution time. If the pipeline is terminated manually in Dagit, obviously the pipeline would appear halted, but the container would still be running. Is there a way to sync up both the container and the solid that spins it up? Is a new RunLauncher necessary for this?

Thank you @daniel, this solution worked perfectly! Here is my final solid for anyone interested:

@solid(
    description="Launch an existing docker container from Dagster",
    input_defs=[
        InputDefinition(name="image_name", dagster_type=String),
        InputDefinition(name="image_tag", dagster_type=Optional[String]),
    ],
    required_resource_keys={'docker'},
)
def launch_container(context, image_name, image_tag):

    # Get the container name from the image name (all characters after last "/")
    container_name = image_name[image_name.rfind('/') + 1:] + "_" + image_tag

    # Build the full image name (include the tag)
    full_image_name = image_name + ":" + image_tag

    # Get list of the names of all the available images
    all_images = context.resources.docker.images.list()
    available_image_names = [image.tags for image in all_images]

    # The list of sub-lists is flattened to create a single flat list
    available_image_names = [item for sublist in available_image_names for item in sublist]

    # Pull image if its not available locally
    if full_image_name not in available_image_names:

        <http://context.log.info|context.log.info>("Info: Pulling container from repo.")

        # Pull the image from the specified dockerhub repo
        context.resources.docker.images.pull(repository=image_name, tag=image_tag)

        <http://context.log.info|context.log.info>("Info: Finished container pull.")

    # Get list of the names of all the available containers
    all_containers = context.resources.docker.containers.list(all=True)
    available_container_names = [container.name for container in all_containers]

    # Try to run the container.
    # If an error or manual termination occurs, stop the container
    try:

        # If container already exists
        if container_name in available_container_names:

            # Restart the container
            container = context.resources.docker.containers.get(container_name)
            container.restart()
            <http://context.log.info|context.log.info>("Info: Restarting existing container.")

            # Block the program here until the container finishes
            container.wait()

        # Build and run container if it does not exist
        else:
            context.resources.docker.containers.run(image=full_image_name, name=container_name)
            <http://context.log.info|context.log.info>("Info: Beginning new container execution.")

        # Signal that the container has finished
        <http://context.log.info|context.log.info>("Info: The container has finished its execution.")

        # Get the previously created container
        container = context.resources.docker.containers.get(container_name)

        # Reload the container's attributes
        container.reload()

        # Remove the container and image once it exits
        if container.status == "exited":
            container.remove()
            context.resources.docker.images.remove(image=full_image_name)

    except:
        container = context.resources.docker.containers.get(container_name)
        container.stop()
        <http://context.log.info|context.log.info>("Info: Stopped container due to an exception")

Also a note: Apparently one can call the detach mode, do something and then "re-block" the program until the container finishes by using the

container.wait()

command.