prefect-community #prefect-community

Howdy, when an Agent triggers a run of a registered flow, is there some standard way to ask Agent to pass some arguments into the flow? e.g. Agent would do:

registerd_flow.run(**runtime_args_passed_by_agent)

Or is this disallowed by design? (registered flow shall be self-sufficient for reproducibility) Context: we have some flow that depends on some value that changes quite often (e.g. dependency package version number). We don’t want to register a new flow each time we update the version number; rather have one flow and have the agent run the flow expecting the package version information will be provided to it at run time. We thought about using env var, but not sure if it’s the best way…

Rob Fowler

04/09/2021, 4:26 AM

damn, seems Dask thread executors are broken again, back to processes and everything is good

Brian Keating

04/09/2021, 5:45 AM

I've written a

ResourceManager

to create and terminate EC2 instances. I test it out with a bare bones workflow:

Copy code

@task
def do_something_on_instance(instance_id):
    prefect.context.get('logger').info(f'Do something on instance {instance_id}')

with Flow('hello-ec2') as flow:
    with EC2Instance('t2.micro') as instance_id:
        do_something_on_instance(instance_id)  # instance_id is a string

This works correctly when using github storage, but when I switch to S3, the flow fails with

TypeError: cannot pickle 'SSLContext' object

. Anyone know what's going on here? Note that the value returned by

EC2Instance.setup

is a

str

Joe McDonald

04/09/2021, 5:52 AM

Anyone had this show up when using the ECS agent? We are running an ECS cluster with three agents and prefect server in a separate ecs cluster all running 3 instances of all services. It seems in the deprecated fargate agent you could have it create new family for each run, but looks like ECS agent doesn’t do that, it reuses the same family for task definition and increments version so when running a lot of flows at the same time we run into this.

Copy code

An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Too many concurrent attempts to create a new revision of the specified family.

xyzz

04/09/2021, 8:54 AM

I'm a bit irritated about the information on https://docs.prefect.io/orchestration/faq/dataflow.html#gotchas-and-caveats , which claims to be an exhaustive list of data that might end up in the prefect database. It doesn't mention anything about the flow metadata like names of flows and tasks and their configuration like run_config, schedules or storage. So I assume this list isn't actually exhaustive?

Noah Holm

04/09/2021, 9:41 AM

I’m using S3 storage for flows running with an ECS Agent. By default the flow’s tasks uses S3Result as outlined here. Is there any way to disable the task results for individual or all tasks in the flow while keeping the S3Storage on the flow?

Marko Jamedzija

04/09/2021, 10:54 AM

Hello 🙂 I recently deployed Prefect Server 0.14.15 on a test k8s cluster. After playing around with it I noticed that k8s agent only runs the flows for the first tenant I created. For the projects I created from other tenants, all their flows are just stuck in

scheduled

state (becoming late runs). Is this a bug, or there’s a way of configuring k8s agent to work with more than one tenant? If so, how? Thanks 🙂 (I deployed the services using the prefecthq helm chart)

Brent Bateman

04/09/2021, 1:45 PM

Who's the admin for this workspace? I want to leave it/ not just sign out of it?

Joe McDonald

04/09/2021, 1:50 PM

So what is the chance of getting official PrefectHQ docker images hosted on https://public.ecr.aws/ so that ECS services can avoid the docker hub rate limits? Would be nice to convince Hasura to do the same thing…

Trevor Kramer

04/09/2021, 1:55 PM

I'm confused about how to use @resource_manager effectively. I am using it to create an RDS instance in AWS. I can create it fine in setup and destroy it in cleanup but I don't understand how to get the endpoint out of it. It seems to wrap it in a ResourceManager so I can't access any attributes of my class from the flow. How can I use the client as instantiated here

with OracleRDS() as client

Oussama Louati

04/09/2021, 2:16 PM

Hello 🙂, I deployed Prefect server 0.14.14. Everything works fine locally. I am a bit confused so if anyone can help me: • I have a workload of Machine learning tasks That i would like to execute inside a docker container • I have everything packaged inside a container • I have an agent running on my machine "`prefect agent docker start --show-flow-logs -l agent-name --no-pull`" • When i do

docker run my_image

, i get this error each time:

