if I give a deployment some cron schedule, if one run fails it just keeps running the later ones. is there a way to make the schedule pause so they're guaranteed to run sequentially?

Hi, pls how can we execute multiple flows one after the other programmatically?

Hi all, Has anyone connected multiple server [ / backend / orchestration / orion ] connected to a single Database? Will there be any issues doing that ?

Hey all, I am trying to build a new project with a docker build and a git pull in the prefect.yaml. I am pretty sure I have this right, I am getting stuck on what the dockerfile itself should look like. I want to run on 3.10, make a virtual env to then run all the code on. I am testing locally first with the plan to ecs task later. If anyone could help me figure out the dockerfile build I would be very grateful! Feel free to dm, thanks!

Hey All, I'm running an EKS cluster where I host the Prefect Agent. However, I'm not able to get flows working. I am using an S3 storage block without the credentials set, so it means that the EKS cluster needs to have permission to pull the code from S3. When I trigger a flow execution I get this in the agent logs:

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


Agent started! Looking for work from queue(s): infra-dev-plexflow-2...
23:25:42.455 | INFO    | prefect.agent - Submitting flow run '69af3c9b-f22b-42db-a0a5-21595ec5408a'
23:25:44.631 | INFO    | prefect.infrastructure.kubernetes-job - Job 'flow-infra-data-engineering-infra-dev-96hss': Pod has status 'Pending'.
23:25:44.705 | INFO    | prefect.agent - Completed submission of flow run '69af3c9b-f22b-42db-a0a5-21595ec5408a'
23:26:44.628 | ERROR   | prefect.infrastructure.kubernetes-job - Job 'flow-infra-data-engineering-infra-dev-96hss': Pod never started.
23:26:44.792 | INFO    | prefect.agent - Reported flow run '69af3c9b-f22b-42db-a0a5-21595ec5408a' as crashed: Flow run infrastructure exited with non-zero status code -1.

.. and these are the logs from the job pod..

kubectl logs flow-infra-data-engineering-infra-dev-jhqhm-74k5l --follow
/usr/local/lib/python3.8/runpy.py:127: 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))
23:31:09.701 | INFO    | Flow run 'economic-jerboa' - Downloading flow code from storage at ''
23:36:17.233 | WARNING | aiobotocore.credentials - Refreshing temporary credentials failed during mandatory refresh period.
Traceback (most recent call last):
...
..
..
raise ConnectTimeoutError(endpoint_url=request.url, error=e)
botocore.exceptions.ConnectTimeoutError: Connect timeout on endpoint URL: "<https://sts.eu-west-1.amazonaws.com/>"

.. it gets stuck trying to download the code. The role I have assigned to the cluster has S3 full access. My S3 Block has this configuration:

Bucket Path: <bucket-name>
AWS Access Key Id: None
AWS Secret Access Key: None

There seem to be multiple problems but.. 1. Why does the flow fail right after it starts with

Pod never started.

? 2. And why it cannot pull the code from S3 and the log shows

Downloading flow code from storage at ''

? which is empty

is it possible to map over 2 variables that are lists of equal length? For example, i’d like to get this functionality:

@task
def task(x: str, y: int):
    ...


@flow
def flow():
    list_1, list_2 = ['a', 'b'], [1, 2]
    for input_1, input_2 in zip(list_1, list_2):
        task(input_1, input_2)

Would this be done via:

task.map(list_1 , list_2)

Or will prefect only iterate only one of the parameters?

Hey folks 👋 We are experiencing an interesting behaviour with Prefect 1.4.0 and flows that run with ECSRun config on an ECS Agent. When a flow is cancelled, either from the UI or by calling

client.cancel_flow_run

, it seems the ECS task of the flow run is not stopped and stays in the Running state indefinitely. Has anyone experienced such behaviour? Is this expected or not? Thanks!

With Prefect 2.10.x I can see that the worker pool has a concept of worker type. However, the Kubernetes one is flagged at Beta. Can we use it to replace the usage of Deployment and infrastructure?

When will the docker images for version 2.10.4 be updated on hub.docker.com? The latest tags are for 2.10.3

Hi again, still testing the new projects feature! What's the pull step needed when building a docker image with

push:True

