https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> for Prefect 3.6.28 write me a flow or...
# ask-marvin
r
@Marvin for Prefect 3.6.28 write me a flow or automation that notifies when the prefect-server-postgres database size is larger than some threshold. Prefect-server-postgres was created using the sub chart from the prefect-helm repo.
m
✅ thought for 153.5 seconds
Yes — for Prefect 3.6.28, the cleanest solution is: 1. run a small monitoring flow on a schedule 2. have that flow query Postgres for database size 3. if the size is over your threshold, emit a custom Prefect event 4. create a Prefect automation that sends a notification when that event appears Why this pattern? Prefect automations can react to Prefect events and built-in Prefect metrics, but database size is not a built-in Prefect metric, so an automation cannot directly watch 
prefect-server-postgres
size by itself. Relevant APIs: - Events / automations: https://docs.prefect.io - Helm chart repo: https://github.com/PrefectHQ/prefect-helm - Prefect pricing / Cloud UI if needed later: app.prefect.cloud Below is a working pattern. --- ## Option A: Recommended — flow emits event, automation sends notification ### 1) Monitoring flow This flow connects to the same Postgres DB your Prefect server uses, checks size with 
pg_database_size
, and emits an event if the threshold is exceeded.
Copy code
python
from datetime import datetime, timezone
import os

import asyncpg
from prefect import flow, get_run_logger
from prefect.events import emit_event


async def get_db_size_bytes(pg_dsn: str) -> int:
    conn = await asyncpg.connect(pg_dsn)
    try:
        return await conn.fetchval(
            "SELECT pg_database_size(current_database())"
        )
    finally:
        await conn.close()


@flow(name="monitor-prefect-server-postgres-size")
async def monitor_prefect_server_postgres_size(
    pg_dsn: str | None = None,
    threshold_gb: float = 10.0,
):
    logger = get_run_logger()

    if pg_dsn is None:
        pg_dsn = os.environ["PREFECT_API_DATABASE_CONNECTION_URL"]

    size_bytes = await get_db_size_bytes(pg_dsn)
    size_gb = size_bytes / 1024**3

    <http://logger.info|logger.info>(f"Prefect DB size: {size_gb:.2f} GiB")

    if size_gb > threshold_gb:
        event = emit_event(
            event="prefect.db.size.threshold-exceeded",
            resource={
                "prefect.resource.id": "postgres.prefect-server-postgres",
                "prefect.resource.name": "prefect-server-postgres",
                "prefect.resource.role": "database",
            },
            payload={
                "size_bytes": size_bytes,
                "size_gb": round(size_gb, 2),
                "threshold_gb": threshold_gb,
                "observed_at": datetime.now(timezone.utc).isoformat(),
            },
        )
        logger.warning(
            f"Threshold exceeded: {size_gb:.2f} GiB > {threshold_gb:.2f} GiB. Event={event}"
        )
    else:
        <http://logger.info|logger.info>(
            f"Threshold not exceeded: {size_gb:.2f} GiB <= {threshold_gb:.2f} GiB"
        )
--- ### 2) Important DSN note for Helm/Postgres subchart Because your DB was created by the 
prefect-helm
Postgres subchart, your server likely already has a connection string in 
PREFECT_API_DATABASE_CONNECTION_URL
. If you run this monitoring flow inside the same Kubernetes cluster, reusing that env var is usually the easiest approach. One thing to watch for: 
asyncpg.connect(...)
expects a normal Postgres DSN like:
Copy code
text
postgresql://user:password@host:5432/dbname
If your env var is in SQLAlchemy form, like:
Copy code
text
postgresql+asyncpg://user:password@host:5432/dbname
then strip the 
+asyncpg
before connecting. Here is a safer version of the flow that handles that: ```python from datetime import datetime, timezone import os import asyncpg from prefect import flow, get_run_logger from prefect.events import emit_event def normalize_postgres_dsn(dsn: str) -> str: return dsn.replace("postgresql+asyncpg://", "postgresql://", 1) async def get_db_size_bytes(pg_dsn: str) -> int: conn = await asyncpg.connect(normalize_postgres_dsn(pg_dsn)) try: return await conn.fetchval( "SELECT pg_database_size(current_database())" ) finally: await conn.close()
@flow(name="monitor-prefect-server-postgres-size") async def monitor_prefect_server_postgres_size( pg_dsn: str | None = None, threshold_gb: float = 10.0, ): logger = get_run_logger() if pg_dsn is None: pg_dsn = os.environ["PREFECT_API_DATABASE_CONNECTION_URL"] size_bytes = await get_db_size_bytes(pg_dsn) size_gb = size_bytes / 1024**3 logger.info(f"Prefect DB size: {size_gb:.2f} GiB") if size_gb > threshold_gb: emit_event( event="prefect.db.size.threshold-exceeded", resource={ "prefect.resource.id": "postgres.prefect-server-postgres", "prefect.resource.name": "prefect-server-postgres", "prefect.resource.role": "database", }, payload={ "size_bytes": size_bytes, "size_gb": round(size_gb, 2), "threshold_gb": threshold_gb, "observed_at": datetime.now(timezone.utc).isoformat(), }, ) logger.warning( f"Threshold exceeded: {size_gb:.2f} GiB > {threshold_gb:.2f} GiB" )
Copy code
---

### 3) Deploy it on a schedule

In Prefect 3.x, use `flow.from_source(...).deploy(...)` or `prefect deploy` — *not* deprecated 2.x deployment build patterns.

I’d recommend deploying this to run every hour or every 6 hours.

Example `prefect.yaml` approach is fine, but here’s a simple Python deployment example:

```python
from monitor_db import monitor_prefect_server_postgres_size

