Release 1.12.20 (tracks 8d1f1cc2d6d384f1121a04eae95e909c8edf7b70)

Author: elementl-devtools

dagster-cloud-examples/dagster_cloud_examples/version.py

@@ -1 +1 @@
-__version__ = "1.12.19"
+__version__ = "1.12.20"

dagster-cloud-examples/pyproject.toml

@@ -7,7 +7,7 @@ build-backend = "hatchling.build"
 [project]
 name = "dagster-cloud-examples"
 requires-python = ">=3.9,<3.13"
-version = "1.12.19"
+version = "1.12.20"
 authors = [
     { name = "Elementl", email = "hello@elementl.com" },
 ]
@@ -17,7 +17,7 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = [
-    "dagster-cloud==1.12.19",
+    "dagster-cloud==1.12.20",
 ]
 
 [project.license]

dagster-cloud/conftest.py

@@ -1,4 +1,7 @@
-def pytest_addoption(parser):
+import pytest
+
+
+def pytest_addoption(parser: pytest.Parser):
     parser.addoption("--force-shard1", action="store_true", default=False, help="Enable sharding")
 
 

dagster-cloud/dagster_cloud/agent/dagster_cloud_agent.py

@@ -319,7 +319,7 @@ def run_loop(
 
         self._check_update_workspace(
             user_code_launcher,
-            upload_all=user_code_launcher.upload_snapshots_on_startup,
+            upload_outdated=user_code_launcher.upload_outdated_snapshots_on_startup,
         )
 
         self._logger.info(
@@ -371,7 +371,7 @@ def run_loop(
                 pass
 
             try:
-                self._check_update_workspace(user_code_launcher, upload_all=False)
+                self._check_update_workspace(user_code_launcher, upload_outdated=False)
 
             except Exception:
                 self._logger.error(
@@ -428,7 +428,7 @@ def _check_for_long_running_reconcile(self, user_code_launcher):
         else:
             self._warned_about_long_in_progress_reconcile = False
 
-    def _check_update_workspace(self, user_code_launcher, upload_all):
+    def _check_update_workspace(self, user_code_launcher, upload_outdated):
         curr_time = get_current_datetime()
 
         if (
@@ -439,7 +439,7 @@ def _check_update_workspace(self, user_code_launcher, upload_all):
             return
 
         self._last_workspace_check_time = curr_time
-        self._query_for_workspace_updates(user_code_launcher, upload_all=upload_all)
+        self._query_for_workspace_updates(user_code_launcher, upload_outdated=upload_outdated)
 
     def _check_add_heartbeat(
         self,
@@ -723,7 +723,7 @@ def _get_locations_with_ttl_to_query(self, user_code_launcher) -> list[tuple[str
     def _query_for_workspace_updates(
         self,
         user_code_launcher: DagsterCloudUserCodeLauncher,
-        upload_all: bool,
+        upload_outdated: bool,
     ):
         locations_with_ttl_to_query = self._get_locations_with_ttl_to_query(user_code_launcher)
 
@@ -828,7 +828,7 @@ def _query_for_workspace_updates(
             deployment_map,
             control_plane_error_locations=tracked_error_locations,
             control_plane_outdated_locations=tracked_outdated_locations,
-            upload_all=upload_all,
+            upload_outdated=upload_outdated,
         )
 
         # Tell run worker monitoring which deployments it should care about

dagster-cloud/dagster_cloud/version.py

@@ -1 +1 @@
-__version__ = "1.12.19"
+__version__ = "1.12.20"

dagster-cloud/dagster_cloud/workspace/config_schema/__init__.py

@@ -161,6 +161,7 @@ def process_workspace_config(workspace_config) -> dict[str, Any]:
                 ),
                 "deployment_metadata": Permissive(),
                 "service_metadata": Permissive(),
+                "service_spec_config": Permissive(),
             }
         ),
         is_required=False,

dagster-cloud/dagster_cloud/workspace/ecs/client.py

@@ -1,8 +1,10 @@
 import asyncio
+import datetime
 import json
 import logging
 import os
 import time
+import uuid
 
 import boto3
 import botocore
@@ -64,11 +66,19 @@ def __init__(
         launch_type: str = "FARGATE",
         show_debug_cluster_info: bool = True,
         assign_public_ip: bool | None = None,
+        service_discovery_role_arn: str | None = None,
     ):
         self.ecs = ecs_client if ecs_client else boto3.client("ecs", config=config)
         self.logs = boto3.client("logs", config=config)
-        self.service_discovery = boto3.client("servicediscovery", config=config)
         self.tags_client = boto3.client("resourcegroupstaggingapi", config=config)
+
+        self._service_discovery_role_arn = service_discovery_role_arn
+        if service_discovery_role_arn:
+            self._sts = boto3.client("sts", config=config)
+            self._sd_session_expires: datetime.datetime | None = None
+            self._create_service_discovery_session()
+        else:
+            self._service_discovery = boto3.client("servicediscovery", config=config)
         self._use_legacy_tag_filtering = False
 
         self.cluster_name = cluster_name.split("/")[-1]
@@ -87,6 +97,35 @@ def __init__(
         self._namespace: str | None = None
         self._assign_public_ip_override = assign_public_ip
 
+    def _create_service_discovery_session(self) -> None:
+        """Assume cross-account role and create a servicediscovery client with temporary credentials."""
+        response = self._sts.assume_role(
+            RoleArn=check.not_none(self._service_discovery_role_arn),
+            RoleSessionName=f"dagster-ecs-sd-{uuid.uuid4()}",
+            DurationSeconds=3600,
+        )
+        self._sd_session_expires = response["Credentials"]["Expiration"]
+        session = boto3.Session(
+            aws_access_key_id=response["Credentials"]["AccessKeyId"],
+            aws_secret_access_key=response["Credentials"]["SecretAccessKey"],
+            aws_session_token=response["Credentials"]["SessionToken"],
+        )
+        self._service_discovery = session.client("servicediscovery", config=config)
+
+    def _refresh_service_discovery_session(self) -> None:
+        """Refresh the cross-account session if it's within 5 minutes of expiration."""
+        if self._sd_session_expires is None:
+            return
+        now = datetime.datetime.now(self._sd_session_expires.tzinfo)
+        if self._sd_session_expires <= now + datetime.timedelta(minutes=5):
+            self._create_service_discovery_session()
+
+    @property
+    def service_discovery(self):
+        if self._service_discovery_role_arn:
+            self._refresh_service_discovery_session()
+        return self._service_discovery
+
     @property
     def ec2(self):
         # Reconstruct the resource on each call because it's not threadsafe

dagster-cloud/dagster_cloud/workspace/ecs/launcher.py

@@ -93,12 +93,14 @@ def __init__(
         server_task_definition_prefix: str = "server",
         run_task_definition_prefix: str = "run",
         assign_public_ip: bool | None = None,
+        service_discovery_role_arn: str | None = None,
         **kwargs,
     ):
         self.ecs = boto3.client("ecs")
         self.logs = boto3.client("logs")
         self.service_discovery = boto3.client("servicediscovery")
         self.secrets_manager = boto3.client("secretsmanager")
+        self.service_discovery_role_arn = service_discovery_role_arn
 
         self.cluster = cluster
         self.subnets = subnets
@@ -186,6 +188,7 @@ def __init__(
             grace_period=self._ecs_grace_period,
             launch_type=self.launch_type,
             assign_public_ip=assign_public_ip,
+            service_discovery_role_arn=self.service_discovery_role_arn,
         )
         super().__init__(**kwargs)
 
@@ -311,6 +314,17 @@ def config_type(cls):
                         "determination may not be accurate. In this case, you can explicitly set this value to True or False."
                     ),
                 ),
+                "service_discovery_role_arn": Field(
+                    StringSource,
+                    is_required=False,
+                    description=(
+                        "An optional IAM role ARN to assume when making AWS Cloud Map (service discovery) "
+                        "API calls. Use this when the Cloud Map namespace lives in a different AWS account "
+                        "than the ECS agent (e.g., centralized networking in AWS Organizations). The agent's "
+                        "task role must have sts:AssumeRole permission for this role, and the target role "
+                        "must trust the agent's task role."
+                    ),
+                ),
             },
             SHARED_ECS_CONFIG,
             SHARED_USER_CODE_LAUNCHER_CONFIG,

dagster-cloud/dagster_cloud/workspace/kubernetes/launcher.py

@@ -77,6 +77,7 @@ def __init__(
         inst_data=None,
         namespace=None,
         kubeconfig_file=None,
+        k8s_api_ssl_ca_cert_file=None,
         pull_policy=None,
         env_config_maps=None,
         env_secrets=None,
@@ -147,6 +148,13 @@ def __init__(
         else:
             kubernetes.config.load_incluster_config()
 
+        # Override the SSL CA cert if a custom CA bundle is provided
+        if k8s_api_ssl_ca_cert_file:
+            # Get the active configuration and override the ssl_ca_cert
+            config = kubernetes.client.Configuration.get_default_copy()
+            config.ssl_ca_cert = k8s_api_ssl_ca_cert_file
+            kubernetes.client.Configuration.set_default(config)
+
         self._k8s_apps_api_client = k8s_apps_api_client
         self._k8s_core_api_client = k8s_core_api_client
 
@@ -202,14 +210,15 @@ def __init__(
             run_k8s_config=self._run_k8s_config,
             fail_pod_on_run_failure=fail_pod_on_run_failure,
             kubeconfig_file=kubeconfig_file,
+            k8s_api_ssl_ca_cert_file=k8s_api_ssl_ca_cert_file,
             load_incluster_config=not kubeconfig_file,
             security_context=self._security_context,
             # leave out the code server specific fields from only_allow_user_defined_k8s_config_fields
             only_allow_user_defined_k8s_config_fields=(
                 {
                     key: val
                     for key, val in only_allow_user_defined_k8s_config_fields.items()
-                    if key not in {"service_metadata", "deployment_metadata"}
+                    if key not in {"service_metadata", "deployment_metadata", "service_spec_config"}
                 }
                 if only_allow_user_defined_k8s_config_fields
                 else only_allow_user_defined_k8s_config_fields
@@ -244,6 +253,15 @@ def config_type(cls):
                 "dagster_home": Field(StringSource, is_required=True),
                 "instance_config_map": Field(StringSource, is_required=True),
                 "kubeconfig_file": Field(StringSource, is_required=False),
+                "k8s_api_ssl_ca_cert_file": Field(
+                    StringSource,
+                    is_required=False,
+                    description=(
+                        "Path to a custom CA bundle file for TLS verification when connecting to "
+                        "the Kubernetes API. Use this in enterprise environments with custom CA "
+                        "chains where the default service account CA cert is not sufficient."
+                    ),
+                ),
                 "deployment_startup_timeout": Field(
                     IntSource,
                     is_required=False,
@@ -302,6 +320,7 @@ def config_type(cls):
                             "pod_template_spec_metadata": Permissive(),
                             "deployment_metadata": Permissive(),
                             "service_metadata": Permissive(),
+                            "service_spec_config": Permissive(),
                         }
                     ),
                     is_required=False,
@@ -331,6 +350,9 @@ def config_type(cls):
                             "service_metadata": Field(
                                 Map(key_type=str, inner_type=bool), is_required=False
                             ),
+                            "service_spec_config": Field(
+                                Map(key_type=str, inner_type=bool), is_required=False
+                            ),
                             "namespace": Field(bool, is_required=False),
                         }
                     ),

dagster-cloud/dagster_cloud/workspace/kubernetes/utils.py

@@ -87,6 +87,9 @@ def construct_code_location_service(
     user_defined_k8s_config = container_context.server_k8s_config
     user_defined_service_metadata = copy.deepcopy(dict(user_defined_k8s_config.service_metadata))
     user_defined_service_labels = user_defined_service_metadata.pop("labels", {})
+    user_defined_service_spec_config = copy.deepcopy(
+        dict(user_defined_k8s_config.service_spec_config)
+    )
 
     return k8s_model_from_dict(
         client.V1Service,
@@ -104,6 +107,7 @@ def construct_code_location_service(
                 },
             },
             "spec": {
+                **user_defined_service_spec_config,
                 "selector": {"user-deployment": service_name},
                 "ports": [{"name": "grpc", "protocol": "TCP", "port": get_code_server_port()}],
             },