to a private registry ? Right now, my flow is built and push, the

flow.py

is inside the docker image, and I tried to configure the pull step with

prefect.projects.steps.set_working_directory

but I get

FileNotFoundError: [Errno 2] No such file or directory: '/opt/prefect/flow'

It seems that the pull step is executed outside of the docker container ? my code in thread

Hi, So we use the below code to deploy our prefect flows, and you can see the

skip_upload=True

here which we set because our s3 bucket policy requires to have encryption set while uploading to s3, so we copy the flow code using aws s3 cp cmd and we set -sse AES256 parameter there. I'm wondering is this something that can be supported in prefect in future, or if it is already supported what is the parameter I need to add? Thanks!

deployment = Deployment.build_from_flow(
    flow=flow,
    name=FLOW_NAME,
    parameters={},
    infra_overrides={
        "image": SHARED_CONFIG.value["prefect_image_name"],
        "service_account_name": "sa",
        "image_pull_policy": "Always",
        "namespace": "nspc",
    },
    infrastructure={"block_type_slug": "kubernetes-job"},
    work_queue_name="k8s",
    storage=storage,
    path=FLOW_NAME,
    skip_upload=True,
)

im using the helm server and work charts and i've isolated the pods with nodeselector as usual, but is there a way to isolate the job containers that spin up? they seem to create on any node they want by default. how can i control what nodes these spawn on?

Hello everyone, I have a Python Version question that i'm hoping someone can provide clarity on for me. In the Docker Image

prefecthq/prefect:2-python3.8

, the python version is

ENV PYTHON_VERSION=3.8.16

-- unfortunately, 3.8.16 breaks my code. I need to run 3.8.0 through the docker image. Does anyone have thoughts on how I can get the base image set back to 3.8.0? Thx much!

Hi Community, I'm trying to create a deployment over the client-api. The Code looks like the following

async def deploy_ki_pipeline_controller():
    logger = get_run_logger()
    async with get_client() as client:
        flow_id: UUID = await client.create_flow(infere_pipeline_handler_flow)
        <http://logger.info|logger.info>(f"Der Flow infere_pipeline_handler_flow wurde erfolgreich mit der UUID {flow_id} erstellt")
        params = {
            "watch_folder_path" : "test",
            "dead_letter_path": 'test/dead_letter',
            "on_success_move_path": 'test/bearbeitet',
            "lego_api_url": 'test/annehmen',
            "lego_api_user": 'test',
            "log2File": False,
            "logging_file_path": 'test',
            "dry_run_api": False,
            "dry_run_clean": False,
            "delete_on_success": True,
            "create_images": False,
            "extended_logs": False,
            "extended_anonym_error_logs": True
        }
        deployment_id: UUID = await client.create_deployment(
            flow_id=flow_id,
            name="infere_pipeline_handler_deployment",
            storage_document_id="5bf39961-0f81-4a4b-9751-2d7e132066a8",
            parameters=params,
            work_pool_name="first-worker-pool",
            work_queue_name="test-queue-master",
            parameter_openapi_schema=params,
            entrypoint="prefect_module/flows/infere_pipeline_handler_flow.py:infere_pipeline_handler_flow",
        )

All works fine, the flow and the deployment are created and I can access it via the prefect UI. But the parameters for the deployment are empty Is there a step I'm missing or I do something wrong here?

Hi, my Deployments in Prefect Cloud UI are not displaying the deployment parameters. The deployment for my flow always says 0 parameters. However, when I run

prefect deployment inspect flow_name/deployment_name

for my deployment, I can see the parameters in the returned JSON. Additionally, this means that I can’t override the Deployment parameters for ad-hoc runs of my flows. Is this an expected behaviour in any of the Prefect Cloud UIs? question also posted in #prefect-ui

how do you specify an infrastructure block using projects with

prefect deploy

?

prefect deployment build

has the

-ib

flag but

prefect deploy

does not

Been having a bit of trouble with Dask crashing recently, and I came across this old thread from a few months ago. I think my issue is very similar to the one described, where a large number of tasks being submitted at once causes the flow to crash. Has there been any updates on this and/or a simple way to solve this?

Hi Guys i got this error from git, what i am doing wrong?

