Hi there, How to stop using inactive AWS ECS TaskDefinition revision by a Prefect agent? The Flow has been failing every time and cannot register another ECS Task Definition revision by itself. Thanks!

State Message
Submission failed. botocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameterException) when calling the RunTask operation: TaskDefinition is inactive

Hi everyone. I suppose with Prefect 2 monthly plans based on usage are gone, right? Is new pricing based on only workspaces and users? Finally, what is the exact usage of workspaces? Can we use same workspace for multiple projects?

Hi, thanks for a great product! I’m going through the Prefect First Steps tutorial, and have noticed something funny/wrong about the Task Inputs report for the call_api task run when doing the Basic Flow With Tasks example in the tutorial. While the call_api task is running correctly and showing interactive output as in the tutorial, if you inspect the Task Inputs with the Orion GUI, instead of the expected true url value as in the Parameters for the Flow Run, you see just an empty array, which clearly is wrong because the url was actually fetched and response returned

{"url":[]}

. Is this a bug? I am running Prefect 2.7.12 locally on Python 3.10.9 on sqlite 3.40.1.

Hello, is there any way to get the Flow name from within a running task?

prefect.context.get_run_context()

returns a TaskRunContext when triggered from a running task which does not have Flow name. Thanks!

Hi, I've been having a lot of crazy reliability issues with Prefect lately. As soon as I ran >5 flows on an agent, it starts crashing. The flows are not doing much, polling an API and waiting in between (min 20s intervals) - and it's all using ConcurrentTaskRunner, all functions are async. Why is this happening? Any configuration I can tweak to make this run better? If not, I'll have to move off Prefect, reliability here really is the most important feature

Hello guys , I’m blocked with the problem MissingResult: State data is missing (DaskTaskRunner). When I try to run multiple flows (~30) in the same moment, some flows are failed with this error. It should not be the problem of Dask cluster, because on the same cluster I can run ~50 flows with prefect1 run without any problem! The problem is, that it’s a random error the flow which had this problem previous run, won’t have it on the next one. I add a stack trace in threads I see on GitHub that problem exists is more than one year: MissingResult State data is missing. · Issue #67 · PrefectHQ/prefect-dask · GitHub Any suggestions? I’m on the last prefect 2.7.12

When it comes to deployments, what strategies are people using to ensure only required files are loaded? I've got a directory full of flows that I want to deploy a single flow from but it would be nice if I wasn't copying the entire directory just for a single flow

Hello hello, i am running into this error in Prefect 1 when using

get_task_run_result

from a dependent flow. The child flow runs and is successful, but the parent flow seems to be unaware of that? we are using local storage to persist task results. this error only happens sometimes.

Task 'get_task_run_result': Exception encountered during task execution!
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 880, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/usr/local/lib/python3.9/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/usr/local/lib/python3.9/site-packages/prefect/tasks/prefect/flow_run.py", line 239, in get_task_run_result
    return task_run.get_result()
  File "/usr/local/lib/python3.9/site-packages/prefect/backend/task_run.py", line 73, in get_result
    raise ValueError("The task result cannot be loaded if it is not finished.")
ValueError: The task result cannot be loaded if it is not finished.

Task 'get_task_run_result': Finished task run for task with final state: 'Failed'

