https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Write a Prefect 3.7.0 flow that check...
# ask-marvin
r
@Marvin Write a Prefect 3.7.0 flow that checks Blocks that are saved in self hosted prefect-server against a list of expected block_types and block_names. A block_name can appear more than once if it's under a different block_type. The list of expected Blocks will be passed as a parameter into the flow. The flow should emit events if an expected Block is not found in prefect-server, but also an event if prefect-server has a Block not found in expected. The payload of these events should have the details, i.e. explaining the gap and giving the block_type and block_name. Be sure to add docstrings and use the rules that a python linter, such as ruff, would use.
m
✅ thought for 252.4 seconds
Here’s a Prefect 3.7.0 flow that: - accepts a parameterized list of expected blocks - checks saved blocks in a self-hosted Prefect server - emits an event for: - each expected block that is missing from the server - each unexpected block found on the server but not in the expected list I’ve written it to be linter-friendly and added docstrings throughout. ```python from future import annotations from collections.abc import Iterable from typing import Any from prefect import flow, get_run_logger from prefect.client.orchestration import get_client from prefect.events import emit_event EVENT_PREFIX = "custom.prefect.block-audit" RESOURCE_ID = "custom.prefect.block-audit" def _normalize_block_key(block_type: str, block_name: str) -> tuple[str, str]: """Normalize a block identity for reliable comparisons. This function strips surrounding whitespace from both values. It preserves case because Prefect block names and type names should be compared exactly as stored unless your organization has a different naming convention. Args: block_type: The Prefect block type name. block_name: The saved block document name. Returns: A tuple of normalized 
(block_type, block_name)
. """ return block_type.strip(), block_name.strip() def _build_expected_set( expected_blocks: Iterable[dict[str, str]], ) -> set[tuple[str, str]]: """Convert the expected block input into a comparable set. Each expected item must provide: - 
block_type
- 
block_name
Args: expected_blocks: Iterable of dictionaries describing expected blocks. Returns: A set of 
(block_type, block_name)
tuples. Raises: ValueError: If any item is missing required keys or contains blanks. TypeError: If any item is not a dictionary. """ normalized: set[tuple[str, str]] = set() for index, item in enumerate(expected_blocks): if not isinstance(item, dict): msg = ( f"Each expected block must be a dictionary. " f"Item at index {index} has type {type(item)!r}." ) raise TypeError(msg) block_type = item.get("block_type") block_name = item.get("block_name") if not block_type or not block_name: msg = ( "Each expected block must include non-empty 'block_type' and " f"'block_name'. Invalid item at index {index}: {item!r}" ) raise ValueError(msg) normalized.add(_normalize_block_key(block_type, block_name)) return normalized async def _fetch_saved_blocks() -> set[tuple[str, str]]: """Fetch all non-anonymous saved blocks from the Prefect API. This uses offset-based pagination via 
read_block_documents()
. Returns: A set of 
(block_type, block_name)
tuples for all saved blocks. """ saved_blocks: set[tuple[str, str]] = set() offset = 0 limit = 200 async with get_client() as client: while True: block_documents = await client.read_block_documents( offset=offset, limit=limit, include_secrets=False, ) if not block_documents: break for block_document in block_documents: if not block_document.block_type_name or not block_document.name: continue saved_blocks.add( _normalize_block_key( block_type=block_document.block_type_name, block_name=block_document.name, ) ) offset += len(block_documents) return saved_blocks def _emit_gap_event( *, event_name: str, gap_direction: str, block_type: str, block_name: str, message: str, ) -> None: """Emit a Prefect event describing a block inventory gap. Args:
event_name: The full event name to emit. gap_direction: Either 
missing_expected
or 
unexpected_actual
. block_type: The Prefect block type name. block_name: The saved block document name. message: Human-readable explanation of the gap. """ emit_event( event=event_name, resource={ "prefect.resource.id": RESOURCE_ID, "prefect.resource.name": "prefect-block-audit", "block_type": block_type, "block_name": block_name, "gap_direction": gap_direction, }, payload={ "message": message, "block_type": block_type, "block_name": block_name, "gap_direction": gap_direction, }, ) @flow(name="audit-prefect-blocks") async def audit_prefect_blocks( expected_blocks: list[dict[str, str]], ) -> dict[str, Any]: """Audit saved Prefect blocks against an expected inventory. This flow compares the current saved blocks in a self-hosted Prefect server against a caller-provided expected list. It emits events for: - expected blocks that are missing from Prefect server - blocks that exist in Prefect server but are not in the expected list Input format: expected_blocks = [ {"block_type": "S3 Bucket", "block_name": "raw-data"}, {"block_type": "Secret", "block_name": "db-password"}, {"block_type": "Webhook", "block_name": "alerts"}, ] Notes: - A 
block_name
may appear under multiple block types. - Comparison is performed on the pair 
(block_type, block_name)
. - Saved block discovery uses 
read_block_documents()
, which returns non-anonymous block documents by default. Args: expected_blocks: List of expected block definitions. Returns: A summary dictionary containing expected, actual, missing, and unexpected block inventories. """ logger = get_run_logger() expected_set = _build_expected_set(expected_blocks) actual_set = await _fetch_saved_blocks() missing_blocks = sorted(expected_set - actual_set) unexpected_blocks = sorted(actual_set - expected_set) logger.info("Expected blocks: %d", len(expected_set)) logger.info("Actual saved blocks: %d", len(actual_set)) logger.info("Missing expected blocks: %d", len(missing_blocks)) logger.info("Unexpected actual blocks: %d", len(unexpected_blocks)) for block_type, block_name in missing_blocks: message = ( "Expected block was not found in Prefect server: " f"block_type={block_type!r}, block_name={block_name!r}." ) _emit_gap_event( event_name=f"{EVENT_PREFIX}.expected-block-missing", gap_direction="missing_expected", block_type=block_type, block_name=block_name, message=message, ) for block_type, block_name in unexpected_blocks: message = ( "Prefect server contains a block that was not found in the expected " f"inventory: block_type={block_type!r}, block_name={block_name!r}." ) _emit_gap_event( event_name=f"{EVENT_PREFIX}.unexpected-block-found", gap_direction="unexpected_actual", block_type=block_type, block_name=block_name, message=message, ) summary = { "expected_blocks": [ {"block_type": block_type, "block_name": block_name} for block_type, block_name in sorted(expected_set) ], "actual_blocks": [ {"block_type": block_type, "block_name": block_name} for block_type, block_name in sorted(actual_set) ], "missing_blocks": [ {"block_type": block_type, "block_name": block_name} for block_type, block_name in missing_blocks ], "unexpected_blocks": [
{"block_type": block_type, "block_name": block_name} for block_type, block_name in unexpected_blocks ], "missing_count": len(missing_blocks), "unexpected_count": len(unexpected_blocks), } logger.info("Block audit summary: %s", summary) return summary if name == "__main__": audit_prefect_blocks( expected_blocks=[ {"block_type": "S3 Bucket", "block_name": "raw-data"}, {"block_type": "Secret", "block_name": "db-password"}, {"block_type": "Webhook", "block_name": "alerts"}, {"block_type": "Webhook", "block_name": "raw-data"}, ] ) ``` Notes - This compares by 
(block_type, block_name)
, so the same 
block_name
can appear under different 
block_type
values. - It uses 
block_document.block_type_name
, which is the block type name returned by the API. - If you want to compare against block type slug instead of type name, I can rewrite it to do that too. Slugs are often more stable than display names. Relevant APIs - Prefect client: 
get_client()
and block document listing via 
read_block_documents()
- Event emission: 
emit_event()
If you want, I can also provide: - a version using a 
pydantic
model for the 
expected_blocks
parameter - a deployment-ready 
prefect deploy
setup - an automation example that reacts to these emitted events
r
@Marvin Build on the script above adding your suggestions to use slugs, the pydantic model for expect_blocks and also the separate example automation that should send a notification
m
✅ thought for 154.4 seconds
Absolutely — below is a more complete version that adds: - block type slug comparison instead of display name comparison - a Pydantic model for 
expected_blocks
- a separate automation example that sends a notification when one of the custom block-audit events is emitted I’ve also made the event payload a bit richer so the automation notification has useful fields to reference. ```python from future import annotations from collections.abc import Iterable from typing import Any from pydantic import BaseModel, Field, field_validator from prefect import flow, get_run_logger from prefect.client.orchestration import get_client from prefect.events import emit_event EVENT_PREFIX = "custom.prefect.block-audit" RESOURCE_ID = "custom.prefect.block-audit" class ExpectedBlock(BaseModel): """Represents one expected Prefect block document. The comparison key is the pair 
(block_type_slug, block_name)
, which allows the same block name to appear under multiple block types. Attributes: block_type_slug: The Prefect block type slug, such as 
secret
, 
s3-bucket
, or 
webhook
. block_name: The saved block document name. """ block_type_slug: str = Field( ..., description="The Prefect block type slug.", examples=["secret", "s3-bucket", "webhook"], ) block_name: str = Field( ..., description="The saved Prefect block document name.", examples=["db-password", "raw-data", "alerts"], ) @field_validator("block_type_slug", "block_name") @classmethod def validate_not_blank(cls, value: str) -> str: """Validate that a field contains a non-blank string. Args: value: The string field value. Returns: The stripped string. Raises: ValueError: If the value is blank after stripping whitespace. """ normalized = value.strip() if not normalized: msg = "Value must not be blank." raise ValueError(msg) return normalized def _normalize_block_key(block_type_slug: str, block_name: str) -> tuple[str, str]: """Normalize a block identity for comparisons. Args: block_type_slug: The block type slug. block_name: The block document name. Returns: A normalized 
(block_type_slug, block_name)
tuple. """ return block_type_slug.strip(), block_name.strip() def _build_expected_set( expected_blocks: Iterable[ExpectedBlock], ) -> set[tuple[str, str]]: """Convert expected block models into a set of comparable keys. Args: expected_blocks: Expected block definitions. Returns: A set of 
(block_type_slug, block_name)
tuples. """ return { _normalize_block_key( block.block_type_slug, block.block_name, ) for block in expected_blocks } async def _fetch_saved_blocks() -> set[tuple[str, str]]: """Fetch all saved non-anonymous blocks from the Prefect API. This function paginates through 
read_block_documents()
and extracts the 
(block_type_slug, block_name)
pair for each block. If a returned block document does not include a type slug, it is skipped. Returns: A set of 
(block_type_slug, block_name)
tuples. """ saved_blocks: set[tuple[str, str]] = set() offset = 0 limit = 200 async with get_client() as client: while True: block_documents = await client.read_block_documents( offset=offset, limit=limit, include_secrets=False, ) if not block_documents: break for block_document in block_documents: if ( not block_document.name or not block_document.block_type or not block_document.block_type.slug ): continue
saved_blocks.add( _normalize_block_key( block_type_slug=block_document.block_type.slug, block_name=block_document.name, ) ) offset += len(block_documents) return saved_blocks def _emit_gap_event( *, event_name: str, gap_direction: str, block_type_slug: str, block_name: str, message: str, ) -> None: """Emit a custom event describing a block inventory gap. Args: event_name: Full event name. gap_direction: Gap direction such as 
missing_expected
or 
unexpected_actual
. block_type_slug: The Prefect block type slug. block_name: The saved block document name. message: Human-readable explanation of the gap. """ emit_event( event=event_name, resource={ "prefect.resource.id": RESOURCE_ID, "prefect.resource.name": "prefect-block-audit", "block_type_slug": block_type_slug, "block_name": block_name, "gap_direction": gap_direction, }, payload={ "message": message, "block_type_slug": block_type_slug, "block_name": block_name, "gap_direction": gap_direction, }, ) @flow(name="audit-prefect-blocks") async def audit_prefect_blocks( expected_blocks: list[ExpectedBlock], ) -> dict[str, Any]: """Audit saved Prefect blocks against an expected inventory. This flow compares the current saved blocks in a self-hosted Prefect server against an expected list provided as a flow parameter. The comparison uses: - 
block_type_slug
- 
block_name
It emits events for: - expected blocks that are missing from Prefect server - blocks present in Prefect server that are not in the expected inventory Args: expected_blocks: Expected block definitions. Returns: A summary dictionary describing the expected, actual, missing, and unexpected block inventories. """ logger = get_run_logger() expected_set = _build_expected_set(expected_blocks) actual_set = await _fetch_saved_blocks() missing_blocks = sorted(expected_set - actual_set) unexpected_blocks = sorted(actual_set - expected_set) logger.info("Expected blocks: %d", len(expected_set)) logger.info("Actual saved blocks: %d", len(actual_set)) logger.info("Missing expected blocks: %d", len(missing_blocks)) logger.info("Unexpected actual blocks: %d", len(unexpected_blocks)) for block_type_slug, block_name in missing_blocks: message = ( "Expected block was not found in Prefect server: " f"block_type_slug={block_type_slug!r}, block_name={block_name!r}." ) _emit_gap_event( event_name=f"{EVENT_PREFIX}.expected-block-missing", gap_direction="missing_expected", block_type_slug=block_type_slug, block_name=block_name, message=message, ) for block_type_slug, block_name in unexpected_blocks: message = ( "Prefect server contains a block that was not found in the expected " "inventory: " f"block_type_slug={block_type_slug!r}, block_name={block_name!r}." ) _emit_gap_event( event_name=f"{EVENT_PREFIX}.unexpected-block-found", gap_direction="unexpected_actual", block_type_slug=block_type_slug, block_name=block_name, message=message, ) summary = { "expected_blocks": [ { "block_type_slug": block_type_slug, "block_name": block_name, } for block_type_slug, block_name in sorted(expected_set) ], "actual_blocks": [ { "block_type_slug": block_type_slug,
"block_name": block_name, } for block_type_slug, block_name in sorted(actual_set) ], "missing_blocks": [ { "block_type_slug": block_type_slug, "block_name": block_name, } for block_type_slug, block_name in missing_blocks ], "unexpected_blocks": [ { "block_type_slug": block_type_slug, "block_name": block_name, } for block_type_slug, block_name in unexpected_blocks ], "missing_count": len(missing_blocks), "unexpected_count": len(unexpected_blocks), } logger.info("Block audit summary: %s", summary) return summary if name == "__main__": audit_prefect_blocks( expected_blocks=[ ExpectedBlock( block_type_slug="s3-bucket", block_name="raw-data", ), ExpectedBlock( block_type_slug="secret", block_name="db-password", ), ExpectedBlock( block_type_slug="webhook", block_name="alerts", ), ExpectedBlock( block_type_slug="webhook", block_name="raw-data", ), ] )
Copy code
*Example automation that sends a notification*

This example creates a Prefect Automation that listens for either:

- `custom.prefect.block-audit.expected-block-missing`
- `custom.prefect.block-audit.unexpected-block-found`

and sends a notification using a saved *notification block*.

```python
from __future__ import annotations

