Pipeline File Format and Step Reference

A pipeline is a YAML file with one or more entries. Each entry names the data types it handles, its place in the running order and a list of steps that turn the raw log into a standard event. This page describes the file format, how the engine runs pipelines, and every step you can use.

File layout

This is the complete shipped pipeline for the generic data type:

pipeline:
  - dataTypes: [generic]
    order: 14
    steps:
      - grok:
          patterns:
            - fieldName: log.message
              pattern: '(.*)'
          source: raw
      - json:
          source: log.message
          where: 'startsWith("log.message", "{")'

When you save a pipeline from the product, UTMStack also writes a tenantId into each entry so that it runs only for your tenant. You do not need to write it yourself.

Pipeline Examples has complete pipelines for JSON, key=value, free text, CSV, CEF and LEEF logs.

From raw log to event

1. A log arrives with its data type, data source, tenant, time and the original text in raw.

2. The engine makes a working copy of the log. Every step reads this copy and writes its results back into it, so a step sees what the steps before it wrote.

3. The matching pipeline entries run one after the other (see below), each one's steps from top to bottom.

4. A drop step whose condition is true ends the work: the log is discarded.

5. After the last step, the working copy is turned into an event in the Standard Event Schema and passed on to storage and the alerting rules.

Which entries run, and in what order

1. The engine takes every entry, in every pipeline file, whose dataTypes include the log's data type. One log can go through several entries.

2. It skips pipelines that are switched off for the log's tenant and custom pipelines of other tenants.

3. It runs first the pipelines your tenant placed on the Pipelines page, in that sequence, then the rest by their order value.

The shipped pipelines use the orders 0 to 37, each once. A custom pipeline saved without an order, or with 0, gets 100, so it runs after the shipped ones. Only the syslog data type has more than one shipped entry: generic syslog (order 31) runs first, then the CEF (35) and LEEF (36) stages.

Several entries for one data type

Because every matching entry runs, you can split the work into stages, in one file or across several. Each stage sees the fields the earlier stages wrote:

pipeline:
  - dataTypes: [my-app]
    order: 100
    steps:
      - grok:
          source: raw
          patterns:
            - fieldName: log.message
              pattern: '{{.greedy}}'
  - dataTypes: [my-app]
    order: 101
    steps:
      - json:
          source: log.message
          where: 'startsWith("log.message", "{")'

Conditions on steps (where)

Every step accepts a where condition, written in the same language as rule conditions. See Condition Functions Reference.

• The condition is checked against the log as it stands at that step, so it can read fields that earlier steps wrote.

• When the condition is false, the step is skipped.

• Field names inside a condition are always quoted: exists("log.message"), not exists(log.message). An unquoted name either fails to compile or silently never matches.

• In YAML, wrap a condition in single quotes, or start it on its own line after |, when it begins with ! or contains : . Without the quotes, YAML reads a leading ! as a tag of its own and the file does not mean what you wrote.

• Inside a condition, text can be in double or single quotes: equals("log.result", "OK") and equals('log.result', 'OK') are the same. A condition in double quotes in YAML therefore uses single quotes inside.

- drop:
    where: 'startsWith("raw", "[") || contains("raw", "RunExecutor") || contains("raw", "ETL pool")'

When a step fails

• A step that fails (for example a json step whose source is missing) adds its message to the event's errors field. The next steps still run.

• A where condition that cannot be compiled skips the step and adds an error to the event.

• Saving a pipeline does not check conditions or regular expressions, so a broken one fails on every log of that data type. Test before you save, and look at the errors field in Log Explorer after you save.

• After the last step, the engine builds the event. Unknown field names are dropped and a value of the wrong type drops the whole event. See Standard Event Schema.

Building a pipeline

A well-structured pipeline has four phases.

Phase 1: extraction

Pull the data out of the raw text with json, kv, csv or grok. If the log is JSON, use json: it is simpler and faster than patterns.

- json:
    source: raw
    where: 'startsWith("raw", "{")'

Phase 2: normalization

Move the extracted values to the standard fields.

- rename:
    from:
      - log.HOSTNAME
      - log._HOSTNAME
    to: origin.host
    where: exists("log.HOSTNAME") || exists("log._HOSTNAME")

Phase 3: enrichment and conversion

Add context and fix types and formats.

- dynamic:
    plugin: com.utmstack.geolocation
    params:
      source: origin.ip
      destination: origin.geolocation
    where: exists("origin.ip")
- cast:
    fields:
      - origin.port
    to: int
    where: exists("origin.port")

Phase 4: cleanup

Remove helper fields you no longer need.

- delete:
    fields:
      - log.jsonMessage
      - log.eventType

Step reference

The product accepts 12 step types in the pipelines you save. The "Required fields" column shows what saving checks.

json

Parses the JSON text in source and writes each top-level key as log.<key>. Nested objects stay objects, so you can reach their values with longer paths such as log.status.errorCode.

If source does not exist, the step fails with "source was not found". Guard it with a condition:

- json:
    source: raw
    where: 'startsWith("raw", "{")'

kv