Has anyone encountered the below warning while registring flows using the cache_for option Task has cache settings but does not have a result type I am using the cache_for option in the create_flow_run as below Flowname = create_flow_run( flow_name=FlowName_Exporter, project_name=ProjectName, task_args={"name":"Test Flow","trigger":"all_successful","cache_for":datetime.timedelta(hours=2)}

Hello I run flows in dask cluster. How it can be possible, the flow state is

Completed

, orion graph has

red parts

, and all my subtasks

crashed

? And in logs I see

crashed

. Maybe it changes smth: • I submit with options every task and I add str(uuid.uuid4())[:8] before every task name. • error:

Crash detected! Execution was interrupted by an unexpected exception: concurrent.futures._base.CancelledError: 2ae637b0currency_symbol-26-99893272733d461fba8a3051f42a13a8-1

Hi, I am encountering something new/weird with result-caching. In the past I could just delete an existing flow-run and start a new flow-run with the same parameters and it would recompute everything. My understanding was/is that by deleting the flow-run also all cache-entries are deleted. Now I have tried this approach again, but the cache-entries are still there i.e. all tasks are marked as cached. The problem I am facing is the following: I have deleted all entries in the cache-storage location and now prefect complains that it can't find the cached result. I could set the cache_refresh env variable to force recomputation, but I wanted to ask how this actually works? Are the cache-entries deleted when I delete a flow-run and it just takes some time? Or are they not deleted and I should not touch the cache-storage location?

Hi, I am using python to run flow from deployment with timeout=0 (prefect.deployment.run_deployment) , I like to inspect the flow run status, what API should I use?

Recently my agents (2.7.9) have been unable to push logs to the prefect API and display them in the prefect cloud UI. Anyone know what might be causing this?

--- Orion logging error ---
Traceback (most recent call last):
  File "/root/micromamba/envs/prefect/lib/python3.7/site-packages/prefect/logging/handlers.py", line 151, in send_logs
    await client.create_logs(self._pending_logs)
  File "/root/micromamba/envs/prefect/lib/python3.7/site-packages/prefect/client/orion.py", line 1843, in create_logs
    await <http://self._client.post|self._client.post>(f"/logs/", json=serialized_logs)
  File "/root/micromamba/envs/prefect/lib/python3.7/site-packages/httpx/_client.py", line 1855, in post
    extensions=extensions,
  File "/root/micromamba/envs/prefect/lib/python3.7/site-packages/httpx/_client.py", line 1527, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/root/micromamba/envs/prefect/lib/python3.7/site-packages/prefect/client/base.py", line 253, in send
    response.raise_for_status()
  File "/root/micromamba/envs/prefect/lib/python3.7/site-packages/httpx/_models.py", line 736, in raise_for_status
    raise HTTPStatusError(message, request=request, response=self)
httpx.HTTPStatusError: Client error '429 Too Many Requests' for url '<https://api.prefect.cloud/api/accounts/><redacted>/workspaces/<redacted>/logs/'
For more information check: <https://httpstatuses.com/429>
Worker information:
    Approximate queue length: 8
    Pending log batch length: 3
    Pending log batch size: 1128

What could be possible reasons for agent to not log any logs? My agent just went to sleep. like, no logs , no errors , just my flows suddenly going into late state and on inspection i saw, no logs were recorded during that time. i paused my workqueue and suddenly the agent is back

Hi Community, I am new to Prefect 2. Have a question regarding creating service accounts using Rest API. Can someone point me to any documentation?

Hello, I'm trying to run a KubernetesJob with a PRIVATE image from dockerhub, but I have not success with that 😕 I'm using prefect 2.7.11. I'm also using prefect-orion and prefect-agent helm charts. I add my docker-cred secret using agent.image.pullSecrets. I have tried creating my deployment with two options: • a KubernetesJob

infrastructure=KubernetesJob(
        namespace="prefect-orion",
        image="<http://docker.io/myrepo/myimage:tag|docker.io/myrepo/myimage:tag>"
    ),

but I'm getting this error:

Flow run infrastructure exited with non-zero status code -1.

• a DockerContainer I passed my credentials directed in a Docker Registry and I created a DockerContainer. but I'm getting this error:

State Message
Submission failed. Traceback (most recent call last): File "/usr/local/lib/python3.10/site-packages/urllib3/connectionpool.py", line 703, in urlopen httplib_response = self._make_request( File "/usr/local/lib/python3.10/site-packages/urllib3/connectionpool.py", line 398, in _make_request conn.request(method, url, **httplib_request_kw) File "/usr/local/lib/python3.10/http/client.py", line 1282, in request self._send_request(method, url, body, headers, encode_chunked) File "/usr/local/lib/python3.10/http/client.py", line 1328, in _send_request self.endheaders(body, encode_chunked=encode_chunked) File "/usr/local/lib/python3.10/http/client.py", line 1277, in endheaders self._send_output(message_body, encode_chunked=encode_chunked) File "/usr/local/lib/python3.10/http/client.py", line 1037, in _send_output self.send(msg) File "/usr/local/lib/python3.10/http/client.py", line 975, in send self.connect() File "/usr/local/lib/python3.10/site-packages/docker/transport/unixconn.py", line 30, in connect sock.connect(self.unix_socket) FileNotFoundError: [Errno 2] No such file or directory During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.10/site-packages/requests/adapters.py", line 489, in send resp = conn.urlopen( File "/usr/local/lib/python3.10/site-packages/urllib3/connectionpool.py", line 787, in urlopen retries = retries.increment( File "/usr/local/lib/python3.10/site-packages/urllib3/util/retry.py", line 550, in increment raise six.reraise(type(error), error, _stacktrace) File "/usr/local/lib/python3.10/site-packages/urllib3/packages/six.py", line 769, in reraise raise value.with_traceback(tb) File "/usr/local/lib/python3.10/site-packages/urllib3/connectionpool.py", line 703, in urlopen httplib_response = self._make_request( File "/usr/local/lib/python3.10/site-packages/urllib3/connectionpool.py", line 398, in _make_request conn.request(method, url, **httplib_request_kw) File "/usr/local/lib/python3.10/http/client.py", line 1282, in request self._send_request(method, url, body, headers, encode_chunked) File "/usr/local/lib/python3.10/http/client.py", line 1328, in _send_request self.endheaders(body, encode_chunked=encode_chunked) File "/usr/local/lib/python3.10/http/client.py", line 1277, in endheaders self._send_output(message_body, encode_chunked=encode_chunked) File "/usr/local/lib/python3.10/http/client.py", line 1037, in _send_output self.send(msg) File "/usr/local/lib/python3.10/http/client.py", line 975, in send self.connect() File "/usr/local/lib/python3.10/site-packages/docker/transport/unixconn.py", line 30, in connect sock.connect(self.unix_socket) urllib3.exceptions.ProtocolError: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory')) During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.10/site-packages/docker/api/client.py", line 214, in _retrieve_server_version return self.version(api_version=False)["ApiVersion"] File "/usr/local/lib/python3.10/site-packages/docker/api/daemon.py", line 181, in version return self._result(self._get(url), json=True) File "/usr/local/lib/python3.10/site-packages/docker/utils/decorators.py", line 46, in inner return f(self, *args, **kwargs) File "/usr/local/lib/python3.10/site-packages/docker/api/client.py", line 237, in _get return self.get(url, **self._set_request_timeout(kwargs)) File "/usr/local/lib/python3.10/site-packages/requests/sessions.py", line 600, in get return self.request("GET", url, **kwargs) File "/usr/local/lib/python3.10/site-packages/requests/sessions.py", line 587, in request resp = self.send(prep, **send_kwargs) File "/usr/local/lib/python3.10/site-packages/requests/sessions.py", line 701, in send r = adapter.send(request, **kwargs) File "/usr/local/lib/python3.10/site-packages/requests/adapters.py", line 547, in send raise ConnectionError(err, request=request) requests.exceptions.ConnectionError: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory')) During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.10/site-packages/prefect/infrastructure/docker.py", line 89, in _get_docker_client docker_client = docker.from_env() File "/usr/local/lib/python3.10/site-packages/docker/client.py", line 96, in from_env return cls( File "/usr/local/lib/python3.10/site-packages/docker/client.py", line 45, in __init__ self.api = APIClient(*args, **kwargs) File "/usr/local/lib/python3.10/site-packages/docker/api/client.py", line 197, in __init__ self._version = self._retrieve_server_version() File "/usr/local/lib/python3.10/site-packages/docker/api/client.py", line 221, in _retrieve_server_version raise DockerException( docker.errors.DockerException: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory')) The above exception was the direct cause of the following exception: RuntimeError: Could not connect to Docker.

