we also have a feature that allows triggering a flow run from an already existing (i.e. already created) deployment via API call - is that what you are trying to do? if so, you can do this via

OrionClient.create_flow_run_from_deployment

or via a direct API call. If not, can you describe the problem you try to solve a bit more? What would be needed for your use case - triggering a flow run that gets executed on the agent you spun up on your EC2 instance? if so, then the CLI command above should have the same effect as it will create a flow run scheduled for now If you attach a schedule on your

DeploymentSpec

this will also trigger a flow run running on your agent, provided that the

DeploymentSpec

definition matches the work queue filter criteria

also one more thing i am having an issue wtih, is my ssh tunnel gets disonnected and any active processes i have running in the background gets stopped. i have 2 ssh tunnels running, one where the prefect agent is listening for active deployments, and another one that has a redis-server running. my problem is after i loginto the cloud UI, and the ssh tunnel gets disconnected, i notice i have a http error on the prefect side, a 403 HTTP response. not sure whats happening here

@Anna Geller, if i have a file called deployments.py that contains all the deployments, how do i call

OrionClient.create_flow_run_from_deployment

? right now, i am crating a hackey way of doing this by creating another python file called execute_deployments that runs prefect shell commands of creating deployments and running them.

each flow run is independent of each other

not sure whats happening here

me neither. hard to debug this like this remotely. Can you try Cloud 2.0 maybe https://beta.prefect.io/? this would give you UI available from the public Internet from anywhere once you attach the schedule, no need to invoke your flow run from a client, Prefect will automatically invoke the flow run based on your schedule.

...or with the while loop approach you wouldn't invoke any deployment, you would just do:

python your_never_ending_script_creating_millions_of_flow_runs.py

The deployment file has about 10 deployment specs, reprsenting 10 coins. to automate this from a shell call, i created another file called stream_deployments.py with contents as such:

import os
import subprocess
from time import sleep
from typing import List

from exchange_feeds.constants import ACCOUNT, FLOW_NAME, WORKSPACE_NAME

ID = os.path.join(ACCOUNT, WORKSPACE_NAME)


def cloud_login():
    key = os.environ["PREFECTCLOUDKEY"]
    cmd = f"prefect cloud login --key {key} -w {ID}"
    subprocess.run(cmd, shell=True)


def create_deployments(deployment_name):
    cmd = f"prefect deployment create {deployment_name}"
    subprocess.run(cmd, shell=True)


def run_unscheduled_deployments(deployment_names: List[str]):
    for d in deployment_names:
        cmd = f"prefect deployment run {d}"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        print(result.stdout)


def start_prefect_agent(agent_name):
    cmd = f"prefect work-queue create {agent_name}"
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    uiud = result.stdout.strip("UIUD()").replace("(", "").replace(")", "")
    cmd = f"prefect agent start {uiud}"
    subprocess.run(cmd, shell=True)


coins = (
    "avaxusd",
    "bnbusd",
    "btc-perp",
    "btcusd",
    "dogeusd",
    "dotusd",
    "ethusd",
    "lunausd",
    "solusd",
    "xrpusd",
)
ftx_L1_deployment = [f"ftx_L1_{coin}" for coin in coins]
ftx_blotter_deployment = [f"ftx_blotter_{coin}" for coin in coins]


def sign_into_cloud_and_create_deployments():
    print("logging into prefect cloud")
    cloud_login()
    print("Successfully logged into prefect cloud")
    print("Creating redis to postgres deployment")
    create_deployments("redis_to_postgres_deployment.py")
    sleep(5)
    print("Creating FTX deployments")
    create_deployments("ftx/orderbook_deployment.py")
    sleep(5)
    create_deployments("ftx/blotter_deployment.py")
    sleep(5)
    create_deployments("kraken/orderbook_deployment.py")
    sleep(5)
    create_deployments("kraken/blotter_deployment.py")
    sleep(5)


if __name__ == "__main__":
    sign_into_cloud_and_create_deployments()
    activate_coin = ["ethusd", "btcusd"]
    inactivate_flows = [f"{FLOW_NAME}/ftx_L1_{coin}" for coin in activate_coin]
    run_unscheduled_deployments(inactivate_flows)
    inactivate_flows = [f"{FLOW_NAME}/ftx_blotter_{coin}" for coin in activate_coin]
    run_unscheduled_deployments(inactivate_flows)
    inactivate_flows = [f"{FLOW_NAME}/kraken_L1_{coin}" for coin in activate_coin]
    run_unscheduled_deployments(inactivate_flows)
    inactivate_flows = [f"{FLOW_NAME}/kraken_blotter_{coin}" for coin in activate_coin]
    run_unscheduled_deployments(inactivate_flows)
    start_prefect_agent("feed_to_redis_agent")

It mainly freezes due to memory issues. in the above file, i am testing

activate_coin = ["ethusd", "btcusd"]

these two coins. and hence noticed all the memory usage. I am using prefect work-queue agent to deploy my tasks so thats the only service im using i believe.

and btw, thanks so much for sharing your code! 🙏 This helps troubleshooting more than any explanation 🙂 given that your tasks are mainly using IO operations (grabbing some data from the exchange, writing this to Redis, waiting in between API responses), I think that the default ConcurrentTaskRunner is a better choice for your use case than a DaskTaskRunner. I would try switching that first before troubleshooting any further. You can benchmark the performance - I think in your use case, Dask adds more overhead than it helps, but I could be wrong

@Anna Geller, thanks for that rec. I tried it and now the number of processes, if i deploy 10 python processes, theres only 10 other prefect agents, so much better. before it was 2 and it went to 120.

also adding the default concurrent task runner has sped up the performance dramatically, thanks