im getting intermittent 404s on the service account page…

Hello everyone, I have a prefect 1.0 Server running on K8 cluster with Git as storage. Is it possible for me to run a flow on a remote server with all the scripts and dependencies being on remote server and only using prefect on K8 to track the progress of the flow run.

Hi, bumping this one - maybe someone has an idea if I'm doing something wrong or if there really is a problem in ConcurrentTaskRunner/prefect-snowflake

Hello 🙂 I have a problem with a newly created flow. It is a relative straightforward one and pretty easy but it only works locally (flow.run()) and not on the cloud 😞 The logs don't show any information, just reference task fail. The task is using a custom class for processing and loading the data like this:

@task(log_stdout=True)
def etl(
    connector,
    exec_file,
    environment: Optional[str] = None
):
    master_data = MasterDataSet(connector, exec_file)
    master_data.get_dataset(environment)
    master_data.load_dataset()

My flow looks like this:

.... as flow:
    source = SOME_SOURCE
    destination = SOME_DESTINATION

    file_list = ['SOME/PATH/TO/FILE', 'ANOTHER/PATH/TO/FILE']

    master_dataset = etl.map(
        connector=unmapped(source),
        exec_file=file_list
    )

The cloud logs look like this:

Task 'etl_master_data': Starting task run...
Task 'etl_master_data': Finished task run for task with final state: 'Mapped'
Flow run FAILED: some reference tasks failed.

There is no failed reference task. Does anybody has any idea why this is not working on the cloud but locally? It is configured like all the other flows, even the data handling by class is already implemented for other flows. I really have absolutely no clue why 😄

(Prefect 2.0 question) my flow has a

start_date

parameter. In the flow code the default of this value is set like this

def sync_circleci(start_date: datetime.date = datetime.datetime.today().date()):
   ...

and it will come out in the YYYY-MM-DD format. Now when I deploy the flow, it seems that the default value is not recognized by Prefect UI. When I run this flow "using default param values", the flow fails with this error

prefect.exceptions.SignatureMismatchError: Function expects parameters ['start_date'] but was provided with parameters []

which means the default value is not recognized (?). And when I select a date value using the date picker, looks like the date format is not recognized by my flow code with this error

- start_date: invalid date format

How should I handle this?

Upgraded to 2.1.0, ran locally works fine but deployments don't work anymore as libraries could not be find. Do you have to somehow install libraries now to run the deployments???

apologies if this is a silly question, but I’ve been going around and around for an hour now How can we trigger a flow run from the REST API (Prefect REST API - Prefect 2.0) using a POST

/api/flow_runs/

request from an external source? This was something we’d do regularly with the GraphQL api in Prefect 1.0

2.1.0 seems great, I especially like being able to set schedules when building! However, as far as I can tell, there's no way to set the timezone, unless I'm missing something. If I'm not missing something I'll go ahead and create a github issue, just figured I'd check first.

Hi all, we are running the Prefect 2.1.0 agent on Kubernetes. The API in the deployment has no errors, but the agent throws the following error: prefect.exceptions.PrefectHTTPStatusError: Client error ‘404 Not Found’ for url ‘https://api.prefect.cloud/account/[my accountid]/workspace/[my workspace id]/block_types/’ Can someone please explain what is going on?

I have an issue running deployments in 2.1.0. My dbt flows take two parameters that have default values set: loglevel and n_dbt_threads. For some reason I am now getting the following message when running the deployments: prefect.exceptions.SignatureMismatchError: Function expects parameters ['loglevel', 'n_dbt_threads'] but was provided with parameters []. You can see the parameters listed on my deployment page in the screenshot attached. If I run the deployment with custom parameters, manually setting the values for the two parameters, everything's fine. Has anyone found a workaround for this problem, or am I meant to be doing something different compared to 2.0.4?

Hi 👋 For

KubernetesJob

is it possible to specify a

node-selector

and

tolerations

to be able to assign the worker pod to a specific node pool ? I can’t see it in the doc Thanks for your help

I’ve got a question RE: deployments with

null

(local) storage

Is there a way to stop a Dask cluster's queue to do work when an exception happens?

Are there performance implications between • starting several agents each with their own single work queue • starting one agent with several work queues (e.g. moving all work queues to one agent like

prefect agent start -q queue1 -q queue2 …

within the script below) I am bootstrapping agents within a docker container with this python script:

import prefect
import asyncio
import subprocess
import time
import sys

async def bootstrap_agents():

    # get list of work queues
    pc = prefect.get_client()
    work_queues = await pc.read_work_queues()

    # create agents by ID 
    for wq in work_queues:
        proc = subprocess.Popen(['prefect', 'agent', 'start', str(wq.id)])
        print(f'Starting agent associated with {wq.name}. PID number is {proc.pid}', file=sys.stdout)

if __name__ == '__main__':
    asyncio.run(bootstrap_agents())
    while True:
        time.sleep(1)

If there is a better deployment pattern I’d love to get insight on that.

