Implementing Filters

The Parsing Lifecycle

1. Ingestion: A log enters the system (via Input Plugin) with basic metadata: dataSource, dataType, and tenantId.

2. Draft Creation: The engine creates a Draft The original log resides within the object. raw field.

3. Pipeline Matching: The engine iterates through the pipeline configuration.

   • Stages are evaluated in order.

• A stage executes if the log's dataType is included in the stage's dataTypes array.

• Multiple Matches: A log can match and run through multiple stages if they all contain its dataType.

1. Step ExecutionIn a stage, steps execute in sequence. Every step changes the Draft's internal JSON string.

2. Finalization: Once all matching stages finish, the final Draft is converted into a structured Event and then forwarded to the Analysis stage.

Pipeline Architecture

A well-designed pipeline follows these four phases:

phase 1: extraction

Use json, csv, kv, or grok to pull data out of the raw string.

- json:
    source: raw

phase 2: normalization

Map extracted fields to the Standard Event Schema.

- rename:
    from: [log.source_ip, log.src]
    to: origin.ip

phase 3: enrichment & logic

Add context or transform data using dynamic plugins or reformat.

- dynamic:
    plugin: com.synoti.geolocation
    params: { source: origin.ip, destination: origin.geolocation }
- cast:
    fields: [origin.port]
    to: int

phase 4: cleanup

Delete transient fields to enhance storage and indexing.

- delete:
    fields: [log.temp_id, log.unnecessary_meta]

Technical Details

Conditional Execution (where)

Every step can be made conditional using CEL expressions. When the expression yields a result false, the step is bypassed. For the complete list of available comparison functions, see the CEL Overloads Guide.

Example: Only run a grok when a specific field is present.

- grok:
    source: log.message
    patterns: [...]
    where: 'exists("log.message")'

Stopping Processing (drop)

If you encounter "noise" (logs that shouldn't be stored or analyzed), use the drop step. It immediately halts the pipeline and drops the log.

- drop:
    where: 'contains(raw, "HealthCheck")'

Annotated Full Example

This example handles a firewall log received as a Key-Value string.

pipeline:
  - dataTypes: [firewall-fortigate-traffic]
    steps:
      # 1. Extract KV pairs (automatically goes into log.*)
      - kv:
          source: raw
          fieldSplit: " "
          valueSplit: "="

      # 2. Normalize to standard schema
      - rename:
          from: [log.src, log.s_ip]
          to: origin.ip
      - rename:
          from: [log.dst, log.d_ip]
          to: target.ip

      # 3. Handle data types for better indexing
      - cast:
          fields: [origin.port, target.port]
          to: int

      # 4. Logical enrichment: Determine action
      - add:
          function: string
          params: { key: action, value: denied }
          where: 'equals("log.policy", "block")'

      # 5. Cleanup temporary metadata
      - delete:
          fields: [log.policy]