Skip to main content

Basic Log Processing Pipeline

This example demonstrates a simple log processing pipeline that reads from a file, parses JSON logs, and outputs to stdout. It's a great starting point for understanding how Expanso pipelines work.

Download & Run​

Quick Start:

# 1. Download the pipeline and check it offline
curl -sSLo pipeline.yaml \
  https://docs.expanso.io/examples/basic-pipeline.yaml
expanso-edge validate pipeline.yaml

# 2. Start a local edge agent (its state stays in ./.edge)
expanso-edge run --local --data-dir ./.edge

# 3. In a second terminal, deploy the pipeline to it as a job
export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli"
{ printf 'name: basic-pipeline\ntype: pipeline\nconfig:\n'
  sed 's/^/  /' pipeline.yaml; } |
  expanso-cli --endpoint http://localhost:9010 job deploy -

This runs on your machine with no account. Step 3 wraps the pipeline in the name, type and config fields that a job needs, and a stdout output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See Local Mode for more, or Deploy to Cloud to run it on managed nodes.

Download: basic-pipeline.yaml

What This Pipeline Does​

  1. Reads log lines from a file
  2. Parses each line as JSON
  3. Extracts the timestamp field
  4. Outputs the processed logs to stdout

Complete Pipeline​

input:
  file:
    paths: [/var/log/app.log]
    codec: lines

pipeline:
  processors:
    - mapping: |
        # Parse the log line as JSON
        root = this.parse_json()

    - mapping: |
        # Parse timestamp field if it exists
        root.timestamp = this.timestamp.parse_timestamp("2006-01-02T15:04:05Z")

output:
  stdout:
    codec: lines

Configuration Breakdown​

Input: File​

input:
  file:
    paths: [/var/log/app.log]
    codec: lines

The file input reads from /var/log/app.log, processing each line individually. The lines codec tells Expanso to treat each line as a separate message.

See: File Input Documentation

Processors: Parse & Transform​

pipeline:
  processors:
    - mapping: |
        # Parse the log line as JSON
        root = this.parse_json()

    - mapping: |
        # Parse timestamp field if it exists
        root.timestamp = this.timestamp.parse_timestamp("2006-01-02T15:04:05Z")

The pipeline uses two mapping processors (which use the Bloblang language):

  1. JSON Parsing: Converts each log line from a string to a JSON object
  2. Timestamp Parsing: Converts the timestamp string to a proper timestamp type

See: Mapping Processor | Bloblang Guide

Output: Stdout​

output:
  stdout:
    codec: lines

Outputs each processed message to stdout in JSON format. Useful for testing and debugging.

See: Stdout Output Documentation

Example Input/Output​

Input Log Line:

{"timestamp":"2024-01-15T10:30:00Z","level":"INFO","message":"User login successful","user_id":"12345"}

Output:

{
  "timestamp": "2024-01-15T10:30:00Z",
  "level": "INFO",
  "message": "User login successful",
  "user_id": "12345"
}

Common Variations​

Add Filtering​

Filter out debug-level logs:

pipeline:
  processors:
    - mapping: |
        root = this.parse_json()

        # Drop debug logs
        root = if this.level == "DEBUG" { deleted() }

Output to File​

Change the output to write to a file instead of stdout:

output:
  file:
    path: /var/log/processed-${! timestamp_unix() }.json
    codec: lines

Add Timestamps​

Add a processing timestamp:

pipeline:
  processors:
    - mapping: |
        root = this.parse_json()
        root.processed_at = now()

Next Steps​