job_variables

from prefect.deployments import run_deployment

run_deployment(
    name="my-flow/my-deployment",
    parameters={"some_param": "value"},
    job_variables={"image": "<http://ghcr.io/your-org/prefect-ray:py310-cu121|ghcr.io/your-org/prefect-ray:py310-cu121>"},
)

# sketch
my_flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="flows/ray_flow.py:my_flow",
).deploy(
    name="ray-flow",
    work_pool_name="kubernetes-pool",  # or docker, ecs, etc.
    image="<http://ghcr.io/your-org/prefect-ray:py310|ghcr.io/your-org/prefect-ray:py310>",
)

RayTaskRunner(address=...)

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(task_runner=RayTaskRunner(address="<ray://your-ray-head:10001>"))
def my_flow():
    # your tasks here; they will run on the Ray cluster
    ...

# Deploy this flow with different images per deployment or override per run via job_variables

init_kwargs

RayTaskRunner

address="ray://..."

init_kwargs={"runtime_env": {"image_uri": ...}}

init_kwargs={"runtime_env": {"container": {"image": ...}}}

RayTaskRunner(address=...)

address

init_kwargs={"runtime_env": {"container": {"image": "your-image"}}}

image_uri

RayTaskRunner

ray.init(...)

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

# Connect to an existing cluster (won’t change its pod image)
@flow(task_runner=RayTaskRunner(address="<ray://ray-head:10001>",
                                init_kwargs={"runtime_env": {"env_vars": {"FOO": "bar"}}}))
def my_flow():
    ...

# Start a local Ray runtime and run tasks in a Docker image (host must have Docker)
@flow(task_runner=RayTaskRunner(init_kwargs={
    "runtime_env": {"container": {"image": "<http://ghcr.io/your-org/ray-runtime:py310|ghcr.io/your-org/ray-runtime:py310>"}}
}))
def local_containerized_flow():
    ...

from prefect.deployments import run_deployment

run_deployment(
    name="my-flow/my-deployment",
    job_variables={"image": "<http://ghcr.io/your-org/prefect-flow:py310-cu121|ghcr.io/your-org/prefect-flow:py310-cu121>"},
)

RayTaskRunner(address=...)

init_kwargs={"runtime_env": {"image_uri": ...}}

init_kwargs={"runtime_env": {"container": {"image": ...}}}

ray.init(...)

docker

RayTaskRunner(address=...)

# flow.py
import os
from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

RAY_ADDRESS = os.getenv("RAY_ADDRESS")  # e.g., ray://<head-ip>:10001

@flow(task_runner=RayTaskRunner(address=RAY_ADDRESS))
def my_flow():
    ...

# When deploying, set the env var per deployment/run
# e.g., via job_variables
# job_variables={"env": {"RAY_ADDRESS": "<ray://ec2-gpu-head:10001>"}}

runtime_env={"image_uri": "<your-image>"}

from ray.job_submission import JobSubmissionClient