good morning everybody - i have an "execution environment" question: is it possible to have an agent running on a micro EC2 instance that deploys flow runs as ECS tasks? i know in 1.x we could spin up an ECS agent that would automatically do that, but in 2.x it seems like we would manually have to stitch that piece together? this repo shows how to spin up an agent running on ECS in 2.x which is cool, but the infra in the deployment is still

Process

- which means the flow run will be executed in the same container that the agent is running - am i understanding that correctly?

Looking to get back into prefect now that it is more stable. I have multiple deployments each with their own flow. Can i prevent the same flow from running twice at the same time while using just one work queue? In the early betas the solution was one work queue or agent per deployment. Edit: dang. Just noticed #5623 is still open

I am running a flow in Prefect 1.0 with a

DaskExecutor

with

cluster_class="dask_jobqueue.SLURMCluster"

. Everything works fine, unless SLURM kills the resources (e.g. due to time limit). Then the computation is naturally terminated, but in the Prefect cloud I still see the flow and task running. I did an experiment where SLURM kills the resouces after 2min, but the flow is still shown as running for 1h 38min and counting. I thought that the heartbeat option should kick in at some point to flag the flow as

Failed

. Has anyone experience with a similar setup?

Hi everyone, I'm trying to use Docker Storage with our dockerhub account and I keep getting an error that I'm not given permission to push to our account:

InterruptedError: denied: requested access to the resource is denied

. If I

docker push

outside of prefect it'll push no problem. Maybe it's an incorrect

registry_url

?

flow.storage = Docker(   
    registry_url="<http://registry.hub.docker.com|registry.hub.docker.com>", 
    image_name="atddocker/atd-microstrategy",
    image_tag=ENV,
)

when you call

prefect agent start -q misspelledname

it shouldn’t auto create the queue

It would be pretty nice to be able to share certain blocks across workspaces. Like if you're using a dev and prod workspace that share the same secrets, having two different sets of secrets adds the possibility for there to be discrepancies between the two

Question: I want to have a number that increases every time I run a deployment and access it for logging to know which files correspond to which deployment run. How to go about it? Using remote storage (GCS) I tried using a pickle but it appears that it doesn't save on the bucket. Maybe use another bucket or manually pull and push the object to the bucket. Figured there should be an easier way with Blocks.

I was trying to explore solutions for our use case.... We have a use case where we want to run our jobs in hybrid agents (#PrefectIsPerfect 🙂 ) So I'm trying to accomplish this in prefect 2.0 knowing that we have a little bit of a runway before the solution hits primetime... The cloud options seem to be well documented so I'm less concerned about it Our on prem infrastructure is primarily Openshift. We could spin up VMs to host Docker agents but, was curious how have people gone about running production loads on-prem

Ran into this bug earlier. The UI tells me to use the flag

--work-queue

but the CLI says no such option exists. Removing the

--work-queue

flagged resolved the issue.

Is Prefect Cloud HIPAA compliant? If so, does it have to be Enterprise plan tier?

A function to check if a block already exists would be helpful

What replaces the old

flatten

call in Prefect 2.0? I have a list of lists after mapping tasks and I used to be able to use

flatten

to coalesce that into a single large list.

Hi, just thought I’d say that the logging configuration in 2.0 is nice. I have a flow where I need to debug a

requests-oauthlib

connection, and now all I need to do is: 1. Copy

logging.yaml

2. Set my logger:

requests_oauthlib:
        level: DEBUG
        handlers: [console]
        propagate: no

3. Set

PREFECT_LOGGING_SETTINGS_PATH=logging.yaml

in my

.env

file. And it works!

What happened to

create_markdown_artifact

in Prefect 2.0? I don’t see markdown or artifacts mentioned in the migration guide.

Is it possibly to dynamically assign tags to task runs based on parameters? I’m using Prefect 1.x and want to do something like the following across multiple flows.

my_task.map(databases)

But I don’t want to overload my databases, so I want to set concurrency limits at the task level to make sure there are only ever

N < limit

tasks running against a given database at one time. I know you can set task tags when I’m creating the task object, but can I do so as the flow is executing against mapped tasks?

Thanks for the new release today! Very exciting. I'm trying to use the new python

Deployment

class. Previously, I defined a

KubernetesJob

infrastructure block in a different python script and used the

.save()

method to instantiate a block in the UI. It looks like this was removed in 2.1... How can I do this now? Moreover, I'm trying to understand what I pass in for the

Deployment

class in the

infrastructure

parameter. With the

storage

parameter, I just load the

storage

block that I previously defined. Can I do the same with the

infrastructure

block? If so, how do I load an existing

KubernetesJob

block? Or, is the new workflow to just input the

KubernetesJob

dictionary directly in your

Deployment

class? If that's the case, how do I tie this to a work queue? In general, I'm a big fan of the

KubernetesJob

block. idc whether it's an actual block or just a python dictionary I pass into the deployment. Either way, it's been extremely helpful to manage compute depending on the deployment. Would love to streamline it with CICD. I know that's in the works and I'm eager to see best practices with this and implement it.