Great question. Here’s a concise, “state of the art” approach for passing secrets to ephemeral containers (Docker/Kubernetes), with rotation and late selection in mind. TL;DR (recommended patterns) - Best on Kubernetes: use workload identity to fetch secrets at runtime from a secrets manager (no secrets injected). If you can’t, use the Secrets Store CSI driver to mount secrets as files with rotation. - Best on plain Docker: fetch from an external secrets manager at runtime (short‑lived creds). Avoid baking secrets into images or passing long‑lived secrets via env vars. - Rotation: rely on the external manager’s rotation, and always resolve secrets at runtime by stable logical names (late binding). Kubernetes options (ranked) 1) Workload identity + runtime fetch (no secret injection) - Use cloud-native identity so the pod can call the secret manager directly: - AWS IRSA: EKS IRSA - GCP Workload Identity: GKE Workload Identity - Azure AD Workload Identity: Azure Workload Identity - App loads the secret at startup (or per use) via SDK. Rotation is automatic: the next retrieval returns the new value. 2) Secrets Store CSI driver (mount secrets as files, with rotation) - Use the CSI driver + provider for AWS/GCP/Azure/Vault: - CSI driver: Secrets Store CSI driver - AWS provider: AWS provider - GCP provider: GCP provider - Azure provider: Azure provider - Vault provider: Vault CSI provider - Configure

SecretProviderClass

with

rotationPollInterval

and mount to your pod as a tmpfs-backed volume. Your code reads from files; rotation happens without redeploy. Example (sketch):

apiVersion: <http://secrets-store.csi.x-k8s.io/v1|secrets-store.csi.x-k8s.io/v1>
kind: SecretProviderClass
metadata:
  name: app-secrets
spec:
  provider: aws
  parameters:
    objects: |
      - objectName: "prod/my-api-key"
        objectType: "secretsmanager"
        jmesPath:
          - path: secret
            objectAlias: api_key
---
apiVersion: v1
kind: Pod
spec:
  serviceAccountName: app-sa  # with IRSA/Workload Identity
  volumes:
    - name: secrets-store
      csi:
        driver: <http://secrets-store.csi.k8s.io|secrets-store.csi.k8s.io>
        volumeAttributes:
          secretProviderClass: app-secrets
  containers:
    - name: app
      volumeMounts:
        - name: secrets-store
          mountPath: "/var/run/secrets/app"
          readOnly: true

3) Vault Agent sidecar (dynamic secrets + auto-renewal) - Sidecar authenticates (KSA/JWT), renders secrets to a shared tmpfs, and renews/rotates: - Vault Agent Injector - Good when you need dynamic DB creds or fine-grained TTLs. Notes for Kubernetes - Prefer files over env vars for highly sensitive secrets (env vars can leak into crash dumps and diagnostics). - If you must use Kubernetes Secrets: mount as volumes (tmpfs), reference by name, and rotate by updating the Secret or syncing from an external manager. Still less robust than CSI/identity patterns. Docker (standalone) - There’s no first-class “Docker secrets” outside Swarm. For ephemeral “docker run” or Compose: - Fetch from an external manager at runtime (AWS Secrets Manager, GCP Secret Manager, Azure Key Vault, Vault) using an identity/token you bootstrap into the container (OIDC/JWT, short-lived STS, or a Vault AppRole/oidc login). - Store the fetched secret only in memory or write to a tmpfs path.

- Avoid: embedding secrets in images, passing long-lived secrets via env, or build-time secrets for runtime needs (BuildKit

--secret

is build-time only). - If you control the host: bind-mount a tmpfs file populated by a local secret helper, or run a sidecar/agent (Vault Agent) that writes to a shared volume. Rotation and late selection (based on usage) - Always reference secrets by a stable logical name (e.g., “payments/api-key”) and resolve at runtime. - Let the secret manager rotate the underlying value (or use dynamic secrets). Your app does a fresh read per run (or per use), so it naturally picks up the rotated value. - If you need to choose secrets at runtime (e.g., tenant-based): - Pass a non-sensitive selector (tenant, environment) to the container. - Compute the secret name/key from the selector and fetch from the manager. Using Prefect (3.x) to run ephemeral containers - Keep sensitive values out of

job_variables

. Load secrets at runtime in your flow: - Prefect Secret block (generic): resolves centrally, easy to rotate via block update - Cloud integrations: AWS/GCP/Azure credentials and secret blocks for direct reads Example: late-binding in Prefect

from prefect import flow
from prefect.blocks.system import Secret

@flow
def run(tenant: str, env: str = "prod"):
    # Late-binding secret name
    secret_block_name = f"{tenant}-{env}-api-key"
    api_key = Secret.load(secret_block_name).get()
    # use api_key...

Example: read from cloud secret manager in Prefect

from prefect import flow
from prefect_aws import AwsCredentials, AwsSecret

