Skip to main content

Object Store Setup for dfe-fetcher

What a cloud administrator needs to configure so dfe-fetcher can read log objects (gzipped JSON-lines and similar) that cloud services drop into a storage bucket.

Overview​

Many cloud services do not expose an API for their logs - they only deliver them as objects into a bucket. The object_store source family polls bucket prefixes for new objects, decodes them (gzip aware), and emits one record per line. Typical feeds:

  • AWS CloudTrail -> S3
  • AWS VPC Flow Logs -> S3
  • AWS WAF logs -> S3
  • GCP Cloud Logging -> GCS storage sink (backend not implemented)
  • Azure diagnostic settings -> Blob container (backend not implemented)

The source is a family of backends selected per entry by a provider discriminator:

  • s3 - FULLY IMPLEMENTED and tested live. SigV4-signed ListObjectsV2 + GetObject. Also works against S3-compatible stores (MinIO, Cloudflare R2, Backblaze B2) via endpoint_override.
  • gcs - NOT IMPLEMENTED. The config parses, but the backend is skipped at load with a warning naming it. NOT USABLE.
  • azure_blob - NOT IMPLEMENTED. Same status as gcs. NOT USABLE.

The S3 backend runs on the shipped object_store profile (crates/fetcher/profiles/object_store.yaml); the sources.object_store block below maps onto one instance of it at load, one unit per prefix. Each object is read exactly once: the highest last_modified a tick delivered is the prefix's checkpoint, committed after the batch is acknowledged, and the next listing starts there (a first tick starts a day back). A refused listing (401 or 403) fails that prefix's tick and the other prefixes still run; an object that keeps failing is retried per the profile's policy and then fails the prefix's tick, so the checkpoint never advances past it.

Supported object formats: json_gz, jsonl, json, text_gz, text. Per tick, each (bucket, prefix) is capped at 1000 objects and 10 list pages; the remainder waits for the next tick. One S3 backend per block: a second bucket account runs as a sources.rest instance of the object_store profile with its own key pair.

dfe-fetcher only reads. It never writes, modifies, deletes, or lifecycles objects, buckets, or containers.

Prerequisites​

For the live S3 backend:

  • An S3 bucket (or S3-compatible store) that already receives log objects from the producing service. dfe-fetcher does not provision delivery - see Source-Side Setup for links on configuring each producer.
  • An IAM user with read-only access to the bucket and prefixes you intend to poll.
  • Static AWS access keys for that user (or a vault secret holding them). The S3 backend authenticates with SigV4 static credentials, resolved once per instance.
  • The bucket region (used for SigV4 signing and endpoint construction).

For GCS / Azure Blob: nothing to do. Those backends cannot fetch data. The permission notes below are documented so the eventual rollout is unblocked, but leave those backends out of your config.

Required Permissions​

S3 (live)​