from datetime import timedelta
from uuid import UUID

from prefect.automations import Automation, EventTrigger, Posture, SendNotification


def create_block_audit_notification_automation(
    notification_block_document_id: UUID,
) -> Automation:
    """Create an automation for block audit gap events.

    This automation reacts to block audit events emitted by the
    `audit_prefect_blocks` flow and sends a notification through the provided
    notification block.

    Args:
        notification_block_document_id: The block document ID of a saved
            notification block, such as a Slack webhook notification block.

    Returns:
        The created automation object.
    """
    automation = Automation(
        name="Notify on Prefect block audit gaps",
        description=(
            "Send a notification whenever the block audit flow detects a "
            "missing expected block or an unexpected saved block."
        ),
        enabled=True,
        trigger=EventTrigger(
            expect={
                "custom.prefect.block-audit.expected-block-missing",
                "custom.prefect.block-audit.unexpected-block-found",
            },
            match={
                "prefect.resource.id": "custom.prefect.block-audit",
            },
            posture=Posture.Reactive,
            threshold=1,
            within=timedelta(seconds=0),
        ),
        actions=[
            SendNotification(
                block_document_id=notification_block_document_id,
                subject="Prefect block audit detected a gap",
                body=(
                    "A Prefect block audit event was emitted.\n\n"
                    "Event: {{ event.event }}\n"
                    "Message: {{ event.payload['message'] }}\n"
                    "Block type slug: {{ event.payload['block_type_slug'] }}\n"
                    "Block name: {{ event.payload['block_name'] }}\n"
                    "Gap direction: {{ event.payload['gap_direction'] }}\n"
                ),
            )
        ],
    )
    return automation.create()