@flow
def run(secret_name: str = "prod/my-api-key"):
    creds = AwsCredentials.load("aws-default")
    api_key = AwsSecret(secret_name=secret_name, aws_credentials=creds).read_secret()
    # use api_key...

Relevant Prefect integration docs - Prefect AWS: prefect-aws - Prefect GCP: prefect-gcp - Prefect Azure: prefect-azure Quick do/don’t - Do: identity-based access (IRSA/Workload Identity), fetch at runtime, mount via CSI or sidecar if needed, keep secrets out of images and env vars, use short TTLs/dynamic secrets where possible. - Don’t: bake secrets into images, pass static long-lived secrets via env, or resolve secrets at deploy-time. If you can share: - Are you on Kubernetes or just Docker? - Which cloud/provider? - Will these containers run via Prefect workers? With that, I can give you a minimal, concrete config and code snippet tailored to your stack.

Great — you can absolutely use a single Secret block to hold a structured payload (id + multiple tokens) and implement “next token” selection across runs. The simplest, robust pattern is to store JSON in the Secret’s value and manage a pointer (cursor) or usage counts in that JSON. Key points - Secret blocks store a single (sensitive) string. Put your structured data in there as JSON. - Select a token at runtime (round-robin or least-used), then update the block with

overwrite=True

so the next run picks the next token. - If multiple runs may update at once, you’ll need to avoid races (see notes below). Create the Secret block with multiple tokens

import json
from prefect.blocks.system import Secret

data = {
  "id": "my-service",
  "tokens": ["tokA", "tokB", "tokC"],
  "cursor": 0  # points to the next token to use
}

Secret(value=json.dumps(data)).save("my-token-pool", overwrite=True)

Round-robin selection and update

import json
from prefect.blocks.system import Secret

def claim_token(block_name: str) -> str:
    # Load and parse
    block = Secret.load(block_name)
    data = json.loads(block.get())

    tokens = data["tokens"]
    if not tokens:
        raise RuntimeError("No tokens available in secret block")

    cursor = int(data.get("cursor", 0)) % len(tokens)
    token = tokens[cursor]

    # Advance cursor and save back for next run
    data["cursor"] = (cursor + 1) % len(tokens)
    Secret(value=json.dumps(data)).save(block_name, overwrite=True)

    return token

Use from a flow

from prefect import flow, get_run_logger

@flow
def use_token_flow():
    logger = get_run_logger()
    token = claim_token("my-token-pool")
    <http://logger.info|logger.info>("Got a token; using it now")
    # ...use token...

Selecting a specific token (by key/tier) If you want selection based on “usage/type/tier”, store tokens as a dict and pick by a selector passed at runtime (no cursor needed):

# Secret payload example
{
  "id": "my-service",
  "tokens": {
    "low": "tokA",
    "medium": "tokB",
    "high": "tokC"
  }
}

# Selection
def select_token(block_name: str, tier: str) -> str:
    data = json.loads(Secret.load(block_name).get())
    return data["tokens"][tier]

Least-used selection (fairness by usage count)

# Secret payload example
# keep an array of objects with usage counts
{
  "id": "my-service",
  "tokens": [
    {"value": "tokA", "uses": 0},
    {"value": "tokB", "uses": 3},
    {"value": "tokC", "uses": 1}
  ]
}

def claim_least_used(block_name: str) -> str:
    block = Secret.load(block_name)
    data = json.loads(block.get())

    tokens = data["tokens"]
    # pick token with min uses
    idx = min(range(len(tokens)), key=lambda i: tokens[i]["uses"])
    token_val = tokens[idx]["value"]

    # increment and save
    tokens[idx]["uses"] = int(tokens[idx]["uses"]) + 1
    Secret(value=json.dumps(data)).save(block_name, overwrite=True)

    return token_val

Important notes about concurrency - Secret blocks don’t provide atomic “read-modify-write.” If multiple ephemeral containers may claim/update at once, you can have races. - Options to mitigate: - Ensure only one run updates at a time (single-writer). In Prefect Cloud, apply a concurrency limit to the flow/task that performs the mutation so only one instance runs concurrently. - Use a non-mutating selection strategy (e.g., derive the index from a stable input like flow_run_id: index = hash(flow_run_id) % len(tokens)). No writes → no races, but you won’t get strict round-robin. - If strict fairness and high concurrency are required, use an external store with atomic ops (e.g., Redis, DB, Vault w/ leasing) to manage token claims. Updating/rotating tokens in the block (out of band) - You can rotate tokens by updating the JSON and resaving:

data = json.loads(Secret.load("my-token-pool").get())
# modify tokens/cursor however you like
Secret(value=json.dumps(data)).save("my-token-pool", overwrite=True)

- Because flows read the block at runtime, the next run will see the updated set. Caveats - Don’t log the token or the full JSON. If you log, redact. - Keep the “secret selection” code as a small helper function so it’s easy to maintain and test. If you expect multiple containers to select/update simultaneously, do you want me to show a pattern that’s race-resistant (e.g., hash-based selection or a simple Redis-backed claim)?

