コンテンツにスキップ

Pipeline (Log Ingestion)

A Pipeline defines how external log data is ingested, stored, and made searchable in CSIRT-Pro. Create one Pipeline per log source type.

CSIRT-Pro stores the raw log line and extracts fields at search time rather than at ingestion. This schema-on-read approach keeps ingestion fast and lets the same data be reinterpreted when parse rules change. The parse rules described below are applied when a search runs, not when data is written.

Screen Layout

The Pipeline configuration screen has the following components.

  1. Parse settings for the regular-expression parsing rule.
  2. Field extraction for the field names assigned to extracted values.
  3. TTL settings for the data retention period.
  4. Data import for a manual bulk upload.
  5. AI parse generation for producing a parse rule from a sample log.

Creating a Pipeline

1. Create a New Pipeline

  1. Click Create New from the Pipeline list.
  2. Enter a pipeline name, which must be unique within your organization.
  3. Save.

Automatic table creation

When a pipeline is created, a backing table named {org_uuid}-{pipeline_name} is created automatically in the columnar log analytics store.

2. Configure Parse Rules

Configure the parse rule to match your log format.

Parse Settings Format

A pipeline holds parse configuration in two separate fields.

druid_parse (regex pattern)

A regular expression with numbered capture groups, written with standard parentheses. Each capture group extracts one field value.

druid_parse_field (field names)

A comma-separated string of field names. Each name maps to the corresponding numbered capture group, in order.

Important

Do not use Python named groups such as (?P<name>...) in druid_parse. Use standard numbered capture groups such as (\S+) instead. Field names are defined separately in druid_parse_field.

Example: Apache Access Log

Sample log:

192.168.1.100 - - [15/Jan/2025:10:30:00 +0900] "GET /api/users HTTP/1.1" 200 1234

druid_parse (regex with numbered groups):

(\S+) \S+ \S+ \[([^\]]+)\] "(\S+) (\S+) \S+" (\d+) (\d+)

druid_parse_field (comma-separated field names):

remote_host, timestamp, method, path, status, bytes

Mapping:

Capture Group Regex Pattern Field Name Extracted Value
Group 1 (\S+) remote_host 192.168.1.100
Group 2 ([^\]]+) timestamp 15/Jan/2025:10:30:00 +0900
Group 3 (\S+) method GET
Group 4 (\S+) path /api/users
Group 5 (\d+) status 200
Group 6 (\d+) bytes 1234

Example: Syslog

Sample log:

Jan 15 10:30:00 server01 sshd[12345]: Failed password for root from 10.0.0.50 port 22

druid_parse:

(\w+ \d+ \d+:\d+:\d+) (\S+) (\S+): (.+)

druid_parse_field:

timestamp, hostname, process, message

Example: Firewall Log

Sample log:

2025-01-15T10:30:00Z DENY 192.168.1.100 10.0.0.1 443 1024 TCP

druid_parse:

(\S+) (\S+) (\S+) (\S+) (\d+) (\d+) (\S+)

druid_parse_field:

timestamp, action, src_ip, dst_ip, port, bytes, protocol

JSON Logs

When the regex in druid_parse is left empty, the pipeline runs in JSON mode. The raw log line is treated as JSON, and fields are extracted by the names in druid_parse_field at search time, so no regex is needed.

{
  "timestamp": "2025-01-15T10:30:00Z",
  "src_ip": "192.168.1.100",
  "action": "DENY",
  "bytes": 1024
}

A field name can carry a type annotation in the form name(Type), so that values are extracted with the right type. This works in both regex mode and JSON mode. For example, dst_port(Int32) and bytes(Int64) let you compare numerically (dst_port == 22) and aggregate (sum bytes) without conversion functions. A name without an annotation is extracted as a string. Supported types: String (default), Int8/Int16/Int32/Int64, UInt8/UInt16/UInt32/UInt64, Float32/Float64, Bool, Date, DateTime. Extraction happens at search time (schema-on-read), so changing a type applies to existing logs on the next search. A value that does not match its type becomes null.

AI Parse Generation

Paste a sample log line and have generative AI suggest a parse rule.

Steps

  1. Click the AI Generate button.
  2. Paste a sample log line.
  3. Review the suggested regex pattern and field names.
  4. Click Apply to save the rule.

Generative AI analyzes the log structure and proposes both the druid_parse regex and the druid_parse_field names, with field names normalized to the ECS schema. The suggestion is an aid: the operator reviews and applies it, so generation does not change live ingestion on its own.

Data Retention (TTL)

Set how long data is retained for each pipeline.

Setting Description
Unlimited Data is stored permanently (TTL = 0)
1 to 3650 days Data is deleted automatically after the specified number of days

Data Ingestion Methods

Send data programmatically through the REST API.

curl -X POST "https://<domain>/api/v2/ingest?pipeline_id=<pipeline_id>" \
  -H "Content-Type: application/json" \
  -H "X-API-Key: <api_key>" \
  -d '{"timestamp": "2025-01-15T10:30:00Z", "src_ip": "10.0.0.1", "action": "DENY"}'

curl -X POST "https://<domain>/api/v2/ingest/batch" \
  -H "Content-Type: application/json" \
  -H "X-API-Key: <api_key>" \
  -d '{
    "pipeline_id": "<pipeline_id>",
    "records": [
      {"timestamp": "2025-01-15T10:30:00Z", "src_ip": "10.0.0.1"},
      {"timestamp": "2025-01-15T10:30:01Z", "src_ip": "10.0.0.2"}
    ]
  }'

curl -X POST "https://<domain>/api/user/<user_id>/bulk/<pipeline_name>/" \
  -H "Content-Type: application/json" \
  -H "X-API-Key: <api_key>" \
  -d '[{"log": "record1"}, {"log": "record2"}]'

UI Upload

Files can be uploaded directly from the Pipeline configuration screen.

Supported formats:

Format Description
JSON JSON array
NDJSON One JSON record per line
Plain Text Newline-delimited text logs

Data Flow

Log ingestion runs through the following stages.

  1. Receive. Log data arrives at the high-performance ingest endpoint over the REST API.
  2. Buffer. Records are placed on the message bus for asynchronous processing.
  3. Consume. An ingestion worker reads from the message bus and writes records to the columnar log analytics store.
  4. Store. The raw log line is retained, so fields can be extracted later according to the pipeline's parse rule.
  5. Search. Data becomes searchable on the Search screen, where the parse rule is applied to extract fields at query time.

Pipeline Sharing

A pipeline can be shared with another organization so that users there can access its data, subject to SharedAccess permissions.

POST /api/user/<user_id>/share/pipeline/

API Operations

Pipelines can also be managed through the REST API. See the API Reference for full endpoint documentation, including the following.

  • GET /api/user/<user_id>/pipeline/ lists pipelines.
  • POST /api/user/<user_id>/pipeline/ creates a pipeline.
  • PUT /api/user/<user_id>/pipeline/<pipeline_id>/ updates a pipeline.
  • POST /api/user/<user_id>/generate-parse/ runs AI parse generation.
  • POST /api/user/<user_id>/share/pipeline/ shares a pipeline.