Example usage ```python from uuid import UUID automation = create_block_audit_notification_automation( notification_block_document_id=UUID("11111111-2222-3333-4444-555555555555") )
print(f"Created automation with ID: {automation.id}") ``` A few practical notes - The automation’s 
notification_block_document_id
must be the UUID of a saved notification block document. - For example, you might use a Slack webhook notification block. - The event matching is based on: - exact event names in 
expect
- 
prefect.resource.id == "custom.prefect.block-audit"
Why slugs are better here Using 
block_type.slug
is usually preferable to 
block_type_name
because: - slugs are more stable than display names - they are better for automation and code-based comparisons - they avoid formatting inconsistencies like spaces/capitalization If you want, I can next give you either: 1. a deployment-ready version with 
prefect deploy
, or 2. a version that fails the flow run when gaps are found, in addition to emitting events.
r
@Marvin Can you convert the create automations python script to a yaml format?
m
✅ thought for 123.3 seconds
Yes — for general automations, the YAML should be in a separate file, not in 
prefect.yaml
. You can then create it with:
Copy code
bash
prefect automation create --from-file automation.yaml
Verified CLI: - 
prefect automation create --from-file ...
Here is the YAML equivalent of the Python automation script from above:
Copy code
yaml
name: Notify on Prefect block audit gaps
description: Send a notification whenever the block audit flow detects a missing expected block or an unexpected saved block.
enabled: true

trigger:
  type: event
  posture: Reactive
  expect:
    - custom.prefect.block-audit.expected-block-missing
    - custom.prefect.block-audit.unexpected-block-found
  match:
    prefect.resource.id: custom.prefect.block-audit
  threshold: 1
  within: 0

actions:
  - type: send-notification
    block_document_id: "11111111-2222-3333-4444-555555555555"
    subject: "Prefect block audit detected a gap"
    body: |
      A Prefect block audit event was emitted.

      Event: {{ event.event }}
      Message: {{ event.payload['message'] }}
      Block type slug: {{ event.payload['block_type_slug'] }}
      Block name: {{ event.payload['block_name'] }}
      Gap direction: {{ event.payload['gap_direction'] }}
Create it with:
Copy code
bash
prefect automation create --from-file automation.yaml
If you want multiple automations in one file, wrap them in an 
automations:
list:
Copy code
yaml
automations:
  - name: Notify on Prefect block audit gaps
    description: Send a notification whenever the block audit flow detects a missing expected block or an unexpected saved block.
    enabled: true
    trigger:
      type: event
      posture: Reactive
      expect:
        - custom.prefect.block-audit.expected-block-missing
        - custom.prefect.block-audit.unexpected-block-found
      match:
        prefect.resource.id: custom.prefect.block-audit
      threshold: 1
      within: 0
    actions:
      - type: send-notification
        block_document_id: "11111111-2222-3333-4444-555555555555"
        subject: "Prefect block audit detected a gap"
        body: |
          A Prefect block audit event was emitted.

          Event: {{ event.event }}
          Message: {{ event.payload['message'] }}
          Block type slug: {{ event.payload['block_type_slug'] }}
          Block name: {{ event.payload['block_name'] }}
          Gap direction: {{ event.payload['gap_direction'] }}
A few important notes: - 
block_document_id
must be the UUID of your saved notification block document. - This automation YAML is not part of 
prefect.yaml
. - 
prefect.yaml
supports deployment triggers, but not general automations like 
send-notification
. Relevant command: - 
prefect automation create --from-file automation.yaml
If you want, I can also convert this into: - a Slack notification-specific example, or - a full package with the flow Python file plus the automation YAML side-by-side.
r
@Marvin rewrite your audit_prefect_blocks flow from above Yesterday at 10:27 AM to use the full block slug. By full block slug, I mean the slug that is output from calling prefect block ls and combines the block_type and the block_name already as separated by a / That should simplify the code. Otherwise try to stay true to what you've already done Yesterday at 10:27 AM
m
✅ thought for 27.6 seconds
Yep — using the full block slug in the 
block_type_slug/block_name
form is a nice simplification. Below is a rewritten version of 
audit_prefect_blocks
that: - uses a single 
block_slug
field instead of separate 
block_type_slug
and 
block_name
- compares expected vs actual blocks using the full slug shown by 
prefect block ls
- still emits events for: - missing expected blocks - unexpected actual blocks - keeps the same general structure, docstrings, and lint-friendly style ```python from future import annotations from collections.abc import Iterable from typing import Any from pydantic import BaseModel, Field, field_validator from prefect import flow, get_run_logger from prefect.client.orchestration import get_client from prefect.events import emit_event EVENT_PREFIX = "custom.prefect.block-audit" RESOURCE_ID = "custom.prefect.block-audit" class ExpectedBlock(BaseModel): """Represents one expected Prefect block using its full block slug. The 
block_slug
must match the format shown by 
prefect block ls
, combining the block type slug and block name as: <block_type_slug>/<block_name> Example: - 
secret/db-password
- 
s3-bucket/raw-data
- 
webhook/alerts
""" block_slug: str = Field( ..., description=( "The full Prefect block slug in the form " "'<block_type_slug>/<block_name>'." ), examples=["secret/db-password", "s3-bucket/raw-data", "webhook/alerts"], ) @field_validator("block_slug") @classmethod def validate_block_slug(cls, value: str) -> str: """Validate and normalize a full block slug. Args: value: A block slug in the form 
<block_type_slug>/<block_name>
. Returns: The stripped block slug. Raises: ValueError: If the block slug is blank or does not contain exactly one 
/
separator. """ normalized = value.strip() if not normalized: msg = "block_slug must not be blank." raise ValueError(msg) parts = normalized.split("/", maxsplit=1) if len(parts) != 2 or not parts[0].strip() or not parts[1].strip(): msg = ( "block_slug must be in the form " "'<block_type_slug>/<block_name>'." ) raise ValueError(msg) return normalized def _build_expected_set( expected_blocks: Iterable[ExpectedBlock], ) -> set[str]: """Convert expected block models into a set of comparable full block slugs. Args: expected_blocks: Expected block definitions. Returns: A set of full block slugs. """ return {block.block_slug for block in expected_blocks} def _split_block_slug(block_slug: str) -> tuple[str, str]: """Split a full block slug into its block type slug and block name. Args: block_slug: Full block slug in the form 
<block_type_slug>/<block_name>
. Returns: A tuple of 
(block_type_slug, block_name)
. """ return tuple(block_slug.split("/", maxsplit=1)) # type: ignore[return-value] async def _fetch_saved_blocks() -> set[str]: """Fetch all saved non-anonymous blocks from the Prefect API. This function paginates through 
read_block_documents()
and extracts the full block slug for each block as: <block_type_slug>/<block_name> Returns: A set of full block slugs. """ saved_blocks: set[str] = set() offset = 0 limit = 200 async with get_client() as client: while True: block_documents = await client.read_block_documents( offset=offset, limit=limit, include_secrets=False, ) if not block_documents: break for block_document in block_documents: if ( not block_document.name
or not block_document.block_type or not block_document.block_type.slug ): continue saved_blocks.add( f"{block_document.block_type.slug}/{block_document.name}" ) offset += len(block_documents) return saved_blocks def _emit_gap_event( *, event_name: str, gap_direction: str, block_slug: str, message: str, ) -> None: """Emit a custom event describing a block inventory gap. Args: event_name: Full event name. gap_direction: Gap direction such as 
missing_expected
or 
unexpected_actual
. block_slug: Full block slug in the form 
<block_type_slug>/<block_name>
. message: Human-readable explanation of the gap. """ block_type_slug, block_name = _split_block_slug(block_slug) emit_event( event=event_name, resource={ "prefect.resource.id": RESOURCE_ID, "prefect.resource.name": "prefect-block-audit", "block_slug": block_slug, "block_type_slug": block_type_slug, "block_name": block_name, "gap_direction": gap_direction, }, payload={ "message": message, "block_slug": block_slug, "block_type_slug": block_type_slug, "block_name": block_name, "gap_direction": gap_direction, }, ) @flow(name="audit-prefect-blocks") async def audit_prefect_blocks( expected_blocks: list[ExpectedBlock], ) -> dict[str, Any]: """Audit saved Prefect blocks against an expected inventory. This flow compares the current saved blocks in a self-hosted Prefect server against an expected list provided as a flow parameter. The comparison uses the full block slug shown by `prefect block ls`: <block_type_slug>/<block_name> It emits events for: - expected blocks that are missing from Prefect server - blocks present in Prefect server that are not in the expected inventory Args: expected_blocks: Expected block definitions using full block slugs. Returns: A summary dictionary describing the expected, actual, missing, and unexpected block inventories. """ logger = get_run_logger() expected_set = _build_expected_set(expected_blocks) actual_set = await _fetch_saved_blocks() missing_blocks = sorted(expected_set - actual_set) unexpected_blocks = sorted(actual_set - expected_set) logger.info("Expected blocks: %d", len(expected_set)) logger.info("Actual saved blocks: %d", len(actual_set)) logger.info("Missing expected blocks: %d", len(missing_blocks)) logger.info("Unexpected actual blocks: %d", len(unexpected_blocks)) for block_slug in missing_blocks: message = ( "Expected block was not found in Prefect server: " f"block_slug={block_slug!r}." ) _emit_gap_event( event_name=f"{EVENT_PREFIX}.expected-block-missing", gap_direction="missing_expected", block_slug=block_slug, message=message, ) for block_slug in unexpected_blocks: message = ( "Prefect server contains a block that was not found in the expected " f"inventory: block_slug={block_slug!r}." ) _emit_gap_event( event_name=f"{EVENT_PREFIX}.unexpected-block-found", gap_direction="unexpected_actual", block_slug=block_slug, message=message, ) summary = { "expected_blocks": [ {"block_slug": block_slug} for block_slug in sorted(expected_set) ], "actual_blocks": [ {"block_slug": block_slug} for block_slug in sorted(actual_set) ], "missing_blocks": [ {"block_slug": block_slug}
for block_slug in missing_blocks ], "unexpected_blocks": [ {"block_slug": block_slug} for block_slug in unexpected_blocks ], "missing_count": len(missing_blocks), "unexpected_count": len(unexpected_blocks), } logger.info("Block audit summary: %s", summary) return summary if name == "__main__": audit_prefect_blocks( expected_blocks=[ ExpectedBlock(block_slug="s3-bucket/raw-data"), ExpectedBlock(block_slug="secret/db-password"), ExpectedBlock(block_slug="webhook/alerts"), ExpectedBlock(block_slug="webhook/raw-data"), ] ) ``` What changed - Replaced: - 
block_type_slug
- 
block_name
with just: - 
block_slug
- Comparison is now just set math on strings like: - 
secret/db-password
- 
s3-bucket/raw-data
- The event payload still includes: - 
block_slug
- 
block_type_slug
- 
block_name
so downstream automations and notifications can still reference either the full slug or the separated parts. One small note I kept 
_split_block_slug()
so emitted events still contain both the full slug and its components. That makes the flow simpler without losing detail in event payloads. If you want, I can also rewrite the automation YAML to reference just 
block_slug
in the message body.
r
@Marvin why does _emit
m
✅ thought for 9.7 seconds
Do you mean why does 
_emit_gap_event()
still exist now that you’re using 
block_slug
? If so: it’s mostly for encapsulation and consistency. It centralizes: - the 
emit_event(...)
call - the resource structure - the payload structure - splitting 
block_slug
into: - 
block_type_slug
- 
block_name
So instead of repeating this in both loops, you do it once. Without 
_emit_gap_event()
, you’d repeat something like this twice:
Copy code
python
block_type_slug, block_name = _split_block_slug(block_slug)

