Hey all, wanted to post here before creating an issue on github. But currently having an issue getting my ci/cd registration job working when trying to update

for flow in ${flows[@]}; do prefect register flow --file flows/$flow --project my_project --skip-if-flow-metadata-unchanged; done

(which currently works) to

prefect register --path flows/ --project my_project

where my flows are utilizing azure storage and are located within flows folder. Using the new register command does create a new blob, but then errors out

Error building storage: azure.core.exceptions.ResourceExistsError: Operation returned an invalid status 'The specified blob already exists.'

for each flow ultimately exiting out the job. For now I will just continue to use ``prefect register flow``, but wanted to see if anyone was having similar issues.

Hi! Just discovered that Prefect.io/slack link is on longer active, so new friends will not be able to join this workspace. Is this something that can be fixed soon?

Hey everyone — I’m trying to use a shellTask and it is working but I’m getting duplicated logs, with slightly different formatting but identical content. I can’t seem to find a way to disable/hide the duplicated logs. Here’s an example of duplication:

[2021-06-22 13:25:46+0100] INFO - prefect.ShellTask | 13:25:46 | Concurrency: 12 threads (target='dev')
INFO:prefect.ShellTask:13:25:46 | Concurrency: 12 threads (target='dev')

and I have

stream_output=True

on the shelltask

Hi, I’m trying to get on a secret on my cloud account, and having trouble with that I can successfully set a secret using

from prefect import Client

client = Client()
client.set_secret(name="MYSECRET", value="MY SECRET VALUE")

I can see ‘MYSECRET’ on my cloud UI, so I’m sure I’m authorized etc. But I am unable to get a secret with:

from prefect.client import Secret

# Load the value of `MYSECRET`
my_secret_value = Secret("MYSECRET").get()

Where I get a

*** ValueError: Local Secret "MYSECRET" was not found.

error. Reading the docs, I see that it should be looking for a local secret first, and if not found, will look for a cloud secret. I think it is not looking for a cloud secret

Does anyone know if I just run

prefect agent kubernetes start --job-template

again to amend my existing agent? I'm trying to edit the safe-to-evict setting.

Is there a way for me to define a Flow in such a way as to be generic and then I can subclass it for each child flow? I am looking to reduce code repetition for some flows I have that follow a general pattern, but have small differences, or added steps/tasks. I don't think the imperative API is what I am looking for.

Second question is whether or not there is a way to provide state (and define within the Flow) state than using the Parameter tasks. I know the

context

exists, but I am looking for something that can read a file and set the state. I don't think the context object is setup for that. It seems to be more runtime to me

Hello, we are having some issues around the logging. Some, but not all of the logs are appearing in the dashboard. Any ideas how I might troubleshoot this? We are using ECS.

TL;DR I’m seeking a source for converting AirFlow to Prefect. Hi All. I have inherited what amounts to an AirFlow demo that I would like to sort out how to convert over to a Prefect demo. I am a graduate student in a practicum and the previous cohort wrote the AF stuff. Now I am trying to answer the question as to whether AF should remain in place or if Prefect could possibly be snapped in AF’s place before the project gets truly up and running. As a point of reference, I did get a Prefect Cloud demo type of thing up and running last week (it just spun up a container in AWS and spit some stuff out in the Prefect logs). So I’m not a complete newb, but my background is more on the networking than the development side.

Hi, everyone! I'm new to prefect and have a question regarding some use cases that we saw whilst testing it in our environment: - Use of connection pool to connect to databases: - We actually saw this link here (https://github.com/PrefectHQ/prefect/issues/1876) and tried using it, but when we tried to register the flow to the server, it threw the error: " flow.register(project_name='name') TypeError: can't pickle _thread._local objects " - For us this is a major issue as within our flow, ideally we would have some light queries in a lot of tasks, using connection pooling is mandatory for this use case in particular. If anyone could point us in the right direction it would be very helpful! Thanks in advance, Luis

Hey all, does anyone have experience using Great Expectations? We’re having trouble running checkpoints using the prefect tasks, running into ‘object is not subscriptable’ errors. We’ve tried V2 and V3, but have not attempted going down to 0.12 version. Noticed someone else was getting a different error here but made it successfully https://github.com/PrefectHQ/prefect/issues/4411 any ideas?

Hi all! I've just started using Prefect and I was wondering if it is possible to programmatically access task logs using Python - at the moment I'm using the web interface but it would be great if I could grab and parse them via an API. Is this currently possible? Thanks in advance!

How do people set up their repos? We had a monorepo for all our Flows at my last gig, so I'm inclined towards that. But curious as to how others do it!