client = JobSubmissionClient("http://<head-node-ip>:8265")
job_id = client.submit_job(
    entrypoint="python -c 'import your_module; your_module.main()'",
    runtime_env={"image_uri": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/ray:gpu|123456789012.dkr.ecr.us-east-1.amazonaws.com/ray:gpu>"},
    # env_vars / working_dir / pip, etc. as needed
)

RayTaskRunner

ray.init(...)

@flow(task_runner=RayTaskRunner(
    address="ray://<head-ip>:10001",
    init_kwargs={"runtime_env": {"container": {"image": "your-image:tag"}}},
))
def my_flow():
    ...

address

from prefect.deployments import run_deployment

run_deployment(
    name="my-flow/my-deployment",
    job_variables={"image": "<http://ghcr.io/your-org/prefect-flow:py310-cu121|ghcr.io/your-org/prefect-flow:py310-cu121>"},
)

ray.init()

address="ray://<head-ip>:10001"

RayTaskRunner

ray.init(...)

ray.init

ray.init(runtime_env=...)

remote_options(runtime_env=...)

ray

init_kwargs={"runtime_env": {"container": {"image": "...", ...}}}

RayTaskRunner

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(
    task_runner=RayTaskRunner(
        address="ray://<head-ip>:10001",
        init_kwargs={
            "runtime_env": {
                "container": {
                    "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>",
                    # optional: pass Docker run flags
                    "run_options": [
                        "--gpus=all",           # if you need GPUs
                        "--ipc=host",
                        # "--volume=/host/path:/container/path"
                    ],
                    "env_vars": {"EXTRA_FLAG": "1"},
                }
            }
        },
    )
)
def my_flow():
    # Prefect tasks in this flow will run as Ray tasks inside the container image above
    ...

remote_options

.options(...)

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def heavy_gpu_work(x): ...
@task
def cpu_only_work(y): ...

@flow(task_runner=RayTaskRunner(address="ray://<head-ip>:10001"))
def my_flow(image_for_gpu: str, image_for_cpu: str):
    # GPU section
    with remote_options(
        runtime_env={
            "container": {
                "image": image_for_gpu,
                "run_options": ["--gpus=all"],
            }
        },
        num_gpus=1,
    ):
        a = heavy_gpu_work.submit(1)

    # CPU section
    with remote_options(
        runtime_env={"container": {"image": image_for_cpu}},
        num_cpus=2,
    ):
        b = cpu_only_work.submit(2)

    return a, b

--gpus=all

run_options

run_options

remote_options(runtime_env=...)

ray.init(runtime_env=...)

runtime_env

runtime_env

run_options

sudo apt-get update
      sudo apt-get install -y <http://docker.io|docker.io>
      sudo usermod -aG docker $USER
      # re-login or: newgrp docker
      docker info  # should succeed without sudo

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(
    task_runner=RayTaskRunner(
        address="ray://<head-ip>:10001",
        init_kwargs={
            "runtime_env": {
                "container": {
                    "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>",
                    # optional: docker run flags
                    "run_options": [
                        "--gpus=all",     # if you need GPUs
                        "--ipc=host",
                        # "--volume=/host/path:/container/path"
                    ],
                    # pass env to the container
                    "env_vars": {"EXTRA_FLAG": "1"},
                }
            }
        },
    )
)
def my_flow():
    ...

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def heavy_gpu_work(x): ...
@task
def cpu_only_work(y): ...

@flow(task_runner=RayTaskRunner(address="ray://<head-ip>:10001"))
def my_flow(image_for_gpu: str, image_for_cpu: str):
    # GPU section
    with remote_options(
        runtime_env={"container": {"image": image_for_gpu, "run_options": ["--gpus=all"]}},
        num_gpus=1,
    ):
        a = heavy_gpu_work.submit(1)

    # CPU section
    with remote_options(
        runtime_env={"container": {"image": image_for_cpu}},
        num_cpus=2,
    ):
        b = cpu_only_work.submit(2)

    return a, b

ray.init()

ray://...

ray.init(...)

address="ray://<head-ip>:10001"

--gpus=all

docker pull <your-image>

ray.init(address="ray://...")

runtime_env={"container": {"image": "<your-image>"}}

echo $PATH
which docker

which docker

sudo yum install -y docker
sudo systemctl enable docker
sudo systemctl start docker
sudo usermod -aG docker ec2-user

ec2-user

docker

newgrp docker

groups

docker

which docker

/usr/bin

docker

which docker
docker ps
groups

docker

docker ps

docker

newgrp docker

What to look for:
- The “Without container” and “With container” sections should differ in `os_release` (confirming your container image ran).
- If you get errors like “podman not found” or “docker: permission denied”:
  - Make sure `which docker` returns a path in both the driver and inside the non-container task.
  - Re-check group membership took effect (new login/newgrp).
  - Ensure Docker is installed on every worker node, not just the head.
  - Try pre-pulling the image on nodes (`docker pull <IMAGE>`) to rule out registry/auth issues.

Using this with Prefect + Ray
Once the node-level test works, apply the same runtime_env via Prefect-Ray:

- Default container for all tasks in a flow:

- Different images per section/task:

docker

ray attach ray-cluster-config-docker.yaml

which docker
docker ps
id -nG

docker ps

docker

newgrp docker
docker ps

which docker

/usr/bin

export PATH="/usr/bin:$PATH"
which docker

docker pull <http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>

ray status
python -c "import ray; print(ray.__version__)"

python3

python

python - << 'PY'
import os, shutil, ray, subprocess, json

IMAGE = "123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310"

print("Driver PATH:", os.environ.get("PATH"))
print("Driver which(docker):", shutil.which("docker"))

# Connect to the local running cluster on the head
ray.init(address="auto")

@ray.remote
def info():
    import os, subprocess, shutil
    return {
        "PATH": os.environ.get("PATH"),
        "which_docker": shutil.which("docker"),
        "whoami": subprocess.getoutput("whoami"),
        "python": subprocess.getoutput("python -V 2>&1"),
        "os_release": subprocess.getoutput("cat /etc/os-release || true"),
        "has_nvidia_smi": subprocess.call("nvidia-smi >/dev/null 2>&1", shell=True) == 0,
    }

print("\n--- Without container ---")
print(json.dumps(ray.get(info.remote()), indent=2))

print("\n--- With container runtime_env ---")
task = info.options(
    runtime_env={
        "container": {
            "image": IMAGE,
            # Uncomment/add if needed:
            # "run_options": ["--gpus=all", "--ipc=host"],
            # "env_vars": {"EXAMPLE": "1"},
        }
    }
).remote()

print(json.dumps(ray.get(task), indent=2))
PY

os_release

which_docker

/usr/bin

docker

newgrp docker

ray stop

ray up ray-cluster-config-docker.yaml

ray up

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(task_runner=RayTaskRunner(
    address="ray://<head-ip>:10001",  # or address="auto" if running on head
    init_kwargs={
        "runtime_env": {
            "container": {
                "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>",
                # "run_options": ["--gpus=all"]
            }
        }
    }
))
def my_flow():
    ...

from prefect_ray.context import remote_options
with remote_options(runtime_env={"container": {"image": "your-other-image"}}):
    ...

which docker

docker ps

id -nG

ssh -i /path/to/your/key.pem ec2-user@172.31.25.199

which docker
docker ps
id -nG

sudo yum install -y docker
sudo systemctl enable --now docker
sudo usermod -aG docker ec2-user
# Open a fresh login OR run:
newgrp docker
docker ps

# ray-cluster-config-docker.yaml (excerpt)
auth:
  ssh_user: ec2-user

setup_commands:
  - sudo yum install -y docker
  - sudo systemctl enable --now docker
  - sudo usermod -aG docker ec2-user
  # ensure current shell has docker group for subsequent commands
  - newgrp docker || true
  - docker --version

ray up

RayTaskRunner(init_kwargs={...})

docker ps | grep -i ray

ps -ef | grep raylet

sudo nsenter -t <raylet-pid> -m -u -i -n env | grep PATH

docker exec -it <ray-container-name> bash -lc 'which docker || echo no-docker; ls -l /var/run/docker.sock || echo no-sock'

FROM rayproject/ray:2.48.0-py310
      RUN apt-get update && apt-get install -y <http://docker.io|docker.io> && rm -rf /var/lib/apt/lists/*

--volume=/var/run/docker.sock:/var/run/docker.sock

--ipc=host

--privileged

pgrep -f raylet

sudo tr '\0' '\n' </proc/$(pgrep -f raylet | head -n1)/environ | grep PATH

# 1) Confirm raylet PID (on the worker)
pgrep -f raylet

# 2) Show the PATH visible to the raylet process (replace <PID> with output above)
sudo tr '\0' '\n' </proc/<PID>/environ | grep ^PATH=

# 3) Sanity check your current shell
echo $PATH
which docker
id -nG

# 4) Check runtime env agent logs for clues
ls -1 /tmp/ray/session_*/logs | grep runtime_env
grep -i -E 'podman|docker' /tmp/ray/session_*/logs/runtime_env* 2>/dev/null | tail -n 100

python - << 'PY'
import ray, json, os, shutil

ray.init(address="auto")

# Make an info task
@ray.remote
def info():
    import os, shutil, subprocess
    return {
        "PATH": os.environ.get("PATH"),
        "which_docker": shutil.which("docker"),
        "whoami": subprocess.getoutput("whoami"),
    }

# Force the task to run on the target worker via node resource
target_ip = "172.31.25.199"
res = ray.get(info.options(resources={f"node:{target_ip}": 0.001}).remote())
print(json.dumps(res, indent=2))
PY

which_docker

head_start_ray_commands:
      - 'export PATH=/usr/bin:$PATH; ray stop'
      - 'export PATH=/usr/bin:$PATH; ray start --head --dashboard-host=0.0.0.0 --port=6379'
    worker_start_ray_commands:
      - 'export PATH=/usr/bin:$PATH; ray stop'
      - 'export PATH=/usr/bin:$PATH; ray start --address=$RAY_HEAD_IP:6379'

ray up

groups

newgrp docker

ray start

E) Why “podman: not found” despite docker being installed
- Ray’s containerized tasks feature shells out from the raylet/runtime-env agent. If the raylet’s environment cannot find docker (binary not on PATH at the time the process was started), Ray may try podman. Your interactive shell is fine, but the raylet process environment may be missing PATH entries.
- Ensuring PATH (and group membership) for the process that starts Ray fixes this.