emit_event(
    event=event_name,
    resource={
        "prefect.resource.id": RESOURCE_ID,
        "prefect.resource.name": "prefect-block-audit",
        "block_slug": block_slug,
        "block_type_slug": block_type_slug,
        "block_name": block_name,
        "gap_direction": gap_direction,
    },
    payload={
        "message": message,
        "block_slug": block_slug,
        "block_type_slug": block_type_slug,
        "block_name": block_name,
        "gap_direction": gap_direction,
    },
)
So 
_emit_gap_event()
helps keep the flow body cleaner. If your real question is instead why does 
_emit_gap_event()
split the slug at all, the answer is: - it doesn’t need to for the comparison - it only does that so the emitted event includes both: - the full slug - the individual components That can be useful for: - notification templates - automation filters - downstream consumers If you want the simplest possible version, you can remove 
_split_block_slug()
entirely and keep only 
block_slug
in the event:
Copy code
python
def _emit_gap_event(
    *,
    event_name: str,
    gap_direction: str,
    block_slug: str,
    message: str,
) -> None:
    """Emit a custom event describing a block inventory gap."""
    emit_event(
        event=event_name,
        resource={
            "prefect.resource.id": RESOURCE_ID,
            "prefect.resource.name": "prefect-block-audit",
            "block_slug": block_slug,
            "gap_direction": gap_direction,
        },
        payload={
            "message": message,
            "block_slug": block_slug,
            "gap_direction": gap_direction,
        },
    )
If you want, I can rewrite the whole flow into that minimal block_slug-only version.
r
@Marvin Why does emit_gap_event have a wildcard * as it's first parameter. Surely, that is not best practice. Is it really needed? And if it is needed can you solve it with a more precise approach?
m
✅ thought for 14.3 seconds
Good question. The 
*
in this function:
Copy code
python
def _emit_gap_event(
    *,
    event_name: str,
    gap_direction: str,
    block_slug: str,
    message: str,
) -> None:
means: all following arguments must be passed by keyword, not position. So this is allowed:
Copy code
python
_emit_gap_event(
    event_name="custom.prefect.block-audit.expected-block-missing",
    gap_direction="missing_expected",
    block_slug="secret/db-password",
    message="Expected block was not found.",
)
and this is not:
Copy code
python
_emit_gap_event(
    "custom.prefect.block-audit.expected-block-missing",
    "missing_expected",
    "secret/db-password",
    "Expected block was not found.",
)
Why people use it It’s often used when: - several parameters are the same type, especially multiple `str`s - positional ordering would be easy to mix up - readability at call sites matters In your case, all four parameters are strings, so keyword-only arguments do help prevent mistakes. Is it required? No — it is not required. Is it bad practice? Also no — keyword-only arguments are a legitimate Python feature and often considered good API design. That said, for a small private helper like 
_emit_gap_event()
, I agree it may be more ceremony than necessary. If you want a more precise and cleaner approach, I’d suggest one of these: *Option 1: remove the 
*
entirely* This is the simplest approach.
Copy code
python
def _emit_gap_event(
    event_name: str,
    gap_direction: str,
    block_slug: str,
    message: str,
) -> None:
    """Emit a custom event describing a block inventory gap."""
    emit_event(
        event=event_name,
        resource={
            "prefect.resource.id": RESOURCE_ID,
            "prefect.resource.name": "prefect-block-audit",
            "block_slug": block_slug,
            "gap_direction": gap_direction,
        },
        payload={
            "message": message,
            "block_slug": block_slug,
            "gap_direction": gap_direction,
        },
    )
This works fine, but it makes it easier to accidentally swap string arguments. Option 2: use a typed model for the event payload This is the more precise approach if you want stronger structure without relying on keyword-only args. Example:
Copy code
python
from pydantic import BaseModel


class BlockGapEvent(BaseModel):
    """Represents one block inventory gap event."""

    event_name: str
    gap_direction: str
    block_slug: str
    message: str


def _emit_gap_event(event: BlockGapEvent) -> None:
    """Emit a custom event describing a block inventory gap."""
    emit_event(
        event=event.event_name,
        resource={
            "prefect.resource.id": RESOURCE_ID,
            "prefect.resource.name": "prefect-block-audit",
            "block_slug": event.block_slug,
            "gap_direction": event.gap_direction,
        },
        payload={
            "message": event.message,
            "block_slug": event.block_slug,
            "gap_direction": event.gap_direction,
        },
    )
