Skip to main content

Log Reduction

Parse an access log, drop health-check lines, keep every error (status 400 and above) as its own structured event, and collapse the remaining successful requests into one count per minute, route and status.

Components: broker input batching a file input (lines scanner) → mapping (regex parse, health-check drop) → switch processor: errors get a mapping, successes go through group_by_value → archive → mapping → file output.

Needs: a log file on the node, in combined log format with the request time in seconds appended as the last field. A line from the run: 10.0.0.0 - - [17/Sep/2026:12:00:00 +0000] "GET /healthz HTTP/1.1" 200 100 "-" "synthetic-agent/1.0" 0.000

Proved on a local-mode node, with synthetic data

This job ran end to end on an expanso-edge v2.1.21 node in local mode, not through Expanso Cloud, against a generated 5,000-line log.

Complete job​

log-reduction.yaml
name: log-reduction
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: logs
config:
  input:
    broker:
      inputs:
        - file:
            paths: ["/var/log/app/access.log"]
            scanner:
              lines: {}
      batching:
        count: 100000
        period: 2s
  pipeline:
    processors:
      - mapping: |
          let m = content().string().re_find_object(
            "^(?P<ip>\\S+) \\S+ \\S+ " +
            "\\[(?P<ts>[^\\]]+)\\] " +
            "\"(?P<method>\\S+) (?P<path>\\S+) " +
            "[^\"]*\" (?P<status>\\d{3}) (?P<bytes>\\d+) " +
            "\"[^\"]*\" \"[^\"]*\" (?P<latency>[0-9.]+)$"
          )
          root = if $m.path == "/healthz" { deleted() } else {
            {
              "minute": $m.ts.slice(0, 17),
              "method": $m.method,
              "path": $m.path,
              "status": $m.status.number(),
              "latency_ms": ($m.latency.number() * 1000).round(),
              "ip": $m.ip
            }
          }
      - switch:
          - check: this.status >= 400
            processors:
              - mapping: 'root = this.merge({"kind": "error"})'
          - processors:
              - group_by_value:
                  value: '${! json("minute") } ${! json("path") }
                    ${! json("status") }'
              - archive:
                  format: json_array
              - mapping: |
                  root = {
                    "kind": "rollup",
                    "minute": this.index(0).minute,
                    "path": this.index(0).path,
                    "status": this.index(0).status,
                    "count": this.length()
                  }
  output:
    file:
      path: "/var/tmp/expanso-logs/out/reduced.jsonl"
      codec: lines

The selector sends the job to a node labelled pipeline_role: logs; set that label on your node, or change the selector to a label it already carries. With no matching node, the job is stored and never runs.

Replace /var/log/app/access.log with your log. Both paths are on the node that runs the job, and the user running the agent must be able to read the log and write the output. Create the output directory there first:

mkdir -p /var/tmp/expanso-logs/out

If your log format differs, change the regular expression in the first mapping; the named groups become the fields.

Deploy and check it​

expanso-cli job deploy log-reduction.yaml
wc -l < /var/log/app/access.log
wc -l < /var/tmp/expanso-logs/out/reduced.jsonl

Run the last two commands on the node once the job has completed. The output holds one event per error and one rollup per minute, route and status. Two lines from the run, pretty-printed:

{
  "ip": "10.0.4.95",
  "kind": "error",
  "latency_ms": 95,
  "method": "GET",
  "minute": "17/Sep/2026:12:00",
  "path": "/api/users",
  "status": 404
}
{
  "count": 40,
  "kind": "rollup",
  "minute": "17/Sep/2026:12:00",
  "path": "/api/orders",
  "status": 200
}

What the run proved​

  • 5,000 synthetic lines went in and 330 events came out. The 3,000 health-check lines were dropped.
  • All 250 lines with status 400 or above were kept, each as its own event.
  • The 80 rollups (one per minute, route and status) matched an independent recount from the raw log, and every non-health line was accounted for.
  • The file shrank from 546,200 to 39,725 bytes, 92.7% smaller. That figure belongs to this synthetic log, in which 60% of lines were health checks. It is not a benchmark; your reduction depends on your traffic.
  • The job exactly as published, restart_policy: never included, ran on a local-mode node running expanso-edge v2.1.21 and completed after 1 execution.

Limits​

  • Proved on a local-mode node only, not through Expanso Cloud.
  • With restart_policy: never, a failed run stops after one execution instead of being re-run. See Proving your own run.
  • The file is read in batches of up to 100,000 lines or 2 seconds, and rollups are counted per batch. The run's 5,000-line file fit in one batch.
  • Successful requests are reduced to counts: individual IPs, latencies and timestamps for those requests are not kept.
  • The output is a file on the node. Replace the file output to ship the reduced events elsewhere.