Once this test works, you can use the same runtime_env in Prefect-Ray’s RayTaskRunner:
- Global for the flow:

remote_options

/flow_runs/{id}/labels

PREFECT_SERVER_API_AUTH_STRING

PREFECT_API_AUTH_STRING

location /api/ {
      proxy_pass <http://prefect-server:4200/api/;>
      proxy_http_version 1.1;
      proxy_set_header Upgrade $http_upgrade;
      proxy_set_header Connection "upgrade";
      proxy_set_header Host $host;
    }

PREFECT_API_URL

export PREFECT_DEBUG_MODE=1

PREFECT_API_AUTH_STRING

/api/

PREFECT_DEBUG_MODE=1

/api/

PREFECT_API_AUTH_STRING

/flow_runs/{id}/labels

map $http_upgrade $connection_upgrade {
  default upgrade;
  ''      close;
}

location /api/ {
  proxy_pass <http://prefect_api/;>
  proxy_http_version 1.1;

  # Forward client IP/origin
  proxy_set_header Host $host;
  proxy_set_header X-Real-IP $remote_addr;
  proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
  proxy_set_header X-Forwarded-Proto $scheme;
  proxy_set_header X-Forwarded-Host $host;

  # WebSocket upgrade
  proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection $connection_upgrade;

  # Auth headers (for basic auth on the Prefect API)
  proxy_set_header Authorization $http_authorization;
  proxy_pass_header Authorization;

  # Keep WS connections alive
  proxy_read_timeout 3600s;
  proxy_send_timeout 3600s;
  # optional: proxy_buffering off;
}

