Передача логов в ELK / SIEM

Как отдавать события qf во внешний лог-конвейер (Elastic/ELK, Splunk, QRadar, Wazuh, ArcSight и пр.). Что уже работает, как настроить, где границы и что дорабатывать.

TL;DR

CP пересылает rule-логи (per-rule allow/deny) и system-события во внешний sink через одну переменную QF_FORWARDER_DSN. Транспорты: syslog (TCP/UDP/TLS) и HTTP(S). Форматы: RFC5424 / CEF / LEEF / ECS-JSON. По умолчанию tee — события пишутся И в PostgreSQL, И в sink. Audit-log и flow-события наружу пока не форвардятся (см. «Что дорабатывать»).

Что форвардится

Поток событий	Наружу	Хранение	REST
rule-логи (allow/deny/log по правилу)	✅	PostgreSQL (tee)	`GET /hosts/{id}/events`
system-события (agent_started, cert_rotated, attach_failed)	✅	PostgreSQL (всегда)	`GET /hosts/{id}/system-events`
audit-log (действия операторов)	✅	PostgreSQL (всегда)	`GET /audit-log`
flow-события (conntrack-флоу)	❌	PostgreSQL	`GET /hosts/{id}/flows`
счётчики правил	❌	PostgreSQL	`GET /hosts/{id}/counters`

Режим tee vs exclusive

QF_FORWARDER_MODE (default tee):

tee — rule-логи пишутся в PostgreSQL и в sink. GET /events и UI-история живут, SIEM получает копию. Дефолт — не терять историю молча.
exclusive — rule-логи уходят только в sink (основным хранилищем становится SIEM, GET /events пуст, SSE live-tail работает). System-события всегда дублятся в PostgreSQL независимо от режима (иначе сломался бы /system-events).

Настройка форвардинга

Переменные в env CP (cp.env для standalone, secrets/env в Helm-values):

# syslog TCP (по умолчанию), фасилити local0, формат RFC5424
QF_FORWARDER_DSN=syslog://collector.example.com:514

# syslog UDP (без фрейминга, сообщение ≤ ~1 КБ)
QF_FORWARDER_DSN=syslog+udp://collector.example.com:514

# syslog over TLS (RFC5425) — шифрованный канал до SIEM
QF_FORWARDER_DSN=syslog+tls://collector.example.com:6514

# HTTP(S) POST — формат по умолчанию ECS-JSON (NDJSON-батч в теле)
QF_FORWARDER_DSN=https://ingest.example.com/qf

# Kafka — по одному сообщению на событие в топик, формат ECS-JSON
QF_FORWARDER_DSN=kafka://broker1:9092,broker2:9092/qf-events

# явный формат + фасилити
QF_FORWARDER_DSN=syslog://collector:514?format=cef&facility=local5

# режим + дисковый спул (пережить простой приёмника)
QF_FORWARDER_MODE=tee                       # или exclusive
QF_FORWARDER_SPOOL=/var/lib/qf/forward.spool

Параметр	Значения	Дефолт
схема	`syslog://` (TCP), `syslog+udp://`, `syslog+tls://`, `http://`, `https://`, `kafka://`	—
`host:port`	адрес приёмника (kafka: список брокеров через `,` + `/topic`)	—
`?format=`	`rfc5424` / `cef` / `leef` / `ecs`	`rfc5424` (syslog), `ecs` (http/kafka)
`?facility=`	`kern`..`local7` (только rfc5424)	`local0`
`?insecure=true`	`syslog+tls`/`https`: пропустить проверку серт-цепи	`false`
`QF_FORWARDER_MODE`	`tee` / `exclusive`	`tee`
`QF_FORWARDER_SPOOL`	путь к файлу диск-спула (пусто = без спула, drop на overflow)	пусто

Соответствие формат→SIEM: CEF — ArcSight; LEEF — QRadar; ECS-JSON — Elastic; RFC5424 — универсальный syslog.

Поведение транспорта:

TCP/TLS — пред-коннект на старте CP (кривой DSN → CP не поднимется, fail-fast). При обрыве — reconnect + один retry на запись. TCP keepalive (30s) + write-deadline (5s) → half-open (приёмник исчез без RST, напр. под коллектора в k8s) детектится и переподключается сам, без рестарта CP.
UDP/HTTP — соединение на пачку (потеря при недоступности приёмника).
Async-очередь — Forward не блокирует ingest hot-path: события кладутся в bounded in-mem очередь (10k). При недоступности приёмника — буфер на диск (QF_FORWARDER_SPOOL, до 100 МиБ, реплей при восстановлении); без спула или при переполнении — дроп с метрикой qf_ingester_events_dropped_total{type="forward"}.

Пример строки RFC5424 (default). Severity: deny → 3 (error), остальное → 6 (info).

Пример строки (allow на порт 22, с обогащением именами):

<134>1 2026-07-01T10:15:30Z qt1 qf log-ports-a allow [qf@32473 kind="rule" direction="ingress" src="203.0.113.7" src_port="51000" dst="10.0.0.5" dst_port="22" proto="2" proto_name="tcp" hostname="qt1" rule_id="<uuid>" rule_name="log-ports-a" policy_name="log-k8s-ports" tenant_id="<uuid>" ct_state="new"] allow src=203.0.113.7:51000 dst=10.0.0.5:22 proto=2

