Dagster k8s-celery executor

[hajiuduman-starcom]

Hi,

We are developing a data quality software, where we have multiple data processing stages. We are planning to execute each stage as separate celery-k8s executor. We are using helm chart with below values.

dagster-user-deployments:
  deployments:
    - name: "k8s-example-user-code-1"
           image:             
              repository: "cusom/repo.image"
runLauncher:
  type: 
     celeryK8sRunLauncher

Repositories are loaded from our custom user code deployment image.

Executor configuration in UI

execution:
    celery-k8s:
      config:
        job_image: 'my_repo.com/stage1:latest'

But while running the pipeline we are getting below error message from k8s (my_repo.com/stage1 pod) logs.

Error 1: Post processing at path root:instance_config_map of original value {'env': 'DAGSTER_K8S_INSTANCE_CONFIG_MAP'} failed:
dagster.config.errors.PostProcessingError: You have attempted to fetch the environment variable "DAGSTER_K8S_INSTANCE_CONFIG_MAP" which is not set. In order for this execution to succeed it must be set in this environment.

Stack Trace:
  File "/usr/local/lib/python3.9/site-packages/dagster/config/post_process.py", line 77, in _post_process
    new_value = context.config_type.post_process(config_value)
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 42, in post_process
    return str(_ensure_env_variable(cfg))
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 16, in _ensure_env_variable
    raise PostProcessingError(

Error 2: Post processing at path root:postgres_password_secret of original value {'env': 'DAGSTER_K8S_PG_PASSWORD_SECRET'} failed:
dagster.config.errors.PostProcessingError: You have attempted to fetch the environment variable "DAGSTER_K8S_PG_PASSWORD_SECRET" which is not set. In order for this execution to succeed it must be set in this environment.

Stack Trace:
  File "/usr/local/lib/python3.9/site-packages/dagster/config/post_process.py", line 77, in _post_process
    new_value = context.config_type.post_process(config_value)
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 42, in post_process
    return str(_ensure_env_variable(cfg))
  File "/usr/local/lib/python3.9/site-packages/dagster/config/source.py", line 16, in _ensure_env_variable

Can you please help us in solving this issue?

[rexledesma]

These environment variables should be passed in from the pipeline-env configmap that's generated for the chart.

What Helm chart version are you using, and could you give your exact values for dagster-user-deployments and runLauncher?
Installing your current values wouldn't work, since it does not pass validation, since CeleryK8sRunLauncher must be capitalized.

~/elementl/dagster master ⇣
internal-3.8.7 ❯ helm template dagster/dagster -g --version 0.11.13 --set "runLauncher.type=celeryK8sRunLauncher"
Error: values don't meet the specifications of the schema(s) in the following chart(s):
dagster:
- runLauncher.type: runLauncher.type must be one of the following: "CeleryK8sRunLauncher", "K8sRunLauncher", "CustomRunLauncher"

[hajiuduman]

Hi,

Sorry for the late reply.

Am using helm 3.6.0. Please find below config for dagster-user-deployments and runLauncher.

dagster-user-deployments:
  # Creates a workspace file with the gRPC servers hosting your user code.
  enabled: true

  # If you plan on deploying user code in a separate Helm release, set this to false.
  enableSubchart: true

  # List of unique deployments
  deployments:
    - name: "qualiz-job-mgmt"
      image:
        repository: "qualiz/user_code"
        tag: latest
        # Change with caution! If you're using a fixed tag for pipeline run images, changing the
        # image pull policy to anything other than "Always" will use a cached/stale image, which is
        # almost certainly not what you want.
        pullPolicy: Always
      # Arguments to `dagster api grpc`.
      # Ex: "dagster api grpc -m dagster_test.test_project.test_pipelines.repo -a define_demo_execution_repo"
      # would translate to:
      # dagsterApiGrpcArgs:
      #   - "-m"
      #   - "dagster_test.test_project.test_pipelines.repo"
      #   - "-a"
      #   - "define_demo_execution_repo"
      dagsterApiGrpcArgs:
        - "-f"
        - "/repos.py"
      port: 3030


runLauncher:
  # Type can be one of [K8sRunLauncher, CeleryK8sRunLauncher, CustomRunLauncher]
  type: CeleryK8sRunLauncher

  config:
    # This configuration will only be used if the K8sRunLauncher is selected
    k8sRunLauncher:
      ## The image to use for the launched Job's Dagster container.
      ## This image should only be specified when user code deployments are disabled. Otherwise,
      ## it will be ignored.
      # image:
      #   repository: ""
      #   tag: ""
      #   pullPolicy: Always

      # The K8s namespace where new jobs will be launched.
      # By default, the release namespace is used.
      jobNamespace: ~

      # Set to true to load kubeconfig from within cluster.
      loadInclusterConfig: true

      # File to load kubeconfig from. Only set this if loadInclusterConfig is false.
      kubeconfigFile: ~

      # Additional environment variables can be retrieved and set from ConfigMaps for the Job. See:
      # https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#configure-all-key-value-pairs-in-a-configmap-as-container-environment-variables
      #
      # Example:
      #
      # envConfigMaps:
      #   - name: config-map
      envConfigMaps: []

      # Additional environment variables can be retrieved and set from Secrets for the Job. See:
      # https://kubernetes.io/docs/concepts/configuration/secret/#use-case-as-container-environment-variables
      #
      # Example:
      #
      # envSecrets:
      #   - name: secret
      envSecrets: []

    # This configuration will only be used if the CeleryK8sRunLauncher is selected
    celeryK8sRunLauncher:
      # The Celery workers can be deployed with a fixed image (no user code included)
      image:
        repository: "docker.io/dagster/dagster-celery-k8s"
        tag: "latest"
        pullPolicy: Always

      # Support overriding the name prefix of Celery worker pods
      nameOverride: "celery-workers"

      # Additional config options for Celery, applied to all queues.
      # For available options, see:
      # https://docs.celeryproject.org/en/stable/userguide/configuration.html
      configSource: {}

      # Additional Celery worker queues can be configured here. When overriding, be sure to
      # provision a "dagster" worker queue, as this is the default queue used by Dagster.
      #
      # Optionally, labels and node selectors can be set on the Celery queue's workers.
      # Specifying a queue's node selector will override any existing node selector defaults.
      workerQueues:
        - name: "dagster"
          replicaCount: 1
          labels: {}
          nodeSelector: {}

      # Additional environment variables to set on the celery/job containers
      # A Kubernetes ConfigMap will be created with these environment variables. See:
      # https://kubernetes.io/docs/concepts/configuration/configmap/
      #
      # Example:
      #
      # env:
      #   ENV_ONE: one
      #   ENV_TWO: two
      env: 
        DAGSTER_K8S_PG_PASSWORD_SECRET: test

      # Additional environment variables can be retrieved and set from ConfigMaps. See:
      # https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#configure-all-key-value-pairs-in-a-configmap-as-container-environment-variables
      #
      # Example:
      #
      # envConfigMaps:
      #   - name: config-map
      envConfigMaps: []

      # Additional environment variables can be retrieved and set from Secrets. See:
      # https://kubernetes.io/docs/concepts/configuration/secret/#use-case-as-container-environment-variables
      #
      # Example:
      #
      # envSecrets:
      #   - name: secret
      envSecrets: []

      annotations: {}

      # Sets a node selector as a default for all Celery queues.
      #
      # See:
      # https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#nodeselector
      nodeSelector: {}

      # Support affinity and tolerations for Celery pod assignment. See:
      # https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#affinity-and-anti-affinity
      # https://kubernetes.io/docs/concepts/configuration/taint-and-toleration/
      affinity: {}
      tolerations: []
      podSecurityContext: {}
      securityContext: {}

      # Specify resources.
      # Example:
      #
      # resources:
      #   limits:
      #     cpu: 100m
      #     memory: 128Mi
      #   requests:
      #     cpu: 100m
      #     memory: 128Mi
      resources: {}

      # If `livenessProbe` does not contain `exec` field, then we will default to using:
      # exec:
      #   command:
      #     - /bin/sh
      #     - -c
      #     - celery status -A dagster_celery_k8s.app -b {{ include "dagster.celery.broker_url" . }} | grep "${HOSTNAME}:.*OK"
      livenessProbe:
        initialDelaySeconds: 15
        periodSeconds: 10
        timeoutSeconds: 10
        successThreshold: 1
        failureThreshold: 3

    ## Uncomment this configuration will only be used if the CustomRunLauncher is selected.
    ## Using this setting requires a custom Dagit image that defines the user specified
    ## run launcher in an installed python module.
    # customRunLauncher:
    #   module: ~
    #   class: ~
    #   config: {}

Can you please let me how to configure pipeline-env? OR please provide the example link

[rexledesma]

This was an issue of execution configuration. When specifying the run config when using the CeleryK8sRunLauncher, a configmap should be provided that has environment variables required for the run. In the helm, for a release named dagster, a configmap with these values is created for you, named dagster-pipeline-env. This should be passed into the configuration.

execution:
  celery-k8s:
    config:
      env_config_maps:
      - dagster-pipeline-env
      ... other config