Copy code

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named '/root/'")

Should i run the agent inside the container ? Thank you

Paul Prescod

04/09/2021, 3:00 PM

This may be outside of the design criteria for Prefect. The use-case consists of two tasks. Task A is CPU-bound. Task B is network bound. Task A generates data that Task B uploads to a slow server. I want a pool of Task A workers to feed a pool of Task B workers. For now, they are all subprocesses of a single parent process. The number of Task A processes should be the same as the number of Cores (not including "hyperthreads"). I'm more flexible about the number of Task B processes.

Hygor Knust

04/09/2021, 3:32 PM

Is there a way to deploy custom Kubernetes CRDs using the K8s Task Library? I'm trying to deploy a SparkApplication CRD but *i*t seems that there is no task that accomplishes this. Should I use the kubernetes-python inside a Python Task instead? Thanks

Jonathan Buys

04/09/2021, 3:55 PM

Is there a way to setup SSO with Office365 in Prefect Cloud?

xyzy

04/09/2021, 4:04 PM

I tried writing a SQLiteConnection ResourceManager as a demo, but when executing with DaskExecutor it leads to problems with having different threads:

Copy code

ProgrammingError('SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 8032 and this is thread id 2308.')

Is there a way to work around this other than using a LocalExecutor? e.g. a parameter for the ResourceManager to tell it to keep everything that uses it in the same thread?

Josh

04/09/2021, 4:27 PM

Not sure if this should be considered a feature request or discussion idea, but I would hope for updates to patch versions not to introduce breaking changes. I understand we’re still in initial development (0.y.z), it is disheartening to login to monitor flows to see that everything has failed in the last 24 hours because I haven’t updated updated the flows to the newest release. Specifically 0.14.15 introduced things like •

terminal_state_handlers

on Flow objects that will cause a flow to fail if they aren’t present even though the 0.14.15 version defaults to None. •

log_output

on the

ExecuteNotebook

class that defaults to False on the 0.14.15 version, but the execution fails if a flow on a previous version is being run with the attribute not specified

Kevin Kho

04/09/2021, 5:28 PM

Hello everyone! Just wanted to announce that we now have a #random channel for those who want to chat about other stuff.

👍 4

Sean Harkins

04/09/2021, 5:33 PM

We would like to utilize Dask worker plugins (specifically

PipInstall

https://distributed.dask.org/en/latest/plugins.html#distributed.diagnostics.plugin.PipInstall to install dependencies on our workers at worker creation time). We are using a temporary cluster model as described here https://docs.prefect.io/orchestration/flow_config/executors.html#using-a-temporary-cluster. Is it possible to somehow obtain a reference the `DaskExecutor`’s distributed

client

https://github.com/PrefectHQ/prefect/blob/master/src/prefect/executors/dask.py#L209 from within our Flow code (or Task code) so that we can call

register_worker_plugin

? I’ve browsed through the codebase but I’m not sure where in the Flow execution process that the `DaskExecutor`’s

client

is yielded and used?

Trevor Kramer

04/09/2021, 8:50 PM

I'm getting throttling exceptions when trying to submit a batch of jobs to run. What are the limits I need to stay within? An error occurred (ThrottlingException) when calling the RunTask operation (reached max retries: 2): Rate exceeded.

KIRYL BUCHA

04/10/2021, 11:51 AM

Hi to all, I'm trying to use Prefect Jobs as an orchestrator for AWS GLUE Jobs. Does anyone try to implement start of aws glue jobs & monitoring the execution?

Trevor Kramer

04/10/2021, 1:42 PM

Can concurrency tagging be used with resource_managers? We are creating an RDS instance as a @resource_manager and due to AWS limits there can only be 10 running at a time. The resource is used across several tasks so I don't think tagging the tasks would work. Is there another way to limit concurrency to avoid creating more than 10 of the resources protected by @resource_manager at a time?

Slackbot

04/10/2021, 5:44 PM

This message was deleted.

Sean Harkins

04/11/2021, 11:17 PM

We have a somewhat edge case where an external library is generating Prefect Flows https://github.com/pangeo-data/rechunker/blob/master/rechunker/executors/prefect.py#L42 that we register. Prior to registration we need to apply a decorator to each of the Flow’s tasks (in this case to register Dask

WorkerPlugins

). To test this initially I applied our decorator to a test