Hi - I had this problem a few times. Where I am not able to log on to the webpage. I keeps loading. Do you know why that happens, and how to fix it?

Hey folks, I'm getting a weird error on my Kubernetes cluster (AKS) when my agent kicks off its first job. The agent does start successfully, and subsequent jobs are ran to completion, so doesn't appear to stop anything functionally. Just wondered if anyone else has came across this? I am packaging up my dependencies to docker storage, and storing my flows in blob storage. Cheers

[2021-06-23 08:53:35,179] ERROR - agent | Error while managing existing k8s jobs
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 384, in heartbeat
    self.manage_jobs()
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 230, in manage_jobs
    pod_events.items, key=lambda x: x.last_timestamp
TypeError: '<' not supported between instances of 'datetime.datetime' and 'NoneType'
ERROR:agent:Error while managing existing k8s jobs
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 384, in heartbeat
    self.manage_jobs()
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 230, in manage_jobs
    pod_events.items, key=lambda x: x.last_timestamp
TypeError: '<' not supported between instances of 'datetime.datetime' and 'NoneType'

Hi, I have suddenly started to get this error on my flows using the Kubernetes Agent.

ERROR - prefect.CloudFlowRunner | Unexpected error: AttributeError("'Log' object has no attribute 'splitlines'")

I’m running prefect self hosted. Server, agent & flow versions are 0.14.22 Nothing has changed, any idea what could be the reason?

Hi folks, do flow concurrency limits take the flow run parameters into account ? i.e. if I don’t want two flow runs with the same parameters to run at the same time - how do I go about achieving this ?

For those not part of the #show-us-what-you-got channel.

On the Task Run views, why can’t we sort the tasks by Duration?

Good evening from Germany, is there a difference how I activate cloud secret storage on different clients? Is there an example of how I turn in on in the config? The documentation is not very helpful about how to actually do it, for me at least.

Hi — I’m in process of trying to deploy a KubernetesAgent to EKS, I have an EKS cluster with 3 different nodes. Where would I specify which node group I would want to use for a flow run? Would it be in the flow.run_config = KubernetesRun(…)?

Hi are there any best practice recommendations for prefect repositories set up in production?

Messing with the GitHub Action again.  Does this sound right for the sequence of Actions to register flows after a Commit? 1. Checkout repo into Action Runner 2. Install Prefect on the Action Runner 3. Make a config file that has a 

config.toml

 with my auth token (stored as a secret in GitHub) 4. run 

prefect register

 through the CLI Thanks!

Hey, I am using

Docker

storage to push an image to ECR. I am authenticated with

aws sts assume-role

locally in my CLI. I am trying to push the image locally at the moment and getting an error with:

File "/Users/benmuller/code/es-betfair-flows/.venv/lib/python3.8/site-packages/prefect/storage/docker.py", line 585, in push_image

raise InterruptedError(line.get("error"))

InterruptedError: denied: Your authorization token has expired. Reauthenticate and try again.

How can I make sure that my

sts

creds are honoured ?

Hi folks, is it possible to delete or "unregister" a flow with a graphql call? I can see in the docs about deleting flow runs, but I want to programmatically remove a flow from a server altogether.

Hi everyone, I’ve tried registering a prefect flow using docker storage and ECS run config today and received a

pip

and

python 2.7

error? (pls see screenshots) The dockerfile I’m using for the docker storage (

flow.storage = Docker(dockerfile="Dockerfile", …)

looks like this

FROM python:3.9
WORKDIR /app
COPY . .
ENV PYTHONPATH=$PYTHONPATH:`pwd`
FROM <http://gcr.io/google-appengine/python|gcr.io/google-appengine/python>
RUN apt-get update
RUN apt-get install poppler-utils -y

I’ve set the python version to 3.9 in the dockerfile, why doesn’t it use that?

Hey Prefect peeps, Do you have an example of how we can use the

prefect.context.get("logger")

to log throught our entire project ? eg. when sub modules are called by calling their classes in that flow?

I got the following error when I try to invite a new user in Prefect Cloud. (excuse if this is not the place for this).

Hi, I’m having a DAG with basic fan-out, fan-in idea. Some of the fanned out tasks may fail, but I still want to continue the fan-in task for the succesful tasks. Thus if I have upstream: [x, y, z, FAILURE], the downstream still should process for input [x, y, z]. What I did; • Added trigger=all_finished to downstream task • Iterate over the input, filter out exception Although this worked fine locally while developing, I now got an edge case that one of the upstream tasks is TriggerFailed. How can I mitigate this? And, still process the succesfull upstreams?