I have cloud server, kubernetes AKS agent and a flow in Azure storage with a customised container image in azure container registry. When I run the flow it fails on authorisation 401. I have supplied the password to the image in the imagepullsecrets env variable. The error is : Kubernetes Error: rpc error: code = Unknown desc = failed to pull and unpack image "prefregistry.azurecr.io/mstools:v1": failed to resolve reference "prefregistry.azurecr.io/mstools:v1": failed to authorize: failed to fetch anonymous token: unexpected status: 401 Unauthorized.

Currently wrapping a function with

@task

removes all type signatures that the function was given and a bare

FunctionTask

is returned. Any thoughts of improving the

@task

decorator such that all type signatures persist? That would help tremendously with type-checking for flows. Happy to contribute too, if such a proposal is somewhat sensible.

I'm having some issues using the

Github

storage. I'm configuring it like

GitHub(
        repo="some_org/repo_name",
        path="path/to/flow.py",
        access_token_secret="secret_name", 
        ref="stg"
)

When I execute the flow I get the following failure:

Failed to load and execute Flow's environment: GithubException(422, {'message': 'No commit found for SHA: stg', 'documentation_url': '<https://docs.github.com/rest/reference/repos#get-a-commit'}>).

That repo definitely has the branch, do I need to reference it in a special way?

Hi All, I’m having a strange problem executing SQL command within tasks in a Flow. To summerise, I’m using sqlalchemy to execute those commands for me on the database. the issue is that commands are getting executed with no error returned, but later looking at the database, none of those commands went into effect (as if the were not executed at all). Surprsingly, another task within the same Flow is able to successfully utilize the engine (instantiated the same way), when pandas is using that to dump a dataframe into the database. I did a test on the docker environment, where the tasks are executed, and surprisigly, all those sql command are getting executed successfully. I’m a little puzzled as to why a task can utilize sqlalchemy engine and another task fails to do so? please help.

Hi all, I want to make a Flow where I run certain tasks on different agents (some on Local, some on Kubernetes). For this purpose I've understood I need to wrap my Tasks in Flows, which have different run_configs depending on where they are run. I now have one parent Flow from which I launch child Flows using StartFlowRun. My question is how do I pass Parameters to these child Flows?

extract_flow_task = StartFlowRun(flow_name=extract_flow.name, wait=True, project_name=PROJECT_NAME)
transform_flow_task = StartFlowRun(flow_name=transform_flow.name, wait=True, project_name=PROJECT_NAME)

#
# Parent flow, executed on Kubernetes agent
#
with Flow("parent_flow", storage=storage, run_config=KubernetesRun(labels=["kubernetes"])) as parent_flow:
    date_YYYYmmdd = Parameter("date_YYYYmmdd")
    extract_task = extract_flow_task() # How to pass date_YYYYmmdd to this flow/task?
    transform_flow_task(upstream_tasks=[extract_task]) # How to pass date_YYYYmmdd to this flow/task?

Hey community, would there be a way to set an

Automation

for notifications for all flows except

flow_name

as an option ? Or would I need to do something custom to do that?

