prefect-community #prefect-community

Hi everyone, I found a weird behaviour when I use a timeout with a LocalDaskExecutor and manually raise a Success. The task will always fail with the timeout error message, even if it takes less time than the specified timeout. Minimal example:

Copy code

from prefect import Flow, task
from prefect.executors import LocalDaskExecutor

from prefect.engine.signals import SUCCESS


@task(timeout=60)
def succeed():
    raise SUCCESS()


with Flow("timeout-test") as flow:
    succeed()

flow.executor = LocalDaskExecutor()
flow.register(project_name="tutorial")

Is this the intended behaviour? Can anyone reproduce it?

Tomoyuki NAKAMURA

08/31/2021, 9:08 AM

Hi, We are thinking of migrating from Airflow to Prefect Cloud and would like to use Datadog to monitor the number of SUCCES and FAILURES of tasks, task execution time etc in Prefect Cloud. Here are the metrics we can get from Airflow’s Datadog integration: https://docs.datadoghq.com/integrations/airflow/?tab=host Is it possible to get the similar metrics in Prefect Cloud and integrate it with Datadog?

Seonghwan Hong

08/31/2021, 10:06 AM

I set the flow schedule 5minutely by cron string. However, the upcoming flow list is not visible in the UI. And It is not running. Prefect doesn't actually run the flow at the time.

Seonghwan Hong

08/31/2021, 10:06 AM

Simon Gasse

08/31/2021, 12:59 PM

Hi! Can I somehow ignore skipped tasks of a

map

in downstream tasks? Let's say a mapped task skips for some inputs. In this case, I would still like the downstream task that runs on the return values of the mapped task to continue with the results of all successful task runs. An example is in the thread 🙂

Wilson Bilkovich

08/31/2021, 1:19 PM

I'd like to use a container that defines its own `ENTRYPOINT`as the image that prefect/dask workers and schedulers run, but when I try it, Prefect seems to expect

/bin/sh

as an entrypoint. Is there any way to override that?

Paulo Maia

08/31/2021, 2:45 PM

Hey everyone, I'm trying to run two prefect flows with different schedules within the same file. Is this possible? I saw the documentation and I see there can be different clocks for the same schedule, but didnt see anything about my use case. Example: • Machine Learning application where you want to train a model every week, generate predictions every day and run a reporting script every 3 days. I could separate this into three prefect files/flows but I don't want to I'm using the open source version, offline

Nicholas Chammas

08/31/2021, 3:56 PM

I think Prefect currently lacks an organized way to quiesce agents in preparation for an upgrade. i.e. I want a way to tell agents “finish what you’re doing but don’t accept any new work” so I can shut them down and replace them in an organized manner without interrupting any flows. Should I file an issue about this, or is there an accepted workflow for accomplishing this?

Didier Marin

08/31/2021, 4:06 PM

Hey everyone, We are trying to add some fine grained authorization to our Prefect server (such as which user can perform which action and which flow). I understand the Cloud version as roles and custom roles but I don't think this would be detailed enough for our needs (for example the read permission on flows can be granted, but I don't think it can be limited to a particular flow group id for example). It seems like the easiest way to tackle this is at the GraphQL level, using resolvers to customize exactly what request can be performed by who. By any chance, does anyone has some experience doing this ?

Mark McDonald

08/31/2021, 5:07 PM

Hi - we have a system in place wherein we collect all state change events via webhook triggered by automations. We're collecting this data to identify and alert us if a flow run's duration is anomalous. It would be really helpful if we had the flow's parameters in the webhook payload, because these are the best indicator of how long a flow should run. It's not currently part of the default payload. Should I open an issue in the github repo?

Mark McDonald

08/31/2021, 7:48 PM

Hi again - we're having trouble logging into Prefect Cloud. We've been seeing this screen for a while. I've confirmed the issue with multiple people at my company.

👀 1

Tony Yun

08/31/2021, 8:35 PM

hi, I’m having some trouble setting up the Docker container correctly for Flow to register. Could someone take a look? The Dockerfile setup correctly and working on local testing (with flow.run()); • I’m using Docker Storage like this, which the dockerfile is working correctly to call my modules:

Copy code

storage = Docker(
    dockerfile="./Dockerfile"
)

But the register always report the module not exists.

Copy code

ModuleNotFoundError: No module named 'criteo'

Wilson Bilkovich

08/31/2021, 8:41 PM

I'm struggling to get local+docker 'as script' flows to work, and I can't find a working value for 'script path'

Wilson Bilkovich

08/31/2021, 11:36 PM

Can anyone explain the correspondence of KubernetesRun and pods that get spun up. Is KubeRun what the Agent runs to create a scheduler pod? Does it have anything to do with the worker environment?

Hamza Ahmed

09/01/2021, 12:37 AM

Running into issues after upgrading Prefect (flows+kubernetes deployment) from 0.14.6 to 0.15.4 and I am running into an issue where a flow gets scheduled but errors out after a couple hours with an error

