Are there any suggestions for where to put code for instrumentation? I'd like to wrap my flows and tasks. I've found two spots that seem ok:

def patch_task():
    method = getattr(CloudTaskRunner, "run")

    def patched(self, *args, **kwargs):
        name = self.task.name
        with tracer.trace(f"task", resource=name):
            result = method(self, *args, **kwargs)
        tracer.flush()
        return result

    setattr(CloudTaskRunner, "run", patched)


def patch_flow():
    method = getattr(CloudFlowRunner, "run")

    def patched(self, *args, **kwargs):
        name = self.flow.name
        with tracer.trace(f"flow", resource=name):
            result = method(self, *args, **kwargs)
        tracer.flush()
        return result

    setattr(CloudFlowRunner, "run", patched)

What do you mean by instrumentation? Usually, you shouldn't have to modify Cloud task or flow runners, curious what problem are you trying to solve this way?

The problem I am trying to solve is to get apm/observability by tracing all of my flows and tasks using datadog. My question is not to specific datadog though. An example of another open source tool in this space is open telemetry.

Thanks for explaining. This is not as simple as modifying the cloud flow or task runner though, if you want to, I could open a feature request for sending logs to Datadog - but it would be for Prefect 2.0, not 1.0 and for tracing, we do that already in Cloud 2.0 using https://github.com/trallnag/prometheus-fastapi-instrumentator

if you want to, I could open a feature request for sending logs to Datadog

No need for this, thank you though.

and for tracing, we do that already in Cloud 2.0 using https://github.com/trallnag/prometheus-fastapi-instrumentator

I don't understand the intent of this comment. Specifically I'm not sure what this library would do in regards to a prefect end user that's interested in setting up tracing around their flows and tasks.