if __name__ == "__main__":
    monitor_prefect_server_postgres_size.from_source(
        source=".",
        entrypoint="monitor_db.py:monitor_prefect_server_postgres_size",
    ).deploy(
        name="prefect-db-size-monitor",
        work_pool_name="your-work-pool",
        cron="0 * * * *",
        parameters={"threshold_gb": 10.0},
    )
If you want, I can also write the matching 
prefect.yaml
. --- ### 4) Create a notification block Prefect 3.6.28 includes notification blocks like: - 
SlackWebhook
- 
MicrosoftTeamsWebhook
- 
PagerDutyWebHook
- 
DiscordWebhook
- 
MattermostWebhook
- 
CustomWebhookNotificationBlock
For Slack, for example:
Copy code
python
from prefect.blocks.notifications import SlackWebhook

block = SlackWebhook(url="<https://hooks.slack.com/services/XXX/YYY/ZZZ>")
block.save("db-size-slack", overwrite=True)
--- ### 5) Create the automation Your automation should listen for the custom event: 
prefect.db.size.threshold-exceeded
Example:
Copy code
python
from uuid import UUID

from prefect.automations import EventTrigger
from prefect.events.actions import SendNotification
from prefect.events.schemas.automations import Automation


automation = Automation(
    name="Alert on Prefect DB size threshold exceeded",
    description="Sends a notification when prefect-server-postgres exceeds the configured size threshold.",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.db.size.threshold-exceeded"},
        posture="Reactive",
        threshold=1,
        match={
            "prefect.resource.id": "postgres.prefect-server-postgres",
        },
    ),
    actions=[
        SendNotification(
            block_document_id=UUID("YOUR-NOTIFICATION-BLOCK-DOCUMENT-ID"),
            subject="Prefect DB size threshold exceeded",
            body="""
prefect-server-postgres is above threshold.

Database: {{ event.resource.name }}
Current size: {{ event.payload.size_gb }} GiB
Threshold: {{ event.payload.threshold_gb }} GiB
Observed at: {{ event.payload.observed_at }}
""".strip(),
        )
    ],
)