A Lazarus process attempted to reschedule this run 3 times without success. Marking as failed.

The only other errors are in the prefect towel logs [see comments] Any clue as to what could be wrong?

Aric Huang

09/01/2021, 12:41 AM

Hi, I'm trying to run some flows that use

ProcessPoolExecutor

and am having trouble getting print statements and stack traces from subprocesses to show up in the Prefect Cloud UI. When running the flow locally the logs show up in stdout/stderr, and having the task defined with

@task(log_stdout)

makes print statements in subprocesses show up with a

└── <time> | INFO

prefix (which I thought means it's getting picked up by the Prefect logger). However the subprocess print statements don't show up at all in the Prefect Cloud UI logs. Stack traces from subprocesses that crash also print locally but don't show up in the UI, which only shows a stack trace for the main process with

concurrent.futures.process.BrokenProcessPool

. Is there a way to have a flow capture all subprocess output so it's visible in the UI?

Jason Kim

09/01/2021, 2:21 AM

Is it possible to have a local agent have multiple conda environments? Should I use 'prefect agent local start' for each of my conda envs and then add a label for each agent?

CA Lee

09/01/2021, 4:27 AM

Hello, has anyone run into the below errors while using DbtShellTask?

haf

09/01/2021, 8:42 AM

Is Python 3.9 supported?

haf

09/01/2021, 8:45 AM

And do you have a sample with using python modules with docker and prefect?

Michael

09/01/2021, 9:34 AM

Hey there! I’m trying to implement some sort of rudimentary Sensor (like given here https://github.com/PrefectHQ/prefect/discussions/4156), but I would really like to implement a timeout that spans retries. Or in other words, enforce max_retries even though I am raising the RETRY signal manually (which ignores max_retries). Any ideas how to tackle this? I could store a counter in the context and increment a counter inside each of my task’s run() calls I suppose

Konstantin

09/01/2021, 11:08 AM

Hey! I have registered my flow on my prefect server (core). The start of the task failed with an error. in flow I applied parameters and map. error State Message: Error during execution of task: ValueError("Could not infer an active Flow context while creating edge to <Task: request_huntflow['items']>. This often means you called a task outside a

with Flow(...)

block. If you're trying to run this task outside of a Flow context, you need to call `GetItem(...).run(...)`")

✅ 1

Konstantin

09/01/2021, 12:08 PM

Copy code

Hey there! how to add a token file to the storage in the prefect. for example, the error is this:
Task 'request_web [0]': Calling task.run () method ...
15:05:12
INFO
CloudTaskRunner
[Errno 2] No such file or directory: './Tokens/access_token.xml'

Evan Brown

09/01/2021, 2:23 PM

Hey there! I am having trouble logging standard out with shelltasks like you would with the task decorator. Does anyone know how you would do this?

Dan Stoner

09/01/2021, 2:28 PM

I registered a Flow to the wrong Project. How do I delete it from that Project? The cli seems like it should but it does not seem to understand anything other than

delete project

Copy code

$ prefect delete --help                                                                                                                                                     
Usage: prefect delete [OPTIONS] COMMAND [ARGS]...

  Delete commands that refer to mutations of Prefect API metadata.

  Usage:
      $ prefect delete [OBJECT]

  Arguments:
      project    Delete projects
...

marios

09/01/2021, 2:41 PM

Hey! How can I mount volumes on a specific flow without affecting the global configuration of docker agent as described here https://docs.prefect.io/orchestration/agents/docker.html#mounting-volumes

👀 2

An Hoang

09/01/2021, 3:28 PM

Hi, is there a way to set one

Parameter

as the default for another? Here's what I did

Copy code

#task 
@task(target= {output_folder_path}/result)
def example_task(output_folder_path)
....

#in flow
step1_folder_path = Parameter("step1_folder_path")
output_folder_path = Parameter("output_folder_path", default = step1_folder_path)

When I do this, I got a warning

Copy code

<ipython-input-226-3b472991f1ef>:8: UserWarning: A Task was passed as an argument to Parameter, you likely want to first initialize Parameter with any static (non-Task) arguments, then call the initialized task with any dynamic (Task) arguments instead. For example:

  my_task = Parameter(...)  # static (non-Task) args go here
  res = my_task(...)  # dynamic (Task) args go here

see <https://docs.prefect.io/core/concepts/flows.html#apis> for more info.
  output_folder_path = Parameter("output_folder_path", default = step1_folder_path)

When I execute the flow, it runs but the

target

doesn't work as intended, the task still runs every time even though the target files are there. When I provide a hardcoded string to the default of

output_folder_path

, everything works fine

Samuel Hinton

09/01/2021, 4:01 PM

Hi all! Just wondering if anyone has noticed memory leaking when using prefect with dask workers? I am already deleting the local flow variables, calling gc.collect(), and have completely disabled checkpointing with prefect. And yet, this is typical for my workers until the OOM and die over the course of a few days. All the little bumps and step ups lines (green, red, blues) at the top are the dask workers in the swarm that keep growing.