Step 4: Multi-Format Detection

In a real-world system, you often receive logs in various formats through a single input. This final step teaches you how to build a unified parser that can intelligently detect the format of each message and route it to the correct parsing logic.

The Goal

You will build a single pipeline that parses JSON, five-field CSV, syslog, and Access Log strings.

The "Detect -> Route" Pattern

1. Detect: A mapping processor inspects the raw log string and uses simple heuristics (like checking for { or <) to guess the format, setting a metadata field like detected_format.
2. Route: A switch processor reads this metadata field and selects a Bloblang mapping using .parse_json(), .parse_csv(), or regular-expression captures.

Implementation

1. Create the Unified Parser: Copy the following configuration into a file named unified-parser.yaml. This pipeline combines the logic from the previous steps.

   Complete pipeline

   unified-parser.yaml

   name: unified-log-parser
   description: A pipeline that detects and parses multiple log formats.
   
   config:
     input:
       http_server:
         address: 0.0.0.0:8080
         path: /ingest
   
     pipeline:
       processors:
         # 1. Detect the format from the raw log field.
         - mapping: |
             root = this
             let content = this.raw_log.string()
             meta detected_format = if $content.has_prefix("{") {
               "json"
             } else if $content.has_prefix("<") {
               "syslog"
             } else if $content.split(",").length() >= 3 {
               "csv"
             } else if $content.re_match("\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}") {
               "access_log"
             } else {
               "unknown"
             }
   
         # 2. Route to the matching parser.
         - switch:
             - check: meta("detected_format") == "json"
               processors:
                 - mapping: root = this.raw_log.parse_json()
                 - mapping: |
                     root = this
                     root.parsed_by = "json_parser"
             - check: meta("detected_format") == "csv"
               processors:
                 - mapping: |
                     let header = "timestamp,metric_name,sensor_id,value,unit\n"
                     root = ($header + this.raw_log).parse_csv().index(0)
                     root.parsed_by = "csv_parser"
             - check: meta("detected_format") == "syslog"
               processors:
                 - mapping: |
                     let parts = this.raw_log.re_find_all_submatch("^<([0-9]+)>([A-Z][a-z]{2} +[0-9]+ [0-9:]+) ([^ ]+) ([^ :\\[]+)(?:\\[([0-9]+)\\])?: (.*)$").index(0)
                     root = {
                       "priority": $parts.index(1),
                       "timestamp": $parts.index(2),
                       "hostname": $parts.index(3),
                       "tag": $parts.index(4),
                       "pid": $parts.index(5),
                       "message": $parts.index(6),
                       "parsed_by": "syslog_parser"
                     }
             - check: meta("detected_format") == "access_log"
               processors:
                 - mapping: |
                     let parts = this.raw_log.re_find_all_submatch("^(\\S+) \\S+ \\S+ \\[([^]]+)\\] \"([^\"]+)\" ([0-9]+) ([0-9]+|-)$").index(0)
                     root = {
                       "ip": $parts.index(1),
                       "timestamp": $parts.index(2),
                       "request": $parts.index(3),
                       "status": $parts.index(4).number(),
                       "bytes": if $parts.index(5) == "-" { null } else { $parts.index(5).number() },
                       "parsed_by": "access_log_parser"
                     }
             - processors:
                 - mapping: |
                     root = this
                     root.parsed_by = "unknown"
   
     output:
       stdout:
         codec: lines

2. Deploy and Test:

   # --- Send logs in different formats ---
   curl -X POST http://localhost:8080/ingest -d '{"raw_log": "{\"level\":\"info\"}"}'
   
   curl -X POST http://localhost:8080/ingest \
    -d '{"raw_log": "2025-10-20T14:23:45Z,temperature,temp-sensor-01,35.5,celsius"}'
   curl -X POST http://localhost:8080/ingest -d '{"raw_log": "127.0.0.1 - - [01/Jan/2025:12:00:00 +0000] \"GET / HTTP/1.1\" 200 123"}'
   

3. Verify: Check your logs. You will see three structured messages. Each one will have a parsed_by field indicating that it was correctly routed to and processed by the right set of processors.

You have now built a parser for the three configured log formats.