Call it like:
Copy code
python
_emit_gap_event(
    BlockGapEvent(
        event_name=f"{EVENT_PREFIX}.expected-block-missing",
        gap_direction="missing_expected",
        block_slug=block_slug,
        message=message,
    )
)
I think this is the most precise option if your goal is correctness and maintainability. My recommendation For this specific helper, I’d choose one of these: - simplest: remove the 
*
- most precise: pass a 
BlockGapEvent
model Since you asked for a more precise approach, I’d lean toward the model approach. Here’s a compact replacement: ```python class BlockGapEvent(BaseModel): """Represents one block inventory gap event.""" event_name: str gap_direction: str block_slug: str message: str def _emit_gap_event(event: BlockGapEvent) -> None: """Emit a custom event describing a block inventory gap.""" emit_event( event=event.event_name, resource={
"prefect.resource.id": RESOURCE_ID, "prefect.resource.name": "prefect-block-audit", "block_slug": event.block_slug, "gap_direction": event.gap_direction, }, payload={ "message": event.message, "block_slug": event.block_slug, "gap_direction": event.gap_direction, }, ) ``` If you want, I can rewrite the full flow now using this more precise model-based approach.
r
@Marvin Dottore! I've have learned something new. Take a bow
m
✅ thought for 4.1 seconds
Grazie, Dottore 😄 Glad that was useful. The 
*
is one of those small Python features that feels odd until you see why people use it. If you want, I can do a final cleanup pass on the flow and make it: - simpler - more idiomatic - strictly slug-based - model-driven where it actually helps and trim out any unnecessary cleverness.
r
@Marvin what command to run to get just the list of slugs from the prefect blocks ls command ?
m
✅ thought for 16.9 seconds
Use JSON output, then extract the slug field.
Copy code
bash
prefect block ls --output json
If you want just the slugs, pipe it to `jq`:
Copy code
bash
prefect block ls --output json | jq -r '.[].slug'
If 
jq
is not available, in Python:
Copy code
bash
prefect block ls --output json | python -c "import sys, json; print('\n'.join(item['slug'] for item in json.load(sys.stdin)))"
So the most practical answer is:
Copy code
bash
prefect block ls --output json | jq -r '.[].slug'
There isn’t a built-in CLI flag on 
prefect block ls
to output only slugs directly. Relevant command: - 
prefect block ls --help
r
@Marvin that just gives me a bunch of nulls I suspect that it cannot find "slug" and that it probably some other key word
m
✅ thought for 17.5 seconds
Yep — you were right. 
prefect block ls --output json
does not return a top-level 
slug
field. The slug is nested, and the full block slug needs to be constructed from: - 
.block_schema.fields.block_type_slug
- 
.name
So to get the same full slug style you see in 
prefect block ls
, use:
Copy code
bash
prefect block ls --output json | jq -r '.[] | "\(.block_schema.fields.block_type_slug)/\(.name)"'
A slightly safer variant is:
Copy code
bash
prefect block ls --output json | jq -r '.[] | select(.block_schema.fields.block_type_slug and .name) | "\(.block_schema.fields.block_type_slug)/\(.name)"'
If you prefer using the nested block type object instead, this should also work when present:
Copy code
bash
prefect block ls --output json | jq -r '.[] | "\(.block_type.slug)/\(.name)"'
But based on the output you shared, the most reliable path looks like:
Copy code
bash
prefect block ls --output json | jq -r '.[] | "\(.block_schema.fields.block_type_slug)/\(.name)"'
So the reason you got 
null
is exactly what you suspected: 
slug
is not a top-level key.
r
@Marvin how do I check the size of the redis app as created by the sub chart from prefect-server from prefect-helm? Version is Prefect 3.7.0. And a second question, which redis database is it using, e.g. 1? 2?
m
✅ thought for 113.6 seconds
For Prefect 3.7.0 with the 
prefect-server
Helm chart, the answers are: 1. How to check the size of the Redis app created by the subchart There are a few different “size” questions you might mean: - configured CPU/memory requests and limits - actual runtime CPU/memory usage - persistent volume size - data currently stored in Redis memory Here’s how to check each. Find the Redis resources first
Copy code
bash
kubectl get pods -n <namespace> | grep redis
kubectl get statefulset -n <namespace> | grep redis
kubectl get svc -n <namespace> | grep redis
The Redis subchart is Bitnami Redis, and it is typically deployed as a StatefulSet. Check configured CPU / memory requests and limits
Copy code
bash
kubectl get statefulset -n <namespace> <redis-statefulset-name> -o yaml
Then look under:
Copy code
yaml
spec:
  template:
    spec:
      containers:
        - name: redis
          resources:
            requests:
            limits:
A quicker JSONPath view:
Copy code
bash
kubectl get statefulset -n <namespace> <redis-statefulset-name> \
  -o jsonpath='{.spec.template.spec.containers[0].resources}'
Check actual runtime usage If metrics-server is installed:
Copy code
bash
kubectl top pod -n <namespace> <redis-pod-name>
Check PVC size
Copy code
bash
kubectl get pvc -n <namespace> | grep redis
kubectl describe pvc -n <namespace> <redis-pvc-name>
Check actual Redis memory usage Exec into the Redis pod:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- redis-cli INFO memory
Or just the key fields:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- redis-cli INFO memory | grep used_memory
Useful fields include: - 
used_memory
- 
used_memory_human
- 
used_memory_peak
- 
maxmemory
Check the Helm values actually applied This is often the best way to confirm the configured size from the chart:
Copy code
bash
helm get values <release-name> -n <namespace>
or all computed values:
Copy code
bash
helm get values <release-name> -n <namespace> --all
Then inspect the 
redis:
section. Prefect’s chart uses the Bitnami Redis subchart, so Redis sizing is controlled through that subchart’s values. 2. Which Redis database number is used? For the Prefect Helm chart defaults, the Redis DB for messaging is:
Copy code
text
0
The chart sets:
Copy code
yaml
backgroundServices:
  messaging:
    redis:
      db: 0
