Enrich

Synopsis

Enriches documents by performing SQL queries against lookup tables loaded from CSV files. It can then join the data with externally referenced information to add context and semantic depth.

Schema

- enrich:
    lookup_table: <string[]>
    dataset_name: <string[]>
    query: <string>
    type: <string>
    description: <text>
    if: <script>
    tag: <string>
    on_success: <processor[]>
    on_failure: <processor[]>
    override: <boolean>
    ignore_missing: <boolean>
    ignore_failure: <boolean>

Configuration

The following fields are used to define the processor:

Field	Required	Default	Description
`lookup_table`	N	-	Array of paths to CSV files containing lookup data
`dataset_name`	N	-	Not implemented. The names are accepted and then ignored — no dataset is loaded and no error is raised. Use `lookup_table` instead
`query`	Y	-	SQL or KQL query to execute for enrichment
`type`	N	-	Query language type (`kusto` for KQL, empty for SQL in SQLite's dialect)
`description`	N	-	Explanatory note
`if`	N	-	Condition to run
`ignore_failure`	N	`false`	Continue processing if enrichment fails
`ignore_missing`	N	`false`	Skip if fields don't exist
`on_failure`	N	-	Processors to run on failure
`on_success`	N	-	Processors to run on success
`override`	N	`false`	If true, replaces existing document with query results
`tag`	N	-	Identifier for logging
`disabled`	N	`false`	When `true`, the processor is skipped and the event continues to the next one. Lets you take a processor out of the path without removing its configuration

Details

Information gathering from external sources involves

loading lookup data from CSV files, which may involve using SQL queries, including SQLite and Kusto Query Language (KQL),
extracting and manipulating JSON data, overriding fields, and
resolving conflicts.

warning

KQL queries are automatically converted to SQLite syntax, and some complex KQL features might not be supported.

Queries run with SQLite's semantics, so a query written for SQLite gives the same result here. SQLite has no boolean type: a boolean in a query's result, such as a comparison, is written to the document as the integer 1 or 0.

The processor provides the following special tables and fields for queries:

`_ingest`	Contains the current document's data
`_message`	JSON string of the entire document
`lookup_table`	Contains data loaded from CSV files

CSV files must have headers in the first row as the column names become field names in queries. As in SQLite, a query matches column names without regard to case. An empty value is an empty string, not NULL; a row with fewer values than the header has NULL in the columns it leaves out. When lookup_table names several files, their rows form one table: the first file's header names its columns, and every later file's header must name some of those columns.

warning

As large CSV files are loaded directly into memory, the impact of using very large lookup tables must be carefully considered.

Reloading Lookup Tables

The processor loads the files lookup_table names when the first record reaches it. An edited file is picked up without restarting the pipeline:

As records arrive, and at most once every 5 seconds, the processor checks each file's modification time and size. When either has changed, it loads the files again. A processor that receives no records checks nothing.
When a reload fails, for example because a file is briefly missing or cannot be read as CSV, the table loaded last keeps answering queries, and the next check loads the files again.
A file read while it is being rewritten can also load only in part. Its modification time or size changes again when the write ends, so the next check loads it whole. To avoid that window, write the new file under another name and rename it over the old one.
When the first load fails, there is no table to fall back on. Each record fails as on any other processor error: ignore_failure and on_failure apply. The first record that arrives 5 seconds or more after the failure tries the load again.
Enrich processors that name the same files in the same order share one loaded copy, across every pipeline the same Director or Agent runs. The memory warning above therefore applies once per distinct list of files, not once per processor.
Each enrich processor queries the files its own lookup_table names, also when one pipeline contains several. In earlier releases, every enrich processor in a pipeline queried the table the first one had loaded.

Running on Agents

The processor runs on Agents as well as on the Director, in every pipeline an Agent runs, preprocessing pipelines included.

Lookup tables. lookup_table paths are read on the host that runs the pipeline: on an Agent, the Agent's own host. The Director does not send these files. A table kept in the Library reaches an Agent through the Lookup processor instead.
SQL queries (type empty) run on an Agent as they do on the Director.
KQL queries (type: kusto) run as SQL that the Director translates and sends to the Agent with its configuration, since the Agent carries no KQL converter. Each record fails with the director sent no SQL for this KQL enrich query when
- the Agent's Director predates KQL enrich queries on Agents, or
- the Director cannot convert the query. The Director then logs an error naming the device and the converter's reason.

Examples

CSV Lookup

Enriching logs with data from a CSV table...

{
  "event_code": "1234"
}

- enrich:
    lookup_table: ["event_codes.csv"]
    query: |
        SELECT 
          _ingest.*,
          lookup_table.description as event_description
        FROM _ingest
        LEFT JOIN lookup_table
        ON _ingest.event_code = lookup_table.code

adds description:

{
  "event_code": "1234",
  "event_description": "System startup"
}

KQL Queries

Processing logs using Kusto Query Language...

{
  "DeviceVendor": "Check Point",
  "DeviceProduct": "VPN-1 & FireWall-1",
  "SourceIP": "192.168.1.100"
}

- enrich:
    type: "kusto"
    query: |
        _ingest
        | where DeviceVendor == "Check Point"
        | extend isInternal = SourceIP startswith "192.168."
        | project DeviceVendor, DeviceProduct, SourceIP, isInternal

adds computed fields:

{
  "DeviceVendor": "Check Point",
  "DeviceProduct": "VPN-1 & FireWall-1",
  "SourceIP": "192.168.1.100",
  "isInternal": 1
}

Multiple CSV Files

Joining user and department rows from two CSV files. Both files have id and name columns, and their rows form one lookup_table, so u and d are two aliases of that same table...

{
  "user_id": "U123",
  "department_id": "D456"
}

- enrich:
    lookup_table: 
      - "users.csv"
      - "departments.csv"
    query: |
        SELECT 
          _ingest.*,
          u.name as user_name,
          d.name as department_name
        FROM _ingest
        LEFT JOIN lookup_table u ON _ingest.user_id = u.id
        LEFT JOIN lookup_table d ON _ingest.department_id = d.id

combines the matching user and department rows:

{
  "user_id": "U123",
  "department_id": "D456",
  "user_name": "John Doe",
  "department_name": "IT Support"
}

Override Mode

Replacing the document with query results...

{
  "raw_data": "lots of fields",
  "event_type": "security"
}

- enrich:
    override: true
    query: |
        SELECT event_type, 
              'processed' as status
        FROM _ingest
        WHERE event_type = 'security'

keeps only the selected fields:

{
  "event_type": "security",
  "status": "processed"
}

Synopsis​

Schema​

Configuration​

Details​

Reloading Lookup Tables​

Running on Agents​

Examples​

CSV Lookup​

KQL Queries​

Multiple CSV Files​

Override Mode​