Splits text into key=value pairs and writes each as log.<key>. Values are trimmed, and double quotes around a value are removed. A value in double quotes can contain the pair separator (for example msg="login failed"); a value without quotes ends at the next separator, so msg=login failed gives log.msg the value login. A piece without the value separator is ignored. To split on tab characters, write fieldSplit: "\t" in double quotes.

- kv:
    source: log.cefExtension
    fieldSplit: ' '
    valueSplit: '='
    where: exists("log.cefExtension")

csv

Splits text on a separator and writes the parts to the field names in headers, in order. The names are used exactly as written, so give full paths. Values in double quotes may contain the separator.

If a line has fewer columns than there are headers, the step writes the columns it has and then fails with "column index out of range, number of headers should match the number of resulting columns" in the event's errors field. Columns beyond the last header are ignored.

- csv:
    source: log.csvMsg
    separator: ","
    headers:
      - log.ruleNumber
      - log.subRuleNumber
      - log.anchor
      - log.tracker

grok

Cuts unstructured text into fields with a sequence of regular expressions.

How it works:

1. The patterns are applied one after the other.

2. Each pattern must match at the start of the text that is left (leading spaces are ignored). The matched text is trimmed and stored in fieldName, and the rest of the text goes to the next pattern.

3. The fields are written only if every pattern matched. If one fails, matches nothing, or the text runs out first, the step writes nothing.

4. fieldName is used exactly as written, so give full paths such as origin.ip or log.priority.

5. If source does not exist, the step does nothing.

Patterns can use the built-in named patterns, written {{.name}} (see Named patterns below), mixed with normal regular expression syntax.

- grok:
    patterns:
      - fieldName: log.priority
        pattern: '\<{{.integer}}\>'
      - fieldName: log.deviceTime
        pattern: '{{.monthName}}{{.space}}{{.monthDay}}{{.space}}{{.time}}{{.space}}'
      - fieldName: log.syslogHost
        pattern: '{{.hostname}}{{.space}}'
      - fieldName: log.msgAll
        pattern: '{{.greedy}}'
    source: raw

For the line <134>Nov 11 09:30:00 fw01 sshd[812]: session opened, this writes <134> to log.priority, Nov 11 09:30:00 to log.deviceTime, fw01 to log.syslogHost and the rest of the line to log.msgAll.

rename

Moves a value to another field.

Each source in from that exists is copied to to and then removed. If several exist, the last one in the list wins. Rename single values, not objects or lists: an object or list is stored as JSON text under the new name, so paths below it stop working.

- rename:
    from:
      - log.HOSTNAME
      - log._HOSTNAME
    to: origin.host
    where: exists("log.HOSTNAME") || exists("log._HOSTNAME")

cast

Changes the type of values.

Cast ports and counters to int before the event is built: a value in a number field that is not a plain number (for example -) drops the event.

- cast:
    fields:
      - origin.port
    to: int
    where: exists("origin.port")

trim

Removes part of a text value. prefix removes the text from the start of the value, suffix from the end, substring everywhere in the value, and regex removes every part that matches the regular expression. Spaces around the result are removed.

This pair removes the brackets that the grok example above leaves around log.priority, so <134> becomes 134:

- trim:
    function: prefix
    substring: "<"
    fields:
      - log.priority
- trim:
    function: suffix
    substring: ">"
    fields:
      - log.priority

reformat

Rewrites a time value in another format. It overwrites the value in place.

A Go time layout writes the reference moment Mon Jan 2 15:04:05 MST 2006 the way your value is written. For example, 02-Jan-2006 15:04:05 reads 11-Nov-2026 09:30:00, and 2006-01-02T15:04:05Z07:00 writes RFC 3339.

Always convert a vendor time before you put it in deviceTime. A value the database cannot read as a date loses the event.

- reformat:
    fields:
      - log.esetOccurredTime
    function: time
    fromFormat: 02-Jan-2006 15:04:05
    toFormat: '2006-01-02T15:04:05Z'
    where: exists("log.esetOccurredTime")
- rename:
    from:
      - log.esetOccurredTime
    to: deviceTime
    where: exists("log.esetOccurredTime")

delete

Removes fields.

Fields that do not exist are skipped. raw and @timestamp cannot be deleted: a delete step that names them fails with "field cannot be a reserved field" in the event's errors field, and the value stays.

- delete:
    fields:
      - log.restData
      - log.restMessage

add

Writes a fixed text value into a field.

- add:
    function: string
    params:
      key: actionResult
      value: failed
    where: equals("log.eventType","Audit_Event") && regexMatch("log.result", "(?i)^(failure|failed|fail|error|invalid|timeout)$")

dynamic

Calls an enrichment plugin. The plugin shipped for pipelines is com.utmstack.geolocation, which looks up an IP address and writes its location. See Geolocation Enrichment.

- dynamic:
    plugin: com.utmstack.geolocation
    params:
      source: target.ip
      destination: target.geolocation
    where: exists("target.ip")

drop

Discards the log when its condition is true. Nothing is stored and no rule sees it. The remaining steps, and any later pipelines for the same data type, do not run.

- drop:
    where: 'startsWith("raw", "[") || contains("raw", "RunExecutor") || contains("raw", "ETL pool")'