location / {
  proxy_pass <http://prefect_api/;>
  proxy_set_header Host $host;
  proxy_set_header X-Real-IP $remote_addr;
  proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
  proxy_set_header X-Forwarded-Proto $scheme;
  proxy_set_header X-Forwarded-Host $host;
}

PREFECT_SERVER_API_AUTH_STRING="user:pass"

export PREFECT_API_URL="<http://prefect.example.com/api>"
  export PREFECT_API_AUTH_STRING="user:pass"

PREFECT_SERVER_API_AUTH_STRING

PREFECT_API_URL="<http://prefect.example.com/api>"

export PREFECT_DEBUG_MODE=1

curl -i -N \
    -H "Connection: Upgrade" \
    -H "Upgrade: websocket" \
    -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
    -H "Sec-WebSocket-Version: 13" \
    -H "Authorization: Basic dXNlcjpwYXNz" \
    <http://prefect.example.com/api/events/in>

/api/events/in

PREFECT_SERVER_API_AUTH_STRING

docker exec -it <worker-container> prefect version

kubectl exec -it <worker-pod> -- prefect version

which prefect
  prefect version
  python -c "import prefect, sys; print(prefect.__version__, sys.executable)"

pip uninstall -y prefect
  pip install "prefect==2.14.20"
  prefect version

prefecthq/prefect:2.14.20-python3.11

# Place this in http { } context
map $http_upgrade $connection_upgrade {
  default upgrade;
  ''      close;
}

upstream prefect_api {
  least_conn;
  server 127.0.0.1:4200;  # Prefect API/UI
}