anyone have an idea how to pass docker credentials to be able use a private docker image in a KubernetesJob? 😕 thanks

Hello, Today tryed update to 2.8.0 and got alembic error:

alembic.script.revision.MultipleHeads: Multiple heads are present for given argument 'head'; 898a4c563127, cfdfec5d7557

alembic.util.exc.CommandError: Multiple head revisions are present for given argument 'head'; please specify a specific target revision, '<branchname>@head' to narrow to a specific head, or 'heads' for all heads

Anyone have an idea how to fix it?

Hi everyone, I am running prefect with docker with containers like agent, Orion etc. In the agent i have mentioned 2 queues. The problem is If i make a single inferencing request everything works file. But, if i trigger multiple inferencing request, queue goes unhealthy. Can anyone tell me what I am doing wrong here.

Hi everyone, I am running a flow which has 2 subflows in it. 2nd subflow is supposed to end the flow but it keeps running even after completing all the tasks. Both subflows are taking pandas dataframes from the flow. Is there something I missed?

is it possible to call flow from another flow using run_deployment but not set it as a subflow (parent flow id taken from context)?

I have a flow that has

retries=3

. One of my tasks returns a

SnowflakeConnector

that gets passed to a few other tasks that perform queries on the db. If one of the tasks fails, the flow is retried and throws the following exception:

