What's the issue?

Using the new per_step_k8s_config with dynamic ops should be possible by simply using the op name for configuration. Currently, the step_key is used to index into the config which has the format op_name[mapping_key]. The code suggests(per_op_override) that the intended result was to simply index with op_name

Example:

Given the following job

@job
def dynamic_producer_consumer_job():
    @op(out=DynamicOut(int))
    def dyn_producer():
    for i in range(3):
        return DynamicOutput(i, str(i))
    @op
    def dyn_sink(producer: int): ...

What did you expect to happen?

We expect the following execution config to propagate resource configs to every dyn_sink[0-2] op:

execution:
  config:
    step_k8s_config:
      dyn_sink:
         container_config:
           resources:
             requests:
               cpu: '300m'

Instead, this will just get ignored.

How to reproduce?

I can provide a unit test in a PR to fix this, but essentially take the job & config above and either kubectl describe po <run pod> or observe the K8sContainerContext from the K8sStepHandler and see the configs not propagater properly

Dagster version

dagster, version 1.9.2

Deployment type

Dagster Helm chart

Deployment details

No response

Additional information

No response

Message from the maintainers

This is a small bug, I can contribute a fix if the maintainers confirm this is indeed unintended behavior