Skip to content

DBT: S3 Access Log Events

DBT model name: s3_access_log_events

Explore dependencies/lineage: link

Description

Parsed S3 server access-log events for trase-storage, append-only with a rolling retention window (var s3_monitor_access_log_retention_days). Excluded prefixes are dropped at ingest.

Details

Column Type Description
request_time `` Request timestamp (tz-aware, UTC).
operation `` S3 operation, e.g. REST.GET.OBJECT, REST.PUT.OBJECT.
key `` Object key the request targeted (null for bucket-level ops).
bytes_sent `` Bytes returned to the client.
source_log_key `` Access-log object the event was parsed from (incremental watermark).

Review full report including sample errors records if they exist (link)

Test column Test name Failing rows Last test run Last status Query in Metabase
request_time dbt_expectations_expect_row_values_to_have_recent_data_s3_access_log_events_request_time__day__2 — 2026-07-25 04:06 pass

Models

Models

  • s3_monitor_excluded_prefixes

No called script or script source not found.

"""Parsed S3 server access-log events for the `trase-storage` bucket.

Incremental, append-only. Each run lists access-log objects newer than the
highest `source_log_key` already loaded (filenames are date-prefixed, so lexical
order is chronological), parses them, and appends the events. A post-hook prunes
events older than `s3_monitor_access_log_retention_days` to bound table size.

This feed makes usage stats possible (top prefixes/keys by request count and
bytes sent) and surfaces object creates/deletes that happen between the weekly
inventory snapshots.

First run loads only the retention window (not the full log history). Each run is
capped at `s3_monitor_access_log_max_files` objects; any remainder is picked up
on the next run.

NOTE: the access-log bucket prefix has a 60-day S3 expiration lifecycle, so
`s3_monitor_access_log_retention_days` is kept under 60 (default 55) — otherwise
the first run's cutoff reaches into already-expired objects (NoSuchKey). Long-term
history is retained in `s3_downloads_daily`, not here.

Paths matching the `s3_monitor_excluded_prefixes` seed are dropped at ingest.
"""

import fnmatch

import pandas as pd

from trase.tools.progress import ProgressPrinter

OUT_COLS = [
    "bucket_owner",
    "bucket",
    "request_time",
    "remote_ip",
    "requester",
    "request_id",
    "operation",
    "key",
    "request_uri",
    "http_status",
    "error_code",
    "bytes_sent",
    "object_size",
    "total_time",
    "turn_around_time",
    "referrer",
    "user_agent",
    "source_log_key",
]
_INT_COLS = [
    "http_status",
    "bytes_sent",
    "object_size",
    "total_time",
    "turn_around_time",
]


def _exclude_mask(keys: pd.Series, patterns) -> pd.Series:
    """Vectorised fnmatch: True where a key matches any exclusion glob.
    Rows with a null key (bucket-level operations) are never excluded."""
    patterns = [p for p in patterns if isinstance(p, str) and p]
    if not patterns:
        return pd.Series(False, index=keys.index)
    regex = "|".join(f"(?:{fnmatch.translate(p)})" for p in patterns)
    return keys.fillna("").str.match(regex)


def model(dbt, session):
    dbt.config(
        materialized="incremental",
        incremental_strategy="append",
        tags=["s3_access_logs"],
    )
    # Tunables come from the model's +meta in dbt_project.yml — see note in
    # s3_file_schemas.
    meta = dbt.config.get("meta", {})
    retention_days = int(meta.get("access_log_retention_days", 55))
    max_files = int(meta.get("access_log_max_files", 5000))

    from trase.config import aws_session
    from trase.tools.aws.s3_access_logs import fetch_and_parse_access_logs
    from trase.tools.aws.s3_helpers import ACCESS_LOG_BUCKET, list_access_log_keys

    after_key = None
    if dbt.is_incremental:
        # Partition the DuckLake table by event month so time-range Metabase
        # queries and the retention delete prune partitions. Idempotent; only
        # affects new writes; non-fatal on engines without DuckLake partitioning.
        try:
            session.execute(
                f"ALTER TABLE {dbt.this} "
                "SET PARTITIONED BY (year(request_time), month(request_time))"
            )
        except Exception:  # noqa: BLE001
            pass

        # Apply the rolling retention window before appending the new batch.
        session.execute(
            f"DELETE FROM {dbt.this} "
            f"WHERE request_time < now() - INTERVAL {retention_days} DAY"
        )
        row = session.execute(f"SELECT max(source_log_key) FROM {dbt.this}").fetchone()
        after_key = row[0] if row else None

    if after_key is None:
        # First load (or empty table): only ingest the retention window.
        from trase.tools.aws.s3_helpers import ACCESS_LOG_PREFIX

        cutoff = (pd.Timestamp.utcnow() - pd.Timedelta(days=retention_days)).strftime(
            "%Y-%m-%d"
        )
        after_key = f"{ACCESS_LOG_PREFIX}{cutoff}"

    keys = list_access_log_keys(after_key=after_key)[:max_files]
    if not keys:
        return pd.DataFrame(columns=OUT_COLS)

    # These are many tiny, network-bound GETs, so read them in parallel — a big
    # speedup for backfills (hundreds of thousands of log objects). boto3 clients
    # are thread-safe for low-level calls.
    from concurrent.futures import ThreadPoolExecutor

    from botocore.config import Config

    workers = int(meta.get("access_log_workers", 16))
    interval = float(meta.get("progress_interval_seconds", 60))
    # Size the connection pool to the worker count so threads don't queue.
    client = aws_session().client(
        "s3", config=Config(max_pool_connections=max(workers, 10))
    )

    def _fetch_one(key):
        recs = fetch_and_parse_access_logs([key], ACCESS_LOG_BUCKET, client)
        for rec in recs:
            rec["source_log_key"] = key
        return recs

    records = []
    progress = ProgressPrinter(
        len(keys), "s3_access_log_events: reading logs", interval=interval
    )
    with ThreadPoolExecutor(max_workers=workers) as pool:
        for recs in pool.map(_fetch_one, keys):
            records.extend(recs)
            progress.update()
    progress.done()

    if not records:
        return pd.DataFrame(columns=OUT_COLS)

    events = pd.DataFrame(records)
    # Ensure all expected columns exist even if a batch lacked some fields.
    for col in OUT_COLS:
        if col not in events.columns:
            events[col] = None

    # Keep `dbt.ref(...).df()` on its own line — dbt's Python ref parser fails to
    # register the ref if a subscript (`[...]`) is chained directly onto it.
    excluded = dbt.ref("s3_monitor_excluded_prefixes").df()
    exclusion_patterns = excluded["prefix_glob"].to_list()
    events = events[~_exclude_mask(events["key"], exclusion_patterns)].copy()

    events["request_time"] = pd.to_datetime(
        events["request_time"], errors="coerce", utc=True
    )
    for col in _INT_COLS:
        events[col] = pd.to_numeric(events[col], errors="coerce").astype("Int64")

    return events[OUT_COLS]