automation.create()
--- ## Option B: Simpler flow-only alerting If you do not care about using Prefect automations/events, the simpler option is to have the flow send Slack/Teams/PagerDuty directly when the threshold is exceeded. That avoids the extra automation layer. Example shape: ```python from datetime import datetime, timezone import os import asyncpg
from prefect import flow, get_run_logger from prefect.blocks.notifications import SlackWebhook def normalize_postgres_dsn(dsn: str) -> str: return dsn.replace("postgresql+asyncpg://", "postgresql://", 1) async def get_db_size_bytes(pg_dsn: str) -> int: conn = await asyncpg.connect(normalize_postgres_dsn(pg_dsn)) try: return await conn.fetchval( "SELECT pg_database_size(current_database())" ) finally: await conn.close() @flow(name="monitor-prefect-server-postgres-size") async def monitor_prefect_server_postgres_size( threshold_gb: float = 10.0, ): logger = get_run_logger() pg_dsn = os.environ["PREFECT_API_DATABASE_CONNECTION_URL"] size_bytes = await get_db_size_bytes(pg_dsn) size_gb = size_bytes / 1024**3 logger.info(f"Prefect DB size: {size_gb:.2f} GiB") if size_gb > threshold_gb: slack = await SlackWebhook.load("db-size-slack") await slack.notify( f"⚠️ prefect-server-postgres is {size_gb:.2f} GiB, above threshold {threshold_gb:.2f} GiB at {datetime.now(timezone.utc).isoformat()}" ) ``` This is less “Prefect-native” from an events perspective, but easier to operate. --- ## Which option should you choose? Use Option A if you want: - alert logic separated from check logic - Prefect automations in the UI - easier reuse of the emitted event for multiple downstream actions Use Option B if you just want: - one scheduled flow - one notification destination - minimal moving parts --- ## Kubernetes / Helm practical advice Since 
prefect-server-postgres
came from the Helm subchart, the most reliable deployment pattern is: - run the monitoring flow in-cluster - give it access to the same DB connection secret/env var - schedule it via a Prefect deployment If you prefer not to involve Prefect for the check itself, a Kubernetes 
CronJob
that queries Postgres and posts to Slack is also valid — but if you want the alerting visible inside Prefect, the event + automation pattern is better. --- ## My recommendation I’d use Option A with: - hourly schedule - event name 
prefect.db.size.threshold-exceeded
- Slack or PagerDuty notification block - threshold parameterized per environment --- If you want, I can give you either of these next: 1. a complete 
prefect.yaml
 for deploying this 2. a Helm/Kubernetes example that mounts the DB secret into the worker/job 3. a version that also reports the top largest tables so the alert is more actionable
r
@Marvin I tried using the env variable PREFECT_SERVER_DATABASE_CONNECTION_URL to get the connection string but that is only set in the prefect-server pod and not in the job pod.
m
✅ thought for 26.1 seconds
That makes sense — 
PREFECT_SERVER_DATABASE_CONNECTION_URL
is typically only on the server pod, not automatically injected into flow run job pods. So for your monitoring flow/job, you have two practical options: ## Best option: inject the same DB secret/env into the job pod Since your Postgres comes from the 
prefect-helm
subchart, the connection details are usually already stored in a Kubernetes 
Secret
used by the server deployment. You can reuse that same secret in the worker/job template so the flow run pod gets the env var too. In other words: don’t rely on Prefect to forward that env var from the server pod — instead mount the same secret into the flow run pod. ### Pattern Set an env var like 
PREFECT_DB_MONITOR_CONNECTION_URL
on the flow run job from the Kubernetes secret, then read that in the flow. Your flow would use:
Copy code
python
import os

pg_dsn = os.environ["PREFECT_DB_MONITOR_CONNECTION_URL"]
--- ## Example flow
Copy code
python
from datetime import datetime, timezone
import os

import asyncpg
from prefect import flow, get_run_logger
from prefect.events import emit_event


def normalize_postgres_dsn(dsn: str) -> str:
    return dsn.replace("postgresql+asyncpg://", "postgresql://", 1)


async def get_db_size_bytes(pg_dsn: str) -> int:
    conn = await asyncpg.connect(normalize_postgres_dsn(pg_dsn))
    try:
        return await conn.fetchval(
            "SELECT pg_database_size(current_database())"
        )
    finally:
        await conn.close()


@flow(name="monitor-prefect-server-postgres-size")
async def monitor_prefect_server_postgres_size(threshold_gb: float = 10.0):
    logger = get_run_logger()

    pg_dsn = os.environ["PREFECT_DB_MONITOR_CONNECTION_URL"]

    size_bytes = await get_db_size_bytes(pg_dsn)
    size_gb = size_bytes / 1024**3

    <http://logger.info|logger.info>(f"Prefect DB size: {size_gb:.2f} GiB")

    if size_gb > threshold_gb:
        emit_event(
            event="prefect.db.size.threshold-exceeded",
            resource={
                "prefect.resource.id": "postgres.prefect-server-postgres",
                "prefect.resource.name": "prefect-server-postgres",
                "prefect.resource.role": "database",
            },
            payload={
                "size_bytes": size_bytes,
                "size_gb": round(size_gb, 2),
                "threshold_gb": threshold_gb,
                "observed_at": datetime.now(timezone.utc).isoformat(),
            },
        )
        logger.warning(
            f"Threshold exceeded: {size_gb:.2f} GiB > {threshold_gb:.2f} GiB"
        )
--- ## How to make the env var available in the job pod This depends on how your work pool / Kubernetes worker is configured, but the idea is: - find the Kubernetes secret used by Prefect server for Postgres - add that secret as an env source or env var in the worker job template - deploy this monitoring flow to that work pool ### Typical Kubernetes env snippet
Copy code
yaml
env:
  - name: PREFECT_DB_MONITOR_CONNECTION_URL
    valueFrom:
      secretKeyRef:
        name: prefect-server-postgresql
        key: connection-string
The secret name and key will depend on your Helm release. A common way to inspect them:
Copy code
bash
kubectl get secret -n <namespace>
kubectl describe deployment -n <namespace> <prefect-server-deployment-name>
kubectl get deployment -n <namespace> <prefect-server-deployment-name> -o yaml
Look for the env definition on the server container to see: - which secret name it references - which key contains the connection URL Then reuse that same secret/key in the worker job template. --- ## Alternative: build the DSN from separate secret values Sometimes the chart stores: - username - password - database - host - port in separate secret keys rather than one full connection URL. If so, inject them individually: ```yaml env: - name: PGHOST value: prefect-server-postgresql - name: PGPORT value: "5432" - name: PGDATABASE valueFrom: secretKeyRef:
name: prefect-server-postgresql key: postgres-database - name: PGUSER valueFrom: secretKeyRef: name: prefect-server-postgresql key: postgres-username - name: PGPASSWORD valueFrom: secretKeyRef: name: prefect-server-postgresql key: postgres-password
Copy code
Then construct the DSN in Python:

```python
import os

pg_dsn = (
    f"postgresql://{os.environ['PGUSER']}:{os.environ['PGPASSWORD']}"
    f"@{os.environ['PGHOST']}:{os.environ['PGPORT']}/{os.environ['PGDATABASE']}"
)
--- ## Another good option: run this as a plain Kubernetes CronJob Because this check is really infrastructure monitoring, a plain Kubernetes 
CronJob
is sometimes cleaner than a Prefect flow run job. That CronJob can: - use the same Postgres secret - query DB size - either send Slack directly - or call 
emit_event
to Prefect if you still want a Prefect automation This avoids modifying your worker template globally if this secret should only be available to this one monitoring job. --- ## Recommendation I’d recommend one of these: ### If you want it inside Prefect - create a dedicated work pool for infra-monitoring jobs - add the Postgres secret env var to that pool’s job template - deploy the monitoring flow there ### If you want least privilege / least coupling - create a Kubernetes 
CronJob
- mount the Postgres secret only there - send Slack directly or emit a Prefect event --- ## Important note Also, the env var name you mentioned is 
PREFECT_SERVER_DATABASE_CONNECTION_URL
, but in some setups you may also see other DB-related env names depending on chart/version/config. The safest path is to inspect the actual server deployment YAML and reuse exactly the secret/key it references. --- If you want, I can help with either of these next: 1. a Kubernetes worker job template snippet to inject the secret into flow-run pods 2. a standalone Kubernetes CronJob manifest for this DB size monitor 3. a full Prefect automation + Slack block example wired to this flow
r
@Marvin that seems horribly complicated. Show me how to make it into a sqlalchemy block instead.
m
✅ thought for 19.9 seconds
Yep — that’s much simpler. If you already know the DB connection string, putting it in a SQLAlchemy Connector block is a good fit. Then your flow can load the block and query Postgres directly, without needing pod env wiring. ## Important caveat This only works if the flow-run pod can network-reach the Postgres service and the credentials in the block are valid. If your jobs run in the same cluster/namespace as 
prefect-server-postgres
, that’s usually fine. --- ## 1) Create a SQLAlchemy Connector block You’ll need the 
prefect-sqlalchemy
collection installed wherever you create/use the block. Example:
Copy code
python
from prefect_sqlalchemy import SqlAlchemyConnector

block = SqlAlchemyConnector(
    connection_info={
        "url": "<postgresql+asyncpg://USERNAME:PASSWORD@prefect-server-postgresql:5432/DATABASE>"
    }
)

block.save("prefect-server-postgres-conn", overwrite=True)
If your service name is actually 
prefect-server-postgres
instead of 
prefect-server-postgresql
, use that host instead. If you need the fully-qualified in-cluster DNS name, it would look like:
Copy code
text
prefect-server-postgresql.<namespace>.svc.cluster.local
Prefect SQLAlchemy docs: - https://prefecthq.github.io/prefect-sqlalchemy/ --- ## 2) Use the block in a flow Then load the block and run a size query.
Copy code
python
from prefect import flow, get_run_logger
from prefect.events import emit_event
from prefect_sqlalchemy import SqlAlchemyConnector


@flow(name="monitor-prefect-db-size")
def monitor_prefect_db_size(
    threshold_gb: float = 10.0,
    block_name: str = "prefect-server-postgres-conn",
):
    logger = get_run_logger()

    connector = SqlAlchemyConnector.load(block_name)

    size_bytes = connector.fetch_one(
        "SELECT pg_database_size(current_database())"
    )[0]

    size_gb = size_bytes / 1024**3
    <http://logger.info|logger.info>(f"Prefect DB size is {size_gb:.2f} GiB")

    if size_gb > threshold_gb:
        emit_event(
            event="prefect.db.size.threshold-exceeded",
            resource={
                "prefect.resource.id": "postgres.prefect-server-postgres",
                "prefect.resource.name": "prefect-server-postgres",
                "prefect.resource.role": "database",
            },
            payload={
                "size_bytes": int(size_bytes),
                "size_gb": round(size_gb, 2),
                "threshold_gb": threshold_gb,
            },
        )
        logger.warning(
            f"Threshold exceeded: {size_gb:.2f} GiB > {threshold_gb:.2f} GiB"
        )

    return {"size_gb": round(size_gb, 2), "threshold_gb": threshold_gb}
--- ## 3) If you want notification directly in the flow If you don’t care about automations, this is even simpler:
Copy code
python
from prefect import flow, get_run_logger
from prefect.blocks.notifications import SlackWebhook
from prefect_sqlalchemy import SqlAlchemyConnector


@flow(name="monitor-prefect-db-size")
async def monitor_prefect_db_size(
    threshold_gb: float = 10.0,
    db_block_name: str = "prefect-server-postgres-conn",
    slack_block_name: str = "db-size-slack",
):
    logger = get_run_logger()

    connector = await SqlAlchemyConnector.load(db_block_name)
    result = await connector.fetch_one("SELECT pg_database_size(current_database())")
    size_bytes = result[0]
    size_gb = size_bytes / 1024**3

    <http://logger.info|logger.info>(f"Prefect DB size is {size_gb:.2f} GiB")

    if size_gb > threshold_gb:
        slack = await SlackWebhook.load(slack_block_name)
        await slack.notify(
            f":warning: prefect-server-postgres is {size_gb:.2f} GiB, above threshold {threshold_gb:.2f} GiB"
        )

    return {"size_gb": round(size_gb, 2)}
--- ## 4) Best practical setup If you want the fewest moving parts: - store DB connection in 
SqlAlchemyConnector
- store Slack webhook in 
SlackWebhook
- run one scheduled flow - send Slack directly from the flow That is probably the least complicated solution.
If you want alerts visible/manageable in Prefect automations: - keep the SQLAlchemy block - emit a custom event - attach an automation to that event --- ## 5) One likely gotcha Depending on your installed versions, 
SqlAlchemyConnector.fetch_one
may expect SQLAlchemy text objects rather than a raw string. If that happens, use:
Copy code
python
from sqlalchemy import text

size_bytes = connector.fetch_one(
    text("SELECT pg_database_size(current_database())")
)[0]
or async:
Copy code
python
from sqlalchemy import text

result = await connector.fetch_one(
    text("SELECT pg_database_size(current_database())")
)
--- ## Minimal recommendation If you want simple, I’d do this:
Copy code
python
from sqlalchemy import text
from prefect import flow
from prefect.blocks.notifications import SlackWebhook
from prefect_sqlalchemy import SqlAlchemyConnector


@flow
async def monitor_prefect_db_size(threshold_gb: float = 10.0):
    connector = await SqlAlchemyConnector.load("prefect-server-postgres-conn")
    slack = await SlackWebhook.load("db-size-slack")

    row = await connector.fetch_one(
        text("SELECT pg_database_size(current_database())")
    )
    size_gb = row[0] / 1024**3

    if size_gb > threshold_gb:
        await slack.notify(
            f"Prefect DB size is {size_gb:.2f} GiB, above threshold {threshold_gb:.2f} GiB"
        )
--- If you want, I can turn this into a complete ready-to-run example with: - block creation code - flow code - deployment code for Prefect 3.6.28
Open in Slack
PreviousNext
Powered by