Log collectors

Starting points for picking the output up with common log collectors. They are opinionated on purpose; adapt them to your setup. All of them follow the same plan:

• Sort on kind. Every line the tool writes is JSON: "kind": "event" for the box's log on stdout, "kind": "log" for the tool's own log on stderr. This works even where the two streams arrive mixed.
• Take the time from the line. Events carry the time the box logged them in event_time, the tool's own lines theirs in time. Without this, a log stack files events under the time they were read, which after a restart can be days later.
• Few labels or indices. Loki gets the labels kind, device, group (events) and level (own log); Elasticsearch gets one index for events and one for the tool's log. Everything else stays in the JSON.

The Grafana dashboard expects the Loki labels of the Alloy example.

The files are in examples/ in the repository.

Collector	Source	Target	Checked
Grafana Alloy	Docker/Podman socket, journald, Kubernetes	Loki	Docker source end to end with Podman and Loki; all sources with alloy validate
Vector	Docker	Elasticsearch	Unit tests with vector test (included)
Fluentd	Docker fluentd log driver	Elasticsearch	--dry-run; parsing with sample records
Logstash	Docker gelf log driver	Elasticsearch	--config.test_and_exit; filter with sample records

None of the Elasticsearch examples was run against a real Elasticsearch.

Podman

Podman has no fluentd or gelf log driver. With Podman, read the container's log from journald (Podman's default) or through its Docker-compatible socket, as the Alloy and Vector examples do.

Grafana Alloy → Loki

Alloy loads every .alloy file of a directory (alloy run /etc/alloy/). Put pipeline.alloy there together with one of the source files, and set the Loki URL in pipeline.alloy.

pipeline.alloy

// fritz_logextractor: sort its lines into the box's events and the tool's own
// log, take each line's time from the line itself and send both to Loki.
// Combine this file with one source-*.alloy; Alloy loads every .alloy file of
// a directory: alloy run /etc/alloy/
loki.process "fritzlogextractor" {
    forward_to = [loki.write.default.receiver]

    stage.json {
        expressions = {
            kind       = "",
            device     = "",
            group      = "",
            level      = "",
            event_time = "",
            time       = "",
        }
    }

    // Few labels on purpose: kind ("event" or "log"), device and group for
    // events, level for the tool's own log. Everything else stays in the line;
    // query it with "| json".
    stage.labels {
        values = {
            kind   = "",
            device = "",
            group  = "",
            level  = "",
        }
    }

    // Events: when the box logged them, not when they were read.
    stage.match {
        selector = "{kind=\"event\"}"

        stage.timestamp {
            source = "event_time"
            format = "RFC3339"
        }
    }

    stage.match {
        selector = "{kind=\"log\"}"

        stage.timestamp {
            source = "time"
            format = "RFC3339"
        }
    }
}

loki.write "default" {
    endpoint {
        url = "http://loki:3100/loki/api/v1/push"
    }
}

Docker or Podman socketjournald (Podman, Quadlets)Kubernetes

source-docker.alloy

// The container named "fritzlogextractor", read from Docker or from Podman's
// Docker-compatible socket (mount it at /var/run/docker.sock).
discovery.docker "containers" {
    host = "unix:///var/run/docker.sock"
}

discovery.relabel "fritzlogextractor" {
    targets = discovery.docker.containers.targets

    rule {
        source_labels = ["__meta_docker_container_name"]
        regex         = "/fritzlogextractor"
        action        = "keep"
    }
}

loki.source.docker "fritzlogextractor" {
    host       = "unix:///var/run/docker.sock"
    targets    = discovery.relabel.fritzlogextractor.output
    labels     = {job = "fritzlogextractor"}
    forward_to = [loki.process.fritzlogextractor.receiver]
}

Mount the socket into Alloy's container, for example -v /var/run/docker.sock:/var/run/docker.sock:ro. With rootless Podman, mount $XDG_RUNTIME_DIR/podman/podman.sock there; on hosts with SELinux, also pass --security-opt label=disable.

source-journal.alloy

// The container named "fritzlogextractor" from the systemd journal, where
// Podman writes container logs by default (also for Quadlets). Alloy needs
// read access to /var/log/journal.
loki.relabel "fritzlogextractor" {
    forward_to = []

    rule {
        source_labels = ["__journal_container_name"]
        regex         = "fritzlogextractor"
        action        = "keep"
    }
}