server {
  listen 80;
  server_name <http://prefect.example.com|prefect.example.com>;

  # Increase body size if you upload larger artifacts
  client_max_body_size 50m;

  # Prefect API (includes WebSocket endpoints)
  location /api/ {
    proxy_pass <http://prefect_api/;>
    proxy_http_version 1.1;

    # Forward client details
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_set_header X-Forwarded-Host $host;

    # WebSocket upgrade
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection $connection_upgrade;

    # Pass through Authorization (useful if API auth is enabled)
    proxy_set_header Authorization $http_authorization;
    proxy_pass_header Authorization;

    # Keep WS connections alive
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
    proxy_connect_timeout 60s;

    # Optional: reduce buffering for streaming
    proxy_buffering off;
  }

  # Prefect UI
  location / {
    proxy_pass <http://prefect_api/;>
    proxy_http_version 1.1;

    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_set_header X-Forwarded-Host $host;

    # Optional for SPA routing; uncomment if you see 404s on deep links
    # try_files $uri $uri/ /index.html;
  }
}

listen 443 ssl;

prefect version

/api/flow_runs/{id}/labels

/api/events/in

which prefect
  prefect version
  python -c "import prefect, sys; print(prefect.__version__, sys.executable)"

server_name

server_name <http://prefect.example.com|prefect.example.com>;

/etc/nginx/conf.d/prefect.conf

upstream prefect_api {
  least_conn;
  server 127.0.0.1:4200;  # Prefect API/UI
}

server {
  listen 80;
  server_name <http://prefect.example.com|prefect.example.com>;

  # Optional: increase if you upload artifacts
  client_max_body_size 50m;

  # Prefect API (includes WebSocket endpoints)
  location /api/ {
    proxy_pass <http://prefect_api/;>
    proxy_http_version 1.1;

    # Forward client details
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_set_header X-Forwarded-Host $host;

    # WebSocket upgrade (no map needed)
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";

    # Pass through Authorization (harmless on 2.x, useful if you enable API auth later)
    proxy_set_header Authorization $http_authorization;
    proxy_pass_header Authorization;

    # Keep WS connections alive
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
    proxy_connect_timeout 60s;

    # Optional: reduce buffering for streaming
    proxy_buffering off;
  }

  # Prefect UI
  location / {
    proxy_pass <http://prefect_api/;>
    proxy_http_version 1.1;

    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_set_header X-Forwarded-Host $host;

    # Optional for SPA deep links:
    # try_files $uri $uri/ /index.html;
  }
}

/etc/nginx/nginx.conf

http {}

map $http_upgrade $connection_upgrade {
  default upgrade;
  ''      close;
}

proxy_set_header Connection $connection_upgrade;

sudo nginx -t && sudo systemctl reload nginx

curl -i -N \
  -H "Connection: Upgrade" \
  -H "Upgrade: websocket" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  -H "Sec-WebSocket-Version: 13" \
  <http://prefect.example.com/api/events/in>

export PREFECT_API_URL="<http://prefect.example.com/api>"
  export PREFECT_API_AUTH_STRING="user:pass"

prefect version

<http://prefect.example.com/api>

<https://prefect.example.com/api>

http://<ec2-public-dns>:4200/api

PREFECT_API_URL

{
  "name": "PREFECT_API_URL",
  "value": "<http://prefect.example.com/api>"
}

{
  "name": "PREFECT_API_AUTH_STRING",
  "value": "user:pass"
}

curl -i <http://prefect.example.com/api/health>
curl -i <http://prefect.example.com/api/version>

server_name

server_name <http://prefect.example.com|prefect.example.com>;

sudo nginx -t && sudo systemctl reload nginx

which prefect
prefect version
python -c "import prefect, sys; print(prefect.__version__, sys.executable)"

http(s)://prefect.example.com/api

client.read_work_pool(...)

httpx

json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

prefect config view --show-sources

prefect work-pool ls
prefect work-pool inspect "<YOUR_POOL_NAME>"

inspect

inspect

prefect version

prefect work-pool create "<YOUR_POOL_NAME>" --type process --overwrite
prefect work-pool inspect "<YOUR_POOL_NAME>"

prefect worker start -p "<YOUR_POOL_NAME>"

# For self-hosted server typically:
curl -sv "$PREFECT_API_URL/api/v3/work_pools/<YOUR_POOL_NAME>"

# Check that:
# - Status is 200
# - Body is valid JSON (not HTML, not empty)

prefect version

prefect work-pool inspect "<YOUR_POOL_NAME>"

PREFECT_API_URL

prefect config view --show-sources

prefect[docker]

prefect[kubernetes]