Take caution though- this is an experimental feature and it’s easy to reach confusing errors. All the images need to have identical versions of your job defined in them, and in the same location (could be a module, file, etc. depending on your workspace). We’ll also try to use the same python executable path in all images.

@Dagster Bot docs document how to override the image for an op (along with the caveats)

Ah here it is just a couple of posts up actually, there's some example code here: https://dagster.slack.com/archives/C014N0PK37E/p1639479433039200

Maybe we can use this as the basis for a more generic op for running k8s jobs in the dagster-k8s library

Andrea, that log forwarding is awesome!

I reworked @Andrea Giardini’s example into a more generic one. But I wonder if this is the way to go for these things, this example has • a jobname that also becomes the name of the container (and enables log copying when you have more than 1 container in a pod • In my kubernetes setup I cannot go outside this namespace but I have a env var that tells me what namespace I'm in • I need to supply resources (limits, requests) to my containers otherwise they will not run The cool thing is that I can use this operator to start different jobs, but you need to give it a lot of config, which feels, dirty?

import os
import time
from typing import List

from dagster import In, op, DagsterError
from kubernetes import client, config


@op(
    ins = {
        'jobname':In(str,description="becomes names of container and first part of jobname"),
        "image_name": In(str, description="For instance …"),
        "cmdlineargs": In(List[str], description="The commands that need to run f.i. ['python runthis.py'] "),
        "max_cpu": In(str,default_value="750m"),
        "max_mem": In(str,default_value="256Mi"),
        "env": In(default_value=None, description="use client.V1EnvVar(name='',value=''"),
        "env_from": In(default_value=None, description= "use client.V1envFromSource(config_map_ref, secret_ref)")
        }
)
def create_k8s_job(context, jobname,image_name,cmdlineargs, max_cpu,max_mem, env, env_from):
    """Create and run a kubernetes job.

    For this operator to work we need: A name, a link to the image location and commands that 
    run the container. This operator will create the job on the cluster and
    stream the logs to dagster logging.
    You are not able to add volume mounts to this job.

    For more info about jobs see [kubernetes job docs](<https://kubernetes.io/docs/concepts/workloads/controllers/job/>)
    For more (sparse) info about the python library see [the library docs](<https://github.com/kubernetes-client/python/blob/master/kubernetes/README.md>)
    """
    if os.getenv('KUBERNETES_SERVICE_HOST'):
        context.log.debug("execution happening in kubernetes, so using `load_incluster_config`")
        config.load_incluster_config()
    else:
        context.log.debug("execution happening outside cluster, using `load_kube_config()`")
        config.load_kube_config()

    # Since the bucket and cluster namespace are the same in our case
    # we can use the bucketname as namespace variable too
    namespace = os.getenv('AWS_BUCKET','no-bucket')
    v1_batch = client.BatchV1Api()
    v1_core = client.CoreV1Api()
    # create jobdefinition
    job = client.V1Job(
        api_version="batch/v1",
        kind="Job",
        metadata=client.V1ObjectMeta(generate_name=f"{jobname}-", namespace=namespace),
        spec=client.V1JobSpec(
            # time to live (ttl) is 24 hours, but we delete it earlier when everything is alright
            ttl_seconds_after_finished=86400, 
            template=client.V1PodTemplateSpec(
                metadata=client.V1ObjectMeta(
                    annotations={"job.generated-by":"create_k8s_job"},
                    labels={
                    "app": jobname,
                    "<http://app.kubernetes.io/managed-by|app.kubernetes.io/managed-by>":"dagster"
                    }),
                spec=client.V1PodSpec(
                    containers=[
                        client.V1Container(
                            name=jobname,
                            image=image_name,
                            command=["/bin/sh", "-c"],
                            args=cmdlineargs,
                            resources=client.V1ResourceRequirements(
                                limits={"memory":max_mem, "cpu": max_cpu}, 
                                requests={"memory": "256Mi","cpu": "50m"}
                            ),
                            env=env,
                            env_from=env_from
                        )
                    ],
                    
                    restart_policy="Never",
                    service_account="default-editor"
                ),
            )
        )
    )
    # submit job to cluster
    api_response = v1_batch.create_namespaced_job(namespace, job)
    <http://context.log.info|context.log.info>("'%s' - k8sJob created." % str(api_response.metadata.name))
    # TODO: kill the job when not ready
    # poll and stream logs
    while True:
        job_status = v1_batch.read_namespaced_job_status(name=api_response.metadata.name, namespace=namespace)
        pod_list = v1_core.list_namespaced_pod(namespace=namespace, label_selector=f"app={jobname}").items
        pod_list = [pod for pod in pod_list if pod.metadata.name.startswith(api_response.metadata.name)]
        context.log.debug(f'There are {len(pod_list)} pods running of type "app={jobname}"')
        for pod in pod_list:
            context.log.debug(f'After filtering There are {len(pod_list)} pods running of type "app={jobname}"')
            # Forward available logs to dagster
            try:
                for line in v1_core.read_namespaced_pod_log(name=pod.metadata.name, namespace=namespace, since_seconds=5, container = jobname).splitlines():
                    <http://context.log.info|context.log.info>(pod.metadata.name + " - " + line)
                if api_response.status.conditions:
                    context.log.debug(pod.metadata.name + " - " + api_response.status.conditions)
            except:
                pass

        if job_status.status.succeeded:
            <http://context.log.info|context.log.info>("'%s' - k8sJob completed." % api_response.metadata.name)
            v1_batch.delete_namespaced_job(name=api_response.metadata.name, namespace=namespace)
            break
        elif job_status.status.failed:
            context.log.error("'%s' - k8sJob failed." % api_response.metadata.name)
            v1_batch.delete_namespaced_job(name=api_response.metadata.name, namespace=namespace)
            raise DagsterError("k8sJob failed")
        time.sleep(5)

If you want to run this op in a job I now need to do this:

from dagster import job
from ops.kubernetes import create_k8s_job

@job(config={'ops':{"create_k8s_job":{"inputs":{"jobname":aname, "image_name": server/group/images/image:version, "cmdlineargs": "[sleep 3600 & echo 'job done' ]"}}}})
def run_k8s_job():
    create_k8s_job()

I can't help but feel a bit clumsy with this.

I think I will change the INS to config settings that is cleaner

I guess, this partly answers my question. Will try to set it up!