Hi! Is there a way to grab the worker that polled a specific flow run submission? For example, I have a single work-queue that is polled by 3 different process workers. Deployments that run under this work-queue can get picked up by any of the 3 workers. The worker that submitted the flow run gets logged to the beginning of the flow run logs, but I don't see an explicit way to grab the workers from the flow run that doesn't involve manually looking in the logs.

Hi Carlos! I haven't found a direct way of getting the ID of a worker that submitted a flow. However, I think you may be able to leverage to REST API or Prefect client to fetch that info for you from the flow run logs.

Thank you Bianca. Appreciate it!! I will give it a try tomorrow. I wish there was a way to grab this info in a more concrete way, would be very useful to us that only use process workers and have multiple workers for a single work queue for resiliency. It’s useful to know which how many flow runs a single worker is currently Running.

NP! Never hurts to submit an idea through GitHub, if you'd like

I went ahead and created a script to get workers that were used to run specific flow runs currently in a running state, can be modified for any filters, thought I'd share it for anybody else that runs into this thread:

from prefect import flow, task
from prefect.client.orchestration import get_client
from prefect.server.schemas.filters import DeploymentFilter, FlowRunFilter, FlowRunFilterStateName, FlowRunFilterState, DeploymentFilterName, LogFilter, LogFilterFlowRunId
from prefect.artifacts import create_table_artifact
from prefect.task_runners import ThreadPoolTaskRunner
import asyncio
from collections import Counter

async def list_deployment_running_flow_runs(deployment_name: str):
    async with get_client() as client:
        flow_runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                state=FlowRunFilterState(
                    name=FlowRunFilterStateName(any_=["Running"])
                )
            ),
            deployment_filter=DeploymentFilter(
                name=DeploymentFilterName(any_=[deployment_name])
            )

        )
    return flow_runs

@task
async def get_flow_run_worker(flow_run_id: str):
    async with get_client() as client:
        first_log_line = await client.read_logs(
            log_filter=LogFilter(
                flow_run_id = LogFilterFlowRunId(any_= [flow_run_id])), 
            limit=1, 
            offset=0)
        
    return first_log_line[0].message.split()[1].strip("'")

@task
async def create_worker_artifact(workers: list[str]):
    count_dict = dict(Counter(workers))
    await create_table_artifact(key="v500-long-running-workers", table=[count_dict], description="Table containing workers currently running v500 long-running flows")

@flow(task_runner=ThreadPoolTaskRunner(max_workers=15))
async def get_flow_run_workers():

    flow_runs = await list_deployment_running_flow_runs("V500-UPDATE-TABLE")
    worker_names = []

    for flow_run in flow_runs:
        worker_names.append(get_flow_run_worker.submit(flow_run.id))

    await create_worker_artifact(worker_names)

if __name__ == '__main__':
    asyncio.run(get_flow_run_workers())