Asset op_tags are not propagated to step when materialization runs

[alekseik1]

What's the issue?

Asset materialization can require special pods (e.g. with particular node_selector in k8s). Plugins like dagster-celery-k8s allow for such customization via dagster-k8s/config tag.
However, setting op_tags in @asset decorator does not propagate this tag to steps during run when running from UI.
Running from schedules seem to propagate these tags.

What did you expect to happen?

Setting key "dagster-k8s/config" in op_tags would customize pod creation behavior (e.g. "node_selector") when running from UI (for instance, when clicking "Materialize" button in asset tab).

How to reproduce?

Make a simple asset definition:

from dagster import AssetExecutionContext, asset


@asset(
    op_tags={
        "dagster-celery/queue": "default",
        "dagster-k8s/config": {
            "job_spec_config": {"ttl_seconds_after_finished": 600},
            "pod_spec_config": {
                "node_selector": {"dedicated": "pipelines"},
                "tolerations": [
                    {
                        "effect": "NoSchedule",
                        "key": "dedicated",
                        "operator": "Equal",
                        "value": "pipelines",
                    }
                ],
            },
        }
    }
)
def small_asset_with_tags(context: AssetExecutionContext):
    raise ValueError(f"current run tags: {context.run_tags}")

Go to Dagster UI, find asset named small_asset_with_tags, click "Materialize", wait for pod to be created. Check the pod configuration via kubectl get pod -oyaml POD_NAME | grep dedicated - it will not find such line.

Also, you would not find tag key "dagster-k8s/config" in job logs.

Dagster version

dagster, version 1.7.16

Deployment type

Dagster Helm chart

Deployment details

Deployment in ArgoCD with Celery workers, I use charts from official repo.

Additional information

No response

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.
By submitting this issue, you agree to follow Dagster's Code of Conduct.