Kubernetes (dagster-k8s)

See also the Kubernetes deployment guide.

This library contains utilities for running Dagster with Kubernetes. This includes a Python API allowing Dagit to launch runs as Kubernetes Jobs, as well as a Helm chart you can use as the basis for a Dagster deployment on a Kubernetes cluster.


class dagster_k8s.K8sRunLauncher(service_account_name, instance_config_map, postgres_password_secret=None, dagster_home=None, job_image=None, image_pull_policy=None, image_pull_secrets=None, load_incluster_config=True, kubeconfig_file=None, inst_data=None, job_namespace='default', env_config_maps=None, env_secrets=None, env_vars=None, k8s_client_batch_api=None, volume_mounts=None, volumes=None, labels=None)[source]

RunLauncher that starts a Kubernetes Job for each Dagster job run.

Encapsulates each run in a separate, isolated invocation of dagster-graphql.

You may configure a Dagster instance to use this RunLauncher by adding a section to your dagster.yaml like the following:

    module: dagster_k8s.launcher
    class: K8sRunLauncher
        service_account_name: your_service_account
        job_image: my_project/dagster_image:latest
        instance_config_map: dagster-instance
        postgres_password_secret: dagster-postgresql-secret

As always when using a ConfigurableClass, the values under the config key of this YAML block will be passed to the constructor. The full list of acceptable values is given below by the constructor args.

dagster_k8s.k8s_job_executor ExecutorDefinition[source]

Config Schema:
job_image (Union[dagster.StringSource, None], optional)

Docker image to use for launched Jobs. If this field is empty, the image that was used to originally load the Dagster repository will be used.(Ex: “mycompany.com/dagster-k8s-image:latest”).

image_pull_policy (Union[dagster.StringSource, None], optional)

Image pull policy to set on the launched task Job Pods. Defaults to “IfNotPresent”.

image_pull_secrets (Union[List[strict dict], None], optional)

(Advanced) Specifies that Kubernetes should get the credentials from the Secrets named in this list.

service_account_name (Union[dagster.StringSource, None], optional)

(Advanced) Override the name of the Kubernetes service account under which to run the Job.

env_config_maps (Union[List[dagster.StringSource], None], optional)

A list of custom ConfigMapEnvSource names from which to draw environment variables (using envFrom) for the Job. Default: []. See:https://kubernetes.io/docs/tasks/inject-data-application/define-environment-variable-container/#define-an-environment-variable-for-a-container

env_secrets (Union[List[dagster.StringSource], None], optional)

A list of custom Secret names from which to draw environment variables (using envFrom) for the Job. Default: []. See:https://kubernetes.io/docs/tasks/inject-data-application/distribute-credentials-secure/#configure-all-key-value-pairs-in-a-secret-as-container-environment-variables

env_vars (Union[List[String], None], optional)

A list of environment variables to inject into the Job. Default: []. See: https://kubernetes.io/docs/tasks/inject-data-application/distribute-credentials-secure/#configure-all-key-value-pairs-in-a-secret-as-container-environment-variables

volume_mounts (List[strict dict], optional)

A list of volume mounts to include in the job’s container. Default: []. See: https://v1-18.docs.kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#volumemount-v1-core

Default Value: []

volumes (List[permissive dict], optional)

A list of volumes to include in the Job’s Pod. Default: []. For the many possible volume source types that can be included, see: https://v1-18.docs.kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#volume-v1-core

Default Value: []

labels (permissive dict, optional)

Additional labels that should be included in the Job’s Pod. See: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels

job_namespace (dagster.StringSource, optional)

retries (selector, optional)
Default Value:
    "enabled": {}
Config Schema:
enabled (strict dict, optional)
Default Value:
disabled (strict dict, optional)
Default Value:

Executor which launches steps as Kubernetes Jobs.

To use the k8s_job_executor, set it as the executor_def when defining a job:

from dagster import job
from dagster_k8s import k8s_job_executor

def k8s_job():

Then you can configure the executor with run config as follows:

    job_namespace: 'some-namespace'
    image_pull_policy: ...
    image_pull_secrets: ...
    service_account_name: ...
    env_config_maps: ...
    env_secrets: ...
    job_image: ... # leave out if using userDeployments

Configuration set on the Kubernetes Jobs and Pods created by the K8sRunLauncher will also be set on Kubernetes Jobs and Pods created by the k8s_job_executor.

Python API

The K8sRunLauncher allows Dagit instances to be configured to launch new runs by starting per-run Kubernetes Jobs. To configure the K8sRunLauncher, your dagster.yaml should include a section like:

  module: dagster_k8s.launcher
  class: K8sRunLauncher
    service_account_name: dagster
    job_image: "my-company.com/image:latest"
    dagster_home: "/opt/dagster/dagster_home"
    postgres_password_secret: "dagster-postgresql-secret"
    image_pull_policy: "IfNotPresent"
    job_namespace: "dagster"
    instance_config_map: "dagster-instance"
      - "dagster-k8s-job-runner-env"
      - "dagster-k8s-some-secret"

Helm chart

For local dev (e.g., on kind or minikube):

helm install \
    --set dagit.image.repository="dagster.io/buildkite-test-image" \
    --set dagit.image.tag="py37-latest" \
    --set job_runner.image.repository="dagster.io/buildkite-test-image" \
    --set job_runner.image.tag="py37-latest" \
    --set imagePullPolicy="IfNotPresent" \
    dagster \