State message: Flow could not be retrieved from deployment. subprocess.CalledProcessError: Command '['git', 'clone', 'git@github.com:Oicoteam/prefect.git', '-b', 'main', '--depth', '1']' returned non-zero exit status 128.

Hello @Prefect I am trying to create prefect service accounts though rest api. I am spinning the docker container . It is throwing me the *Value Error("prefect.api.cloud) does not appear to be an IPV4 or IPV6 address". Can someone please throw light on this"?

Hello! I’m looking into setting up a project with the docker recipe where I’m deploying to our internal k8s cluster. Is there a good example of this setup?

Guys, I don't understand how the tab

Task inputs

works. I followed this example here and I still don't get it. Can someone ELI5 it to me? I don't get how certain tasks show inputs, but others dont (see screens)

I am having an issue with an auth token for git. I have the below in my prefect.yaml for my project. It seems to be having an issue authenticating and wondering what I may be doing wrong that it cant get the right authentication

pull:
- prefect.projects.steps.git_clone_project:
    repository: <https://github.com/comp_rep/bi_etl.git>
    branch: DATA-1425-docker-project-deploy
    access_token: "{{ prefect.blocks.github-credentials.git-creds }}"

Hello, I’m getting an error on deploying the prefect-workers pods. They can’t seem to connect to the api.

Copy code

prefect.exceptions.PrefectHTTPStatusError: Client error '405 Method Not Allowed' for url '<http://172.28.36.138:4200/api/work_pools/>'

The prefect-agent’s are using the same URL and have come up correctly.

Is there any way to pass task run tags to the

trigger_dbt_cli_command

pre-built task?

Hello!, wondering if someone has issues applying markups in logs?. I have the following code and settings on the Prefect server, but I cannot seem to make it work.

from prefect import flow, task, get_run_logger
from minio import Minio


@task
def bucket_creation(bucket_name: str = "name-entries"):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Creating bucket name: [bold red]{bucket_name} ==> MINIO_ENDPOINT: {os.getenv('MINIO_ENDPOINT')}[/]")
    client = Minio(
...

The above

<http://logger.info|logger.info>

is using Rich’s markup and these are the output I have on the Prefect Server flow run and at the agent level

Hi, How to create a flow in prefect 2.0? I was using create_flow_run from prefect.tasks.prefect.flow_run in prefect 1.0 ?

Hey folks, is there a way to force a task to run in the main thread? I'm using PMAW's PushShiftAPI to get some historical reddit data, but when I decorate my function with

@task

, I get the following error:

ValueError: signal only works in main thread of the main interpreter

This is, from what I understand, due to the fact that pmaw is configured to listen to a SIGKILL from the CLI, and gracefully interrupt the process, which I guess can't happen if it's running in a separate thread. The moment I remove

@task

decorator from that function, then I can execute it normally inside a flow, but I obviously lose all the advantages of tasks such as caching and whatnot. For the record, here's the function I'm talking about:

def get_submission_ids(start_date: str, end_date: str, subreddit: str):
    reddit = get_reddit_client()  # Reddit client from `praw`
    api_praw = PushshiftAPI(praw=reddit)
    start_datetime = datetime.strptime(start_date, "%Y-%m-%d")
    start_date_ts = int(start_datetime.timestamp())
    end_date_ts = int(datetime.strptime(end_date, "%Y-%m-%d").timestamp())
    search_window_days = (datetime.today() - start_datetime).days

    submissions = await api_praw.search_submissions(
        subreddit=subreddit,
        after=start_date_ts,
        until=end_date_ts,
        search_window=search_window_days,
    )

    return [f'{SUBMISSION}_{sub["id"]}' for sub in submissions]

About log output in Google Kubenetes Engine(Auto Pilot) 　When I run flow with Prefect's scheduling feature, the logs are transferred to Google Cloud Logging, but when I run it manually like python sample.py, the logs are not transferred to Google Cloud Logging. Is this a configuration issue or default behavior?

Hello! I just watched and I was wondering: when is it better to use a Variable in place of a Block (like in this exemple, I now use a String Block to store the environment)?

Hi! Is there a way to invoke lambda function from prefect flow in prefect 2.0? Like there was

LambdaInvoke

in prefect v1.