pydantic.error_wrappers.ValidationError: 1 validation error for ParsingModel[SnowflakeConnector]
__root__ -> schema
  field required (type=value_error.missing)

This is probably because of the

SecretStr

field in the

SnowflakeConnector

that holds the password. In

~/.prefect/storage

I can see the persisted results of the tasks and secret strings are obviously obfuscated. How is the retry mechanism supposed to work if my tasks work with SecretStr of SnowflakeConnections?

Hello, currently migrating from prefect v1 to v2 and I'm trying to find the way of passing this argument to a generic task as I was doing it on v1. Is it possible on v2?

task_args={"name": "current-task-run-name"}

Hi all, I am trying to retrieve the result (what is returned) after executing a local docker deployment. As an example, what I get is type='reference' serializer_type='pickle' storage_block_id=UUID('201be8bb-b742-4645-9b7c-6cdcce007d7c') storage_key='36cbbc4e543649dabe195c3ebde33344') if I ask for state.result() or Completed() if i ask for state. Is there any example I can check? (FYI I use persist_result=True and return_state=True)

I posted this a while ago about

timeout_seconds

not working properly. I expect that after the timeout elapses and the flow is still in a Running state, the flow should be marked as failed. However, if the agent is not able to talk to the flow any more (maybe the pod got evicted) then its state will not be updated any more. Is this expected behaviour or can I log an issue on this one?

Hey guys, question - we’re wondering if we have a limits of tags. We plan to use them to optimize some concurrency issues we have, but we’d need to create tags for each flow that involves an user. Is that doable?

Hi all, I've been having some issues with mapping tasks. I'm only mapping 3 inputs, however the tasks seem to hang or execute very slowly. I've seen some threads previously about issues with using

.map()

- however I am on prefect version

2.7.10

and figured this wouldn't be an issue, wondering if I am missing some sort of configuration somewhere

Is there a way to set the flow concurrency limit to 1? Task level & Queue level concurrency do not work for us. It would be very unproductive to have to create many queues and agents for singleton flows. I saw some threads related to the similar question but I could not find what I want.

How can I cancel a flow from within a task? Right now if I return

Cancelled()

from within a task it only cancels the task, not the entire flow.

Hey, have a question. Have a flow that I am currently running with a

DaskTaskRunner

that and multiple tasks. Prior to setting a

max_runtime

the flow would crash after 90 minutes on whatever task it is currently on but instead of the flow entering the failed state it shows as completed. Is this by design or a bug? If it is by design, is there a way to set the flow to instead enter the failed state to keep incomplete ETLs from being missed?

Is there a way to automatically

set_schedule_inactive

for a deployment when a flow run fails in 2.0?