Minimal read-only access. s3:ListBucket is a bucket-level action and targets the bare bucket ARN; s3:GetObject is an object-level action and targets the /* object ARN. Pointing both at /* is the most common mistake - ListBucket then silently fails.

ActionResource ARNWhy
s3:ListBucketarn:aws:s3:::my-log-bucketList object keys + last-modified under each prefix
s3:ListBucketMultipartUploadsarn:aws:s3:::my-log-bucketHold the listing behind uploads still in progress (see Multipart Uploads)
s3:GetObjectarn:aws:s3:::my-log-bucket/*Download each object body

All actions are read-only. No write, delete, or bucket-config actions are needed. For cross-account delivery buckets, the bucket policy on the target bucket must also allow the fetcher principal.

Multipart Uploads​

S3 lists an object written by multipart upload under the time the upload BEGAN, not when it completed. A large object can land behind newer ones the fetcher has already read, and a checkpoint alone would skip it. So before each listing the fetcher calls ListMultipartUploads on the prefix and reads nothing modified at or after the start of the oldest upload still in progress.

Without s3:ListBucketMultipartUploads (a 403), or on a store that does not support the call, the fetcher logs one warning per prefix naming the permission and falls back to settle_secs: only objects older than that many seconds are read. The default is 300, one default fetch interval. An upload that takes longer than settle_secs can still be skipped on that path, so grant the permission or raise settle_secs past your longest upload.

An upload that is never completed or aborted holds its prefix until it goes, and the fetcher warns once it has held for a day. Add an AbortIncompleteMultipartUpload lifecycle rule to the bucket so abandoned uploads are removed.

MinIO lists a completed multipart object under its completion time, so this case does not arise there, and the hold only delays objects written while an upload runs.

GCS (not implemented)​

When the GCS backend ships, the service account needs roles/storage.objectViewer (read plus list) on the bucket or project. roles/storage.legacyObjectReader is the read-without-list variant if listing is not wanted. Do not configure a gcs backend.

Azure Blob (not implemented)​

When the Azure Blob backend ships, the principal needs the data-plane role Storage Blob Data Reader on the storage account or container. Note that Azure control-plane roles (Owner, Contributor, Reader) do NOT grant blob data access on their own - a data-plane role is required, and reading via Entra credentials without it returns 403 AuthorizationPermissionMismatch. Role assignments can take up to 10 minutes to propagate. Do not configure an azure_blob backend.

Source-Side Setup​

dfe-fetcher reads whatever a producing service has already delivered. You must configure the producer to deliver into the bucket, then point a prefix at it.

  1. Confirm (or enable) log delivery to the bucket. Common producers:

  2. Note the exact key prefix each producer writes under. The prefix is what dfe-fetcher polls; getting it wrong yields zero records.

  3. Note the object format. CloudTrail, VPC Flow Logs (parquet excluded), and most native AWS feeds are gzipped JSON-lines -> use json_gz.

  4. Create the read-only IAM identity (see S3 manual setup below) and confirm it can list and read the prefix.

S3 manual IAM setup​

  1. Create the IAM user:

    aws iam create-user --user-name dfe-fetcher-s3-reader

  2. Attach the least-privilege read-only policy (save as policy.json):

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "ListLogBucket",
          "Effect": "Allow",
          "Action": ["s3:ListBucket", "s3:ListBucketMultipartUploads"],
          "Resource": ["arn:aws:s3:::my-log-bucket"]
        },
        {
          "Sid": "GetLogObjects",
          "Effect": "Allow",
          "Action": ["s3:GetObject"],
          "Resource": ["arn:aws:s3:::my-log-bucket/*"]
        }
      ]
    }
    aws iam put-user-policy \
      --user-name dfe-fetcher-s3-reader \
      --policy-name dfe-fetcher-s3-readonly \
      --policy-document file://policy.json

    To restrict to a single prefix, narrow the GetObject resource to arn:aws:s3:::my-log-bucket/AWSLogs/123456789012/CloudTrail/* and add an s3:prefix condition to the ListBucket statement.

  3. Generate an access key:

    aws iam create-access-key --user-name dfe-fetcher-s3-reader

dfe-fetcher Configuration​

Config File​

S3 backend (live). Each backend lists buckets; each bucket lists prefixes with a format and a source_tag. Records are tagged object_store.<source_tag>.

sources:
  object_store:
    enabled: true
    topic: "object_store"
    # filter: '_dfe_fetcher_object.bucket == "my-log-bucket"'  # hot-reloaded
    backends:
      - provider: s3
        region: "ap-southeast-2"
        access_key_id: "AKIA..."
        secret_access_key: "wJalrXUtnFEMI..."
        # endpoint_override: "http://localhost:9000"  # MinIO / R2 / B2
        buckets:
          - bucket: "my-log-bucket"
            prefixes:
              - prefix: "AWSLogs/123456789012/CloudTrail/ap-southeast-2/"
                format: json_gz
                source_tag: "aws_cloudtrail"
                # topic: "cloudtrail"   # optional per-prefix topic override
              - prefix: "AWSLogs/123456789012/vpcflowlogs/"
                format: json_gz
                source_tag: "aws_vpc_flow"

Field reference (S3 backend):

  • provider - s3 (live), gcs / azure_blob (not implemented).
  • region - AWS region for SigV4 signing and endpoint construction.
  • access_key_id, secret_access_key - static credentials. Each may be an inline value, an env:VAR spec, or a vault: spec.
  • endpoint_override - optional; for MinIO, R2, B2, or VPC endpoints.
  • credential_secret - optional; vault spec holding both keys as JSON (see Secrets Manager below). Wins over the inline key fields.
  • settle_secs - optional, default 300. How old an object must be before it is read, used only where ListMultipartUploads is refused (see Multipart Uploads).
  • buckets[].bucket - bucket name (container name for Azure Blob).
  • buckets[].prefixes[].prefix - key prefix to poll (may be empty to scan the whole bucket).
  • buckets[].prefixes[].format - one of json_gz, jsonl, json, text_gz, text.
  • buckets[].prefixes[].source_tag - record source tag (object_store.<source_tag>). A tag that names a format (json_gz, jsonl, json, text, text_gz) or repeats is refused at load.
  • buckets[].prefixes[].topic - optional per-prefix topic override.

Prefixes run in tag order, not config order. The example config also documents gcs and azure_blob backend shapes, but neither is implemented - if configured, the fetcher warns at load and skips them, and the S3 backend of the same block still runs. Do not enable them.

Environment Variables​

Top-level source fields follow the DFE_FETCHER_SOURCES__OBJECT_STORE__<FIELD> pattern (double underscores):

DFE_FETCHER_SOURCES__OBJECT_STORE__ENABLED="true"
DFE_FETCHER_SOURCES__OBJECT_STORE__TOPIC="object_store"

The backends list (with its nested buckets and prefixes) is awkward to express purely as environment variables. Define backends in the config file and inject only the credentials via env. The credential fields accept an env: indirection so the secret value itself stays in the process environment, not in the YAML:

backends:
  - provider: s3
    region: "ap-southeast-2"
    access_key_id: "env:AWS_ACCESS_KEY_ID"
    secret_access_key: "env:AWS_SECRET_ACCESS_KEY"
AWS_ACCESS_KEY_ID="AKIA..."
AWS_SECRET_ACCESS_KEY="wJalrXUtnFEMI..."

Secrets Manager​

Keep static keys out of the config file with credential_secret. It resolves to a JSON object containing both keys and takes precedence over the inline access_key_id / secret_access_key fields:

sources:
  object_store:
    enabled: true
    topic: "object_store"
    backends:
      - provider: s3
        region: "ap-southeast-2"
        credential_secret: "vault:kv/data/dfe/aws-s3-reader:credentials"
        buckets:
          - bucket: "my-log-bucket"
            prefixes:
              - prefix: "AWSLogs/123456789012/CloudTrail/ap-southeast-2/"
                format: json_gz
                source_tag: "aws_cloudtrail"

The vault secret should contain JSON:

{
  "access_key_id": "AKIA...",
  "secret_access_key": "wJalrXUtnFEMI..."
}

The keys AccessKeyId / SecretAccessKey (AWS casing) are also accepted.

Verification​

  1. Confirm the IAM identity can list and read the prefix directly:

    aws s3 ls s3://my-log-bucket/AWSLogs/123456789012/CloudTrail/ \
      --profile dfe-fetcher-s3-reader

    A non-empty listing confirms s3:ListBucket. Then download one object to confirm s3:GetObject:

    aws s3 cp s3://my-log-bucket/<one-key-from-above> /tmp/probe.gz \
      --profile dfe-fetcher-s3-reader

  2. Start dfe-fetcher with the object_store source enabled and watch the logs. A working prefix logs unit tick complete with its rows, filtered, oversize and flushes counts on every tick; a refused listing logs unit tick failed with the API's text.

  3. The health check is the key pair resolving; there is no probe request. The unimplemented backends never participate in it.

  4. If you (incorrectly) configure a gcs or azure_blob backend, the fetcher warns at load that the backend is not implemented and skips it - remove that backend.

  5. Emitted records carry a _dfe_fetcher_object envelope with provider, bucket, key, last_modified, and size, so you can filter and route downstream.

Cost​

dfe-fetcher only reads objects; it never creates trails, flow logs, sinks, or diagnostic settings. The object LIST/GET calls it makes may generate usage charges (API requests and any cross-region or egress data transfer), so there could be a cost depending on bucket traffic and where the fetcher runs - polling is bounded per tick, and tuning the fetch interval limits request volume. The producing service's own logging/storage is billed separately by the cloud provider. Confirm any cost implications against your own cloud agreement.

References​