Log Pipeline

The Log Pipeline feature (Observability > Log Pipeline) provides a managed Vector observability pipeline. You pick log sources and configure sinks in the UI; the backend generates vector.toml on the shared data volume and the Vector container picks it up via --watch-config. No manual Vector configuration or container restarts are needed.

How It Works

  HAProxy (JSON log line)
       │
       │  syslog → :601
       ▼
  Vector container (vector.toml on shared volume)
       │
       ├── source: corex  (TCP syslog :601)
       │     │
       │     ▼
       │   haproxy_finalize transform
       │   (JSON parse, JA4 decode, req_fp decode,
       │    unique_id decode, @timestamp, _doc_id)
       │     │
       │     ├──→ sink: aws_s3 (per-source)
       │     ├──→ sink: datadog_logs (per-source)
       │     ├──→ sink: elasticsearch (per-source)
       │     └──→ sink: ... (one per enabled sink with source=corex)
       │
       ├── source: waf  (file tail coraza-spoa.log)
       │     │
       │     ▼
       │   waf_parse_json transform
       │     │
       │     └──→ sinks with source=waf
       │
       └── source: mcp  (file tail mcp/events.ndjson)
             │
             ▼
           mcp_parse_json transform
             │
             └──→ sinks with source=mcp

The backend uses a 1-to-1 source-per-sink model: each sink has exactly one source (corex, waf, or mcp). Sources are auto-enabled when at least one enabled sink references them — there are no separate source toggles.

Log Sources

VRL Transforms

Before events reach sinks, Vector applies VRL (Vector Remap Language) transforms that parse and enrich the raw log lines. These transforms are built-in and run automatically — you don’t configure them.

corex (haproxy_finalize)

The HAProxy JSON log line arrives in the .message field via syslog. The transform:

1. Parses JSON — parse_json(.message) and merges into the event. Non-JSON lines (HAProxy internal messages like SSL handshake errors) are parsed with a regex into structured fields and tagged haproxy_internal = true.
2. Null-coerces dash fields — HAProxy sends - for numeric fields when no value is available (e.g. WAF fields when no WAF rules matched, timing fields for error responses). The transform converts these to null so structured sinks don’t reject them.
3. Decodes JA4 — Splits the JA4 fingerprint into ja4_a, ja4_b, ja4_c and extracts ja4_proto, ja4_version, ja4_sni, ja4_cipher_count, ja4_ext_count, ja4_alpn.
4. Decodes req_fp — Splits the request fingerprint into its subfields (req_fp_param_keys, req_fp_param_types, req_fp_path_depth, req_fp_hdr_count, etc.).
5. Decodes unique_id — Splits HAProxy’s unique-id into unique_id_timestamp, unique_id_pid, unique_id_request_counter, unique_id_client_port.
6. Flattens WAF match — Extracts waf_rule_id, waf_action, waf_anomaly_score from the WAF match fields.
7. Sets @timestamp — Parses HAProxy’s ts field (format %d/%b/%Y:%H:%M:%S%.3f) into an ISO 8601 timestamp. Falls back to now() if parsing fails.
8. Sets _doc_id — Uses unique_id as the document ID (for Elasticsearch deduplication); generates a UUID if missing.
9. Tags corex_source — Adds a corex_source field identifying the originating source.

waf (waf_parse_json)

Parses the Coraza SPOA JSON log line, extracts rule ID, severity, message, client IP, URI, and action.

mcp (mcp_parse_json)

Parses the MCP gateway newline-delimited JSON events.

Sinks

Each sink forwards events from one source to an external destination. You can create multiple sinks per source (e.g. send corex logs to both S3 and Datadog). You can also create the same sink type for different sources (e.g. one aws_s3 sink for corex and another for waf).

AWS S3

Forwards logs to an S3 bucket (or S3-compatible storage like MinIO, Cloudflare R2, Wasabi).

Azure Log Analytics

Forwards logs to Azure Monitor via the Data Collection Rules (DCR) ingestion endpoint.

Datadog Logs

Forwards logs to Datadog’s log intake.

Elasticsearch

Forwards logs to Elasticsearch or OpenSearch.

HTTP

Forwards logs to a generic HTTP endpoint via POST/PUT/PATCH.

New Relic

Forwards logs to New Relic’s log API.

Splunk HEC

Forwards logs to Splunk via the HTTP Event Collector.

Secret Handling

Fields marked as secret in the tables above are Fernet-encrypted at rest in the database. The encryption key is derived from VECTOR_SECRETS_KEY (falling back to SECRET_KEY).

• On read — secret fields are masked as ******** in API responses, config previews, diffs, and check-sink output
• On write — the backend re-encrypts the value; sending ******** means “keep the existing secret” (useful when editing a sink without changing credentials)
• In backups — secret values are redacted unless the backup is taken with “include secrets” enabled
• At render time — secrets are decrypted and inlined into the generated vector.toml (Vector needs the plaintext to connect to the sink)