<134> = PRI (local0×8 + 6 info; deny → 131). 1 = версия RFC5424.
HOSTNAME = имя хоста (обогащение; UUID в SD-поле hostname тоже есть при желании), PROCID = имя правила, MSGID = action.
SD-ID qf@32473 — все поля события как key="value" (готово к парсингу).

Обогащение: имена рядом с id

CP резолвит UUID/enum в человекочитаемые имена (name-cache из PG, refresh 30s), но оставляет сырые значения рядом для точного джойна/фильтра:

Сырое	+ имя
`host` (UUID в SD) / HOSTNAME	`hostname` (имя)
`rule_id`	`rule_name`
—	`policy_name` (резолв через rule→policy; агентский LogEvent не несёт policy_id)
`proto` (qf-enum: 1=any/2=tcp/3=udp/4=icmp/5=icmpv6, не IANA)	`proto_name`

По форматам: RFC5424 — SD-поля выше; CEF — dhost/cs1..cs3/proto=имя; LEEF — host/ruleName/policyName/proto=имя; ECS — host.name/rule.name/ rule.ruleset.name/network.transport.

Интеграция с ELK (Elastic Stack)

Три типовых пути. Выбор — по тому, что уже стоит.

Вариант A. Logstash syslog input (прямо)

# logstash.conf
input {
  syslog {
    port  => 5514
    codec => "line"
  }
}
filter {
  # выдрать SD-поля qf@32473 (kv подхватит hostname/rule_name/policy_name/proto_name и пр.)
  grok {
    match => { "message" => "\[qf@32473 %{DATA:qf_sd}\]" }
  }
  kv { source => "qf_sd" field_split => " " value_split => "=" trim_value => "\"" }
  mutate { convert => { "dst_port" => "integer" "src_port" => "integer" "proto" => "integer" } }
}
output { elasticsearch { hosts => ["es:9200"] index => "qf-events-%{+YYYY.MM.dd}" } }

QF_FORWARDER_DSN=syslog://logstash:5514.

Вариант B. rsyslog/syslog-ng → файл → Filebeat

Коллектор пишет в файл, Filebeat подхватывает:

# /etc/rsyslog.d/qf.conf
local0.*  /var/log/qf/events.log

# filebeat.yml
filebeat.inputs:
  - type: filestream
    paths: ["/var/log/qf/events.log"]
output.elasticsearch: { hosts: ["es:9200"], index: "qf-events" }

QF_FORWARDER_DSN=syslog://rsyslog-host:514?facility=local0.

Вариант C. Elastic Agent / Filebeat syslog input

Filebeat syslog input (RFC5424) слушает порт напрямую — CP шлёт на него.

Интеграция с SIEM

SIEM	Как подключить
Splunk	syslog → Splunk Universal Forwarder / HEC-syslog-gateway; либо rsyslog → monitored file → Splunk. `sourcetype=rfc5424`
QRadar	Log Source типа Syslog, протокол Syslog; QRadar парсит RFC5424, поля из SD доступны как custom properties
Wazuh	`<remote>` syslog listener → decoder на `qf@32473`; правила на `action="deny"`
Graylog	Input Syslog TCP/UDP, extractor kv по SD-полям
ArcSight	SmartConnector Syslog Daemon; для нативного CEF — см. гэп ниже (пока не CEF)

Общий рецепт: QF_FORWARDER_DSN=syslog://<siem-syslog-vip>:514, дальше — приёмник и парсер на стороне SIEM. Все поля события уже в structured-data, грок по MSG не нужен.

Полезные детекты: action="deny" всплеск с одного src (скан/брутфорс), dst_port в чувствительном списке, ct_state="invalid".

Границы

Один глобальный DSN, без per-tenant роутинга. Все тенанты валятся в один приёмник (tenant_id присутствует в каждом событии — фильтрация/роутинг на стороне SIEM).
flow-события и счётчики наружу не форвардятся — только REST (GET /hosts/{id}/flows, GET /hosts/{id}/counters) и PostgreSQL.

Быстрая проверка

# приёмник-заглушка
nc -lk 514            # TCP; для UDP: nc -luk 514

# на CP
export QF_FORWARDER_DSN=syslog://127.0.0.1:514
# рестарт CP → создай policy с rule log:true на открытый порт → дай трафик
# в nc польются RFC5424-строки с hostname/rule_name/proto_name

Нет строк? Проверь: rule имеет log:true; агент active и получил bundle; порт приёмника открыт; для TCP — CP поднялся (кривой DSN валит старт с ошибкой forwarder: initial connect).

Имена (rule_name/policy_name/hostname) пустые? name-cache обновляется раз в 30s — свежесозданное правило/хост появятся в логах с именем в пределах интервала.

TL;DR​

Что форвардится​

Режим tee vs exclusive​

Настройка форвардинга​

Обогащение: имена рядом с id​

Интеграция с ELK (Elastic Stack)​

Вариант A. Logstash syslog input (прямо)​

Вариант B. rsyslog/syslog-ng → файл → Filebeat​

Вариант C. Elastic Agent / Filebeat syslog input​

Интеграция с SIEM​

Границы​

Быстрая проверка​