Hi Community, I am trying out the Prefect Fargate Agent. So far I managed to build Flows with a Docker Storage type, schedule them on Prefect Cloud and launch a Prefect Fargate Agent that deploys the Flows after connecting to Prefect Cloud. Now inside AWS ECS I see that the tasks fail due to:

STOPPED (CannotPullContainerError: Error response from daem)

To my understanding the launched Fargate Instance does not have any access to the Docker Image of the Flow. Concerning the solution from https://docs.prefect.io/orchestration/execution/storage_options.html#docker # Non-Docker Storage for Containerized Environments for rapid deployment, which image would the

metadata={"image": "repo/name:tag"}

refer to in the example given?

Hey guys, I’m trying to run a flow on a custom base_image - it’s built from

prefecthq/prefect:python3.7

and installs a bunch of other junk in there too. The Flow builds and registers, but when I kick it off I get the following:

Failed to set task state with error: ConnectionError(MaxRetryError("HTTPConnectionPool(host='host.docker.internal', port=4200): Max retries exceeded with url: /graphql/alpha

Searching in this channel I see an issue that was identified around this and fixed in April, so I’m not sure if I should still be seeing it, or if there’s config that needs to be passed to the Docker agent to run it? It’s running on Docker agent locally on Macbook if that’s any help.. Prefect version installed locally is 0.11.1, [upgrading now to test] and python in local and the docker image is 3.7 UPDATE: Never mind, upgrade to 0.11.5 did it!

Working on upgrade to 0.11.... from a _state_handler_ invocation, we need to access the return value of write() method on a custom ResultHandler (now wrapped inside a ResultHandlerResult instance as I understand it). The state handler receives a _new_state_ instance, but as I read it there's a getter on State.result that gives back a scalar value rather than the result instance from which I might access location attribute where result_handler.write() retval appears to be stored. What is the right way to access this?

Hi, I'm using mapped tasks to run a flow locally using output = flow.run(). My mapped task calls have outputs (which should be an array based on what I'm passing in). Can i please get some assistance with accessing the array outputs from each mapped task call after I've run flow.run() when my result handler looks something like this attached screenshot? The data I need is circled at the bottom but I can't work out the syntax needed to get through the <Task: Save Sim Data> section. Thanks

While trying to set log format in kubernetes for the scheduler, I've tried setting it like this:

- name: PREFECT__LOGGING__FORMAT
              value: '{"level": "%(levelname)s", "message": "%(message)s"}'

And I can see the pod has the environment variable set.

kubectl exec -it prefect-scheduler-6b96994c6-qtqh5 --namespace=data -- /bin/sh -c 'echo "password: $PREFECT__LOGGING__FORMAT"'                                                                                  <aws:default>
password: {"level": "%(levelname)s", "message": "%(message)s"}

But the logs are still in the default format:

kubectl --namespace data logs prefect-scheduler-6b96994c6-qtqh5 -f                                                                                                                                                                   <aws:default>
[2020-06-09 11:36:54,302] INFO - prefect-server.Scheduler | Scheduler will start after an initial delay of 275 seconds...

Does anyone have any idea? Or even better, I'm trying to get logs in datadog to be parsed correctly.

I think Prefect could be poised to be the next big data engineering tool, one gap I see though is that it can really only operate "serverless" on AWS with Fargate which I'm not sure even applies. On GCP it seems like Cloud Run could be a good deploy target for tasks that take less than 15 minutes. You'd need something to supervise the requests or you could utilize cloud pubsub. Any interest in this from the community? I'd really like to not worry about running a full blown k8s cluster to run Prefect. Airflow on the other hand is provided as a hosted service by Google.

Is there a set (or estimated) limit for the amount of data you can cache with a

PrefectResult

? I’ve hit a wall and flows are failing with:

Failed to set task state with error: HTTPError('413 Client Error: Payload Too Large for url: http://???:4200/graphql/graphql/alpha')
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 119, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1096, in set_task_run_state
    result = self.graphql(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 213, in graphql
    result = <http://self.post|self.post>(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 172, in post
    response = self._request(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 318, in _request
    response.raise_for_status()
  File "/usr/local/lib/python3.8/site-packages/requests/models.py", line 941, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 413 Client Error: Payload Too Large for url: http://???:4200/graphql/graphql/alpha

Saw this error for our scheduled flow:

04:15:18 UTC
INFO
prefect-cloud.Lazarus.FlowRun
Rescheduled by a Lazarus process. This is attempt 1.
04:15:45 UTC

ERROR
agent
HTTPSConnectionPool(host='172.20.0.1', port=443): Read timed out. (read timeout=None)

Hey guys, looking for advice on best practice with Secrets for our use case. we’re using an EC2 instance to host our Prefect Server, and the main execution env is Fargate. Fargate tasks will need to be able to access various secrets in order to do stuff like S3Download,/Upload, various other things. From reading the docs on the various ways to inject a Secret into the system this is what it currently seems like… For static items like AWS creds • Configure the EC2 env with variables that can be interpolated into the config.toml • Start Fargate agent with this same env, configure the “secret”s section of the agent with the secrets to inject • Flows on Fargate using S3Tasks can pull those injected vars from the environment and run This will work for a handful of cases but not all, so I’m trying to figure the best method of getting runtime variables/secrets into a Flow. A given Flow can read from prefect.context.secrets… but how do I actually propagate them through the system above without doing it manually/adding them to the Core UI?

Hi there. I’m having some trouble getting parameters to work on my flow comprised of ShellTasks. Is there anyway to get a parameter into a ShellTask? This doesn’t seem to work:

with Flow("toy_flow") as flow:
    a = Parameter("a")
    b = Parameter("b")
    c = Parameter("c")

    job1_task = ShellTask(name="job1", command=f"""echo {a} {b} {c}""")

Trying to track down strange behavior vis-a-vis cacheing; I run a flow 3 times with same input, the first time task in question is run and then cached and flow succeeds; the second time task is satisfied by cache hit and flow succeeds; the third time (and fourth and 5th times) the task seems to neither be satisfied by cache hit, nor run, and downstream dependencies fail -- log from successful cache hit looks like:

Task 'run_...': 1 candidate cached states were found
11:19:32 PDT
INFO
GCSTmpFileHashResultHandler
Starting to download result .....
INFO
GCSTmpFileHashResultHandler
Finished downloading result ....
DEBUG
CloudTaskRunner
Task 'run_...': Handling state change from Pending to Cached

Log from same task with subsequent flow runs neither shows the file download nor logs that no valid cache results were used:

Task 'run_...': Starting task run...
11:23:43 PDT
DEBUG
CloudTaskRunner
Task 'run_...': 3 candidate cached states were found
11:23:43 PDT
DEBUG
CloudTaskRunner
Task 'run_...': Handling state change from Pending to Cached
11:23:43 PDT
DEBUG
CloudTaskRunner
Task 'run_...': can't set state to Running because it isn't Pending; ending run.
11:23:43 PDT
INFO
CloudTaskRunner
Task 'run_...': finished task run for task with final state: 'Cached'

Suggestions for tracing this further would be appreciated -

Hi, I have made flow with few mapped tasks it looks like this:

out_a = task_a(input)
out_b = task_b.map(out_a)
out_c = task_c.map(out_b)

Now the flow is waiting for task_b to end and then starts with task_c. Is it possible to change that behavior and create a Flow where one element of out_a is flowing through the full pipe before the next element is taken by the worker?

Hi all. I just spotted that we now have a first implementation of GreatExpectations in GitHub HEAD! Great stuff... I try to run the example and wonder how to access the GE return json with the validation results?

is this the appropriate venue to report a bug? or is there somewhere better?

Hi, i am running my flows on a RemoteDask Environment. I can monitor how the memory usage accumulates over time (with every scheduled run) until at one point the memory is full, dask becomes unresponsive and logs this:

distributed.worker - WARNING - Memory use is high but worker has no data to store to disk.  Perhaps some other process is leaking memory?  Process memory: 3.43 GB -- Worker memory limit: 4.18 GB

All the runs complete successfully and I have the feeling, but still the memory used does not get released. I am not really sure where to start debugging. Is there a way to force a memory release? The only other option I currently see is to force a dask-worker restart after the flow run finishes, but that feels very hacky.

Hi guys! I've been looking around but I did not find anything: Is there a recommended way to upgrade the Prefect Core Server to a new version while in production? What I think the steps are: 1. Stop the prefect server 2. Run

pip install prefect --upgrade

3. Run

prefect server start

Is this correct? Would that be it or am I missing some steps?

Hi. I have bit of trouble getting the UI to work as it can't connect to the graphql. It keeps saying "Attempting to connect... http://localhost:4200/graphql/". I have set the host I want to use in my config.toml: ############## config.toml ############## backend = "server" [server] host = "http://srvdocker01" port = "4200" host_port = "4200" endpoint = "${server.host}:${server.port}" [server.ui] host = "http://srvdocker01" port = "8081" host_port = "8081" endpoint = "${server.ui.host}:${server.ui.port}" graphql_url = "http://srvdocker01:4200/graphql" ########################################## This setting is reflected when I run the server (see pic):

I am trying to use the Orchestration GraphQL API Python Client and it seems like there is barely any documentation on it. I used the interactive API on the prefect cloud website to build my query, but it has "where" statements, and the small snippet of code in the prefect docs about the python client don't give any details on how to use "where" statements with the python client. https://docs.prefect.io/orchestration/concepts/api.html#getting-started

is there an ETA on the 0.11.6 release? 🙂

Hi All ... prefect newbie here. Sorry if the answer to my question is obvious and somewhere. But I couldn't find it. I am struggling to figure out how to access the state of a task (task a) within another task (task b). I want task b to always run whether or not task a succeeds or fails. I want task b to do-stuff if task A fails, or skip if task A succeeds.  Sort of like this

@task 
def a ():
   < do something that might succeed or fail >
@task (trigger = always_run)
def b():
  < if task a status==FAIL .. do something >
  < if task b status=SUCCESS .. skip >    

with Flow as flow:
    res1 = a()
    b(res1)
flow.run()

This is part of a larger flow with other tasks ahead and after a&b. I tried

trigger=any_failed

on task b, but that causes task b to fail if task a succeeds (due to trigger not satisfied) .. which is not what I want. I want task b to always show success. Again .. sorry if this is obvious and its just a newbie-thing. Feel free to just point me to the right place in the doc. Thanks in advance

Back again, looking for advice on secrets. I’m trying to inject secrets into Fargate containers and not getting very far. As I understand it, the native AWS Secret Manager support allows me to specify a secret ARN, and if my secret name is something like CREDS, I add the following to the FargateAgent config:

"secrets":[{"name": "CREDS", "valueFrom": "arn:aws:secretsmanager:eu-west-1:11111111:secret:local/aws/credentials-abcd"}]

In Flow, I read like this:

creds = prefect.context.secrets.CREDS

But I keep getting the following:

AttributeError: 'dict' object has no attribute 'CREDS'

Confused face.

Has anyone run into the following issue trying to use the Dask Cloud Provider environment?

Failed to load and execute Flow's environment: TypeError("object NoneType can't be used in 'await' expression")

I'm able to run the flow over a local cluster with two workers just fine, it's only when I register a flow using the Dask Cloud Provider environment that I have issues 🤔

Hi, for my company I'm searching for a task management infrastructure (like airflow, luigi, prefect). Is it possible to run the enterprise flavor of prefect on our own hardware (k8s)? Its no option to rely on other people's infrastructure.

how do I make my flows visible in the server ui? (prefect server start) I can start the flows, no problem, but how to connect to the server instance? Can you please point me to the related documentation?

Need help with below error:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 668, in complete_value_catching_error
    return_type, field_nodes, info, path, result
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 733, in complete_value
    raise result
  File "/prefect-server/src/prefect_server/graphql/states.py", line 73, in set_state
    task_run_id=state_input["task_run_id"], state=state,
  File "/prefect-server/src/prefect_server/api/states.py", line 91, in set_task_run_state
    f"State update failed for task run ID {task_run_id}: provided "
graphql.error.graphql_error.GraphQLError: State update failed for task run ID 63293e14-b1d4-4d2e-ae21-e9aeb8edfade: provided a running state but associated flow run 73a41de3-adc3-4a48-9b57-9b7bdb6094f7 is not in a running state.

So my workflow involves running some commands inside docker containers. The workflow itself aren't huge but the docker execution can take several seconds (should be under 1min though). I am currently running with couple of dask workers with limited memory i.e. 500MB. Workflow works fine for small no. of requests but as I start hitting multiple requests, workers starts dying and I see this error in logs prefect server logs. Although this is just a testing system and actual prod environment will have higher memory limits but still would like to know if this error is expected and if there is any way to avoid/handle this?

I have another question about checkpointing, results, and skipping task reruns. In my toy example below it runs each task again despite the presence of a LocalResult. Additionally, some of the tasks seem to run twice in a given flow run. Not sure if I have found a bug or if I am doing something wrong. Version I'm using is

0.11.5+134.g5e4898dde

I am running on Win 10 and have verified I have the environment variable

PREFECT__FLOWS__CHECKPOINTING=true

from prefect import Flow, task, unmapped, Parameter
from prefect.engine.results import LocalResult
from prefect.engine.executors import LocalDaskExecutor
from prefect.engine.cache_validators import all_parameters

lr = LocalResult(location='{flow_name}-{task_name}-{x}-{y}.pkl',
                 validators=all_parameters
)

@task(log_stdout=True, checkpoint=True)
def add(x, y):
    print(f'add ran with {x} {y}')
    try:
        return sum(x) + y
    except TypeError:
        return x + y

with Flow('iterated map', result=lr) as flow:
    y = unmapped(Parameter('y', default=7))
    x = Parameter('x', default=[1,2,3])
    mapped_result = add.map(x, y=y)
    out = add(mapped_result, y)

flow.run(executor=LocalDaskExecutor())

For testing flows: https://docs.prefect.io/core/idioms/testing-flows.html , when you run anything with

flow.run()

do you need to have the prefect server up? As my test just seems to hang once it hits that part.

hey last night I experience an unknown error to me. here is a pic of the logs

Anybody have experience with a different AWS Access Key ID than what’s in their Secrets showing up in the boto3 session?

boto3.Session().get_credentials().access_key

vs.

PrefectSecret("AWS_CREDENTIALS")["ACCESS_KEY"]

Pretty sure I followed these instructions: https://docs.prefect.io/core/concepts/secrets.html#default-secrets. Both values are keys, but the one in the boto3 session is not one that I recognize.