loki.source.journal "fritzlogextractor" {
    path          = "/var/log/journal"
    relabel_rules = loki.relabel.fritzlogextractor.rules
    labels        = {job = "fritzlogextractor"}
    forward_to    = [loki.process.fritzlogextractor.receiver]
}

If Alloy already ships the whole journal, do not add a second loki.source.journal: the lines would arrive twice. Instead, send the lines of the container (__journal_container_name is fritzlogextractor) from your existing source through loki.process.fritzlogextractor.

source-kubernetes.alloy

// Pods labelled app=fritzlogextractor, read through the Kubernetes API.
discovery.kubernetes "pods" {
    role = "pod"
}

discovery.relabel "fritzlogextractor" {
    targets = discovery.kubernetes.pods.targets

    rule {
        source_labels = ["__meta_kubernetes_pod_label_app"]
        regex         = "fritzlogextractor"
        action        = "keep"
    }

    rule {
        source_labels = ["__meta_kubernetes_namespace"]
        target_label  = "namespace"
    }

    // As the other sources; Loki derives service_name="fritzlogextractor" from it,
    // which the Grafana dashboard looks for.
    rule {
        target_label = "job"
        replacement  = "fritzlogextractor"
    }
}

loki.source.kubernetes "fritzlogextractor" {
    targets    = discovery.relabel.fritzlogextractor.output
    forward_to = [loki.process.fritzlogextractor.receiver]
}

Fits the Kubernetes example (app: fritzlogextractor). Alloy needs RBAC to list pods and read their logs. Only checked with alloy validate.

To query in Grafana:

{job="fritzlogextractor", kind="event"} | json
{job="fritzlogextractor", kind="event", group="net"} | json | id = 22
{job="fritzlogextractor", kind="log", level=~"WARNING|ERROR"}

Old entries and Loki

On its first start, the tool prints the whole log of the box, which can reach back weeks. Loki rejects lines older than reject_old_samples_max_age (one week by default), so these are lost unless you raise that limit for the first run. Later lines arrive in order, apart from mesh entries a repeater hands over late; those are seconds or minutes behind, well within the hour Loki accepts out of order.

Vector → Elasticsearch

vector.yaml

# fritz_logextractor with Vector: sort its lines into the box's events and the
# tool's own log, take each line's time from the line itself and send them to
# two Elasticsearch indices.
sources:
  fritzlogextractor:
    type: docker_logs
    include_containers: [fritzlogextractor]

transforms:
  fritz_parsed:
    type: remap
    inputs: [fritzlogextractor]
    source: |
      parsed, err = parse_json(.message)
      if err != null || !is_object(parsed) {
        abort
      }
      . = object!(parsed)
      # Events: when the box logged them; the tool's own log: when it logged.
      # An event_time without UTC offset (box without time zone) falls back to now.
      stamp = if .kind == "event" { .event_time } else { .time }
      .timestamp = parse_timestamp(string(stamp) ?? "", "%+") ?? now()

  fritz_by_kind:
    type: route
    inputs: [fritz_parsed]
    reroute_unmatched: false
    route:
      event: .kind == "event"
      log: .kind == "log"

sinks:
  fritz_events:
    type: elasticsearch
    inputs: [fritz_by_kind.event]
    endpoints: ["http://elasticsearch:9200"]
    bulk:
      index: "fritz-events-%Y.%m.%d"

  fritz_tool_log:
    type: elasticsearch
    inputs: [fritz_by_kind.log]
    endpoints: ["http://elasticsearch:9200"]
    bulk:
      index: "fritzlogextractor-%Y.%m.%d"

The tests check the parsing and routing without Docker or Elasticsearch; run them after changing the transforms:

vector test vector.yaml vector-tests.yaml

vector-tests.yaml