and this becomes the environment variable:
Copy code
text
PREFECT_REDIS_MESSAGING_DB=0
So for Prefect background services messaging, the default Redis DB is 0. Important nuance about Docket There is also separate Docket configuration in the chart, and if you set a Docket Redis URL yourself, that URL can use a different DB, for example 
/1
. Example from chart patterns:
Copy code
text
<redis://host:6379/1>
So: - Prefect messaging Redis DB default: 
0
- Docket Redis DB: whatever is in 
backgroundServices.messaging.docket.url
- if unset, Docket may not use Redis at all How to confirm in your running cluster Check the env vars on the background services pod:
Copy code
bash
kubectl get pod -n <namespace> <background-services-pod> -o yaml
Look for:
Copy code
text
PREFECT_REDIS_MESSAGING_DB
PREFECT_SERVER_DOCKET_URL
Or directly:
Copy code
bash
kubectl exec -n <namespace> <background-services-pod> -- env | grep PREFECT_REDIS_MESSAGING_DB
kubectl exec -n <namespace> <background-services-pod> -- env | grep PREFECT_SERVER_DOCKET_URL
Summary - Redis subchart “size” can be checked via: - 
kubectl get statefulset ... -o yaml
- 
kubectl top pod ...
- 
kubectl describe pvc ...
- 
redis-cli INFO memory
- Prefect messaging uses Redis DB 0 by default - Docket may use a different Redis DB if configured separately in its URL If you want, I can give you a single copy-paste command set tailored to your namespace and release name.
r
@Marvin For the size question, I'm after usage of storage space. I want to monitor if I'm getting too close to the allocated resource size.
m
✅ thought for 21.8 seconds
Got it — you mean storage consumption vs the allocated storage capacity for the Redis subchart, not CPU/memory. For the Bitnami Redis subchart used by the Prefect Helm chart, the thing to monitor is usually the PVC backing Redis persistence. 1. Find the Redis PVC
Copy code
bash
kubectl get pvc -n <namespace> | grep redis
Then inspect it:
Copy code
bash
kubectl describe pvc -n <namespace> <redis-pvc-name>
That shows the requested/allocated size, for example 
8Gi
, but not the used bytes inside the filesystem. 2. Check actual disk usage inside the Redis pod The most direct way is to exec into the Redis pod and run 
df
on the Redis data directory. First find the pod:
Copy code
bash
kubectl get pods -n <namespace> | grep redis
Then run:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- df -h /bitnami/redis/data
That should show something like:
Copy code
text
Filesystem      Size  Used Avail Use% Mounted on
/dev/...         8G   2.1G  5.9G  27% /bitnami/redis/data
That is usually the most useful answer to “how close am I to the allocated size?” You can also inspect directory size:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- du -sh /bitnami/redis/data
3. Compare allocated size vs used size - allocated size: from the PVC - used size: from 
df -h /bitnami/redis/data
So your quick check is:
Copy code
bash
kubectl describe pvc -n <namespace> <redis-pvc-name>
kubectl exec -it -n <namespace> <redis-pod-name> -- df -h /bitnami/redis/data
4. If you want Redis logical memory usage too This is separate from disk persistence, but often useful:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- redis-cli INFO memory
That tells you how much data Redis is holding in memory, which is not the same as PVC usage. Important caveat If Redis persistence is disabled, there may be no PVC at all. In that case: - there is no persistent storage allocation to compare against - Redis data is just in the container filesystem / ephemeral volume - the monitoring approach changes You can check whether persistence is enabled via:
Copy code
bash
helm get values <release-name> -n <namespace> --all
and inspect the 
redis.persistence
section. Recommended monitoring target If your goal is “alert me before Redis storage fills up”, the best signal is: - filesystem usage percentage on 
/bitnami/redis/data
For example, alert at: - 70% - 80% - 90% Most useful one-liner
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- df -h /bitnami/redis/data
If you want, I can also help you turn this into: - a Prometheus metric / alert - a small Prefect flow to check Redis disk usage periodically - a Kubernetes command sequence that auto-discovers the Redis pod and PVC names
r
@Marvin when running kubectl exec -it -n <namespace> <redis-pod-name> -- df -h /bitnami/redis/data how do I pass the password
m
✅ thought for 8.9 seconds
You generally don’t need a Redis password for that command.
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- df -h /bitnami/redis/data
runs 
df
inside the container filesystem. It is a Kubernetes exec into the pod, not a Redis client connection, so Redis authentication is not involved. You only need the Redis password when using 
redis-cli
, for example:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- redis-cli -a '<password>' INFO memory
or better, with an env var to avoid putting the password directly in shell history:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -- sh -c 'redis-cli -a "$REDIS_PASSWORD" INFO memory'
But for 
df
and 
du
, no Redis password is needed. If 
kubectl exec
itself is failing, that would usually be an RBAC or pod/container issue, not a Redis password issue. For example, if the pod has multiple containers, specify the container:
Copy code
bash
kubectl exec -it -n <namespace> <redis-pod-name> -c redis -- df -h /bitnami/redis/data
So the short answer is: - 
df
/ `du`: no password needed - `redis-cli`: password may be needed
r
@Marvin Ok, write a small "check_redis_storage" prefect flow using kubectl exec -it -n <namespace> <redis-pod-name> -c redis -- df -h /bitnami/redis/data to monitor against a threshold as passed as a parameter. namesspace and podname should also be parameters. The flow should emit an event if threshold is exceeded.
m
✅ thought for 112.5 seconds
Absolutely — here’s a small Prefect 3 flow that: - accepts 
namespace
, 
redis_pod_name
, and 
threshold_percent
as parameters - runs:
Copy code
bash
kubectl exec -n <namespace> <redis-pod-name> -c redis -- df -h /bitnami/redis/data
- parses the filesystem usage percentage - emits an event if usage exceeds the threshold I removed 
-it
because interactive TTY flags are usually not appropriate for subprocess execution from automation code. ```python from future import annotations import re import subprocess from typing import Any from pydantic import BaseModel, Field, field_validator from prefect import flow, get_run_logger from prefect.events import emit_event EVENT_NAME = "custom.redis.storage-threshold-exceeded" RESOURCE_PREFIX = "custom.redis.storage" class RedisStorageCheckParameters(BaseModel): """Parameters for checking Redis persistent storage usage.""" namespace: str = Field( ..., description="Kubernetes namespace containing the Redis pod.", ) redis_pod_name: str = Field( ..., description="Name of the Redis pod to inspect.", ) threshold_percent: float = Field( ..., description="Usage percentage threshold that triggers an event.", ge=0.0, le=100.0, ) @field_validator("namespace", "redis_pod_name") @classmethod def validate_not_blank(cls, value: str) -> str: """Validate that string parameters are non-empty. Args: value: Input string value. Returns: The stripped value. Raises: ValueError: If the value is blank. """ normalized = value.strip() if not normalized: msg = "Value must not be blank." raise ValueError(msg) return normalized def _run_df_command(namespace: str, redis_pod_name: str) -> str: """Run 
df -h
in the Redis container and return stdout. Args: namespace: Kubernetes namespace containing the Redis pod. redis_pod_name: Redis pod name. Returns: The command stdout. Raises: RuntimeError: If the command fails. """ command = [ "kubectl", "exec", "-n", namespace, redis_pod_name, "-c", "redis", "--", "df", "-h", "/bitnami/redis/data", ] completed_process = subprocess.run( command, capture_output=True, text=True, check=False, timeout=60, ) if completed_process.returncode != 0: msg = ( "Failed to run kubectl exec for Redis storage check. " f"returncode={completed_process.returncode}, " f"stderr={completed_process.stderr.strip()!r}" ) raise RuntimeError(msg) return completed_process.stdout def _parse_df_output(df_output: str) -> dict[str, Any]: """Parse 
df -h
output and extract storage usage details. Expected output resembles: Filesystem Size Used Avail Use% Mounted on /dev/... 8G 2.1G 5.9G 27% /bitnami/redis/data Args: df_output: Raw stdout from 
df -h
. Returns: A dictionary containing parsed fields. Raises: ValueError: If the output cannot be parsed. """ lines = [line.strip() for line in df_output.splitlines() if line.strip()] if len(lines) < 2: msg = f"Unexpected df output: {df_output!r}" raise ValueError(msg) data_line = lines[-1] columns = data_line.split() if len(columns) < 6: msg = f"Unable to parse df data line: {data_line!r}" raise ValueError(msg) filesystem = columns[0] size = columns[1] used = columns[2] available = columns[3] use_percent_raw = columns[4] mounted_on = columns[5] match = re.fullmatch(r"(\d+)%", use_percent_raw) if not match: msg = f"Unable to parse usage percentage: {use_percent_raw!r}"
raise ValueError(msg) used_percent = float(match.group(1)) return { "filesystem": filesystem, "size": size, "used": used, "available": available, "used_percent": used_percent, "mounted_on": mounted_on, "raw_output": df_output, } def _emit_threshold_event( namespace: str, redis_pod_name: str, threshold_percent: float, storage_details: dict[str, Any], ) -> None: """Emit an event indicating Redis storage usage exceeded the threshold. Args: namespace: Kubernetes namespace containing the Redis pod. redis_pod_name: Redis pod name. threshold_percent: Configured threshold percentage. storage_details: Parsed filesystem usage details. """ used_percent = storage_details["used_percent"] event = emit_event( event=EVENT_NAME, resource={ "prefect.resource.id": ( f"{RESOURCE_PREFIX}.{namespace}.{redis_pod_name}" ), "prefect.resource.name": redis_pod_name, "kubernetes.namespace": namespace, "kubernetes.pod.name": redis_pod_name, }, payload={ "message": ( "Redis storage usage exceeded the configured threshold." ), "namespace": namespace, "redis_pod_name": redis_pod_name, "threshold_percent": threshold_percent, "used_percent": used_percent, "filesystem": storage_details["filesystem"], "size": storage_details["size"], "used": storage_details["used"], "available": storage_details["available"], "mounted_on": storage_details["mounted_on"], }, ) logger = get_run_logger() if event is None: logger.warning("Threshold exceeded, but event emission returned None.") else: logger.info("Emitted event %s", event.id) @flow(name="check-redis-storage") def check_redis_storage( namespace: str, redis_pod_name: str, threshold_percent: float, ) -> dict[str, Any]: """Check Redis persistent storage usage and emit an event if exceeded. This flow executes 
df -h /bitnami/redis/data
inside the Redis container, parses the reported usage, and emits a custom Prefect event when the usage percentage is greater than the configured threshold. Args: namespace: Kubernetes namespace containing the Redis pod. redis_pod_name: Name of the Redis pod. threshold_percent: Usage percentage threshold for event emission. Returns: A dictionary containing the parsed storage details and threshold status. """ logger = get_run_logger() params = RedisStorageCheckParameters( namespace=namespace, redis_pod_name=redis_pod_name, threshold_percent=threshold_percent, ) logger.info( "Checking Redis storage usage for pod %s in namespace %s", params.redis_pod_name, params.namespace, ) df_output = _run_df_command( namespace=params.namespace, redis_pod_name=params.redis_pod_name, ) storage_details = _parse_df_output(df_output) used_percent = storage_details["used_percent"] threshold_exceeded = used_percent > params.threshold_percent logger.info( "Redis storage usage is %.1f%% (threshold %.1f%%)", used_percent, params.threshold_percent, ) if threshold_exceeded: logger.warning( "Redis storage usage exceeded threshold: %.1f%% > %.1f%%", used_percent, params.threshold_percent, ) _emit_threshold_event( namespace=params.namespace, redis_pod_name=params.redis_pod_name, threshold_percent=params.threshold_percent, storage_details=storage_details, ) return { "namespace": params.namespace,
"redis_pod_name": params.redis_pod_name, "threshold_percent": params.threshold_percent, "threshold_exceeded": threshold_exceeded, **storage_details, } if name == "__main__": result = check_redis_storage( namespace="prefect", redis_pod_name="prefect-redis-master-0", threshold_percent=80.0, ) print(result)
Copy code
*Notes*

- I used `subprocess.run(...)` with a timeout of 60 seconds.
- I intentionally did *not* use `-it` because TTY allocation is usually wrong in non-interactive process execution.
- The flow emits this custom event:

```text
custom.redis.storage-threshold-exceeded
- The event payload includes: - 
threshold_percent
- 
used_percent
- 
size
- 
used
- 
available
- 
namespace
- 
redis_pod_name
If you want, I can also provide: - a matching automation YAML that sends a notification when this event is emitted - a version that uses 
df -P
for slightly more stable parsing - an async version using Prefect process utilities instead of 
subprocess.run()
r
@Marvin Building from your reply and script from Yesterday at 9:28 AM, is there another way to get the redis usage other than calling kubectl?
m
✅ thought for 12.1 seconds
Yes — if your code is already running inside the cluster, you often don’t need 
kubectl
at all. A few alternatives: 1. Query Redis directly If by “Redis usage” you mean memory / keys / stats, connect to Redis and use `INFO`:
Copy code
import redis