Hi everyone! I’d like to mark that a task is dependent from the successful completion of another task. The docs (https://docs.prefect.io/core/concepts/tasks.html#triggers) only show examples about how to add a

trigger

in the

run()

method, but these tasks are written using the task decorator only.

@task
def parent():

@task
def child():

What are the best next steps?

Hi Everyone, I'm using the prefect-agent (Docker agent) within Azure as VM. And I connected it to my Prefect Cloud, but I'm unsure how to send my flow triggers to the Docker Agent. So far all my registered flows still go to the Local Agent, but when I'm not inside my VM the Local Agent goes offline, while it looks like the Docker Agent is always online (keeps pinging it every 10-ish seconds with heartbeat) My simple flow is inside the 2nd image. I also tried using, but the flow still went to my Local agent My DockerAgent is using the 'prod' label.

from prefect import task, Flow
from prefect.run_configs import DockerRun


@task
def say_hello(x=1):
    return f"Test {x}'


with Flow("hello-world") as flow:
    first = say_hello()
    second = say_hello(2)

flow.config_run = DockerRun()
flow.register(project_name='test', labels=['prod'])

So question: How do I send the flow created inside my Azure VM to the VM's DockerAgent using the Prefect Cloud orchestration?

Hi, I have a question about Prefect agents and Flows : is there something in Prefect such as Kubernetes'

resources

parameters to specify resource affinities (RAM, CPU cores, GPU environment) a Flow requires to be run ? Ex (Kubernetes) :

resources:
      requests:
        memory: "64Mi"
        cpu: "250m"
      limits:
        memory: "128Mi"
        cpu: "500m"

Thanks !

Hey all, i'm struggling to connect to our Azure Devops Repo using the GIT storage class. I keep getting an HTTP Unauthorized error. I've tried connecting with git credentials and with a PAT and neither work. Has anyone else managed to connect to AZ Devops?

Does anyone know if it is possible to kick off a script in a different git repo in a prefect flow?

hi prefect! I"m struggling to use the

DbtShellTask

along with secrets (more details in thread)

Hi. we are using prefect agents with dask executors... after some time (<1day) log streaming from tasks (logger = prefect.context.get("logger"); logger.info(text)) stops to appear in the cloud flow log.... the log can be still seen in the dask worker log on the machine ... is streaming breaking? is it a rate limit? any pointer how to debug that? (cc @Connor Martin )... though there is a tiny bit of log until it runs on dask ... Flow successfully downloaded. Using commit: YYY Beginning Flow run for 'XXX' Connecting to an existing Dask cluster at tcp://localhost:8786 ... no content here which is the problem ... Flow run FAILED: some reference tasks failed.

Hi there, im trying to build a flow where at the beginning of the flow I have a task that creates a new database then I want to pass that to all of the remaining tasks. (ie one task preps the stage and then all other tasks use that) Is there a way to do this with the context? Can have that task write to the context and then have all my other tasks reference that?

Hi everyone, There is an issue where the first task in the flow is re-running before finishing. This happens 3-4 times and then the run fails with

KilledWorker

error. It keeps restarting the first task. Do you know what the issue could be?

I have a quick question about logging that I can't quite find in the docs. I have a private package that I pip install into my docker environment that runs my Prefect flows. I want to see the logging output from the code run inside that private package on the UI for prefect cloud. It currently shows up on the agent side, but I can't seem to figure out how to get it to log to the UI. Since it is in another package that is used other places it doesn't make sense to use the Prefect logger like I do for code running in the flows. I am using loguru and just writing to stdout. Any easy way to make these logs appear in the UI?

I have a flow that runs every minute - at one point my agent went down and now there are 15 that are in "running" state. Is there a way to cancel them all?

Hi everyone, All my prefect runs are failing with a same error when interacting with any Google Python APIs (ex:

google-cloud-secret-manager

,

google-cloud-logging

, etc.). All the Google python packages are running into a grpc transport error. It may be a package version issue or maybe some conflict.

Hi, anyone have an example using prefect to extract batches of data without knowing the size of the data? Kind of recursive fetching until no more?

Hello Prefect Land! Simple request. I have a notification (new file upload) set up in AWS S3 that goes to SNS. From there, it branches to send an e-mail and also kick off a lambda function that kicks off a flow in Prefect. When I receive the e-mail, I can see the entire file path in S3. How can I get that same info over to my Prefect flow so I can parameterize? I hope I’m making sense… Best All!

Hi, I deployed an ECS agent as a fargate service, i use run_config as ECSRun but when the agent creates the task definition it creates it with EC2 compatibility, and flow fails with exception “task not support fargate launc type” Any suggestions? Am i missing something?

Does anyone else have an issue where the dashboard isn't properly rendering in Google Chrome?

Hi all - we’ve got one system on which the workflows which we’re kicking go into a state of “submitted” but don’t seem to go any further. It just started happening recently, and I’m not sure what might have caused it. Will investigate today but are there any immediate gotchas that we should be looking for?

Hi! My Flow schematic looks funny. I am using StartFlowRun to launch sub-flows. Basically everything works, but the schematic and logs look... funny. Am I doing something wrong?

Hey all, I was wondering if anyone had experience with the

local_agent

slowly leaking memory? Over a day or two the memory usage of our local agent service will grow until its out and then we run into exceptions when new flows try to be scheduled. It might not be a memory error, but our 3GB memory cap on the service seems to be not be sufficient.

@Kevin Kho another different question from me - but is this a known bug in prefect? My dashboard commonly shows inaccurate numbers. 12 failed flows in the past 24 hours, but also 50.3% of 559 failed, which is >>12. The confusing part is we have slack notifications turned on for all flows, and we’ve received only 2 notifications in the last 24 hours, and Im not sure who to trust. 2 failures, 12 failures, or 281 failures.

I’m trying to construct a schedule with

parameter_defaults

, where the defaults are derived from the date that the schedule is triggered on. In other words, I’m looking for something like this:

schedule = Schedule(
    clocks=[
        IntervalClock(
            interval=timedelta(days=1),
            parameter_defaults={"version": date.today().isoformat()},
        ),
    ],
...

Except I don’t want

date.today().isoformat()

to be calculated just once when the flow is registered and then fixed for all time. I want it to be calculated at the time the clock fires. How do I do this? Pass in a lambda as the value in the

parameter_defaults

dictionary? Construct a dict-like object that calculates the value on key lookup? I feel like I’m thinking about this incorrectly.

Hi All, Is there a mechanism to apply on a task so that we would know if a task (or every task if that matters) within a flow is stuck in a specific state for a prolonged amount of time?

Wondering if anyone can help with the following use case. I have a project that contains a number of flows, which I would like to register to an internal backend using a json format, but I also need to run the build in a CI/CD system (which includes multiple environments), this means that I need to configure the

run_config

and

storage

attributes for each flow before registration.

Is it possible to cleanly replicate the

prefect build ...

CLI command directly within a custom python script