Support custom log storage implementations
Describe the current behavior
Currently, flow and task run logs are stored in the Prefect database.
For deployments with large log volumes, it can be useful to store logs in a dedicated backend such as VictoriaLogs, Loki, Elasticsearch/OpenSearch, or ClickHouse while preserving the existing Prefect API and UI behavior.
This is related to #11348, which proposes making log consumption configurable.
Describe the proposed behavior
Introduce a server-side LogStorage abstraction for historical log persistence, querying, and deletion, while keeping live log streaming separate.
For example:
class LogStorage(ABC):
async def write_logs(
self,
logs: Sequence[Log],
) -> None:
...
async def read_logs(
self,
log_filter: LogFilter | None,
offset: int,
limit: int,
sort: LogSort,
) -> Sequence[Log]:
...
async def delete_logs(
self,
log_filter: LogFilter,
) -> None:
...The configured storage would be used by the existing log paths such as:
- POST /logs/
- POST /logs/filter
- flow-run log downloads
- flow-run log cleanup
- task-run log cleanup
Without custom configuration, Prefect would continue using the existing database-backed storage.
Example Use
A self-hosted Prefect user could install a custom storage implementation:
FROM prefect:<version>
RUN uv pip install --system prefect-victorialogsand configure it through an environment variable:
environment:
PREFECT_SERVER_LOGS_STORAGE: prefect_victorialogs.storageThe Prefect UI and clients would continue using the existing log API, while historical logs would be stored and queried through the configured backend.
Additional context
I have a small VictoriaLogs-backed proof of concept demonstrating the basic approach:
https://github.com/cakeisacake/prefect-log-storage-poc
The goal is to keep log storage pluggable without changing the existing Prefect log API or UI behavior.
Source: PrefectHQ/prefect