Log which ops require each resource during initialization#34005
Log which ops require each resource during initialization#34005Mariomsoliman wants to merge 1 commit into
Conversation
The RESOURCE_INIT_STARTED log line previously listed only resource keys:
Starting initialization of resources [io_manager, resource_a, resource_b].
On jobs with many resources this made it impossible to tell which op pulled
in a given resource. It now attributes each resource to the ops that require
it:
Starting initialization of resources:
io_manager - required by the I/O layer
resource_a - required by op instances [op_one]
resource_b - required by op instances [op_two]
get_resource_origins_mapping builds resource_key -> {op instance names} from
op and hook required_resource_keys, keyed by full node handle so nested and
aliased ops are unambiguous, and propagates through resource-to-resource
dependencies. job_def is threaded to the resource init generator to supply
it; when absent (e.g. build_resources) the original message is preserved.
The dagstermill resource manager is updated to match the new signature.
Resolves dagster-io#2307.
Greptile SummaryThis PR addresses issue #2307 by enriching the
Confidence Score: 4/5Safe to merge — changes are confined to a log message and its supporting plumbing; no execution logic is altered and all existing code paths have The falsy-empty-dict check in python_modules/dagster/dagster/_core/events/init.py — the
|
| Filename | Overview |
|---|---|
| python_modules/dagster/dagster/_core/events/init.py | Adds _resource_init_start_message helper and threads it into resource_init_start; falsy-empty-dict check on the mapping may silently suppress the enriched format |
| python_modules/dagster/dagster/_core/execution/resources_init.py | Adds get_resource_origins_mapping and threads job_def through the init call chain; logic is correct for typical jobs |
| python_modules/dagster/dagster/_core/execution/context_creation_job.py | Passes job_def to scoped_resources_builder_cm; dagstermill and all known custom implementations are updated accordingly |
| python_modules/dagster/dagster_tests/core_tests/test_job_init.py | Good coverage: unit test for mapping logic, format tests for both new and fallback paths, and a monkeypatch regression test verifying job_def propagation depth |
| python_modules/libraries/dagstermill/dagstermill/manager.py | Adds optional job_def param to _setup_resources and forwards it to resource_initialization_event_generator, keeping the drop-in contract intact |
Sequence Diagram
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant EG as execution_context_event_generator
participant RIM as resource_initialization_manager
participant RIEG as resource_initialization_event_generator
participant CRIEG as _core_resource_initialization_event_generator
participant GROM as get_resource_origins_mapping
participant MSG as _resource_init_start_message
EG->>RIM: "job_def=job_def (new)"
RIM->>RIEG: "job_def=job_def (new)"
RIEG->>CRIEG: "job_def=job_def (new)"
CRIEG->>GROM: job_def (if not None)
GROM-->>CRIEG: Mapping[resource_key to Set[op_name]]
CRIEG->>MSG: resource_keys, resource_key_to_op_names
MSG-->>CRIEG: enriched log string
CRIEG-->>EG: RESOURCE_INIT_STARTED event with enriched message
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant EG as execution_context_event_generator
participant RIM as resource_initialization_manager
participant RIEG as resource_initialization_event_generator
participant CRIEG as _core_resource_initialization_event_generator
participant GROM as get_resource_origins_mapping
participant MSG as _resource_init_start_message
EG->>RIM: "job_def=job_def (new)"
RIM->>RIEG: "job_def=job_def (new)"
RIEG->>CRIEG: "job_def=job_def (new)"
CRIEG->>GROM: job_def (if not None)
GROM-->>CRIEG: Mapping[resource_key to Set[op_name]]
CRIEG->>MSG: resource_keys, resource_key_to_op_names
MSG-->>CRIEG: enriched log string
CRIEG-->>EG: RESOURCE_INIT_STARTED event with enriched message
Reviews (1): Last reviewed commit: "Log which ops require each resource duri..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| if not resource_key_to_op_names: | ||
| return "Starting initialization of resources [{}].".format(", ".join(sorted(resource_keys))) |
There was a problem hiding this comment.
if not resource_key_to_op_names falls through to the old message format for both None (no job_def) and an empty dict {}. An empty dict is returned by get_resource_origins_mapping when every resource in the job is required only through the I/O layer or type system (e.g., a job whose ops have no explicit required_resource_keys). In that case job_def was provided and the enriched format should still be used — each resource would correctly display as "required by the I/O layer" — but the old flat-list fallback is emitted instead. The check should use an identity comparison with None.
| if not resource_key_to_op_names: | |
| return "Starting initialization of resources [{}].".format(", ".join(sorted(resource_keys))) | |
| if resource_key_to_op_names is None: | |
| return "Starting initialization of resources [{}].".format(", ".join(sorted(resource_keys))) |
| resource_dependencies = resolve_resource_dependencies(job_def.resource_defs) | ||
| for resource_key, op_names in list(origins.items()): | ||
| for dep_key in get_dependencies(resource_key, resource_dependencies): | ||
| if dep_key != resource_key: | ||
| origins[dep_key].update(op_names) |
There was a problem hiding this comment.
get_dependencies indexes into resource_dependencies with every key currently in origins. If an op (or hook) declares a required_resource_keys entry that is not present in job_def.resource_defs, resource_deps[resource_key] will raise a KeyError before the enriched log message is produced. In a valid, fully-validated job this can't happen, but if get_resource_origins_mapping is ever called before full job validation (e.g., during inspection or tooling), an uncaught KeyError would surface here as an obscure internal error rather than a clear validation message. A resource_key in resource_dependencies guard, or a resource_dependencies.get(resource_key, set()) call inside get_dependencies, would make the failure mode explicit.
|
Hey @sryza and @alangenfeld , opening this to fix #2307. It adds the missing context to the resource initialization log, so instead of just listing resource names, it now shows which ops actually need each one. Let me know if you'd want a different approach, happy to adjust. Thanks for taking a look! |
1 participant
This closes #2307.
Right now when Dagster starts initializing resources for a run, the log just prints something like: Starting initialization of resources [io_manager, resource_a, resource_b]. That's fine if we have one or two resources, but on a job with many, it's hard to tell which op actually needed which resource. If resource_a fails to initialize, we have no idea what part of your pipeline that's going to break. This PR fixes that by adding the missing context: Starting initialization of resources: io_manager - required by the I/O layer
resource_a - required by op instances [op_one]
resource_b - required by op instances [op_two]
I added a helper, get_resource_origins_mapping, that walks a job's ops (and their hooks) and builds a mapping from each resource key to the op names that actually require it. It also handles the case where one resource depends on another , so if spark is only used because pyspark needs it, spark still gets attributed to whatever op needed pyspark. That's actually the exact example from the original issue.
I threaded job_def through the resource init call chain so this mapping can actually get built where it's needed, and it defaults to None everywhere so nothing breaks for code paths that don't have a job (like the standalone build_resources() helper).
One thing worth calling out: resources get initialized per step process under the default executor, not once for the whole run. So the log message only looks at the resources that specific process actually needs, not the whole job, otherwise you'd get one step's log showing another step's ops, which would be confusing in the opposite direction.
I also had to update dagstermill's resource setup, since it mirrors this same function and would've broken for notebook-based jobs otherwise.
I also added tests that directly cover the new mapping logic, both the new message format and the old fallback message (for when no job_def is available), and a regression test that checks that the job_def actually makes it all the way through the call chain instead of being silently dropped somewhere. Also ran this against a real job with dagster job execute to confirm the output looks right, and ran the full test suites for both dagster core and dagstermill. Any failures I found were pre-existing on main, not caused by my change.