Add container_context to gRPC server and Helm chart (#7236)

helm/dagster/charts/dagster-user-deployments/templates/configmap-env-shared.yaml

@@ -0,0 +1,14 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ template "dagster.fullname" $ -}}-user-deployments-shared-env
+  labels:
+    app: {{ template "dagster.name" $ }}
+    chart: {{ template "dagster.chart" $ }}
+    release: {{ $.Release.Name }}
+    heritage: {{ $.Release.Service }}
+data:
+  {{ include "dagsterUserDeployments.sharedEnv" $ | nindent 2 }}
+
+---

helm/dagster/charts/dagster-user-deployments/templates/configmap-env-user.yaml

@@ -9,8 +9,6 @@ metadata:
     release: {{ $.Release.Name }}
     heritage: {{ $.Release.Service }}
 data:
-  {{ include "dagsterUserDeployments.sharedEnv" $ | nindent 2 }}
-
   {{- range $name, $value := $deployment.env }}
   {{ $name }}: {{ $value | quote }}
   {{- end }}

helm/dagster/charts/dagster-user-deployments/templates/deployment-user.yaml

@@ -51,7 +51,15 @@ spec:
                 secretKeyRef:
                   name: {{ include "dagsterUserDeployments.postgresql.secretName" $ }}
                   key: postgresql-password
+            {{- if $deployment.includeConfigInLaunchedRuns.enabled }}
+            # uses the auto_envvar_prefix of the dagster cli to set the --container-context arg
+            # on 'dagster api grpc'
+            - name: DAGSTER_CLI_API_GRPC_CONTAINER_CONTEXT
+              value: {{ include "dagsterUserDeployments.k8sContainerContext" (list $ $deployment) | fromYaml | toJson | quote }}
+            {{- end }}
           envFrom:
+            - configMapRef:
+                name: {{ include "dagster.fullname" $ }}-user-deployments-shared-env
             - configMapRef:
                 name: {{ include "dagster.fullname" $ }}-{{ $deployment.name }}-user-env
             {{- range $envConfigMap := $deployment.envConfigMaps }}

helm/dagster/charts/dagster-user-deployments/templates/helpers/_helpers.tpl

@@ -91,3 +91,36 @@ DAGSTER_K8S_INSTANCE_CONFIG_MAP: "{{ template "dagster.fullname" .}}-instance"
 DAGSTER_K8S_PIPELINE_RUN_NAMESPACE: "{{ .Release.Namespace }}"
 DAGSTER_K8S_PIPELINE_RUN_ENV_CONFIGMAP: "{{ template "dagster.fullname" . }}-pipeline-env"
 {{- end -}}
+
+
+{{- define "dagsterUserDeployments.k8sContainerContext" -}}
+  {{- $ := index . 0 }}
+  {{- with index . 1 }}
+  k8s:
+    image_pull_policy: {{ .image.pullPolicy }}
+    env_config_maps:
+    - {{ include "dagster.fullname" $ }}-{{ .name }}-user-env
+    {{- range $envConfigMap := .envConfigMaps }}
+    {{- if hasKey $envConfigMap "name" }}
+    - {{ $envConfigMap.name }}
+    {{- end }}
+    {{- end }}
+    {{- if .envSecrets }}
+    env_secrets:
+    {{- range $envSecret := .envSecrets }}
+    {{- if hasKey $envSecret "name" }}
+    - {{ $envSecret.name }}
+    {{- end }}
+    {{- end }}
+    {{- end }}
+    {{- if .volumeMounts }}
+    volume_mounts: {{- .volumeMounts | toYaml | nindent 6 }}
+    {{- end }}
+    {{- if .volumes }}
+    volumes: {{- .volumes | toYaml | nindent 6 }}
+    {{- end }}
+    {{- if .labels }}
+    labels: {{- .labels | toYaml | nindent 6 }}
+    {{- end }}
+  {{- end }}
+{{- end -}}

helm/dagster/charts/dagster-user-deployments/values.yaml

@@ -43,6 +43,12 @@ deployments:
       - "-f"
       - "/example_project/example_repo/repo.py"
     port: 3030
+
+    # Whether or not to include configuration specified for this user code deployment in the pods
+    # launched for runs from that deployment
+    includeConfigInLaunchedRuns:
+      enabled: false
+
     # Additional volumes that should be included in the Deployment's Pod. See:
     # https://v1-18.docs.kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#volume-v1-core
     #

helm/dagster/schema/schema/charts/dagster_user_deployments/subschema/user_deployments.py

@@ -5,10 +5,15 @@
 from ...utils import kubernetes
 
 
+class UserDeploymentIncludeConfigInLaunchedRuns(BaseModel):
+    enabled: bool
+
+
 class UserDeployment(BaseModel):
     name: str
     image: kubernetes.Image
     dagsterApiGrpcArgs: List[str]
