Howdy! Quick Question! Is there an easy way to see which worker ran the task?

Are there any recipes for managing workpools/queues in a declarative way? Something like defining pools/queues in yaml and syncing with prefect

Hey everyone. I was trying to edit users in Prefect and noticed that I don’t have an organization anymore, even though we signed for one

Hi there. My team has a (paid) Organization under the Prefect Cloud 1.0 application on cloud.prefect.io and would like to open up the same Org on Prefect 2.0 on app.prefect.cloud. Thanks

When I update the base job template json under the advanced tab for a work pool via the prefect cloud UI, save, and then look at the json, my changes are not reflected. Am I updated the base template incorrectly? Is this a bug?

hi guys, is there any way to submit/apply a deployment to prefect cloud without activating it what I meant is that I don’t want my deployment runs right away after submitted to the cloud

Hi guys. I am Geese, and This is my first time joining prefect slack community. I was just wondering if I can get some help from issue I am recently suffering while trying to deploy prefect-agent via Google Kubernetes Engine. While deploying either prefect-agent or prefect-worker using helm chart (

helm install prefect-agent prefect/prefect-agent --namespace==prefect -f values.yaml

), I was constantly getting API authentification error with the following kubernetes pod logs:

Starting v2.12.1 agent connected to 
<https://api.prefect.cloud/api/accounts//workspaces/>...

 ___ ___ ___ ___ ___ ___ _____   _  ___ ___ _ _ _____
 | _ \ _ \ __| __| __/ __|_  _|  /_\ / __| __| \| |_  _|
 | _/  / _|| _|| _| (__ | |  / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|  /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from queue(s): default...
Traceback (most recent call last):
 File "/usr/local/lib/python3.10/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
  return fn(*args, **kwargs)
 File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 255, in coroutine_wrapper
  return call()
 File "/usr/local/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 382, in __call__
  return self.result()
 File "/usr/local/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 282, in result
  return self.future.result(timeout=timeout)
 File "/usr/local/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 168, in result
  return self.__get_result()
 File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
  raise self._exception
 File "/usr/local/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 345, in _run_async
  result = await coro
 File "/usr/local/lib/python3.10/site-packages/prefect/cli/agent.py", line 196, in start
  async with anyio.create_task_group() as tg:
 File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 597, in __aexit__
  raise exceptions[0]
 File "/usr/local/lib/python3.10/site-packages/prefect/utilities/services.py", line 53, in critical_service_loop
  await workload()
 File "/usr/local/lib/python3.10/site-packages/prefect/agent.py", line 271, in check_for_cancelled_flow_runs
  named_cancelling_flow_runs = await self.client.read_flow_runs(
 File "/usr/local/lib/python3.10/site-packages/prefect/client/orchestration.py", line 1776, in read_flow_runs
  response = await <http://self._client.post|self._client.post>("/flow_runs/filter", json=body)
 File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1848, in post
  return await self.request(
 File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1530, in request
  return await self.send(request, auth=auth, follow_redirects=follow_redirects)
 File "/usr/local/lib/python3.10/site-packages/prefect/client/base.py", line 285, in send
  response.raise_for_status()
 File "/usr/local/lib/python3.10/site-packages/prefect/client/base.py", line 138, in raise_for_status
  raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<https://api.prefect.cloud/api/accounts//workspaces/flow_runs/filter>'
Response: {'detail': 'Not Found'}
For more information check: <https://httpstatuses.com/404>
An exception occurred

I tried to check the correctness of API keys by retrieving prefect-api-key from GKE secret, using new API key, but non of these measures worked well. Would it be possible to get any guess on why this is happening?

Hello, I've been having an issue logging into Prefect Cloud since Friday - every magic link/code I receive results in the same error. Should I be reaching out here or is there an email I can reach out to? My teammates can login so it seems to be an isolated issue. Thanks for your help!

Hello Guys! I have a weird problem with prefect2. Recently we ramp up to PROD, But now we have an error that we haven't in Stag. Any clu?

prefect_aws.workers.ecs_worker.TaskFailedToStart: ResourceInitializationError: unable to pull secrets or registry auth: execution resource retrieval failed: unable to retrieve secrets from ssm: service call has been retried 5 time(s): RequestCanceled: request context canceled caused by: context deadline exceeded. Please check your task network configuration.

Hi guys - I have a question on logging: I have many production flows deployed to one local worker pool - if I

tail -f

the pool process, I can see the logs streamed in but mixed together across all flows so not really readable. My question is: is there any way that I could stream in the logs separately for each flow ? for me, this is very important for production monitoring and stuff. Really appreciate if anyone could help!

When I try to create a work pool prefixed with the word "prefect", the GUI fails mysteriously and gives no description. I just tried via the CLI and I get this exception:

prefect.exceptions.PrefectHTTPStatusError: Client error '403 Forbidden' for url '<https://api.prefect.cloud/api/accounts/><...>/work_pools/'
Response: {'detail': "Work pools starting with 'Prefect' are reserved for internal use."}

Could detailed messages like this please be included in the GUI?

Hi, I'm running my flow on a kubernetes pod, I have self-hosted agent and worker. Now I see the error logs like this:

Worker 'KubernetesWorker 7763f002-66a8-475f-90cf-b40e048d3e02' submitting flow run '83edf78a-26ec-4f1a-870b-c82b96f25db2'
Creating Kubernetes job...
Job 'axiomatic-oyster-lcgft': Pod has status 'Pending'.
Completed submission of flow run '83edf78a-26ec-4f1a-870b-c82b96f25db2'
Job 'axiomatic-oyster-lcgft': Pod has status 'Running'.
Crash detected! Execution was interrupted by an unexpected exception: PrefectHTTPStatusError: Client error '403 Forbidden' for url '<https://api.prefect.cloud/api/accounts/xxxx/workspaces/xxxx/block_schemas/>'
Response: {'detail': 'Workspace scopes missing: manage_blocks'}
For more information check: <https://httpstatuses.com/403>
Job 'axiomatic-oyster-lcgft': Job reached backoff limit.

I masked our account id and workspace id. Anyone has any ideas? Thanks.

I am using the ecs:push work pool. When I send a concurrent batch of 100 flows to be ran it is only sending around 80 tasks to my AWS cluster at once. It then just stalls and randomly sends a few tasks with significant delays in between (15 min+). Where can I look either in prefect or AWS to see what the issue is? The AWS cluster as sufficient quota requirements to run up to 4000 vcpu.

Hello, I've been deploying using YAML files via the CLI to Prefect Cloud. I noticed that when I remove a deployment from the YAML and redeploy, the deployment still exists in Prefect Cloud. Does anyone know if there's a way to automatically delete deployments that are no longer in the YAML file during re-deployment? Thanks!

.

Hey hey! I have this small flow I want to try and deploy

@flow(name="LO", flow_run_name="test-LO")
def live_odds_poc():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Starting LO poc flow")

    flow = LiveOdds()
    lo_result = flow.run_flow()

and I’m doing running this

if __name__ == "__main__":
    live_odds_poc.serve(
        name="LO-test",
        tags=["LO"],
        interval=15
    )

but when I run it I get this Attribute Error ->

AttributeError: 'Flow' object has no attribute 'serve'

I have followed the tutorial in Prefect Cloud and I don’t see anything wrong (I also copied the code from the tutorial and got the same error). Can you help me? 🙏

Hi prefect cloud! I've been using prefect core for about a year at work. I'm working on getting vendor approval internally to start a Cloud accont. I understand that the service doesn't access my data. So the two architecture questions I have are: • What about the credentials I store in blocks? These would be stored by Prefect Cloud, correct? • What about logs data? Do these get stored in Prefect Cloud?

Hi prefect cloud! I was wondering if there is any way to modify prefect.yaml file deployment session to assign custom kubernetes cloud resources, including cpu, gpu and memory. Previously, I was using prefect.yaml file of the following format, which did not work (the flow runs well, but it ignores using the resource specifications):

# Generic metadata about this project
name: flows
prefect-version: 2.12.1

# build section allows you to manage and build docker images
build: null

# push section allows you to manage if and how this project is uploaded to remote locations
push:
- prefect_gcp.deployments.steps.push_to_gcs:
    id: push_code
    requires: prefect-gcp>=0.4.3
    bucket: prefect-flows
    folder: prefect_practice
    credentials: "{{ prefect.blocks.gcp-credentials.credential }}"

# prefect.yaml
pull:
- prefect_gcp.deployments.steps.pull_from_gcs:
    id: pull_code
    requires: prefect-gcp>=0.4.3
    bucket: '{{ push_code.bucket }}'
    folder: prefect_practice
    credentials: "{{ prefect.blocks.gcp-credentials.credential }}"

# the definitions section allows you to define reusable components for your deployments
definitions:
  tags: &common_tags
    - "dev"
  work_pool: &gke
    name: "gke"
    work_queue_name: "dev"

# the deployments section allows you to provide configuration for deploying flows
deployments:
- name: "dev: resource allocation with GPU"
  tags: *common_tags
  schedule: null
  entrypoint: "test.py:get_data"
  work_pool: 
    <<: *gke_workpool
    job_variables:
        job_configuration:
          job_manifest:
            spec:
              template:
                spec:
                  containers:
                  - resources:
                      limits: 
                        cpu: "5000m"
                        memory: "4Gi"
                        <http://nvidia.com/gpu|nvidia.com/gpu>: "1"
                      requests: 
                        cpu: "5000m"
                        memory: "4Gi"
                        <http://nvidia.com/gpu|nvidia.com/gpu>: "1"
                  nodeSelector:
                    <http://cloud.google.com/gke-accelerator|cloud.google.com/gke-accelerator>: "nvidia-tesla-t4"
                    <http://iam.gke.io/gke-metadata-server-enabled|iam.gke.io/gke-metadata-server-enabled>: "true"

Did I miss something? Is prefect.serve replacing "prefect deploy" and the prefect.yaml file? I really like the yaml-file and Wizard-style deployment.

Hello all! I am attempting to have a worker deployed on a kubernetes cluster run a few flows for us, but am having some trouble getting it to clone a github repository(followed the guidance here: https://docs.prefect.io/latest/guides/deployment/storage-guide/). Not sure if there is anything else I should do here or if there are any additions needed to the script maybe?

Hi friends. I'm using Prefect2 cloud and trying to follow Laura's video on getting a hello world deployment via GH block to a k8s worker pool. I can create a new deployment in my local env and run a flow, but whenever I move to a GHA workflow (I even checked pip freeze and all look the same), I see this error:

Traceback (most recent call last):
  File "/opt/hostedtoolcache/Python/3.10.13/x64/bin/prefect", line 5, in <module>
    from prefect.cli import app
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/prefect/__init__.py", line 37, in <module>
    from prefect.runner import Runner, serve
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/prefect/runner.py", line 46, in <module>
    from prefect.client.orchestration import get_client
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/prefect/client/orchestration.py", line 17, in <module>
    import prefect.settings
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/prefect/settings.py", line 639, in <module>
    PREFECT_CLOUD_URL = Setting(
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/prefect/settings.py", line 120, in __init__
    generate_deprecation_message(
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/prefect/_internal/compatibility/deprecated.py", line 63, in generate_deprecation_message
    parsed_start_date = pendulum.from_format(start_date, DEPRECATED_DATEFMT)
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/pendulum/__init__.py", line 259, in from_format
    parts = _formatter.parse(string, fmt, now(), locale=locale)
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/pendulum/__init__.py", line 211, in now
    dt = _datetime.datetime.now(local_timezone())
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/pendulum/tz/__init__.py", line 60, in local_timezone
    return get_local_timezone()
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/pendulum/tz/local_timezone.py", line 35, in get_local_timezone
    tz = _get_system_timezone()
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/pendulum/tz/local_timezone.py", line 63, in _get_system_timezone
    return _get_unix_timezone()
  File "/opt/hostedtoolcache/Python/3.10.13/x64/lib/python3.10/site-packages/pendulum/tz/local_timezone.py", line 242, in _get_unix_timezone
    raise RuntimeError("Unable to find any timezone configuration")
RuntimeError: Unable to find any timezone configuration
Error: Process completed with exit code 1.

I know this stack is huge, but I'm completely at a loss why my agent is crashing .. here is the traceback:

12:51:35.373 | INFO    | prefect.runner - Pausing schedules for all deployments...
12:52:40.797 | INFO    | prefect.runner - All deployment schedules have been paused!
Traceback (most recent call last):
  File "/app/scripts/jobs.py", line 140, in <module>
Failed the last 3 attempts. Please check your environment and configuration.
Examples of recent errors:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/h2/connection.py", line 224, in process_input
    func, target_state = self._transitions[(self.state, input_)]
KeyError: (<ConnectionState.CLOSED: 3>, <ConnectionInputs.SEND_HEADERS: 0>)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/http2.py", line 106, in handle_async_request
    await self._send_request_headers(request=request, stream_id=stream_id)
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/http2.py", line 203, in _send_request_headers
    self._h2_state.send_headers(stream_id, headers, end_stream=end_stream)
  File "/usr/local/lib/python3.10/dist-packages/h2/connection.py", line 766, in send_headers
    self.state_machine.process_input(ConnectionInputs.SEND_HEADERS)
  File "/usr/local/lib/python3.10/dist-packages/h2/connection.py", line 228, in process_input
    raise ProtocolError(
h2.exceptions.ProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in state ConnectionState.CLOSED
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/httpx/_transports/default.py", line 60, in map_httpcore_exceptions

d
  File "/usr/local/lib/python3.10/dist-packages/httpx/_transports/default.py", line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/connection_pool.py", line 253, in handle_async_request
    raise exc
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/connection_pool.py", line 237, in handle_async_request
    response = await connection.handle_async_request(request)
    serve(
  File "/usr/local/lib/python3.10/dist-packages/prefect/utilities/asyncutils.py", line 255, in coroutine_wrapper
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/connection.py", line 90, in handle_async_request
    return await self._connection.handle_async_request(request)
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/http2.py", line 142, in handle_async_request
    raise LocalProtocolError(exc)  # pragma: nocover
httpcore.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in state ConnectionState.CLOSED
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/prefect/utilities/services.py", line 53, in critical_service_loop
    await workload()
  File "/usr/local/lib/python3.10/dist-packages/prefect/runner.py", line 501, in _get_and_submit_flow_runs
    runs_response = await self._get_scheduled_flow_runs()
  File "/usr/local/lib/python3.10/dist-packages/prefect/runner.py", line 614, in _get_scheduled_flow_runs
    scheduled_flow_runs = await self._client.read_flow_runs(
  File "/usr/local/lib/python3.10/dist-packages/prefect/client/orchestration.py", line 1778, in read_flow_runs
    response = await <http://self._client.post|self._client.post>("/flow_runs/filter", json=body)
  File "/usr/local/lib/python3.10/dist-packages/httpx/_client.py", line 1848, in post
    return await self.request(
  File "/usr/local/lib/python3.10/dist-packages/httpx/_client.py", line 1533, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/usr/local/lib/python3.10/dist-packages/prefect/client/base.py", line 256, in send
    response = await self._send_with_retry(
  File "/usr/local/lib/python3.10/dist-packages/prefect/client/base.py", line 190, in _send_with_retry
    response = await request()
  File "/usr/local/lib/python3.10/dist-packages/httpx/_client.py", line 1620, in send
    response = await self._send_handling_auth(
    return call()
  File "/usr/local/lib/python3.10/dist-packages/prefect/_internal/concurrency/calls.py", line 382, in __call__
  File "/usr/local/lib/python3.10/dist-packages/httpx/_client.py", line 1648, in _send_handling_auth
    response = await self._send_handling_redirects(
  File "/usr/local/lib/python3.10/dist-packages/httpx/_client.py", line 1685, in _send_handling_redirects
    response = await self._send_single_request(request)
  File "/usr/local/lib/python3.10/dist-packages/httpx/_client.py", line 1722, in _send_single_request
    response = await transport.handle_async_request(request)
  File "/usr/local/lib/python3.10/dist-packages/httpx/_transports/default.py", line 352, in handle_async_request
    with map_httpcore_exceptions():
  File "/usr/lib/python3.10/contextlib.py", line 153, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.10/dist-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in state ConnectionState.CLOSED
    return self.result()
  File "/usr/local/lib/python3.10/dist-packages/prefect/_internal/concurrency/calls.py", line 282, in result
    return self.future.result(timeout=timeout)
  File "/usr/local/lib/python3.10/dist-packages/prefect/_internal/concurrency/calls.py", line 168, in result
    return self.__get_result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.10/dist-packages/prefect/_internal/concurrency/calls.py", line 345, in _run_async
    result = await coro
  File "/usr/local/lib/python3.10/dist-packages/prefect/runner.py", line 989, in serve
    await runner.start()
  File "/usr/local/lib/python3.10/dist-packages/prefect/runner.py", line 283, in start
    async with self._loops_task_group as tg:
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 597, in __aexit__
    raise exceptions[0]
  File "/usr/local/lib/python3.10/dist-packages/prefect/utilities/services.py", line 121, in critical_service_loop
    raise RuntimeError("Service exceeded error threshold.")
RuntimeError: Service exceeded error threshold.

Is there any way to suppress the warning: `

/usr/lib/python3.10/runpy.py:126: RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; this may result in unpredictable behaviour
  warn(RuntimeWarning(msg))

it's driving me nuts in my log files

Hello everyone! 👋 Has someone experienced an endless retries and rerunning of tasks? I have a flow with 10 tasks, which run simultaneously with

.submit()

, and final task with

wait_for

for all previous tasks. I also have decorator

@task

with setting

retries=0

. In addition, I explicitly switched global Prefect settings like this:

prefect config set PREFECT_FLOW_DEFAULT_RETRIES=0
prefect config set PREFECT_TASK_DEFAULT_RETRIES=0

Nevertheless, I still can see

retrying

and

rerunning

states of the tasks. It's happening without any messages in logs. Some tasks cannot actually finish because of "repeating". But I want tasks to be executed only once. Did I do something wrong? Some pictures will be below ⤵️

Hello everyone ! I have this error when I try to deploy a flow using

prefect deploy

:

prefect.exceptions.ScriptError: Script at 'xxxxxxxx.py' encountered an exception: PrefectHTTPStatusError("Client error '401 Unauthorized' for url '<https://api.prefect.cloud/api/accounts/xxxxxxx/workspaces/xxxxxx/block_types/slug/gcp-credentials/block_documents/name/bq-load-creds?include_secrets=true>'\nResponse: {'detail': 'Invalid authentication credentials'}\nFor more information check: <https://httpstatuses.com/401>"

It seems that there is an issue with the gcp-credential block. I also have the same issue with other types of blocks and on flows that I successfully deployed in the past... (I'm using Prefect Cloud) Can somebody help me please 😞?

Quick Question! When I am running prefect deploy --all, it runs the build and push for every deployment. How can those run once and then deploy the flows to Prefect?

I was hoping to programmatically define my work pools so we don't need to update them in the UI. But it looks like the only options are the UI or the API. The CLI can create work pools but not edit their configuration, if I'm reading this right. Is it more typical to just define these via the UI instead?

Hi , I ran a flow a couple of days ago, which runs a databricks job, and wait until it finishes. In the meanwhile it is running, the flow died within this error:

Traceback (most recent call last):
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/prefect/engine.py", line 1729, in orchestrate_task_run
    result = await call.aresult()
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/prefect/_internal/concurrency/calls.py", line 291, in aresult
    return await asyncio.wrap_future(self.future)
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/prefect/_internal/concurrency/calls.py", line 345, in _run_async
    result = await coro
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/prefect_databricks/jobs.py", line 1170, in jobs_runs_get
    response = await execute_endpoint.fn(
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/prefect_databricks/rest.py", line 134, in execute_endpoint
    response = await getattr(client, http_method)(
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_client.py", line 1757, in get
    return await self.request(
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_client.py", line 1530, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_client.py", line 1617, in send
    response = await self._send_handling_auth(
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_client.py", line 1645, in _send_handling_auth
    response = await self._send_handling_redirects(
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_client.py", line 1682, in _send_handling_redirects
    response = await self._send_single_request(request)
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_client.py", line 1719, in _send_single_request
    response = await transport.handle_async_request(request)
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_transports/default.py", line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/usr/lib/python3.8/contextlib.py", line 131, in __exit__
    self.gen.throw(type, value, traceback)
  File "/home/sgonzalez/venv/lib/python3.8/site-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.ConnectTimeout
07:34:11 PM
jobs_runs_get-173
prefect.task_runs
Finished in state Failed('Task run encountered an exception ConnectTimeout: ')

The databricks job still runs, but the flow had already finished.

I have a scenario where have a flow that queries ~ 300 apis as separate tasks. Because 1 or 2 of these tasks fail randomly and intermittently on most flow runs we have found our alerting very noisy and want to resolve this. Our solution we want is that we want to only send an alert to slack if THE SAME task fails on consecutive flow runs. Is there any way to introspect the prefect cloud state programatically (CLI or python client) to work out which tasks failed for a given flow run?