r = redis.Redis(host="your-redis-host", port=6379, decode_responses=True)

info = r.info()
print(info["used_memory_human"])
print(info["connected_clients"])
print(info["db0"])  # if present
Useful fields from 
INFO
include: - 
used_memory
- 
used_memory_human
- 
connected_clients
- 
total_commands_processed
- 
keyspace_hits
- 
keyspace_misses
If you want per-DB key counts, look at sections like 
db0
, 
db1
, etc. 2. Use the Kubernetes Python client instead of shelling out to 
kubectl
 If you were using 
kubectl
just to find the Redis pod or inspect pod/container metrics, you can call the Kubernetes API directly:
Copy code
from kubernetes import client, config

config.load_incluster_config()  # or config.load_kube_config()

v1 = client.CoreV1Api()
pods = v1.list_namespaced_pod(
    namespace="default",
    label_selector="app=redis",
)

for pod in pods.items:
    print(pod.metadata.name)
    print(pod.status.phase)
This is generally better than subprocess + 
kubectl
in automation. 3. Read Prometheus metrics If Redis is exporting metrics via 
redis-exporter
, scrape/query Prometheus instead of touching Kubernetes directly. That’s usually the best option for “usage” over time. Typical metrics include things like: - 
redis_memory_used_bytes
- 
redis_connected_clients
- 
redis_commands_processed_total
4. If you only need container resource usage If you meant CPU/memory usage of the Redis pod/container rather than Redis internals, then 
kubectl top
is just one frontend for the Kubernetes metrics API. You can query that API directly, though it’s a bit more involved than using the core K8s client. So the best alternative depends on what “usage” means here: - Redis internals → use Redis 
INFO
- Pod/container status → Kubernetes Python client - Historical monitoring / dashboards → Prometheus - CPU/memory consumption of pod → metrics API / Prometheus If you want, paste the exact script from yesterday (or the part that uses 
kubectl
) and I can rewrite it to use either: - the Redis Python client, or - the Kubernetes Python client without shelling out.
r
@Marvin Write a modified kubernetes base job template that uses the public prefect-server image version 3.7.0 from dockerhub that copies own certificates from an internal git repo into the correct path, e.g. /usr/local/share/ca-certificates/ , in the perfect-server job image and runs update-ca-certificates. Also apt-get update && apt-get install a list of system packages and clean up after the apt-get install into that image. Also installing python uv and a list of python packages, preferably as dynamically discovered from a repo's pyproject.toml into that image. The goal is to avoid having to make an in house Dockerfile saved to an in house dockerhub, and rather just rely on the publicly available image, but enriching it with certificates, operating system packaes, and also getting a uv virtual environment installed. If there are ways to do this better than the base job template idea, then please suggest those as well.
m
✅ thought for 201.3 seconds
Yes — you can do this in a Kubernetes work pool base job template, but there are two important caveats: 1. Use 
prefecthq/prefect
, not a 
prefect-server
image, for flow run jobs 2. System package installs and 
update-ca-certificates
at pod startup are possible, but they’re usually slower and more fragile than a small custom image So I’ll give you: - a modified base job template pattern - a recommended better approach - caveats so you know what will and won’t work well Important image note For Prefect flow runs on Kubernetes, you generally want the public Prefect image from Docker Hub, e.g.: 
prefecthq/prefect:3.7.0
Not a “prefect-server job image”. The server image is for running Prefect services, not your flow-run execution environment. You can see Prefect deploy docs here: Prefect deployment docs And Kubernetes worker docs here: Prefect Kubernetes integration docs --- Best practical recommendation If your goal is: - add internal CA certs - install OS packages - install 
uv
- install Python deps from 
pyproject.toml
then the best long-term pattern is usually: Option A — Better / more reliable Build a very small bootstrap image based on 
prefecthq/prefect:3.7.0
and publish it somewhere accessible. Why this is better: - much faster pod startup - fewer runtime failures from 
apt-get
- no dependency on outbound network during every flow run - cert handling is cleaner - easier to audit/reproduce Option B — Works without custom image Use the base job template to: - add an initContainer that clones certs into a shared volume - mount that volume into the main Prefect container - override the main container command to run a bootstrap shell script that: - installs apt packages - installs 
uv
if needed - installs Python deps - then `exec`s the original Prefect command This is doable, but a bit hacky. --- Important constraint with certificates If you mount certs into 
/usr/local/share/ca-certificates
and run 
update-ca-certificates
, that updates the certificate bundle inside that container filesystem. An 
initContainer
cannot directly mutate the main container’s root filesystem. So for certs, the workable runtime patterns are: Pattern 1: Shared volume + env vars - init container clones/copies cert files into a shared volume - main container mounts the volume - main container runs 
update-ca-certificates
itself before starting Prefect, or points tools at the cert bundle with env vars like: - 
SSL_CERT_FILE
- 
REQUESTS_CA_BUNDLE
- 
PIP_CERT
- 
GIT_SSL_CAINFO
Pattern 2: Main container bootstrap script - mount cert files into main container - main container startup script copies them into 
/usr/local/share/ca-certificates
- run 
update-ca-certificates
- continue to Prefect engine That’s the closest to what you asked for. --- Example base job template This example assumes: - Kubernetes work pool - base image 
prefecthq/prefect:3.7.0
- cert repo is cloned by an init container - main container does final setup - dependencies are installed at runtime - code comes from your deployment as usual You would paste/adapt this as your work pool’s base job template. ```json { "variables": { "properties": { "namespace": { "title": "Namespace", "type": "string", "default": "default" }, "image": { "title": "Image", "type": "string", "default": "prefecthq/prefect:3.7.0" }, "service_account_name": { "title": "Service Account Name", "type": "string", "default": "default" }, "git_cert_repo": { "title": "Git repo containing CA certs", "type": "string", "default": "https://git.example.com/platform/internal-certs.git" }, "git_cert_branch": { "title": "Git branch for cert repo", "type": "string",
"default": "main" }, "cert_subdir": { "title": "Subdirectory in repo containing .crt files", "type": "string", "default": "certs" }, "apt_packages": { "title": "APT packages to install", "type": "string", "default": "git ca-certificates curl" }, "python_packages": { "title": "Additional Python packages to install with uv", "type": "string", "default": "" }, "pyproject_repo": { "title": "Optional repo containing pyproject.toml for dependency install", "type": "string", "default": "" }, "pyproject_branch": { "title": "Branch for pyproject repo", "type": "string", "default": "main" }, "env": { "title": "Environment Variables", "type": "object", "default": {} } }, "type": "object" }, "job_configuration": { "job_manifest": { "apiVersion": "batch/v1", "kind": "Job", "metadata": { "labels": "{{ labels }}", "namespace": "{{ namespace }}", "generateName": "{{ name }}-" }, "spec": { "ttlSecondsAfterFinished": 3600, "backoffLimit": 0, "template": { "metadata": { "labels": "{{ labels }}", "annotations": "{{ annotations }}" }, "spec": { "serviceAccountName": "{{ service_account_name }}", "restartPolicy": "Never", "initContainers": [ { "name": "clone-certs", "image": "alpine/git:2.45.2", "command": [ "/bin/sh", "-ec" ], "args": [ "rm -rf /work/certs-repo && git clone --depth 1 --branch {{ git_cert_branch }} {{ git_cert_repo }} /work/certs-repo && mkdir -p /certs && find /work/certs-repo/{{ cert_subdir }} -type f \\( -name '*.crt' -o -name '*.pem' \\) -exec cp {} /certs/ \\;" ], "volumeMounts": [ { "name": "certs-volume", "mountPath": "/certs" }, { "name": "working-volume", "mountPath": "/work" } ] } ], "containers": [ { "name": "prefect-job", "image": "{{ image }}", "imagePullPolicy": "IfNotPresent", "command": [ "/bin/bash", "-ec" ], "args": [ "set -euxo pipefail\n\nexport DEBIAN_FRONTEND=noninteractive\n\napt-get update && apt-get install -y --no-install-recommends {{ apt_packages }}\n\nmkdir -p /usr/local/share/ca-certificates/extra\nif compgen -G '/bootstrap/certs/*' > /dev/null; then\n cp /bootstrap/certs/* /usr/local/share/ca-certificates/extra/ || true\n update-ca-certificates\nfi\n\nexport SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt\nexport REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt\nexport GIT_SSL_CAINFO=/etc/ssl/certs/ca-certificates.crt\nexport PIP_CERT=/etc/ssl/certs/ca-certificates.crt\n\nif ! command -v uv >/dev/null 2>&1; then\n curl -LsSf https://astral.sh/uv/install.sh | sh\n export PATH=\"$HOME/.local/bin:$PATH\"\nfi\n\nif [ -n '{{ pyproject_repo }}' ]; then\n rm -rf /tmp/pyproject-repo\n git clone --depth 1 --branch {{ pyproject_branch }} {{ pyproject_repo }} /tmp/pyproject-repo\n cd /tmp/pyproject-repo\n if [ -f pyproject.toml ]; then\n uv pip install --system .\n fi\nfi\n\nif [ -n '{{ python_packages }}' ]; then\n uv pip install --system {{ python_packages }}\nfi\n\napt-get clean\nrm -rf /var/lib/apt/lists/* /root/.cache /tmp/*\n\nexec {{ command }}\n" ], "env": "{{ env }}", "volumeMounts": [
{ "name": "certs-volume", "mountPath": "/bootstrap/certs", "readOnly": true }, { "name": "working-volume", "mountPath": "/work" } ] } ], "volumes": [ { "name": "certs-volume", "emptyDir": {} }, { "name": "working-volume", "emptyDir": {} } ] } } } } } }
Copy code
---

*Very important note about `{{ command }}`*

That line is the key idea, but depending on your work pool template/version, the exact injected variable names may differ from how your current base job template is structured.

So before saving this, I strongly recommend:

1. open your current Kubernetes work pool base job template
2. preserve the existing Prefect-provided fields for:
   - image
   - env
   - labels / annotations
   - namespace
   - command / args

Then only add:
- `initContainers`
- `volumes`
- `volumeMounts`
- bootstrap shell logic

If you paste me your current base job template, I can rewrite it *exactly* instead of giving you a pattern.

---

*What this template does*

- Uses public image `prefecthq/prefect:3.7.0`
- Clones certs from internal git repo in an init container
- Shares those cert files through an `emptyDir`
- Main container:
  - runs `apt-get update`
  - installs requested system packages
  - copies certs into `/usr/local/share/ca-certificates/extra`
  - runs `update-ca-certificates`
  - sets common CA env vars
  - installs `uv` if missing
  - optionally clones another repo containing `pyproject.toml`
  - runs `uv pip install --system .`
  - optionally installs extra packages
  - cleans apt and temp files
  - finally executes the Prefect flow-run command

---

*How to handle Python deps from `pyproject.toml`*

There are a few variants.

*Simplest*
If the same repo that Prefect pulls flow code from already contains `pyproject.toml`, it may be better to install from the checked-out source before starting the flow.

But this depends on how your deployment pulls code.

If you’re using `flow.from_source(...)`, Prefect handles code retrieval for the flow itself. Your runtime bootstrap script may not know where that source is available before flow execution starts.

So the reliable runtime-only choices are:

1. *Clone the dependency repo separately in the bootstrap step*  
2. *Pass an explicit package list via variable*
3. *Use Prefect’s `EXTRA_PIP_PACKAGES` env var for Python-only additions*

The public Prefect image already supports runtime Python package installation via `EXTRA_PIP_PACKAGES`, which is much simpler if you only need Python packages.

Example deployment/job variables:

```json
{
  "env": {
    "EXTRA_PIP_PACKAGES": "pandas==2.2.2 pyarrow sqlalchemy"
  }
}
That won’t help with: - apt packages - CA certs - git-based bootstrap but it does simplify Python package installation. --- Better alternative than job template bootstrapping If you want something cleaner without maintaining a big in-house Dockerfile, here are better patterns: Option 1: Tiny derived image Just create a tiny Dockerfile like:
Copy code
dockerfile
FROM prefecthq/prefect:3.7.0

