Hi All. I'm getting a value error when deploying flows outside of working directory, which is already flagged as an issue https://github.com/PrefectHQ/prefect/issues/6777 created by @Cole Murray. Could you please advise if there is any progress on that? Is there a way to go around this error? 🤔

Hi, I am trying to run a flow from within a flow with result persistence on and get the following error:

line 139, in from_flow
    result_storage=flow.result_storage or ctx.result_factory.storage_block,
AttributeError: 'Flow' object has no attribute 'result_storage'

Any help would be appreciated.

Hi! Just a quick question about prefect v2: is it possible to parametrize flow runs based on the scheduler that triggered its execution, like in Prefect v.1? Thanks!

Hello all, We are using git storage and Prefect version 1.2.4. We are randomly getting the following message that fails our production flows: FAIL signal raised: FAIL('b49f14ec-a3dd-4976-9dd6-8577cfa81ddc finished in state <Failed: "Failed to load and execute flow run: PermissionError(13, \'Access is denied\')">') It appears that the prefect agent cannot access some resource afte rthe cloning operation? Any insight or suggestions are appreciated.

I'm getting a JSONdecodeError (assuming bad response?) when trying to save AWS credential block within a GitHub action https://github.com/anna-geller/dataflow-ops/issues/7 I'm not sure why this is happening. Could there be something wrong with my inputs here?

