Hi - Just upgraded to 2.7.1 and my k8's job runners are reporting the following:

/usr/local/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

Hello I am getting the following error in job runs that take place in kubernetes. I guess that the error happens when the job fails to get the status of a task. After that error in the job, the task is being killed even though it was running without problems. I think that this error occurs when I run 3 or more jobs concurrently but this should not be a load to my nodes. Does anybody know what could be its cause and how could it be fixed?

Task 'TaskPreprocessing': Exception encountered during task execution!
Traceback (most recent call last):
  File "my-file.py", line 105, in run_job
    job.RunNamespacedJob(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/tasks/kubernetes/job.py", line 730, in run
    job = api_client_job.read_namespaced_job_status(
  File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api/batch_v1_api.py", line 1393, in read_namespaced_job_status
    return self.read_namespaced_job_status_with_http_info(name, namespace, **kwargs)  # noqa: E501
  File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api/batch_v1_api.py", line 1480, in read_namespaced_job_status_with_http_info
    return self.api_client.call_api(
  File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api_client.py", line 348, in call_api
    return self.__call_api(resource_path, method,
  File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api_client.py", line 180, in __call_api
    response_data = self.request(
  File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api_client.py", line 373, in request
    return self.rest_client.GET(url,
  File "/usr/local/lib/python3.8/site-packages/kubernetes/client/rest.py", line 239, in GET
    return self.request("GET", url,
  File "/usr/local/lib/python3.8/site-packages/kubernetes/client/rest.py", line 233, in request
    raise ApiException(http_resp=r)
kubernetes.client.exceptions.ApiException: (500)
Reason: Internal Server Error
HTTP response headers: HTTPHeaderDict({'Audit-Id': '<Audit-Id>', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': '<X-Kubernetes-Pf-Flowschema-Uid>', 'X-Kubernetes-Pf-Prioritylevel-Uid': '<X-Kubernetes-Pf-Prioritylevel-Uid>', 'Date': 'Tue, 13 Dec 2022 11:02:40 GMT', 'Content-Length': '150'})
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"rpc error: code = Unavailable desc = transport is closing","code":500}

Hi all! My organization is among other things using Prefect to construct and maintain semantic Knowledge Graphs. Is there anybody else working with RDF, SPARQL and the works? We've been compiling a task library and depending on the interest, make it more accessible to others.

Hello, I am trying to define dependencies between my flows in Prefect 2.0 (e.g. don't run script B until script A finishes). I am running each flow as a Fargate task. Ideally, I want to be able to continue to run the flows as separate Fargate tasks but still define dependencies. The only method i've seen in the documentation is using sub flows. But I think in this case, all flows would need to be run from 1 image. Is there a recommendation on how to define dependencies while still building separate images/tasks?

Hey Everyone! Does somebody know if we can

.submit

a task and use one of the parameters as the task name? I know we can currently do

my_task.submit(name="My Task")

and this will show up as

my_task-a1b2c3-0

in the UI, but what I would like to do is:

from prefect import flow, task

@flow(name="My Flow")
def my_flow():
    my_task("Hello")
    my_task(name="World")

@task(name_from_param="name")
def my_task(name):
    pass

Then in the Prefect UI I will see:

Running Flow: My Flow
Running Task: Hello-a1b2c3-0
Running Task: World-a1b2c3-1

A bit of background for context, we run all of our tasks either using AWS Lambda or DBT. For Lambda, I’ve built a central function (which is decorated as a task) with a signature

run_lambda(lambda_name, payload={})

, and most users are using it like

run_lambda.submit("MyLambdaFunction")

but then in the logs all the task names are

run_lambda-[ascii]-0

which is annoying when things go pop, as you have to click into the task run where I have added

<http://logger.info|logger.info>("Lambda name: %s", lambda_name)

for traceability. For all of my personal Flows I have the following convention

run_lambda.with_options(name="Lambda MyLambdaFunction").submit("MyLambdaFunction")

, which gives me the desired Prefect Task names in the UI, but getting everyone else to follow suit is another matter. I’ve attempted many solutions to dynamically name a task mainly focused on custom decorators, but they have all failed in one way or another… So I wanted to see if the Prefect community has any ideas or solutions! Thanks

Hello, I have 2 questions for prefect v1: • Given a flow running on k8s, is it possible to retrieve its k8s Job name somewhere in prefect ? • How is the heartbeat / zombi killer feature working ? I'm still trying to fix our issue with flows forever stuck in a running state while actually failed, and I don't understand why this feature isn't triggered in our case.

This message was deleted.

Kind of a networking question: • I have the prefect v1 server running locally on my Macbook and can connect to it as normal using localhost:8000 or localhost:8080. • I have a separate computer (raspberry pi4) running that I want to build and run flows on • I want the raspberry pi to be able to register flows to the server running on my macbook over the local network What I need to figure out is how to make it possible for the raspberry pi to make network requests to my prefect server that I’m hosting on my mac book pro.

Hi gang - when migrating to 2, it looks like the deployments to an S3 block upload the entire directory? Whereas the previous behavior with 1's register was just to upload the flow. Can someone clarify this?

Hi there! Quick question regarding Notifications, is it possible to set notification settings on a deployment programmatically, or only in the UI?

Hello, can you share documentation on how to do a deployment build from github actions?

I am trying to run a stitch job using the code exactly how it shows here. https://alessandrolollo.github.io/prefect-stitch/. I'm getting an error that says "TypeError: missing a required argument: 'credentials'".

Hello there, are there any documents or tools which showcase Prefect 1.x to 2.x migrations?

Hi Prefect Community! marvin We'd like to introduce you to @ask-marvin, our new GPT3-powered finely tuned chat bot where you can ask them any questions in #ask-marvin channel. Marvin is still acquiring knowledge and understanding, despite their advanced intelligence and vast experience. Big shout out to @Serina , @Nate, and @Rob Freedy for their incredible engineering skills building Marvin. Here's a few words from Marvin:

EDIT: Join the #ask-marvin to use the @ask-marvin bot. This will help keep the community channel clean while Marvin is still learning marvin

Is there a Prefect 2.0 equivalent for

create_flow_run

in Prefect 1.0 https://docs-v1.prefect.io/api/latest/tasks/prefect.html#create-flow-run ? Trying to translate to prefect 2, and this method

create_flow_run

is somewhat confusing - a task that creates a flow within a flow?

We have been running prefect for quite some time now, but suddenly our kubernetes cluster keeps showing "unhealthy". I have tried the following: 1. Restarted deployment 2. Scaled the number of nodes manually 3. Scaled the size of the nodepool Has this happened to anybody else?

What’s the minimum delay I could expect between triggering a Flow run and the first Task starting execution? And what’s the minimum delay I could expect between a Task finishing execution and the next Task in the Flow starting execution? Context: Every time a new user of my application is created, I want to trigger a Flow run that will scrape this user’s data from multiple sources and populate our database. I need to populate my database within 30 seconds of user creation. This is achievable with my current implementation of orchestrating this pipeline via a python function. I want to make sure that migrating my orchestration logic to Prefect will not add significant time to this process. Ideally I’d keep the additional time under 5 seconds. Is this achievable with Prefect?

Hello all. I have a question about logging: Is there a way to disable console/terminal logging while keeping logs visible on the prefect web UI? I thought something like

PREFECT_LOGGING_HANDLERS_CONSOLE_LEVEL=999 run_flow.py

would do the job, but this seems to turn off the logs on the web UI as well as the console.

Heya! A weird error started popping up where any time I run a Flow with a task that uses Papermill in Prefect Cloud, and there's an Error in the Notebook, Prefect just gives me something along the lines of

RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-0_0'.

instead of propagating the actual error (which it used to do!). It does this both with custom tasks and with the official Jupyter task. This kinda popped up out of left field - any idea what's going on? (it's important cuz we have Slack Alerting based on those errors). On Prefect

1.3.1

btw

Would it be possible to include the workspace name on notifications?

This message was deleted.

Hello everyone, I wanted to know if there are any prefect server graph-ql apis which let me know all the flows that are running currently across all the agents? Would need to call this api from inside a flow. For version 0.15.13

Hey all, I'm having some issues with the ECS task block. Tasks are launching properly but are then hanging for a half hour before executing the flow run. I am not sure if the hanging issue is in the task or the agent. I am running Prefect 2.7.0 and using prefect-aws 0.2.1. I am confused because, in my dev environment (running Prefect 2.6.7 and prefect-aws 0.1.8) the workflows deploy and run properly. Not sure if this is related, but I am also seeing in prod that the prefect agent cannot identify the log stream for new tasks, returning the below error:

botocore.errorfactory.ResourceNotFoundException: An error occurred (ResourceNotFoundException) when calling the GetLogEvents operation: The specified log stream does not exist.

however, the task is generating logs that appear in AWS CloudWatch. Maybe the agent is looking for logs under a different name than the task is outputting them to? I prefer not to revert back to the early versions that appear to be working in the dev environment but I cannot seem to get the newer versions of prefect and prefect-aws to function as expected. @Taylor Curran

🎉 Hey all! You might have seen @Will Raphaelson’s recent message about the exciting automations beta. Will and I will be chatting about automations and showing off some of their cool features during PrefectLive at 3pm ET today on Twitch or YouTube!

Hello Prefect, Is there a thought for providing access controls around who can administer Prefect, deploy flows, run flows, and see logs of flows?

What is the prefect 2 equivalent to

trigger = all_finished

in prefect 1? This thread suggests that it’s using

.submit()

https://discourse.prefect.io/t/how-can-i-trigger-downstream-tasks-based-on-upstream-task-s-state/106 on the task that you want to always run regardless of upstream task failures?

Prefect 2.0 question here - I have three scripts I want to orchestrate in a flow by utilizing the ’shell_run_command” and passing in the command to execute a script. The scripts are not idempotent, and I’d like to cache their completed status if there is a failure between the scripts. Shell_run_command is a ‘task’ object type, and I cannot call a task from within a task to utilize the task cache key. What should I be doing instead?

Hi Prefect community. I've been using Prefect 1.0 with Prefect Cloud and ran into a situation where I can't delete a project through the UI. I was wondering if there's anything obvious that would cause this and if anyone has run into a similar issue. The Interactive API times out when trying to delete via GQL. I deleted all flows in the project but there are quite a few archived flows, and still no luck. Any thoughts?

S The Prefect Slack community has grown by leaps and bounds (now over 23k people!). 🌎 Today we created geographic-based channels so folks could connect with other people nearby and be kept in the loop for community events. We’ve started with locations where we’ve had Prefect Associate Certification courses recently and where we have lots of users. Feel free to join a geo channel if there’s one for your area. 🙂 If there’s isn’t yet a channel for your area, please respond to this thread. If someone has already proposed your area in the thread, please give it a 👍. #geo-nyc #geo-bay-area #geo-dc #geo-chicago #geo-boston #geo-london #geo-berlin #geo-texas #geo-seattle #geo-colorado