# Unit tests for vector.yaml, without Docker or Elasticsearch:
#   vector test vector.yaml vector-tests.yaml
tests:
  - name: an event takes the box's time and goes to the event route
    inputs:
      - insert_at: fritz_parsed
        type: log
        log_fields:
          message: '{"kind": "event", "event_time": "2026-10-09T08:31:57+02:00", "device": "fritz.box", "group": "net", "id": 22, "msg": "Internetverbindung wurde erfolgreich hergestellt.", "repeat": 1}'
    outputs:
      - extract_from: fritz_by_kind.event
        conditions:
          - type: vrl
            source: |
              assert_eq!(.timestamp, t'2026-10-09T06:31:57Z')
              assert_eq!(.id, 22)
              assert!(!exists(.message))

  - name: an own log line takes its time and goes to the log route
    inputs:
      - insert_at: fritz_parsed
        type: log
        log_fields:
          message: '{"time": "2026-10-31T16:30:51.485+00:00", "kind": "log", "level": "WARNING", "logger": "fritzlogextractor.extractor", "msg": "Fritz!Box: could not read the event log (ConnectTimeout), will retry"}'
    outputs:
      - extract_from: fritz_by_kind.log
        conditions:
          - type: vrl
            source: |
              assert_eq!(.timestamp, t'2026-10-31T16:30:51.485Z')
              assert_eq!(.level, "WARNING")

  - name: a line that is not JSON is dropped
    inputs:
      - insert_at: fritz_parsed
        type: log
        log_fields:
          message: "Traceback (most recent call last):"
    no_outputs_from: [fritz_parsed]

Fluentd → Elasticsearch

Run the tool's container with Docker's fluentd log driver, tagged fritzlogextractor:

docker run -d --name fritzlogextractor \
  --log-driver fluentd --log-opt fluentd-address=fluentd:24224 --log-opt tag=fritzlogextractor \
  ...

fluent.conf

# fritz_logextractor with Fluentd: Docker's fluentd log driver sends each line,
# Fluentd parses it, takes the box's time for events and writes events and the
# tool's own log to two Elasticsearch indices.
#
# Container: --log-driver fluentd --log-opt fluentd-address=<fluentd>:24224
#            --log-opt tag=fritzlogextractor
# Plugin:    fluent-plugin-elasticsearch

<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

# The line arrives in "log". Events take their time from event_time; the tool's
# own log lines keep the time Docker read them, which is when they were logged.
<filter fritzlogextractor>
  @type parser
  key_name log
  reserve_data true
  remove_key_name_field true
  <parse>
    @type json
    time_key event_time
    keep_time_key true
    time_type mixed
    time_format %Y-%m-%dT%H:%M:%S%z
    # An event_time without UTC offset (box without time zone).
    time_format_fallbacks %Y-%m-%dT%H:%M:%S%z, %Y-%m-%dT%H:%M:%S
  </parse>
</filter>

# One index per kind and day: fritz-event-YYYY.MM.DD, fritz-log-YYYY.MM.DD.
<match fritzlogextractor>
  @type elasticsearch
  host elasticsearch
  port 9200
  index_name fritz-${kind}-%Y.%m.%d
  include_timestamp true
  <buffer tag, time, kind>
    timekey 1d
    flush_interval 10s
  </buffer>
</match>

Needs fluent-plugin-elasticsearch. The tool's own log lines keep the time Docker read them; they are logged as they happen, so that is the time they were written.

Logstash → Elasticsearch

Run the tool's container with Docker's gelf log driver:

docker run -d --name fritzlogextractor \
  --log-driver gelf --log-opt gelf-address=udp://logstash:12201 \
  ...

logstash.conf

# fritz_logextractor with Logstash: Docker's gelf log driver sends each line,
# Logstash parses it, takes each line's time from the line itself and writes
# events and the tool's own log to two Elasticsearch indices.
#
# Container: --log-driver gelf --log-opt gelf-address=udp://<logstash>:12201

input {
  gelf {
    port => 12201
  }
}

filter {
  if [container_name] == "fritzlogextractor" {
    # Under "fritz", so the line's fields do not clash with GELF's own
    # (GELF has a numeric "level", the tool's log a named one).
    json {
      source => "message"
      target => "fritz"
      remove_field => ["message"]
    }

    if [fritz][kind] == "event" {
      date {
        match => ["[fritz][event_time]", "ISO8601"]
      }
      mutate {
        add_field => { "[@metadata][index]" => "fritz-events" }
      }
    } else if [fritz][kind] == "log" {
      date {
        match => ["[fritz][time]", "ISO8601"]
      }
      mutate {
        add_field => { "[@metadata][index]" => "fritzlogextractor" }
      }
    }
  }
}

output {
  if [@metadata][index] {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      index => "%{[@metadata][index]}-%{+YYYY.MM.dd}"
    }
  }
}

Event times without offset

If the box does not tell its time zone, event_time has no UTC offset. The Fluentd and Logstash examples then read it as UTC (or the collector's time zone), Vector takes the time it read the line, and Alloy the time of the line before in the same stream (its default action_on_failure = "fudge"). See Output.