Upon installation, the Helm chart will provide instructions for port forwarding Dagit and Flower (if configured).

Running tests

To run the unit tests:

pytest -m "not integration"

To run the integration tests, you must have Docker, kind, and helm installed.

On macOS:

brew install kind
brew install helm

Docker must be running.

You may experience slow first test runs thanks to image pulls (run pytest -svv --fulltrace for visibility). Building images and loading them to the kind cluster is slow, and there is no visibility into the progress of the load.

NOTE: This process is quite slow, as it requires bootstrapping a local kind cluster with Docker images and the dagster-k8s Helm chart. For faster development, you can either:

  1. Keep a warm kind cluster

  2. Use a remote K8s cluster, e.g. via AWS EKS or GCP GKE

Instructions are below.

Faster local development (with kind)

You may find that the kind cluster creation, image loading, and kind cluster creation loop is too slow for effective local dev.

You may bypass cluster creation and image loading in the following way. First add the --no-cleanup flag to your pytest invocation:

pytest --no-cleanup -s -vvv -m "not integration"

The tests will run as before, but the kind cluster will be left running after the tests are completed.

For subsequent test runs, you can run:

pytest --kind-cluster="cluster-d9971c84d44d47f382a2928c8c161faa" --existing-helm-namespace="dagster-test-95590a" -s -vvv -m "not integration"

This will bypass cluster creation, image loading, and Helm chart installation, for much faster tests.

The kind cluster name and Helm namespace for this command can be found in the logs, or retrieved via the respective CLIs, using kind get clusters and kubectl get namespaces. Note that for kubectl and helm to work correctly with a kind cluster, you should override your kubeconfig file location with:

kind get kubeconfig --name kind-test > /tmp/kubeconfig
export KUBECONFIG=/tmp/kubeconfig

Manual kind cluster setup

The test fixtures provided by dagster-k8s automate the process described below, but sometimes it’s useful to manually configure a kind cluster and load images onto it.

First, ensure you have a Docker image appropriate for your Python version. Run, from the root of the repo:

./python_modules/dagster-test/dagster_test/test_project/build.sh 3.7.6 \

In the above invocation, the Python majmin version should be appropriate for your desired tests.

Then run the following commands to create the cluster and load the image. Note that there is no feedback from the loading process.

kind create cluster --name kind-test
kind load docker-image --name kind-test dagster.io/dagster-docker-buildkite:py37-latest

If you are deploying the Helm chart with an in-cluster Postgres (rather than an external database), and/or with dagster-celery workers (and a RabbitMQ), you’ll also want to have images present for rabbitmq and postgresql:

docker pull docker.io/bitnami/rabbitmq
docker pull docker.io/bitnami/postgresql

kind load docker-image --name kind-test docker.io/bitnami/rabbitmq:latest
kind load docker-image --name kind-test docker.io/bitnami/postgresql:latest

Then you can run pytest as follows:

pytest --kind-cluster=kind-test

Faster local development (with an existing K8s cluster)

If you already have a development K8s cluster available, you can run tests on that cluster vs. running locally in kind.

For this to work, first build and deploy the test image to a registry available to your cluster. For example, with a private ECR repository:

./python_modules/dagster-test/dagster_test/test_project/build.sh 3.7.6
docker tag dagster-docker-buildkite:latest $AWS_ACCOUNT_ID.dkr.ecr.us-west-2.amazonaws.com/dagster-k8s-tests:2020-04-21T21-04-06

aws ecr get-login --no-include-email --region us-west-1 | sh
docker push $AWS_ACCOUNT_ID.dkr.ecr.us-west-1.amazonaws.com/dagster-k8s-tests:2020-04-21T21-04-06

Then, you can run tests on EKS with:

export DAGSTER_DOCKER_IMAGE_TAG="2020-04-21T21-04-06"
export DAGSTER_DOCKER_REPOSITORY="$AWS_ACCOUNT_ID.dkr.ecr.us-west-2.amazonaws.com"
export DAGSTER_DOCKER_IMAGE="dagster-k8s-tests"

# First run with --no-cleanup to leave Helm chart in place
pytest --cluster-provider="kubeconfig" --no-cleanup -s -vvv

# Subsequent runs against existing Helm chart
pytest --cluster-provider="kubeconfig" --existing-helm-namespace="dagster-test-<some id>" -s -vvv

Validating Helm charts

To test / validate Helm charts, you can run:

helm install dagster --dry-run --debug helm/dagster
helm lint

Enabling GCR access from Minikube

To enable GCR access from Minikube:

kubectl create secret docker-registry element-dev-key \
    --docker-server=https://gcr.io \
    --docker-username=oauth2accesstoken \
    --docker-password="$(gcloud auth print-access-token)" \

A note about PVCs

Both the Postgres and the RabbitMQ Helm charts will store credentials using Persistent Volume Claims, which will outlive test invocations and calls to helm uninstall. These must be deleted if you want to change credentials. To view your pvcs, run:

kubectl get pvc

Testing Redis

The Redis Helm chart installs w/ a randomly-generated password by default; turn this off:

helm install dagredis stable/redis --set usePassword=false

Then, to connect to your database from outside the cluster execute the following commands:

kubectl port-forward --namespace default svc/dagredis-master 6379:6379
redis-cli -h -p 6379