hi, I just moved from prefect1 to prefect2 and having some issues with running agents n flows in ecs. I have flow that deploys another flow at scale (using same task definition, no reregister). the flows are created in the ui but • some of them(quite a few) are stuck in pending n not picked by the agent • I think agent is crashing because all of the logs being pulled there (can I disable agent pulling in logs? I assumed stream_output in infrastructure was meant to do that, but that didn't help much) I also noticed that at a time, only around 40 ecs tasks are being created at a time, n the next set of tasks are created only after these ones are completed. didn't have this issue in prefect1, could easily run around 500 tasks with no issue

@Anna Geller I very much like your article on handling streaming data with prefect. However should I wish to implement same for exchange data that doesn’t run 24/7 but within specific times on week days. Can I use the UI to schedule it in such a manner that the streaming starts and stops within the opening and closing time of the exchange?

Hi, Is there a way to retry a

ShellTask.run(run_id :str)

with a new value of

run_id

?

Hi folks -- I feel like I'm missing something, but is there support for running a docker container as a task in Prefect 2.x? I see https://docs-v1.prefect.io/api/0.15.13/tasks/docker.html which looks like what I want in 1.x, but so far the only Docker-related references I've found is how to run flows within a container. Specifically our use case is that there's a tool (not python) that someone has bundled into an image for portability. It does it's thing, produces some output (tbd on whether that's just to stdout or saving to files in the container that need to get copied out) and we'd like to put this into a prefect-managed pipeline. Edit: I do see the prefect-docker collection (https://prefecthq.github.io/prefect-docker/). This looks like what I want -- or at least pretty close. I'd like ability to pull off files (or maybe return standard out from the command being run?) ... I'll start looking through the source to see how I might add that functionality, since I don't see an obvious answer in the exposed API.

Hi, I am still getting

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

Here’s my setup:

prefect==2.7.0

,

prefect-aws==2.1.0

. ECS block definition:

ecs_task_block = ECSTask(
    task_definition_arn=task_def_arn,
    cluster=f'{env}-prefect-cluster',
    image=f'123456789.dkr.ecr.{regions[env]}.<http://amazonaws.com/{image_region[env]}-stride-enrollment-flows-ecr:{ecr_image_tag}|amazonaws.com/{image_region[env]}-stride-enrollment-flows-ecr:{ecr_image_tag}>',
    vpc_id=vpcs[env],
    task_customizations=[
        {
            "op": "add",
            "path": "/networkConfiguration",
            "value": {
                "awsvpcConfiguration": {
                    "subnets" : subnets[env],
                    "securityGroups" : security_groups[env],
                    "assignPublicIp" : "DISABLED"
                }
            }
        }
    ],
    configure_cloudwatch_logs=True,
    stream_output=True,
)
ecs_task_block.save(f'{env}-ecs-block', overwrite=True)

I can see that prefect is still creating new task definitions; one of the things it is overriding on the task def is

aws-logs-prefix

to be the flow-run-id. Is there a different way to be able to access logs in cloudwatch without setting a different name?

Hey, just confirming because the docs aren't super clear. For the flow decorators retry argument: 'An optional number of times to retry on flow run failure.' Can I confirm that this does NOT then include flows that fail due to a Crash state?

Using prefect 1.0, is it possible to set the concurrency limit for a specific flow to 1 at a time and not the rest?

Sorry, lots of questions because Prefect seems to be erroring much more than usual. I have an automation that cancels and then retries a flow that has been waiting for 30 mins. For the last few days I am being notified with an error of:

State message: Submission failed. IndexError: list index out of range

cc: @Will Raphaelson in case this is good info for you.

Getting a strange new error during flow submission:

Submission failed. botocore.exceptions.ParamValidationError: Parameter validation failed: Unknown parameter in input: "deregisteredAt", must be one of: family, taskRoleArn, executionRoleArn, networkMode, containerDefinitions, volumes, placementConstraints, requiresCompatibilities, cpu, memory, tags, pidMode, ipcMode, proxyConfiguration, inferenceAccelerators, ephemeralStorage, runtimePlatform

I don’t get this error if I deploy the flows locally, but when the flows are deployed from circleci, this error is generated.

Hi, I want to retry a task with updated arguments. The value of these arguments will be dynamically generated inside the task during the run. Is there a way to pass these arguments to the next retry by raising a

FAIL

signal?

class K8sSparkSubmitTask(ShellTask):
    def __init__(
        self,
        command: str = None,
        max_retries: int = 10,
        retry_delay_seconds: int = 3,
        master_uri: str = None,
        deploy_mode: str = None,
        job_name: str = None,
        conf_kwargs: dict = None,
        env: dict = None,
        helper_script: str = None,
        shell: str = "bash",
        return_all: bool = False,
        log_stderr: bool = False,
        **kwargs: Any,
    ):
        self.command = command
        self.master_uri = master_uri
        self.deploy_mode = deploy_mode
        self.job_name = job_name

        self.conf_kwargs = conf_kwargs or {}

        super().__init__(
            **kwargs,
            command=command,
            max_retries=max_retries,
            retry_delay=timedelta(seconds=retry_delay_seconds),
            env=env,
            helper_script=helper_script,
            shell=shell,
            return_all=return_all,
            log_stderr=log_stderr,
        )

    def run(
        self,
        run_id: str = None,
    ) -> str:

        if some_condition:
            # TODO : Update the value of run_id so that
            # new value of run_id will be reflected from next retry onwards
            raise FAIL("Retrying with an updated value of run_id")
        else:
            # Do something
            pass


with Flow("Test") as f:
    k = K8sSparkSubmitTask()
    k.map(run_id=unmapped("123"))


f.run()

Hi Community, Using prefect 2.0, is it possible we can track/export the prefect flows with tools like Opentelemetry or Prometheus? Is there any open pull request for this feature?

Hello guys, I have the task which stay in running state (task in subflow), however inside it’s done. So my main flow is in failed state, but there is no a problem, it shouldn’t have a failed state. How can I manage the running tasks, which didn’t change their state after complet?

Hi Community, We have currently many variations of a task that would fit into a parent flow, but are you able to pass a task existing inside the repo as a parameter to the parent flow?

Hi everyone, if we enable result persistence with Prefect2.0, will the next task always get the data from storage or only in case of restarts? In Prefect1.0, we saw that data always comes from our storage (s3) which adds couple of seconds for each task

Hi everyone, question regarding putting flow runs to different agents, Prefect 1 (Prefect 2 is in future plans 😞 ) - see thread

How can we create pods on different instance types in k8s? Currently I have one work queue and one agent which is scheduling pods based on whatever nodegroup is available, I used the helm installation to setup the cluster. I am probably missing on some configuration

Reposting here for some discussion: Hi Prefect Team, first a thank you to the wonderful job you have done creating this platform! I have a question about heterogeneous flow run: 1. Is there any way for a user to create a flow that runs on different type of hardware in the queue? I need to create a flow that partially run on CPU kubernetes queue and partially on GPU enabled nodes. There doesn't seem to be a way to do this and each flow can only run on pure CPU or all with GPU. 2. If not, what's the best practice for doing this? Should I break a flow into separate subflows and deploy them separately and run each step by hand? Or can I trigger this automatically by having a task run other deployed flow in a different queue. 3. Is there any plan to add per task tagging system for agents so when an agent pulls from a queue it only pulls tasks with a certain tag

Is there a way to access from inside the flow the time it was scheduled to run?

Hello, I hope all prefect community is good. I am launching a

prefect orion start

command in a docker container connected to a postgres server. My prefect server is open to the world with http access. Is my architecture susceptible for an sql injection? Can prefect be attacked by SQL injection? Should I take care of this? Thanks for taking the time for reading my post. I wish you a nice day 🙂

I have a use case that requires limiting the concurrency of a subset of flows on a work queue that match a certain pattern (flows with a prefix in their flow_run_name or tag) I made a post on this https://github.com/PrefectHQ/prefect/issues/8455 Curios to know if anyone has a way around this problem. Creating new work queues is not a viable solution for me. I basically want to create concurrency limits on all flows with a certain parameter value inside of a given work queue and still respect the work queue's top-level concurrency limit.

Hi there, I seem to be having issues running the blob_storage_list() function from the prefect_azure.blob_storage module. I’m using the Azure Blob Storage Credentials block to store and load my connection string, but I also tried hardcoding my connection string and neither seems to work. The errors I’m returning are: KeyError: 0 RunTimeError: Event loop is closed Was wondering if anyone else was running into the same issue? For reference I am on version 2.7.12

Hi all, I’m trying out the new

flow_run_name

parameter in 2.7.12 but I’m having some difficulty and I’m unable to create the deployment. It’s a flow-subflow scenario with iteration:

@flow(name="my_subflow", flow_run_name=f"my_subflow run: {iteration}", log_prints=True)
def my_subflow(iteration: str):
  print(f"my_subflow run: {iteration}")

@flow(name="main_flow", log_prints=True)
def main_flow():
  iteration_list = ["iteration_1", "iteration_2", "iteration_3"]
  for iteration in iteration_list:
    my_subflow(iteration)

When I try to deploy either of the flows with something like

prefect deployment build -n main_flow_run_name_test_deployment -q my_queue ./flows/flow_run_name_test.py:main_flow --apply

I get the error:

Script at './flows/flow_run_name_test.py' encountered an exception: NameError("name 'iteration' is not defined")

VScode is also flagging the

iteration

variable as not defined. How would I pass the iteration dynamically if the flow decorator is outside the function?

I'm struggling to understand how to use a k8s custom job template for a kubernetes-job - I found some great examples from @Anna Geller but I'm not sure how to use/implement it: https://github.com/anna-geller/prefect-deployment-patterns/tree/main/blocks/infrastructure_blocks/kubernetes-job/kubernetes_job_custom_job_template

Any help would be appreciated

Hi. I want to use ECSTask, can I specify a subnet? I found the specification of vpc_id. . . I want to run it in a private subnet, thank you.