Testing and Validation

Validate Config

The Validate button runs vector validate inside the Vector container against the currently-applied vector.toml. This checks TOML syntax and sink configuration without sending data. Returns the validation output (success or error messages).

Test Sink

The Test button (in the sink editor) renders a staging TOML for a candidate sink and runs vector validate inside the Vector container. This exercises real sink healthchecks — it attempts actual connectivity and authentication against the target (S3, Elasticsearch, Datadog, etc.).

With Send test event enabled, it also wires a demo_logs source and runs timeout 15 vector … for end-to-end delivery — a test event is sent through the pipeline to the target sink. The output is secret-redacted.

Returns 503 if the Vector container isn’t reachable.

Preview Config

The Preview button shows the generated vector.toml with secrets redacted (********). Useful for reviewing the full pipeline configuration before applying.

Restart Vector

The Restart button restarts the Vector container (via Docker SDK exec_run or Kubernetes exec). Use this if Vector is in a bad state or after manually editing the config.

Apply Flow

1. You create or edit a sink in the UI
2. Click Save — the sink is written to the vector_sinks database table (secrets encrypted)
3. The unapplied-changes banner appears (config diff compares redacted text so credential changes are detected without leaking secrets)
4. Click Apply Changes — the backend:
   • Generates vector.toml from all enabled sinks (secrets decrypted and inlined)
   • Writes it to the shared data volume
   • Writes a .applied snapshot for diff detection
   • Vector picks up the new config via --watch-config (no restart needed)

Empty Pipeline

When no sinks are configured, the backend emits a no-op config:

[sources.internal_metrics]
type = "internal_metrics"

[sinks.blackhole]
type = "blackhole"
inputs = ["internal_metrics"]

This keeps Vector running healthily without forwarding anything.

Step-by-Step: Send coreX Logs to Datadog

1. Navigate to Observability > Log Pipeline
2. Click Add Sink
3. Name: datadog-corex
4. Type: datadog_logs
5. Source: corex
6. API key: your Datadog API key
7. Site: select your Datadog site (e.g. datadoghq.com for US1)
8. Click Test to verify connectivity (optional: enable Send test event for end-to-end delivery)
9. Click Save
10. Click Apply Changes in the top banner
11. Verify logs appear in Datadog within a few seconds

Step-by-Step: Send WAF Events to Splunk

1. Navigate to Observability > Log Pipeline
2. Click Add Sink
3. Name: splunk-waf
4. Type: splunk_hec_logs
5. Source: waf
6. Endpoint: https://splunk-hec.example.com:8088
7. Token: your HEC token
8. Index: corex-waf (optional)
9. Sourcetype: _json (default)
10. Click Test to verify connectivity
11. Click Save
12. Click Apply Changes
13. Verify WAF events appear in Splunk

Step-by-Step: Send All Sources to Elasticsearch

To send all three sources to the same ES cluster with separate daily indices:

1. Create a sink:
   • Name: es-corex, Type: elasticsearch, Source: corex
   • Endpoints: https://es1:9200
   • Index: corex-log-%Y.%m.%d
   • Auth: set as needed
2. Create a second sink:
   • Name: es-waf, Type: elasticsearch, Source: waf
   • Endpoints: https://es1:9200
   • Index: corex-waf-%Y.%m.%d
3. Create a third sink:
   • Name: es-mcp, Type: elasticsearch, Source: mcp
   • Endpoints: https://es1:9200
   • Index: corex-mcp-%Y.%m.%d
4. Click Apply Changes

Alternatively, create one sink with source=corex and set index_corex, index_waf, and index_mcp overrides — but note this only works if the sink type supports per-source index overrides (Elasticsearch does).

Verification

1. Check pipeline status:

   • Navigate to Observability > Log Pipeline
   • Verify the sources card shows which sources are active
   • Verify the sinks list shows all configured sinks with their enabled status

2. Validate the config:

   • Click Validate
   • Expect valid: true with no error output

3. Test a sink:

   • Click Test on a sink
   • Enable Send test event for end-to-end delivery
   • Expect ok: true with no error output

4. Verify delivery:

   • Send test traffic through HAProxy
   • Check the target sink (S3 bucket, Datadog log explorer, ES index, Splunk search, etc.)
   • Verify the log events appear with the expected fields (JA4, req_fp, risk_score, etc.)

5. Check Vector health:

   • Click Restart if Vector isn’t picking up config changes
   • Verify the pipeline status card shows Vector as reachable

Next Steps

• Metrics & Logging — HAProxy and WAF metrics dashboards
• Audit & Snapshots — Config change tracking
• Architecture — Observability pipeline architecture
• Security Rules — Fields available in log events