RUN apt-get update && apt-get install -y --no-install-recommends \
    git \
    ca-certificates \
    curl \
 && rm -rf /var/lib/apt/lists/*

RUN curl -LsSf <https://astral.sh/uv/install.sh> | sh
ENV PATH="/root/.local/bin:${PATH}"
Then at runtime, only: - mount/clone certs - 
update-ca-certificates
- install project deps if needed That is dramatically more reliable. Option 2: ConfigMap/Secret for certs instead of git clone If internal CA certs don’t change often, store them as: - Kubernetes 
ConfigMap
- or 
Secret
Then mount directly into the pod.
That’s much better than cloning from git every run. Option 3: Bake only certs + OS deps, leave Python dynamic A nice compromise: - image contains: - CA trust setup support - git/curl/build-essential/whatever system libs - uv - runtime only installs Python packages from 
pyproject.toml
/ 
EXTRA_PIP_PACKAGES
This reduces runtime fragility a lot. --- What I would recommend in order For production use, I’d rank the approaches: 1. Best: small custom image derived from 
prefecthq/prefect:3.7.0
2. Next best: public image + mounted ConfigMap/Secret certs + 
EXTRA_PIP_PACKAGES
3. Works but less ideal: public image + initContainer git clone + runtime 
apt-get
+ runtime 
uv pip install
Reason: every runtime install step increases: - startup time - network dependence - failure surface - non-reproducibility --- If you want to avoid a custom image entirely Then I’d suggest this refinement of your plan: - Do not clone certs from git on every run if avoidable - prefer ConfigMap/Secret - Do not dynamically discover dependencies from arbitrary 
pyproject.toml
unless necessary - prefer explicit package list, or a pinned lock file - use public 
prefecthq/prefect:3.7.0
- use a bootstrap shell wrapper in the job template - keep runtime apt installs minimal --- One more practical note If you install Python packages at runtime and some have native extensions, you may also need OS build deps like: - 
build-essential
- 
python3-dev
- 
pkg-config
- library headers such as 
libpq-dev
, 
gcc
, etc. That’s another reason a prebuilt image is usually preferable. --- If you want, send me your current Kubernetes base job template JSON and I’ll rewrite it into a concrete version that preserves Prefect’s exact required placeholders and injects: - cert bootstrap - apt packages - uv install - pyproject-based dependency install
Open in Slack
PreviousNext
Powered by