Anish S
03/26/2026, 6:30 PMMarvin
03/26/2026, 6:30 PMMarvin
03/26/2026, 6:33 PMfrom prefect.artifacts import create_link_artifact
create_link_artifact(
key="run-logs",
link="<https://my-bucket.s3.amazonaws.com/logs/flow-run-123.log>",
link_text="Download full logs"
)
- Create a short log summary as markdown:
from prefect.artifacts import create_markdown_artifact
create_markdown_artifact(
key="run-summary",
markdown="# Run Summary\n\n- Errors: 0\n- Warnings: 3\n- Duration: 12m",
description="Concise summary of this run"
)
- View logs via CLI (preferred for raw logs):
prefect flow-run logs <FLOW_RUN_ID>
# or save to file:
prefect flow-run logs <FLOW_RUN_ID> > flow.log
More on logging: Add loggingAnish S
03/26/2026, 7:31 PMMarvin
03/26/2026, 7:31 PMMarvin
03/26/2026, 7:32 PMimport boto3
from prefect.artifacts import create_link_artifact
# R2 is S3-compatible; use your account’s endpoint and keys
endpoint = "https://<account_id>.<http://r2.cloudflarestorage.com|r2.cloudflarestorage.com>"
region = "auto" # R2 accepts 'auto' or 'us-east-1' for signing
bucket = "my-temp-logs"
key = "runs/123/flow.log"
s3 = boto3.client(
"s3",
endpoint_url=endpoint,
aws_access_key_id="<R2_ACCESS_KEY_ID>",
aws_secret_access_key="<R2_SECRET_ACCESS_KEY>",
region_name=region,
)
# Upload the log file
with open("flow.log", "rb") as f:
s3.upload_fileobj(f, bucket, key, ExtraArgs={"ContentType": "text/plain"})
# Create a short-lived presigned URL
url = s3.generate_presigned_url(
"get_object",
Params={"Bucket": bucket, "Key": key},
ExpiresIn=3600, # 1 hour
)
# Surface it in Prefect UI as a Link artifact
create_link_artifact(
key="full-logs",
link=url,
link_text="Download full logs (expires in 1h)",
description="Temporary link to full run logs",
)
If you share a bit about:
- expected log size (KB/MB/GB),
- how long you need to keep them,
- whether logs are sensitive,
…I can recommend the best fit and a minimal setup (including lifecycle policies and example IaC).Anish S
03/26/2026, 7:36 PMMarvin
03/26/2026, 7:36 PMMarvin
03/26/2026, 7:38 PMimport gzip
import io
import boto3
from prefect.artifacts import create_link_artifact
# Configure your R2 S3-compatible client
endpoint = "https://<account_id>.<http://r2.cloudflarestorage.com|r2.cloudflarestorage.com>"
region = "auto"
bucket = "temp-logs"
key = "runs/123/flow.log.gz"
s3 = boto3.client(
"s3",
endpoint_url=endpoint,
aws_access_key_id="<R2_ACCESS_KEY_ID>",
aws_secret_access_key="<R2_SECRET_ACCESS_KEY>",
region_name=region,
)
# One-time: ensure a 30-day delete lifecycle rule exists on the bucket
# (run this once per bucket setup)
s3.put_bucket_lifecycle_configuration(
Bucket=bucket,
LifecycleConfiguration={
"Rules": [
{
"ID": "delete-after-30-days",
"Status": "Enabled",
"Filter": {"Prefix": ""},
"Expiration": {"Days": 30},
}
]
},
)
# Gzip and upload
log_text = open("flow.log", "rb").read()
buf = io.BytesIO()
with gzip.GzipFile(fileobj=buf, mode="wb") as gz:
gz.write(log_text)
buf.seek(0)
s3.upload_fileobj(
buf,
bucket,
key,
ExtraArgs={
"ContentType": "text/plain",
"ContentEncoding": "gzip",
"CacheControl": "max-age=60", # optional
},
)
# Create a short-lived presigned URL (1 hour)
url = s3.generate_presigned_url(
"get_object",
Params={"Bucket": bucket, "Key": key},
ExpiresIn=3600,
)
# Surface in Prefect as a Link artifact
create_link_artifact(
key="full-logs",
link=url,
link_text="Download full logs (expires in 1h)",
description="Temporary log storage (auto-deletes in 30 days)",
)
Backblaze B2 notes
- Use endpoint like https://s3.us-west-004.backblazeb2.com with your region.
- Same boto3 flow works (S3-compatible).
- Lifecycle rules can be configured via B2 UI or via the S3 lifecycle API.
Why this works well
- Free/low-cost for your size and retention.
- Automatic cleanup after 30 days.
- Prefect UI shows a clean link via an artifact without storing large blobs in the orchestration DB.
If you want, I can provide a minimal Terraform snippet for either R2 or B2 to create the bucket + 30-day lifecycle in one step.Anish S
03/26/2026, 8:51 PM{
"cpu": 1024,
"image": "xyz.dkr.ecr.us-east-1.amazonaws.com/image:tag",
"memory": 2048,
"cluster": "prefect-worker",
"mount_point": {
"readOnly": false,
"sourceVolume": "efs-volume",
"containerPath": "/data/"
},
"execution_role_arn": "arn:aws:iam::123:role/prefect-worker-task-role",
...
"cloudwatch_logs_prefix": "workflow-log",
"cloudwatch_logs_options": {
"awslogs-group": "prefect-worker",
"awslogs-create-group": "true"
}
}
Does prefect fetch logs from cloudwatch? Or does it store the logs in the db?
@MarvinMarvin
03/26/2026, 8:51 PMMarvin
03/26/2026, 8:53 PMPREFECT_LOGGING_TO_API_ENABLED=true
PREFECT_LOGGING_LEVEL=INFO
# Optional tuning
PREFECT_LOGGING_TO_API_BATCH_INTERVAL=1.0
PREFECT_LOGGING_TO_API_MAX_LOG_SIZE=1000000
- Capture prints as logs:
@flow(log_prints=True)
def my_flow(): ...
- View logs via CLI (from the Prefect DB):
prefect flow-run logs <FLOW_RUN_ID>
If you want CloudWatch-only or both
- CloudWatch only: set PREFECT_LOGGING_TO_API_ENABLED=false in the task env (logs won’t appear in Prefect UI).
- Both (typical): keep Prefect API logging enabled and keep the awslogs driver for infrastructure observability in CloudWatch.
Docs
- Add logging: Add logging
- Artifacts (if you want to link CloudWatch or external logs into the Prefect UI): Artifacts how-to