Logs discarded by a drop step are not counted in the drop figures of Data Processing; those count logs the engine discarded because it was overloaded.

Named patterns

The grok step can use 26 built-in named patterns. They are read-only: UTMStack restores the built-in set every time it starts, and you cannot add or change patterns. In the pipeline editor's Code tab, Insert regex pattern inserts one.

The exact regular expressions:

ipv4: (((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\.)){3}((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))
ipv6: ([0-9a-fA-F]{1,4}(:[0-9a-fA-F]{0,4}){1,7}|::[0-1]?)
hostname: (\b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b))
domain: ((?:[_a-z0-9](?:[_a-z0-9-]{0,61}[a-z0-9])?\.)+(?:[a-z](?:[a-z0-9-]{0,61}[a-z0-9])?)?)
email: ((?P<name>[a-zA-Z0-9.!#$%&'*+/=?^_`{|}~-]+)@(?P<domain>[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*))
uuid: ([A-Fa-f0-9]{8}-(?:[A-Fa-f0-9]{4}-){3}[A-Fa-f0-9]{12})
integer: (?:[+-]?(?:[0-9]+))
word: \b\w+\b
greedy: .*
data: (.*?)
space: \s+
notSpace: \S+
commonMacAddr: (?:(?:[A-Fa-f0-9]{2}:){5}[A-Fa-f0-9]{2})|(?:(?:[A-Fa-f0-9]{2}-){5}[A-Fa-f0-9]{2})
winMacAddr: (?:(?:[A-Fa-f0-9]{2}-){5}[A-Fa-f0-9]{2})
ciscoMacAddr: (?:(?:[A-Fa-f0-9]{4}\.){2}[A-Fa-f0-9]{4})
syslogDate: '[A-Z][a-z]{2} \d{1,2} \d{2}:\d{2}:\d{2}'
time: ((([01][0-9])|2[0-4]):(?:[0-5][0-9])(?::(?:(?:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?)))
hour: (([01][0-9])|2[0-4])
minute: (?:[0-5][0-9])
seconds: (?:(?:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?)
iso8601Timezone: (Z|([+-](([01][0-9])|2[0-4]):?([0-5][0-9])))
year: (([1-9])[0-9]{1,3})
monthName: \b(?:[Jj]an(?:uary|uar)?|[Ff]eb(?:ruary|ruar)?|[Mm](?:a|ä)?r(?:ch|z)?|[Aa]pr(?:il)?|[Mm]a(?:y|i)?|[Jj]un(?:e|i)?|[Jj]ul(?:y|i)?|[Aa]ug(?:ust)?|[Ss]ep(?:tember)?|[Oo](?:c|k)?t(?:ober)?|[Nn]ov(?:ember)?|[Dd]e(?:c|z)(?:ember)?)\b
monthNumber: (?:0[1-9]|1[0-2])
monthDay: (?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9])
day: (?:Mon(?:day)?|Tue(?:sday)?|Wed(?:nesday)?|Thu(?:rsday)?|Fri(?:day)?|Sat(?:urday)?|Sun(?:day)?)

Annotated example

This custom pipeline handles a firewall connected through a custom integration named "Acme Firewall", so its events have the data type acme-firewall. See Custom Integrations. The firewall sends syslog lines such as:

<134>Nov 11 09:30:00 fw01 acmefw: action=block src=203.0.113.7 dst=10.0.0.5 dport=443 policy=deny-ext

pipeline:
  - dataTypes: [acme-firewall]
    order: 100
    steps:
      # 1. Split the line: the syslog header up to "acmefw:", then the key=value part
      - grok:
          source: raw
          patterns:
            - fieldName: log.header
              pattern: '.*acmefw:'
            - fieldName: log.kvData
              pattern: '{{.greedy}}'

      # 2. Extract the pairs; they are written as log.action, log.src, log.dst, ...
      - kv:
          source: log.kvData
          fieldSplit: " "
          valueSplit: "="
          where: 'exists("log.kvData")'

      # 3. Map to the standard schema
      - rename:
          from: [log.src]
          to: origin.ip
      - rename:
          from: [log.dst]
          to: target.ip
      - rename:
          from: [log.dport]
          to: target.port

      # 4. Ports are whole numbers in the schema
      - cast:
          fields: [target.port]
          to: int

      # 5. Record the outcome with an exact standard value
      - add:
          function: string
          params:
            key: actionResult
            value: denied
          where: 'equals("log.action", "block")'
      - add:
          function: string
          params:
            key: actionResult
            value: success
          where: 'equals("log.action", "accept")'

      # 6. Add the location of the source address
      - dynamic:
          plugin: com.utmstack.geolocation
          params:
            source: origin.ip
            destination: origin.geolocation
          where: 'exists("origin.ip")'

      # 7. Remove the helper fields
      - delete:
          fields: [log.header, log.kvData]

For the sample line, the event gets origin.ip 203.0.113.7, target.ip 10.0.0.5, target.port 443 and actionResult denied, and keeps log.action and log.policy.

Test it on Configure > Pipelines before you save it. See Testing Pipelines and Rules. Writing and Testing Rules and Pipelines shows a rule for these events.