ccc2e94c2a
The reference graph processed every work unit inside a single op, so exactly one worker was ever reported regardless of executor. That made the guide's claim that k8s_job_executor yields several distinct contributing_hosts false, and left the reference implementations unable to demonstrate the executor choice at all. generate_work_units is now a DynamicOut and both graphs map over it, so one step is created per unit and the loosely coupled pattern dispatches one external workload per unit. Evidence is split into contributing_workers (host and pid, differs per process) and contributing_hosts (differs only across machines), because the previous single field could not distinguish multiprocess fan-out from no fan-out. Tests now assert the mapped step keys rather than a host count, since execute_in_process ignores executor_def and cannot prove executor behaviour on its own. Adds a Windows note: multiprocess_executor did not complete during authoring and left orphaned processes. Changelog: fixed
65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
"""Readiness checks backing the pre-run validation checklist.
|
|
|
|
Each function returns a plain dict so the same result can be logged as Dagster
|
|
metadata (evidence for the checklist) or asserted in a test.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import socket
|
|
from urllib.parse import urlparse
|
|
|
|
# Runtime dependencies a tightly coupled execution pod must be able to resolve.
|
|
TIGHTLY_COUPLED_ENV_VARS = (
|
|
"DAGSTER_POSTGRES_HOST",
|
|
"DAGSTER_POSTGRES_USER",
|
|
"DAGSTER_POSTGRES_DB",
|
|
)
|
|
|
|
|
|
def check_env_vars(names: tuple[str, ...] | list[str]) -> dict:
|
|
"""Report which of ``names`` are present, without echoing their values."""
|
|
present = [name for name in names if os.environ.get(name)]
|
|
missing = [name for name in names if not os.environ.get(name)]
|
|
return {
|
|
"check": "env_vars",
|
|
"passed": not missing,
|
|
"present": present,
|
|
"missing": missing,
|
|
}
|
|
|
|
|
|
def check_tcp_reachable(host: str, port: int, timeout: float = 3.0) -> dict:
|
|
"""Open a TCP connection to prove the execution pod can reach a dependency."""
|
|
result = {"check": "tcp_reachable", "target": f"{host}:{port}", "passed": False, "error": None}
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout):
|
|
result["passed"] = True
|
|
except OSError as exc:
|
|
result["error"] = str(exc)
|
|
return result
|
|
|
|
|
|
def check_endpoint_reachable(url: str, timeout: float = 3.0) -> dict:
|
|
"""TCP-level reachability for an endpoint expressed as a URL."""
|
|
parsed = urlparse(url)
|
|
if not parsed.hostname:
|
|
return {"check": "tcp_reachable", "target": url, "passed": False, "error": "no hostname in URL"}
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
return check_tcp_reachable(parsed.hostname, port, timeout=timeout)
|
|
|
|
|
|
def describe_pod_identity() -> dict:
|
|
"""Where this process is actually running - the primary execution-target evidence."""
|
|
hostname = socket.gethostname()
|
|
return {
|
|
"hostname": hostname,
|
|
"pid": os.getpid(),
|
|
# Distinguishes processes on one host, so multiprocess fan-out is visible locally.
|
|
"worker": f"{hostname}#{os.getpid()}",
|
|
"namespace": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_NAMESPACE", "<not-in-kubernetes>"),
|
|
"run_id": os.environ.get("DAGSTER_RUN_ID", "<unset>"),
|
|
"image": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_IMAGE", "<unset>"),
|
|
}
|