task

Copy code

def register_plugin(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        client = distributed.get_client()
        plugin = PipInstall(packages=["xarray"])
        client.register_worker_plugin(plugin)
        result = func(*args, **kwargs)
        return result
    return wrapper


@task
@register_plugin
def say_hello():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hello, Cloud")
    return "hello result"

This works as expected and the Dask worker logs show the plugin use.

Copy code

[2021-04-11 23:11:49+0000] INFO - prefect.CloudTaskRunner | Task 'say_hello': Starting task run...
distributed.worker - INFO - Starting Worker plugin pip
distributed.diagnostics.plugin - INFO - Pip installing the following packages: ['xarray']
[2021-04-11 23:11:50+0000] INFO - prefect.say_hello | Hello, Cloud
[2021-04-11 23:11:50+0000] INFO - prefect.CloudTaskRunner | Task 'say_hello': Finished task run for task with final state: 'Success'

However, we don’t have access to the underlying tasks so instead we need to access them from an existing flow, wrap them and replace them within the flow prior to flow registration with

Copy code

for flow_task in flow.tasks:
    wrapped_task = register_plugin(flow_task)
    flow.replace(flow_task, wrapped_task)

But using this approach seems to alter the task’s execution as the worker logs do not report the plugin use, or the Prefect

logger

statements.

Copy code

distributed.core - INFO - Starting established connection
[2021-04-11 22:43:26+0000] INFO - prefect.CloudTaskRunner | Task 'Constant[function]': Starting task run...
[2021-04-11 22:43:27+0000] INFO - prefect.CloudTaskRunner | Task 'Constant[function]': Finished task run for task with final state: 'Success'

Am I missing something obvious in my task wrapping approach and replacement in the flow? Is there a better approach for accomplishing this?

Svyat

04/12/2021, 12:02 AM

Got a general question about configuring a flow to execute on AWS ECS EC2 instance instead of ECS Fargate. Followed this guide to deploy everything using the latter but have discovered I need more resources than what’s allowed for Fargate task definitions. Is there an example/documentation for creating tasks that will execute on a custom EC2 instance? Thanks much!

Rob Fowler

04/12/2021, 3:49 AM

has anyone else used LocalDaskExecutor in k8s? It seems the dask parent management process does not actually work under k8s. I'm getting all sorts of wacky errors. If it is not supported, it might be an anti-pattern, parallel processes inside k8s for prefect?

Rob Fowler

04/12/2021, 9:49 AM

Copy code

docker run -it prefecthq/prefect:latest-python3.6  /bin/bash
root@c281a603b87e:/# prefect version
0.14.12

xyzz

04/12/2021, 11:38 AM

is there a way to set the project_name parameter of the StartFlowRun Task automatically to the current project of the flow or None if its run with core flow.run?

Jérémy Trudel

04/12/2021, 12:52 PM

Hi! I'm trying to register a project using CLI with the following command:

Copy code

prefect register --project "my project" --path myflow.py

All it returns is:

Project 'my project' does not exist

I know it does not exist since I'm trying to register it. Is there something I'm missing?

👀 1

Philip MacMenamin

04/12/2021, 1:23 PM

I can't seem to find the docs for how to exit a flow if a specific task fails. Currently I'm raising

signals.FAIL

within the task, but for some tasks I just want to exit a flow if they don't work, and not have to mark every other task dependent on that task succeeding. Anything I can look at?

ciaran

04/12/2021, 2:25 PM

cc. @Kevin Kho I'm trying to lock down our CDK deployment so that our Prefect Agent, Flow Runner, Executor and workers are in security groups. I.E. Agent & Runner in Sec1, Executor and Workers in Sec 2. Sec1 can talk to Sec 2, they can both talk to the outside world, but they do not allow external ingress. My issue I'm having is that Prefect seems to try to connect to the Public IP of the Dask Scheduler task in ECS, rather than the Local (within the VPC) IP of the Dask Scheduler task. Is there a way to get Prefect/Dask Executor to use the Local (AWS Private IP) of the Scheduler? It would really help trying to make our infra a bit more secure.