+    includeConfigInLaunchedRuns: UserDeploymentIncludeConfigInLaunchedRuns
     port: int
     env: Optional[Dict[str, str]]
     envConfigMaps: Optional[List[kubernetes.ConfigMapEnvSource]]

helm/dagster/schema/schema_tests/test_user_deployments.py

@@ -10,6 +10,7 @@
 from schema.charts.dagster.values import DagsterHelmValues
 from schema.charts.dagster_user_deployments.subschema.user_deployments import (
     UserDeployment,
+    UserDeploymentIncludeConfigInLaunchedRuns,
     UserDeployments,
 )
 from schema.charts.dagster_user_deployments.values import DagsterUserDeploymentsHelmValues
@@ -491,6 +492,12 @@ def test_user_deployment_default_image_tag_is_chart_version(
     assert image_tag == chart_version
 
 
+def _assert_no_container_context(user_deployment):
+    # No container context set by default
+    env_names = [env.name for env in user_deployment.spec.template.spec.containers[0].env]
+    assert "DAGSTER_CLI_API_GRPC_CONTAINER_CONTEXT" not in env_names
+
+
 def test_user_deployment_image(template: HelmTemplate):
     deployment = create_simple_user_deployment("foo")
     helm_values = DagsterHelmValues.construct(
@@ -511,9 +518,35 @@ def test_user_deployment_image(template: HelmTemplate):
     assert image_name == deployment.image.repository
     assert image_tag == deployment.image.tag
 
+    _assert_no_container_context(user_deployments[0])
+
+
+def test_user_deployment_include_config(template: HelmTemplate):
+    deployment = create_simple_user_deployment("foo", include_config_in_launched_runs=True)
+    helm_values = DagsterHelmValues.construct(
+        dagsterUserDeployments=UserDeployments(
+            enabled=True,
+            enableSubchart=True,
+            deployments=[deployment],
+        )
+    )
+
+    user_deployments = template.render(helm_values)
+
+    assert len(user_deployments) == 1
+
+    container_context = user_deployments[0].spec.template.spec.containers[0].env[2]
+    assert container_context.name == "DAGSTER_CLI_API_GRPC_CONTAINER_CONTEXT"
+    assert json.loads(container_context.value) == {
+        "k8s": {
+            "image_pull_policy": "Always",
+            "env_config_maps": ["release-name-dagster-user-deployments-foo-user-env"],
+        }
+    }
 
-def test_user_deployment_volumes(template: HelmTemplate):
 
+@pytest.mark.parametrize("include_config_in_launched_runs", [False, True])
+def test_user_deployment_volumes(template: HelmTemplate, include_config_in_launched_runs: bool):
     name = "foo"
 
     volumes = [
@@ -538,6 +571,9 @@ def test_user_deployment_volumes(template: HelmTemplate):
         volumeMounts=[
             kubernetes.VolumeMount.construct(None, **volume_mount) for volume_mount in volume_mounts
         ],
+        includeConfigInLaunchedRuns=UserDeploymentIncludeConfigInLaunchedRuns(
+            enabled=include_config_in_launched_runs
+        ),
     )
 
     helm_values = DagsterHelmValues.construct(
@@ -575,6 +611,121 @@ def test_user_deployment_volumes(template: HelmTemplate):
     assert image_name == deployment.image.repository
     assert image_tag == deployment.image.tag
 
+    if include_config_in_launched_runs:
+        container_context = user_deployments[0].spec.template.spec.containers[0].env[2]
+        assert container_context.name == "DAGSTER_CLI_API_GRPC_CONTAINER_CONTEXT"
+        assert json.loads(container_context.value) == {
+            "k8s": {
+                "env_config_maps": [
+                    "release-name-dagster-user-deployments-foo-user-env",
+                ],
+                "image_pull_policy": "Always",
+                "volume_mounts": volume_mounts,
+                "volumes": volumes,
+            }
+        }
+    else:
+        _assert_no_container_context(user_deployments[0])
+
+
+@pytest.mark.parametrize("include_config_in_launched_runs", [False, True])
+def test_user_deployment_secrets_and_configmaps(
+    template: HelmTemplate, include_config_in_launched_runs: bool
+):
+    name = "foo"
+
+    secrets = [{"name": "my-secret"}, {"name": "my-other-secret"}]
+
+    configmaps = [{"name": "my-configmap"}, {"name": "my-other-configmap"}]
+
+    deployment = UserDeployment(
+        name=name,
+        image=kubernetes.Image(repository=f"repo/{name}", tag="tag1", pullPolicy="Always"),
+        dagsterApiGrpcArgs=["-m", name],
+        port=3030,
+        envConfigMaps=[
+            kubernetes.ConfigMapEnvSource.construct(None, **configmap) for configmap in configmaps
+        ],
+        envSecrets=[kubernetes.SecretEnvSource.construct(None, **secret) for secret in secrets],
+        includeConfigInLaunchedRuns=UserDeploymentIncludeConfigInLaunchedRuns(
+            enabled=include_config_in_launched_runs
+        ),
+    )
+
+    helm_values = DagsterHelmValues.construct(
+        dagsterUserDeployments=UserDeployments(
+            enabled=True,
+            enableSubchart=True,
+            deployments=[deployment],
+        )
+    )
+
+    user_deployments = template.render(helm_values)
+
+    assert len(user_deployments) == 1
+
+    if include_config_in_launched_runs:
+        container_context = user_deployments[0].spec.template.spec.containers[0].env[2]
+        assert container_context.name == "DAGSTER_CLI_API_GRPC_CONTAINER_CONTEXT"
+        assert json.loads(container_context.value) == {
+            "k8s": {
+                "image_pull_policy": "Always",
+                "env_secrets": ["my-secret", "my-other-secret"],
+                "env_config_maps": [
+                    "release-name-dagster-user-deployments-foo-user-env",
+                    "my-configmap",
+                    "my-other-configmap",
+                ],
+            }
+        }
+    else:
+        _assert_no_container_context(user_deployments[0])
+
+
+@pytest.mark.parametrize("include_config_in_launched_runs", [False, True])
+def test_user_deployment_labels(template: HelmTemplate, include_config_in_launched_runs: bool):
+    name = "foo"
+
+    labels = {"my-label-key": "my-label-val", "my-other-label-key": "my-other-label-val"}
+
+    deployment = UserDeployment(
+        name=name,
+        image=kubernetes.Image(repository=f"repo/{name}", tag="tag1", pullPolicy="Always"),
+        dagsterApiGrpcArgs=["-m", name],
+        port=3030,
+        labels=labels,
+        includeConfigInLaunchedRuns=UserDeploymentIncludeConfigInLaunchedRuns(
+            enabled=include_config_in_launched_runs
+        ),
+    )
+
+    helm_values = DagsterHelmValues.construct(
+        dagsterUserDeployments=UserDeployments(
+            enabled=True,
+            enableSubchart=True,
+            deployments=[deployment],
+        )
+    )
+
+    user_deployments = template.render(helm_values)
+
+    assert len(user_deployments) == 1
+
+    if include_config_in_launched_runs:
+        container_context = user_deployments[0].spec.template.spec.containers[0].env[2]
+        assert container_context.name == "DAGSTER_CLI_API_GRPC_CONTAINER_CONTEXT"
+        assert json.loads(container_context.value) == {
+            "k8s": {
+                "image_pull_policy": "Always",
+                "env_config_maps": [
+                    "release-name-dagster-user-deployments-foo-user-env",
+                ],
+                "labels": labels,
+            }
+        }
+    else:
+        _assert_no_container_context(user_deployments[0])
+
 
 def test_subchart_image_pull_secrets(subchart_template: HelmTemplate):
     image_pull_secrets = [{"name": "super-duper-secret"}]

helm/dagster/schema/schema_tests/utils.py

@@ -1,17 +1,27 @@
-from schema.charts.dagster_user_deployments.subschema.user_deployments import UserDeployment
+from schema.charts.dagster_user_deployments.subschema.user_deployments import (
+    UserDeployment,
+    UserDeploymentIncludeConfigInLaunchedRuns,
+)
 from schema.charts.utils import kubernetes
 
 
-def create_simple_user_deployment(name: str) -> UserDeployment:
+def create_simple_user_deployment(
+    name: str, include_config_in_launched_runs: bool = False
+) -> UserDeployment:
     return UserDeployment(
         name=name,
         image=kubernetes.Image(repository=f"repo/{name}", tag="tag1", pullPolicy="Always"),
         dagsterApiGrpcArgs=["-m", name],
         port=3030,
+        includeConfigInLaunchedRuns=UserDeploymentIncludeConfigInLaunchedRuns(
+            enabled=include_config_in_launched_runs
+        ),
     )
 
 
-def create_complex_user_deployment(name: str) -> UserDeployment:
+def create_complex_user_deployment(
+    name: str, include_config_in_launched_runs: bool = False
+) -> UserDeployment:
     return UserDeployment(
         name=name,
         image=kubernetes.Image(repository=f"repo/{name}", tag="tag1", pullPolicy="Always"),
@@ -61,4 +71,7 @@ def create_complex_user_deployment(name: str) -> UserDeployment:
             "label_one": "one",
             "label_two": "two",
         },
+        includeConfigInLaunchedRuns=UserDeploymentIncludeConfigInLaunchedRuns(
+            enabled=include_config_in_launched_runs
+        ),
     )