Got it — let’s store “id + its tokens” together and rotate both the id and the token with an easy path to add new id/token pairs. Proposed payload (single Secret block) - `id_order`: the rotation order of ids (explicit list so it’s stable and easy to insert new ids) - `pools`: each id maps to its token list and an internal cursor - `id_cursor`: which id is next Example payload:

{
  "id_order": [
    "service-a",
    "service-b"
  ],
  "pools": {
    "service-a": {
      "tokens": [
        "tokA1",
        "tokA2"
      ],
      "cursor": 0,
      "meta": {
        "region": "us-east-1"
      }
    },
    "service-b": {
      "tokens": [
        "tokB1"
      ],
      "cursor": 0
    }
  },
  "id_cursor": 0
}

Helper: create or update the Secret block

import json
from prefect.blocks.system import Secret

def save_secret_block(block_name: str, data: dict) -> None:
    Secret(value=json.dumps(data)).save(block_name, overwrite=True)

def init_block(block_name: str) -> None:
    data = {"id_order": [], "pools": {}, "id_cursor": 0}
    save_secret_block(block_name, data)

def load_data(block_name: str) -> dict:
    return json.loads(Secret.load(block_name).get())

Add or update an id + tokens (easy to append new pairs)

def add_or_update_id(block_name: str, id_: str, tokens: list[str], meta: dict | None = None) -> None:
    data = load_data(block_name)

    if id_ not in data["pools"]:
        # New id: add to pools and rotation order
        data["pools"][id_] = {"tokens": list(tokens), "cursor": 0}
        if meta:
            data["pools"][id_]["meta"] = meta
        data["id_order"].append(id_)
    else:
        # Update existing id: replace tokens, keep cursor in range
        pool = data["pools"][id_]
        pool["tokens"] = list(tokens)
        if meta is not None:
            pool["meta"] = meta
        pool["cursor"] = min(pool.get("cursor", 0), max(len(tokens) - 1, 0))

    # Keep id_cursor in range
    if data["id_order"]:
        data["id_cursor"] = data["id_cursor"] % len(data["id_order"])
    else:
        data["id_cursor"] = 0

    save_secret_block(block_name, data)

Claim the next (id, token) and advance both rotations - Alternating strategy: each claim moves to the next id and uses the next token for that id (round‑robin across ids and within each id). - Skips ids that have no tokens.

def claim_next(block_name: str) -> tuple[str, str]:
    data = load_data(block_name)
    id_order = data.get("id_order", [])
    pools = data.get("pools", {})
    if not id_order:
        raise RuntimeError("No ids configured in secret block")

    n = len(id_order)
    start = data.get("id_cursor", 0) % n
    attempts = 0

    while attempts < n:
        idx = (start + attempts) % n
        id_ = id_order[idx]
        pool = pools.get(id_)
        tokens = (pool or {}).get("tokens", [])
        if tokens:
            # Use current token and advance
            t_idx = pool.get("cursor", 0) % len(tokens)
            token = tokens[t_idx]
            pool["cursor"] = (t_idx + 1) % len(tokens)

            # Advance id cursor to next id for the next claim
            data["id_cursor"] = (idx + 1) % n
            save_secret_block(block_name, data)
            return id_, token

        attempts += 1

    raise RuntimeError("No tokens available for any id")

Using from a Prefect 3.x flow

from prefect import flow, get_run_logger

@flow
def use_creds():
    logger = get_run_logger()
    id_, token = claim_next("my-rotating-ids-and-tokens")
    <http://logger.info|logger.info>("Claimed an id+token pair (redacted).")
    # Use id_ and token...

Rotate/replace an id’s tokens, or add a new id on the fly

# Add a brand-new id with two tokens
add_or_update_id("my-rotating-ids-and-tokens", "service-c", ["tokC1", "tokC2"], meta={"owner":"team-x"})

# Replace tokens for an existing id
add_or_update_id("my-rotating-ids-and-tokens", "service-a", ["tokA3", "tokA4"])

Notes on behavior and alternatives

- Alternating strategy: claims rotate ids every call and also rotate each id’s tokens. If you prefer “drain all tokens for an id then move to next id,” I can share a nested strategy instead. - Concurrency: Secret blocks do not provide atomic updates. If many runs may claim simultaneously, add a Prefect concurrency limit to the task/flow that performs the mutation, or switch to a deterministic, non‑mutating selection (e.g., hash-based by flow_run_id) at the cost of strict fairness. For strict fairness at scale, use a store with atomic ops (Redis/DB). - Redaction: never log ids/tokens; only log meta or high‑level info. Docs - Blocks overview: Blocks concept - Secret block API: Secret block API If you tell me your preferred rotation strategy (alternate vs nested), I’ll tailor the claim function accordingly.