G-BlackHole Documentation

Tổng quan cấu hình

Forwarder (blackhole-fwd) đọc cấu hình từ một file YAML. Trong file đó bạn khai báo các source (nhận dữ liệu), transform (xử lý), sink (gửi đi), các embedded server (listener nội bộ) và các section cấu hình chung.

Mọi ví dụ YAML trên trang này đều có thể copy trực tiếp vào file cấu hình. Khi một trường có giá trị mặc định, bạn chỉ cần viết nó khi muốn đổi giá trị đó.

Vị trí file cấu hình

Phương thức	Hành vi
`blackhole-fwd` (không chỉ định gì)	Đọc `blackhole-fwd.yml` trong thư mục làm việc; nếu không có thì thử `./config.yml`, rồi `./config.yaml`, cuối cùng dùng cấu hình rỗng built-in
`-c, --config <PATH>`	Chỉ định chính xác file cấu hình phải dùng
`--config-dir <DIR>`	Ưu tiên hơn `--config`: tìm `config.yml` hoặc `config.yaml` trong `DIR`, rồi merge mọi file `.blh.yml` / `.blh.yaml` nằm bên trong `DIR` (kể cả thư mục con)

Nếu --config-dir chỉ định thư mục mà không có config.yml lẫn config.yaml, tiến trình báo lỗi:

text

No config file (config.yml or config.yaml) found in directory: <DIR>

⚠️

Khi chạy `blackhole-fwd start`, file tại đúng đường dẫn `--config` **phải tồn tại trước khi khởi động**. Watcher hot-reload luôn băm (hash) file cấu hình lúc khởi động, nên nếu file không có bạn sẽ gặp lỗi:

text

Failed to read file for hashing: blackhole-fwd.yml

Hãy tạo file tại chính xác đường dẫn bạn truyền cho --config (hoặc tạo config.yml/config.yaml trong thư mục mà --config-dir trỏ tới).

Các section cấp cao nhất

Section	Vai trò	Chi tiết
`logging`	File log, mức log, xoay vòng	Cấu hình logging
`channel_buffers`	Bộ đệm kênh giữa pipeline	Channel buffers
`inventory`	Hàng đợi tin nhắn bền vững, retry, backpressure	Inventory
`hot_reload`	Tự nạp lại cấu hình khi file đổi	Hot-reload
`resources_threshold`	Ngưỡng CPU/RAM/đĩa để tạm dừng pipeline	Ngưỡng tài nguyên
`registry`	Kết nối Registry + cổng REST next-hop	Registry
`sources`	Các nguồn dữ liệu (map đặt tên)	Sources
`transforms`	Pipeline xử lý sự kiện	Transforms
`sinks`	Các đích gửi dữ liệu (map đặt tên)	Sinks
`rsyslog_server`	Listener syslog UDP/TCP	rsyslog_server
`snmp_server`	Listener SNMP trap	snmp_server
`mqtt_server`	MQTT broker nhúng	mqtt_server
`proxy_server`	HTTP proxy nhúng	proxy_server
`grpc_server`	gRPC server nhúng	grpc_server

Mỗi embedded server chỉ khởi động khi section tương ứng tồn tại trong file cấu hình.

Lệnh CLI liên quan tới cấu hình

text

blackhole-fwd [-c|--config <PATH>] [--config-dir <DIR>] [-v|--verbose] [--service]
              [--service-name <N>] [-V|--version] [-h] [COMMAND]

  (không có COMMAND)                In banner

  configure | config  [-f|--force] [--output <PATH>] <ACTION>
      pull              Đọc cấu hình từ Registry, lưu ra file local
                        (--output: tùy chọn, dump YAML ra file khác)
      push              Đẩy file --config lên Registry
      schema            Viết file config.jsonschema cạnh file --config

  service [-f|--force] [--service-name <N>] <install|uninstall|start|stop|status>
  auth -k|--key <KEY>
  start [-s|--service]

Ví dụ:

bash

# Kéo cấu hình từ Registry về file local
blackhole-fwd -c config.yml configure pull

# Kéo và đồng thời dump YAML đã merge ra một file khác
blackhole-fwd -c config.yml configure pull --output resolved.yml

# Đẩy file cấu hình local lên Registry
blackhole-fwd -c config.yml configure push

# Sinh JSON schema cạnh file cấu hình
blackhole-fwd -c config.yml configure schema

📝

Các lệnh sau **không tồn tại** — đừng gõ chúng: `configure default`, `config validate`, `service restart`, `--path`, `--log-level`, `--dry-run`. Muốn có cấu hình mẫu, hãy viết YAML thủ công theo các ví dụ trên trang này, hoặc dùng `configure pull` để lấy cấu hình từ Registry.

Không có --log-level. Muốn log debug, dùng cờ toàn cục -v / --verbose (đặt logging.level thành debug) hoặc khai báo logging.level trong YAML.
-c/--config và --config-dir là cờ toàn cục, đặt trước tên subcommand. --output chỉ có ý nghĩa với configure pull.
auth --key <KEY> dùng để khai báo API key; key không bao giờ được ghi vào log hay file cấu hình.

Yêu cầu khi dùng Registry

⚠️

Khi section `registry` tồn tại, `registry.http_server.enabled` mặc định là `true`. Trong trường hợp đó **bạn bắt buộc phải cấu hình `registry.api_key` và thiết bị phải đã có device id**, nếu không tiến trình không khởi động được và báo:

text

registry.rest_forwarder.enabled=true requires registry.api_key

Đây là lỗi thường gặp khi chạy lần đầu. Nếu chưa muốn bật lớp next-hop, đặt registry.http_server.enabled: false.

Cấu hình logging

Toàn bộ section này là tùy chọn — thiếu section logging thì mọi trường dùng giá trị mặc định.

Cú pháp

yaml

logging:
  level: "info"
  dir: "./logs"
  file: "default.log"
  error_file: "default-error.log"
  max_size_mb: 10

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`level`	String	Không	`"info"`	Mức log: `trace`, `debug`, `info`, `warn`, `error`. Cũng chấp nhận chỉ định theo target kiểu `collectors=trace`
`dir`	String (đường dẫn)	Không	`"./logs"`	Thư mục chứa file log, tính tương đối so với thư mục làm việc
`file`	String	Không	`"default.log"`	Tên file log chính
`error_file`	String	Không	`"default-error.log"`	Tên file log riêng cho lỗi/cảnh báo
`max_size_mb`	Int (MB)	Không	`10`	Tổng dung lượng file log được phép trước khi xoay vòng (xem Lưu ý)

Ví dụ

Cơ bản:

yaml

logging:
  level: "info"

Nâng cao (log chi tiết cho một module, đổi thư mục và tên file):

yaml

logging:
  level: "supervisor=debug"
  dir: "/var/log/blackhole-fwd"
  file: "forwarder.log"
  error_file: "forwarder-error.log"
  max_size_mb: 50

Lưu ý

Xoay vòng luôn bật. max_size_mb là tổng dung lượng cho phép, không phải ngưỡng của từng file:
- File log chính (file): xoay thành 5 file cũ, mỗi file bị cắt ở max_size_mb / 5.
- File log lỗi (error_file): xoay thành 10 file cũ, mỗi file cắt ở max_size_mb (tức dung lượng tối đa có thể lên tới ~10 lần giá trị bạn đặt).
Đường dẫn thực tế của file log chính là <dir>/<file> (ví dụ ./logs/default.log), file lỗi là <dir>/<error_file> (./logs/default-error.log).
dir là tương đối so với thư mục làm việc của tiến trình, không phải /var/log/.... Khi chạy dưới systemd, thư mục làm việc mặc định là thư mục chứa binary — nên nếu muốn ghi vào /var/log, hãy khai báo đường dẫn tuyệt đối cho dir.
Ở chế độ service, stdout bị tắt — file log là nguồn output duy nhất.
-v/--verbose trên dòng lệnh sẽ đặt logging.level thành debug.

Channel buffers

Điều chỉnh độ sâu của các kênh nội bộ giữa collector và sink. Tùy chọn toàn bộ.

Cú pháp

yaml

channel_buffers:
  orchestrator_capacity: 100000
  shipper_capacity: 100000
  router_max_in_flight_routes: 4

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`orchestrator_capacity`	Int	Không	`100000`	Số sự kiện tối đa xếp hàng giữa nguồn và orchestrator
`shipper_capacity`	Int	Không	`100000`	Số sự kiện tối đa xếp hàng giữa orchestrator và từng sink
`router_max_in_flight_routes`	Int	Không	`4`	Số tuyến định tuyến chạy song song tối đa khi phát sự kiện sang nhiều sink

Ví dụ

Cơ bản:

yaml

channel_buffers:
  orchestrator_capacity: 100000
  shipper_capacity: 100000

Nâng cao (dự phòng burst lớn hơn, CPU cao hơn):

yaml

channel_buffers:
  orchestrator_capacity: 500000
  shipper_capacity: 500000
  router_max_in_flight_routes: 8

Lưu ý

Tăng capacity giúp xử lý burst tốt hơn nhưng tốn nhiều bộ nhớ hơn.
router_max_in_flight_routes càng lớn càng nhanh nhưng càng tốn CPU.

Inventory

inventory là hàng đợi bền vững giữ tin nhắn khi sink chưa gửi được. Bỏ trống toàn bộ section thì giá trị mặc định là max_messages: 100000, max_bytes: 268435456 (256 MB), backpressure_on_limit: false.

⚠️

Nếu bạn viết section `inventory:` thì **`backpressure_on_limit` và `per_shipper` là bắt buộc** — thiếu một trong hai, file cấu hình không parse được. Hãy copy nguyên ví dụ bên dưới thay vì viết một phần.

Cú pháp

yaml

inventory:
  max_messages: 100000
  max_bytes: 268435456
  backpressure_on_limit: false
  per_shipper:
    file:
      ttl: 3600
      max_retries: 10
      flush_interval_secs: 10

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`max_messages`	Int hoặc `null`	Khi viết section: có	`100000`	Số tin nhắn tối đa trong hàng đợi; `null` = không giới hạn
`max_bytes`	Int hoặc `null`	Khi viết section: có	`268435456`	Tổng byte tối đa; `null` = không giới hạn
`backpressure_on_limit`	Bool	Có (khi viết section)	`false`	`true`: chặn đầu vào khi đầy (không mất dữ liệu); `false`: bỏ tin cũ nhất
`per_shipper`	Map	Có (khi viết section)	—	Chính sách giao cho từng loại sink

Khối per_shipper — mỗi entry có 3 trường bắt buộc:

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`ttl`	Int (giây)	Có	—	Tuổi tin nhắn tối đa trong hàng đợi, quá hạn bị loại
`max_retries`	Int	Có	—	Số lần thử gửi tối đa trước khi bỏ tin
`flush_interval_secs`	Int (giây)	Có	—	Chu kỳ nền thử gửi lại hàng đợi

Các khóa hợp lệ trong per_shipper: file, mqtt, kafka, grpc, blackhole, alert. Một sink không có entry sẽ dùng ttl 3600 (1 giờ), max_retries 10, flush_interval_secs 10.

Ví dụ

Cơ bản (chỉ khai báo hạn mức, dùng chính sách mặc định cho sink thiếu entry):

yaml

inventory:
  max_messages: 100000
  max_bytes: 268435456
  backpressure_on_limit: false
  per_shipper:
    grpc:
      ttl: 3600
      max_retries: 10
      flush_interval_secs: 10

Nâng cao (giữ dữ liệu khi sink tạm ngưng, retry lâu hơn cho Kafka):

yaml

inventory:
  max_messages: 1000000
  max_bytes: 1073741824
  backpressure_on_limit: true
  per_shipper:
    kafka:
      ttl: 86400
      max_retries: 20
      flush_interval_secs: 5
    mqtt:
      ttl: 86400
      max_retries: 20
      flush_interval_secs: 5
    grpc:
      ttl: 3600
      max_retries: 10
      flush_interval_secs: 10

Lưu ý

backpressure_on_limit: true đổi mất mát dữ liệu lấy độ trễ: hàng đợi đầy thì pipeline bị chặn thay vì bỏ tin cũ.
max_messages/max_bytes đặt null để không giới hạn — chỉ nên làm khi biết chắc bộ nhớ máy đủ.

Hot-reload

Tự nạp lại cấu hình khi file YAML trên đĩa thay đổi. Tùy chọn.

Cú pháp

yaml

hot_reload:
  enabled: true
  poll_interval_ms: 5000
  debounce_ms: 1000

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`enabled`	Bool	Không	`true`	Bật/tắt hot-reload
`poll_interval_ms`	Int (ms)	Không	`5000`	Chu kỳ quét file cấu hình bằng SHA-256
`debounce_ms`	Int (ms)	Không	`1000`	Thời gian chờ sau khi phát hiện đổi để tránh reload liên tục

Ví dụ

Cơ bản:

yaml

hot_reload:
  enabled: true

Nâng cao (quét chậm hơn để giảm I/O):

yaml

hot_reload:
  enabled: true
  poll_interval_ms: 15000
  debounce_ms: 2000

Lưu ý

Cơ chế là quét SHA-256 định kỳ trên file cấu hình chính và mọi *.blh.yml / *.blh.yaml bên trong thư mục cấu hình. Không có inotify, không có SIGHUP — gửi SIGHUP không làm gì.
Sửa cấu hình local sẽ nạp vào bộ nhớ và ghi lại file .blackhole-resolved.yaml, nhưng không dựng lại collector/transform/sink đang chạy. Nếu nguồn cấu hình đang active là Registry, mọi sửa local bị bỏ qua hoàn toàn với thông báo: Remote config is active — ignoring local config change.
Chỉ cập nhật đẩi từ Registry mới chạy diff theo từng identifier để thêm/bỏ/restart từng component riêng lẻ.
Những thứ không hot-reload được (phải khởi động lại service): logging, hot_reload, registry, resources_threshold và mọi section embedded server (rsyslog_server, snmp_server, mqtt_server, proxy_server, grpc_server).
Khi hot-reload bật, file cấu hình chính phải tồn tại từ trước lúc khởi động (xem Vị trí file cấu hình).

Ngưỡng tài nguyên

Khi CPU/RAM/đĩa vượt ngưỡng đủ lâu, pipeline tạm dừng để máy thở. Tùy chọn toàn bộ.

Cú pháp

yaml

resources_threshold:
  cpu:
    threshold_percentage: 80.0
    sustained_secs: 60
  memory:
    threshold_percentage: 80.0
    sustained_secs: 60
  disk:
    threshold_percentage: 101.0
    sustained_secs: 60
  check_interval: 10

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`cpu.threshold_percentage`	Float	Không	`80.0`	Ngưỡng CPU (%)
`cpu.sustained_secs`	Int (giây)	Không	`60`	Thời gian phải duy trì liên tục ở ngưỡng mới kích hoạt
`memory.threshold_percentage`	Float	Không	`80.0`	Ngưỡng bộ nhớ (%)
`memory.sustained_secs`	Int (giây)	Không	`60`	Thời gian duy trì cho bộ nhớ
`disk.threshold_percentage`	Float	Không	`101.0`	Ngưỡng đĩa (%). Mặc định `101.0` = tắt
`disk.sustained_secs`	Int (giây)	Không	`60`	Thời gian duy trì cho đĩa
`check_interval`	Int (giây)	Không	`10`	Chu kỳ đo

Ví dụ

Cơ bản (dùng mặc định):

yaml

resources_threshold:
  check_interval: 10

Nâng cao (bật giám sát đĩa, nới ngưỡng CPU):

yaml

resources_threshold:
  cpu:
    threshold_percentage: 90.0
    sustained_secs: 120
  memory:
    threshold_percentage: 85.0
    sustained_secs: 60
  disk:
    threshold_percentage: 95.0
    sustained_secs: 60
  check_interval: 30

Lưu ý

Guard chỉ hoạt động khi threshold_percentage <= 100; đặt trên 100 để tắt giám sát resource đó. Disk mặc định tắt theo đúng cơ chế này.
CPU/RAM vượt ngưỡng trong sustained_secs → pipeline tạm dừng; giảm xuống dưới ngưỡng → tự chạy lại.
resources_threshold không hot-reload — đổi phải khởi động lại service.

Registry

Section registry mô tả cách Forwarder kết nối Registry (điểm điều phối trung tâm) để nhận cấu hình, đăng ký thiết bị và chuyển tiếp request cho thiết bị downstream. Section này tùy chọn — không có thì Forwarder chạy độc lập với cấu hình local.

Cú pháp

yaml

registry:
  api_url: "https://blackhole.glabs.one"
  api_key: "<api-key>"
  config_update_interval: "15s"
  tls:
    ca_cert_path: null
    insecure_skip_verify_https: false
  proxy:
    enable: true
    http: null
    https: null
  http_server:
    enabled: true
    listen_addr: "0.0.0.0:18080"

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`api_url`	String (URL)	Có	—	REST API của Registry. Giá trị built-in dùng khi lấy mẫu là `https://blackhole.glabs.one`
`api_key`	String	Xem Warning bên dưới	`null`	API key xác thực với Registry (riêng bản Forwarder)
`config_update_interval`	String (duration)	Không	`"15s"`	Chu kỳ pull cấu hình từ Registry, dạng `15s`, `5m`, `1h`
`tls.ca_cert_path`	String	Không	`null`	CA bundle (PEM) để validate chứng chỉ của Registry
`tls.insecure_skip_verify_https`	Bool	Không	`false`	Bỏ qua xác thực HTTPS (không khuyến nghị). Chỉ có ý nghĩa với `https://`
`proxy.enable`	Bool	Không	`true`	Bật/tắt proxy (chỉ áp dụng khi block `proxy` được khai báo)
`proxy.http`	String (URL)	Không	`null`	HTTP proxy, dạng `http://user:pass@proxy:8080`
`proxy.https`	String (URL)	Không	`null`	HTTPS proxy, dạng `https://user:pass@proxy:8443`
`http_server`	Object	Không	bật sẵn	Cổng REST next-hop — xem Cổng REST next-hop

Ví dụ

Cơ bản (kết nối Registry, không bật lớp next-hop):

yaml

registry:
  api_url: "https://blackhole.glabs.one"
  api_key: "your-api-key-here"
  config_update_interval: "15s"
  http_server:
    enabled: false

Nâng cao (qua proxy nội bộ, CA riêng, poll chậm hơn):

yaml

registry:
  api_url: "https://registry.company.com"
  api_key: "your-api-key-here"
  config_update_interval: "60s"
  tls:
    ca_cert_path: "/etc/ssl/certs/company-ca.pem"
    insecure_skip_verify_https: false
  proxy:
    enable: true
    http: "http://proxy.company.com:8080"
    https: "https://proxy.company.com:8443"

Lưu ý

registry không hot-reload — đổi xong phải khởi động lại service.
Không có trường mqtt_url, identifier hay rate_limit trong section registry; các trường đó không tồn tại.
Chi tiết cổng REST next-hop và yêu cầu api_url/api_key khi khởi động: xem section tiếp theo.

Cổng REST next-hop

Khi section registry tồn tại, Forwarder có thể chạy một HTTP server nhỏ để chuyển tiếp (next-hop) một số endpoint của Registry cho các thiết bị downstream trong mạng. Đây là lớp trung gian: thiết bị con trỏ vào Forwarder, Forwarder ủy quyền lên registry.api_url.

Cú pháp

yaml

registry:
  api_url: "https://blackhole.glabs.one"
  api_key: "<api-key>"
  http_server:
    enabled: true
    listen_addr: "0.0.0.0:18080"
    cache_ttl: "30s"
    cache_max_entries: 10000
    cache_max_body_bytes: 262144
    request_timeout_sec: 30
    retry_on_statuses: [401, 404]
    tls:
      enabled: false
      cert_path: null
      key_path: null

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`enabled`	Bool	Không	`true`	Bật/tắt cổng next-hop
`listen_addr`	String	Không	`"0.0.0.0:18080"`	Địa chỉ và cổng lắng nghe
`cache_ttl`	String (duration)	Không	`"30s"`	TTL cache cho các request GET
`cache_max_entries`	Int	Không	`10000`	Số entry cache tối đa
`cache_max_body_bytes`	Int (bytes)	Không	`262144`	Kích thước body GET tối đa được cache
`request_timeout_sec`	Int (giây)	Không	`30`	Timeout gọi lên Registry
`retry_on_statuses`	Danh sách Int	Không	`[401, 404]`	Status khiến server tự join thiết bị rồi retry 1 lần
`tls.enabled`	Bool	Không	`false`	Bật HTTPS cho cổng next-hop
`tls.cert_path`	String	Khi `tls.enabled: true`	`null`	Server certificate (PEM)
`tls.key_path`	String	Khi `tls.enabled: true`	`null`	Private key (PEM)

Các endpoint

Method	Path	Header yêu cầu	Ghi chú
`POST`	`/devices/join`	`x-device-name`, `x-device-type`	Đăng ký thiết bị downstream (tự join khi upstream trả `401`/`404`)
`GET`	`/client/config`	`x-device-ip-address`	Lấy cấu hình cho thiết bị
`PATCH`	`/client/config`	`x-device-ip-address`	Cập nhật cấu hình cho thiết bị
`POST`	`/client/alerts`	`x-device-ip-address`	Chuyển tiếp alert; body tối đa 1 MiB
`GET`	`/health`	—	Trả về `{"status":"ok"}`

Header x-device-ip-address là bắt buộc với các route config và alerts; thiếu sẽ trả lỗi JSON missing required header x-device-ip-address. Header x-device-name và x-device-type dùng cho auto-join khi server nhận 401/404 từ upstream (theo retry_on_statuses).

Ví dụ

Kiểm tra sức khỏe của cổng next-hop:

bash

curl -s http://127.0.0.1:18080/health
# {"status":"ok"}

Lấy cấu hình cho một thiết bị:

bash

curl -s http://127.0.0.1:18080/client/config \
  -H "x-device-ip-address: 10.20.30.40"

Đẩy alert từ thiết bị con:

bash

curl -s -X POST http://127.0.0.1:18080/client/alerts \
  -H "x-device-ip-address: 10.20.30.40" \
  -H "Content-Type: application/json" \
  -d '{"alerts":[{"title":"Disk full","severity":"high"}]}'

Join thiết bị downstream (thường do server tự gọi khi retry):

bash

curl -s -X POST http://127.0.0.1:18080/devices/join \
  -H "x-device-name: edge-fw-01" \
  -H "x-device-type: firewall" \
  -H "x-device-ip-address: 10.20.30.40"

Lưu ý

⚠️

`http_server.enabled` mặc định là `true`. Khi section `registry` có mặt mà `registry.api_key` chưa được khai báo (hoặc thiết bị chưa có device id), tiến trình **khởi động thất bại** với:

text

registry.rest_forwarder.enabled=true requires registry.api_key

Chạy lần đầu chưa có key thì đặt http_server.enabled: false.

listen_addr mặc định bind mọi interface (0.0.0.0:18080) — nếu chỉ muốn nội bộ, đổi thành 127.0.0.1:18080.
tls.enabled mặc định false (HTTP trần). Chỉ bật khi đã có cert_path + key_path.
Cổng này không hot-reload — đổi xong phải khởi động lại.

Device guard (danh sách chặn token)

Device guard là cơ chế chặn truy cập ở runtime dựa trên trạng thái thiết bị từ Registry.

Hoạt động

Đây là danh sách chặn (deny-list) các device token giữ trong bộ nhớ — không phải allow-list và không có trường nào trong file YAML cấu hình nó. Danh sách được nạp runtime theo trạng thái thiết bị do Registry trả về.
Danh sách rỗng = mọi token đều được phép. Đây là điều operators thường hiểu ngược: một allow-list rỗng sẽ chặn tất cả, nhưng deny-list rỗng thì không chặn gì cả.
Khi một thiết bị không ở trạng thái Active, token của nó bị đưa vào danh sách và mọi request mang token đó bị từ chối với gRPC permission_denied. Nếu thiết bị đang có stream gRPC mở, stream đó bị ngắt.
Danh sách được cập nhật lại theo polling — token bị chặn sẽ hết chặn khi thiết bị trở lại Active.

Header liên quan

Header	Dùng ở đâu	Ý nghĩa
`x-device-token`	Metadata gRPC và header HTTP upstream	Token của thiết bị. Đi qua được bước kiểm tra API key (`x-api-key`) vì được xử lý riêng
`x-api-key`	gRPC server (tên header lấy từ `grpc_server.auth.header_name`)	API key tĩnh; `api_keys` rỗng nghĩa là tắt kiểm tra API key

Lưu ý

Không thể cấu hình device guard từ YAML — đừng tìm kiếm section device_guard trong file cấu hình.
Muốn chặn một thiết bị, đổi trạng thái của nó trên Registry; Forwarder tự đồng bộ về.
Token không bao giờ được ghi ra log ở mức INFO.

Sources

Tổng quan sources

sources là một map đặt tên: khóa là identifier của source, giá trị là object có trường type chọn loại source.

yaml

sources:
  syslog_in:
    type: "rsyslog"
    includes: []

Các type hợp lệ của Forwarder: rsyslog, snmp, mqtt, kafka, grpc.

Quy tắc với identifier (khóa):

Không được là * (dành cho wildcard của sink) và không được để rỗng.
Identifier do bạn đặt ở khóa; các field identifier bên trong object được điền tự động từ khóa đó — không cần (và không nên) khai báo thủ công.
inputs của transform/sink tham chiếu tới các identifier này.

Rsyslog source

Nhận syslog từ các host khác qua UDP/TCP (cần bật rsyslog_server).

Cú pháp

yaml

sources:
  syslog_in:
    type: "rsyslog"
    includes: []
    index: null
    sourcetype: null

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`type`	String	Có	—	Luôn là `"rsyslog"`
`includes`	Danh sách String	Không	`[]`	Cho phép theo IP hoặc CIDR, ví dụ `["192.168.1.1", "10.1.1.0/24"]`. Rỗng = nhận tất cả
`index`	String hoặc `null`	Không	`null`	Tên index gán cho dữ liệu nhận được
`sourcetype`	String hoặc `null`	Không	`null`	Sourcetype gán cho dữ liệu nhận được

Ví dụ

Cơ bản:

yaml

sources:
  syslog_in:
    type: "rsyslog"

Nâng cao (chỉ nhận từ dải mạng nội bộ, gắn index/sourcetype):

yaml

sources:
  syslog_in:
    type: "rsyslog"
    includes:
      - "192.168.10.0/24"
      - "10.0.0.5"
    index: "network_logs"
    sourcetype: "syslog_rfc5424"

Lưu ý

Source chỉ nhận dữ liệu khi rsyslog_server được khai báo (listener mở cổng).
includes chỉ lọc theo địa chỉ nguồn; không có expression phức tạp.

SNMP source

Thu thập dữ liệu thiết bị SNMP theo hai nhánh: poll (định kỳ truy vấn OID) và trap (nhận sự kiện đẩy).

Cú pháp

yaml

sources:
  net_devices:
    type: "snmp"
    includes: []
    port: 161
    version: "v3"
    credentials: {}
    poll: {}
    poll_enabled: true
    trap_enabled: true
    oids:
      - "1.3.6.1.2.1.1.3.0"
    index: null
    sourcetype: null

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`type`	String	Có	—	Luôn là `"snmp"`
`includes`	Danh sách object	Không	`[]`	Mỗi entry là `{ target, port?, index?, sourcetype? }`. `target` nhận IP, hostname, `hostname:port` hoặc CIDR
`port`	Int	Không	`161`	Cổng poll mặc định (entry `includes` có thể ghi đè)
`version`	String	Không	`"v3"`	`v1`, `v2c` hoặc `v3`
`credentials.community`	String	Với v1/v2c	`null`	Community string
`credentials.v3`	Object	Với v3	`null`	Khối SNMPv3 (xem bảng dưới)
`poll`	Object	Không	xem bảng dưới	Cấu hình poll
`poll_enabled`	Bool	Không	`true`	Bật/tắt poll định kỳ
`trap_enabled`	Bool	Không	`true`	Bật/tắt nhận trap cho các include của source
`oids`	Danh sách (String hoặc object)	Không	`["1.3.6.1.2.1.1.3.0"]`	OID cần poll. Dạng string (`"1.3.6..."`) hoặc `{ name, oid }`
`index`	String hoặc `null`	Không	`null`	Index cho dữ liệu poll
`sourcetype`	String hoặc `null`	Không	`null`	Sourcetype cho dữ liệu poll

Khối poll:

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`interval_secs`	Int (giây)	Không	`60`	Khoảng cách giữa hai chu kỳ poll
`timeout_ms`	Int (ms)	Không	`2000`	Timeout mỗi request
`retries`	Int	Không	`2`	Số lần thử lại trước khi coi là lỗi
`max_oids_per_request`	Int	Không	`20`	Số OID tối đa mỗi request
`jitter_pct`	Int (0–100)	Không	`null`	Phần trăm jitter cộng vào interval

Khối credentials.v3 (SNMPv3):

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`username`	String	Có	—	Security username
`auth.protocol`	String	Không	`"sha256"`	`md5`, `sha`, `sha224`, `sha256`, `sha384`, `sha512`
`auth.key`	String	Với auth	—	Auth key (hỗ trợ tham chiếu môi trường như `${SNMP_AUTH_KEY}`)
`privacy.protocol`	String	Không	`"aes128"`	`des`, `aes128`, `aes192`, `aes256`
`privacy.key`	String	Với privacy	—	Privacy key
`context_engine_id`	String	Không	`null`	Engine ID dạng hex, ví dụ `"80003a8c04"`
`context_name`	String	Không	`null`	Context name (hiếm khi cần)

Ví dụ

Cơ bản (SNMPv3, chỉ poll uptime):

yaml

sources:
  net_devices:
    type: "snmp"
    includes:
      - target: "10.0.10.5"
      - target: "edge-fw-1"
    credentials:
      v3:
        username: "monitor"
        auth:
          protocol: "sha256"
          key: "${SNMP_AUTH_KEY}"
        privacy:
          protocol: "aes128"
          key: "${SNMP_PRIV_KEY}"
    oids:
      - name: "sysUpTime"
        oid: "1.3.6.1.2.1.1.3.0"

Nâng cao (v2c, poll nhanh hơn, trap tắt, override index từng host):

yaml

sources:
  legacy_switches:
    type: "snmp"
    includes:
      - target: "10.0.20.11"
        port: 1161
        index: "switch_metrics"
      - target: "10.0.30.0/24"
    port: 161
    version: "v2c"
    credentials:
      community: "public"
    poll_enabled: true
    trap_enabled: false
    poll:
      interval_secs: 30
      timeout_ms: 1500
      retries: 1
      max_oids_per_request: 40
      jitter_pct: 10
    oids:
      - name: "cpu_load"
        oid: "1.3.6.1.2.1.25.3.3.1.2.1"
      - "1.3.6.1.2.1.1.3.0"
    index: "network_metrics"

Lưu ý

⚠️

- `snmp_server` **phải được khai báo trước** khi source `snmp` chạy được. Thiếu section này, collector báo lỗi: `SNMP server config required`. - Khi `snmp_server.require_v3: true` (mặc định), các source chỉ khai báo `v1`/`v2c` sẽ bị từ chối khởi động poller.

includes là danh sách object { target, port?, index?, sourcetype? }, không phải danh sách chuỗi như rsyslog.
OID dạng chuỗi không có name sẽ dùng chính OID làm tên trong sự kiện.

MQTT source

Subscribe vào broker MQTT và nhận message.

Cú pháp

yaml

sources:
  mqtt_in:
    type: "mqtt"
    broker:
      url: "mqtt://broker.company.com:1883"
      auth:
        type: "none"
    topics: []
    index: null
    sourcetype: null
    source_override: false

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`type`	String	Có	—	Luôn là `"mqtt"`
`broker`	Object	Có (xem Lưu ý)	`null`	Thông tin kết nối broker
`broker.url`	String	Không (trong block broker)	`"mqtt://localhost:1883"`	URL broker, `mqtt://` hoặc `mqtts://`
`broker.auth`	Object	Không	`type: "none"`	`{"type":"none"}` hoặc `{"type":"basic","username":...,"password":...}`
`broker.proxy`	Object	Không	`null`	`{ enable, http, https }`
`broker.lwt`	Object	Không	`null`	Last Will: `{ topic, payload, qos, retain }`
`broker.tls`	Object	Không	`null`	`{ capath, certpath, keypath, insecure }` (đồng nghĩa `ca_cert_path`, `client_cert_path`, `client_key_path`)
`topics`	Danh sách String	Không	`[]`	Topic cần subscribe, hỗ trợ `+` và `#`. Alias: `includes`
`index`	String hoặc `null`	Không	`null`	Index cho dữ liệu
`sourcetype`	String hoặc `null`	Không	`null`	Sourcetype cho dữ liệu
`source_override`	Bool	Không	`false`	`true`: luôn ghi đè index/sourcetype downstream bằng giá trị source
`request_queue_capacity`	Int	Không	runtime = `100`	Kích thước hàng đợi request của client MQTT
`keep_alive_seconds`	Int	Không	theo thư viện	Keep-alive (giây)
`max_incoming_packet_size`	Int (bytes)	Không	theo thư viện	Kích thước packet vào tối đa
`max_outgoing_packet_size`	Int (bytes)	Không	theo thư viện	Kích thước packet ra tối đa
`subscription_qos`	String	Không	`"at_most_once"`	`at_most_once`, `at_least_once`, `exactly_once`
`network.connection_timeout_seconds`	Int	Không	theo thư viện	Timeout kết nối TCP (giây)
`network.nodelay`	Bool	Không	theo thư viện	TCP_NODELAY
`clean_session`	Bool	Không	theo thư viện	Clean session
`inflight`	Int	Không	theo thư viện	Số message QoS>0 đang bay tối đa
`pending_throttle_micros`	Int (µs)	Không	theo thư viện	Khoảng nghỉ giữa hai packet gửi ra
`manual_acks`	Bool	Không	theo thư viện	Tự ack message nhận vào

Ví dụ

Cơ bản:

yaml

sources:
  mqtt_in:
    type: "mqtt"
    broker:
      url: "mqtt://broker.company.com:1883"
      auth:
        type: "basic"
        username: "fwd_user"
        password: "fwd_pass"
    topics:
      - "sensors/+/temperature"
      - "devices/#"

Nâng cao (TLS, QoS 1, bật manual ack):

yaml

sources:
  mqtt_secure:
    type: "mqtt"
    broker:
      url: "mqtts://broker.company.com:8883"
      auth:
        type: "none"
      tls:
        ca_cert_path: "/etc/ssl/mqtt/ca-cert.pem"
        insecure: false
    topics:
      - "logs/#"
    subscription_qos: "at_least_once"
    manual_acks: true
    request_queue_capacity: 1000
    source_override: true

Lưu ý

⚠️

`broker` **gần như bắt buộc**: nếu không có block `broker`, collector không có URL broker và báo lỗi `No broker URL found`. Hãy luôn khai báo `broker.url` trỏ tới broker thật.

topics rỗng ([]) nghĩa là chưa subscribe gì — hãy liệt kê topic cần lấy.
request_queue_capacity không để trống thì mặc định runtime là 100.

Kafka source

Đọc message từ một hoặc nhiều topic Kafka.

Cú pháp

yaml

sources:
  kafka_in:
    type: "kafka"
    topics: []
    bootstrap_servers:
      - "localhost:9092"
    group_id: "siem-agent-consumer-group"

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`type`	String	Có	—	Luôn là `"kafka"`
`topics`	Danh sách String	Không	`[]`	Topic cần đọc. Alias: `includes`
`bootstrap_servers`	Danh sách String	Không	`["localhost:9092"]`	Danh sách broker
`group_id`	String	Không	`"siem-agent-consumer-group"`	Consumer group
`client_id`	String	Không	`null`	Client id (tự sinh nếu bỏ trống)
`auto_commit_enabled`	Bool	Không	`true`	Tự commit offset
`auto_commit_interval_ms`	Int (ms)	Không	`5000`	Chu kỳ tự commit
`auto_offset_reset`	String	Không	`"latest"`	`earliest`, `latest`, `none` — dùng khi chưa có offset
`fetch_min_bytes`	Int (bytes)	Không	`1`	Số byte tối thiểu broker trả về mỗi request
`max_partition_fetch_bytes`	Int (bytes)	Không	`1048576`	Byte tối đa mỗi partition mỗi request
`session_timeout_ms`	Int (ms)	Không	`60000`	Session timeout
`heartbeat_interval_ms`	Int (ms)	Không	`10000`	Chu kỳ heartbeat
`max_poll_interval_ms`	Int (ms)	Không	`300000`	Thời gian tối đa giữa hai lần `poll()`
`security`	Object	Không	`type: "none"`	`none`, `tls` hoặc `sasl` (xem kafka sink cho chi tiết block)
`source_override`	Bool	Không	`false`	Ghi đè index/sourcetype downstream
`proxy`	Object	Không	`null`	`{ enable, http, https }` (hỗ trợ có giới hạn)
`extra`	Map String→String	Không	`{}`	Thuộc tính librdkafka truyền thêm, ánh xạ 1:1

Ví dụ

Cơ bản:

yaml

sources:
  kafka_in:
    type: "kafka"
    topics:
      - "logs"
      - "events"
    bootstrap_servers:
      - "kafka1.company.com:9092"
      - "kafka2.company.com:9092"
    group_id: "fwd-consumer"

Nâng cao (đọc từ đầu topic, SASL, extra):

yaml

sources:
  kafka_secure:
    type: "kafka"
    topics:
      - "security-events"
    bootstrap_servers:
      - "kafka1.company.com:9093"
    group_id: "fwd-secure"
    auto_offset_reset: "earliest"
    auto_commit_interval_ms: 2000
    security:
      type: "sasl"
      sasl:
        mechanism: "SCRAM_SHA256"
        username: "fwd"
        password: "secret"
    extra:
      "isolation.level": "read_committed"

Lưu ý

topics rỗng = không đọc gì; hãy khai báo ít nhất một topic.
auto_offset_reset: "none" sẽ báo lỗi khi không tìm thấy offset — chỉ dùng khi bạn đã quản lý offset riêng.

gRPC source

Nhận sự kiện qua embedded gRPC server. Phần transport (cổng, TLS, HTTP/2) nằm ở grpc_server; source này chỉ lo lọc theo topic và định tuyến.

Cú pháp

yaml

sources:
  grpc_in:
    type: "grpc"
    includes:
      - "*"
    index: null
    sourcetype: null
    source_override: false

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`type`	String	Có	—	Luôn là `"grpc"`
`includes`	Danh sách String	Không	`["*"]`	Mẫu lọc theo trường `topic` của request gRPC
`index`	String hoặc `null`	Không	`null`	Index cho dữ liệu
`sourcetype`	String hoặc `null`	Không	`null`	Sourcetype cho dữ liệu
`source_override`	Bool	Không	`false`	Ghi đè index/sourcetype downstream
`extra`	Map String→String	Không	`{}`	Tùy chọn nâng cao cho collector

Ví dụ

Cơ bản (nhận mọi topic):

yaml

sources:
  grpc_in:
    type: "grpc"
    includes:
      - "*"

Nâng cao (chỉ nhận topic bắt đầu bằng logs. hoặc metrics.):

yaml

sources:
  grpc_in:
    type: "grpc"
    includes:
      - "logs.*"
      - "metrics.*"
    index: "grpc_events"
    sourcetype: "grpc_json"

Lưu ý

Pattern * được chuyển thành .* (nhận tất cả). Các pattern còn lại được dịch thành regex và so với topic của request.
Regex không hợp lệ không làm chết tiến trình — pattern đó bị bỏ qua, các pattern khác vẫn chạy.
Source grpc chỉ chạy khi có grpc_server được khai báo.

Embedded servers

Tổng quan và cổng lắng nghe

Forwarder có thể đóng vai trò server trong mạng. Mỗi server chỉ khởi động khi section tương ứng tồn tại trong file cấu hình. TLS có sẵn ở mọi server nhưng tắt mặc định.

Section	Cổng mặc định	Ghi nhớ
`rsyslog_server`	UDP/TCP `514`	Cần có source `rsyslog` để nhận dữ liệu
`snmp_server`	UDP `162` (trap)	Bắt buộc trước khi source `snmp` chạy được
`mqtt_server`	TCP `1883`	Broker MQTT nhúng
`proxy_server`	TCP `8080`	HTTP proxy nhúng
`grpc_server`	TCP `50051`	Chỉ chạy khi có ít nhất một source `grpc`

⚠️

- `proxy_server` mặc định bind `0.0.0.0:8080` — mở ra **mọi interface**, không chỉ localhost. Chỉ expose khi bạn chủ động muốn; nếu chỉ cần nội bộ, đổi thành `127.0.0.1:8080`. - Cổng `8080` rất phổ biến; trước khi bật `proxy_server`, kiểm tra không có service khác (web server, công cụ gỡ lỗi…) đang chiếm cổng này.

rsyslog server

Listener syslog UDP/TCP cho hệ thống gửi log về.

Cú pháp

yaml

rsyslog_server:
  udp_addr: "0.0.0.0:514"
  tcp_addr: "0.0.0.0:514"
  default_source: "rsyslog"

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`udp_addr`	String hoặc `null`	Không	`null` — listener UDP không mở	Địa chỉ bind UDP, ví dụ `"0.0.0.0:514"`. Bỏ trống hoặc đặt `null` = không mở UDP listener
`tcp_addr`	String hoặc `null`	Không	`null` — listener TCP không mở	Địa chỉ bind TCP, ví dụ `"0.0.0.0:514"`. Bỏ trống hoặc đặt `null` = không mở TCP listener
`default_source`	String	Không	`"rsyslog"`	Source id gán cho message không khớp include nào

Ví dụ

Mở cả UDP và TCP:

yaml

rsyslog_server:
  udp_addr: "0.0.0.0:514"
  tcp_addr: "0.0.0.0:514"

Chỉ TCP, giới hạn interface, đổi source mặc định:

yaml

rsyslog_server:
  udp_addr: null
  tcp_addr: "10.0.0.5:514"
  default_source: "network_logs"

Lưu ý

Khi viết section, hãy khai báo rõ udp_addr/tcp_addr bạn muốn — bỏ trống trường thì listener tương ứng không mở (null cũng có nghĩa là tắt).
Port 514 thường yêu cầu quyền đặc biệt (root/CAP_NET_BIND_SERVICE) — hoặc đổi sang port > 1024.
Cần có source type: "rsyslog" để dữ liệu đi vào pipeline.

SNMP server

Cấu hình chung cho listener trap và giới hạn tài nguyên của SNMP.

Cú pháp

yaml

snmp_server:
  trap_addr: "0.0.0.0:162"
  max_datagram_size: 8192
  trap_channel_capacity: 1024
  max_inflight_traps: 2048
  max_poll_concurrency: 64
  default_source: "snmp_default"
  require_v3: true
  warn_insecure_versions: true

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`trap_addr`	String	Không	`"0.0.0.0:162"`	Địa chỉ bind UDP nhận SNMP trap
`max_datagram_size`	Int (bytes)	Không	`8192`	Kích thước datagram tối đa; gói lớn bị cắt
`trap_channel_capacity`	Int	Không	`1024`	Độ sâu kênh trap; đầy thì tạo backpressure
`max_inflight_traps`	Int	Không	`2048`	Số task parse trap chạy song song tối đa
`max_poll_concurrency`	Int	Không	`64`	Số request poll chạy song song tối đa
`default_source`	String	Không	`"snmp_default"`	Source id khi IP không khớp include nào
`require_v3`	Bool	Không	`true`	Từ chối start poller chỉ cấu hình v1/v2c
`warn_insecure_versions`	Bool	Không	`true`	Cảnh báo khi cấu hình v1/v2c

Ví dụ

Cơ bản:

yaml

snmp_server:
  trap_addr: "0.0.0.0:162"

Nâng cao (môi trường legacy cho phép v2c, tăng dung lượng trap):

yaml

snmp_server:
  trap_addr: "10.0.0.5:162"
  max_datagram_size: 16384
  trap_channel_capacity: 4096
  max_inflight_traps: 4096
  max_poll_concurrency: 128
  default_source: "snmp_default"
  require_v3: false
  warn_insecure_versions: true

Lưu ý

⚠️

Section `snmp_server` là **điều kiện tiên quyết** để source `snmp` hoạt động. Không có nó, collector SNMP báo lỗi: `SNMP server config required`.

Section này không có các trường default_index hay default_sourcetype — index/sourcetype khai báo ở SNMP source.
require_v3: true (mặc định) là hành vi nghiêm ngặt: source version: v1/v2c sẽ không được start poller. Đặt false nếu bắt buộc dùng thiết bị cũ.
Port 162 cũng cần quyền đặc biệt hoặc đổi sang port cao.

MQTT server

Broker MQTT nhúng, cho phép thiết bị khác kết nối để gửi/nhận message.

Cú pháp

yaml

mqtt_server:
  router:
    max_connections: 10000
    max_outgoing_packet_count: 200
    max_segment_size: 104857600
    max_segment_count: 10
    shared_subscriptions_strategy: "random"
  server:
    name: "v4-1"
    listen: "0.0.0.0:1883"
    next_connection_delay_ms: 1
    connections:
      connection_timeout_ms: 60000
      max_payload_size: 104857600
      max_inflight_count: 100
      dynamic_filters: true
  auth: []

Các trường cấu hình

Khối server:

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`server.listen`	String	Không	`"0.0.0.0:1883"`	Địa chỉ và cổng lắng nghe
`server.name`	String	Không	`"v4-1"`	Tên instance server
`server.next_connection_delay_ms`	Int (ms)	Không	`1`	Khoảng nghỉ giữa các lần chấp nhận kết nối
`server.tls`	Object	Không	`null`	`{ capath, certpath, keypath }` — `certpath` và `keypath` bắt buộc khi bật block này
`server.connections.connection_timeout_ms`	Int (ms)	Không	`60000`	Timeout kết nối
`server.connections.max_payload_size`	Int (bytes)	Không	`104857600`	Kích thước payload tối đa (100 MB)
`server.connections.max_inflight_count`	Int	Không	`100`	Số message in-flight tối đa mỗi kết nối
`server.connections.dynamic_filters`	Bool	Không	`true`	Cho phép subscribe filter động

Khối router:

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`router.max_connections`	Int	Không	`10000`	Số client đồng thời tối đa
`router.max_outgoing_packet_count`	Int	Không	`200`	Số packet gửi ra xếp hàng mỗi kết nối
`router.max_segment_size`	Int (bytes)	Không	`104857600`	Kích thước segment tối đa (100 MB)
`router.max_segment_count`	Int	Không	`10`	Số segment tối đa mỗi topic filter
`router.shared_subscriptions_strategy`	String	Không	`"random"`	`random`, `roundrobin` hoặc `sticky`
`router.initialized_filters`	Danh sách String	Không	`[]`	Topic filter khởi tạo sẵn khi start
`router.custom_segments`	Danh sách object	Không	`[]`	`{ filter, max_segment_size, max_segment_count }`

Khối auth (toplevel của mqtt_server):

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`auth`	Danh sách object	Không	`[]`	Phương thức xác thực client. `[]` = cho phép anonymous

Ví dụ

Cơ bản (broker anonymous, chỉ đổi địa chỉ):

yaml

mqtt_server:
  server:
    listen: "0.0.0.0:1883"
  auth: []

Nâng cao (yêu cầu username/password, tăng giới hạn):

yaml

mqtt_server:
  auth:
    - type: "basic"
      username: "device"
      password: "secret"
  router:
    max_connections: 50000
    max_outgoing_packet_count: 500
    max_segment_size: 209715200
    max_segment_count: 20
    shared_subscriptions_strategy: "roundrobin"
    initialized_filters:
      - "sensors/#"
  server:
    name: "edge-broker"
    listen: "0.0.0.0:1883"
    connections:
      connection_timeout_ms: 30000
      max_payload_size: 104857600
      max_inflight_count: 200
      dynamic_filters: true

TLS cho broker:

yaml

mqtt_server:
  server:
    listen: "0.0.0.0:8883"
    tls:
      capath: "/etc/ssl/mqtt/ca-cert.pem"
      certpath: "/etc/ssl/mqtt/server-cert.pem"
      keypath: "/etc/ssl/mqtt/server-key.pem"
  auth:
    - type: "basic"
      username: "device"
      password: "secret"

Lưu ý

⚠️

`auth` là **tùy chọn** và mặc định là danh sách rỗng — broker **chấp nhận client ẩn danh, không kiểm tra xác thực nào**. Chỉ bật `mqtt_server` trên mạng tin cậy, hoặc khai báo `auth` với `type: "basic"`.

server.tls yêu cầu đủ certpath + keypath (đồng nghĩa cert_path/key_path); capath là tùy chọn.
Broker chỉ chạy khi section mqtt_server tồn tại.

Proxy server

HTTP proxy nhúng (CONNECT/forward) cho các client nội bộ đi ra ngoài qua Forwarder.

Cú pháp

yaml

proxy_server:
  addr: "0.0.0.0:8080"
  username: null
  password: null
  max_connections: 1000
  connect_timeout_sec: 10
  idle_timeout_sec: 300
  auth_timeout_sec: 30
  keepalive_sec: 60
  nodelay: true
  realm: "Proxy Authentication Required"
  buffer_size: 8192

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`addr`	String	Có	—	Địa chỉ và cổng proxy, ví dụ `"0.0.0.0:8080"`. Bắt buộc: khai báo section `proxy_server` mà thiếu `addr` thì config không load được (`missing field \`addr``)
`username`	String hoặc `null`	Không	`null`	User cho HTTP Basic auth
`password`	String hoặc `null`	Không	`null`	Mật khẩu cho HTTP Basic auth
`max_connections`	Int	Không	`1000`	Số kết nối đồng thời (`0` = không giới hạn)
`connect_timeout_sec`	Int (giây)	Không	`10`	Timeout kết nối upstream (`0` = tắt)
`idle_timeout_sec`	Int (giây)	Không	`300`	Timeout kết nối nhàn rỗi (`0` = tắt)
`auth_timeout_sec`	Int (giây)	Không	`30`	Timeout xác thực client
`keepalive_sec`	Int (giây)	Không	`60`	TCP keepalive (`0` = tắt)
`nodelay`	Bool	Không	`true`	Bật TCP_NODELAY
`realm`	String	Không	`"Proxy Authentication Required"`	Realm hiển thị khi yêu cầu auth
`buffer_size`	Int (bytes)	Không	`8192`	Bộ đệm I/O

Ví dụ

Cơ bản (ẩn trên localhost, không auth):

yaml

proxy_server:
  addr: "127.0.0.1:8080"

Nâng cao (mở cho mạng nội bộ, có xác thực):

yaml

proxy_server:
  addr: "0.0.0.0:3128"
  username: "proxyuser"
  password: "proxypass"
  max_connections: 2000
  connect_timeout_sec: 15
  idle_timeout_sec: 600
  auth_timeout_sec: 20
  keepalive_sec: 30
  nodelay: true
  realm: "Company Proxy"
  buffer_size: 16384

Lưu ý

Xác thực chỉ có hiệu lực khi cả username và password cùng được set; thiếu một trong hai thì proxy chạy không auth.
addr mặc định 0.0.0.0:8080 (mọi interface). Đổi sang 127.0.0.1:... nếu chỉ dùng nội bộ.
Section không hot-reload — đổi phải khởi động lại.

gRPC server

gRPC server nhúng — transport cho gRPC source.

Cú pháp

yaml

grpc_server:
  listen_addr: "0.0.0.0:50051"
  auth:
    header_name: "x-api-key"
    api_keys: []
  reflection:
    enabled: true
    include_health_service: true

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`listen_addr`	String	Không	`"0.0.0.0:50051"`	Địa chỉ và cổng lắng nghe
`auth.header_name`	String	Không	`"x-api-key"`	Tên header mang API key
`auth.api_keys`	Danh sách String	Không	`[]`	Các API key hợp lệ. `[]` = tắt kiểm tra API key
`tls.ca_cert_path`	String	Không	`null`	CA bundle (PEM) để kiểm tra client certificate
`tls.cert_path`	String	Không	`null`	Server certificate (PEM)
`tls.key_path`	String	Không	`null`	Server private key (PEM)
`tls.require_client_cert`	Bool	Không	`false`	Bắt buộc client certificate (mTLS)
`tls.insecure_skip_verify`	Bool	Không	`false`	Bỏ qua xác thực (không khuyến nghị)
`reflection.enabled`	Bool	Không	`true`	Bật gRPC server reflection
`reflection.include_health_service`	Bool	Không	`true`	Expose gRPC health service cùng reflection
`tcp_nodelay`	Bool hoặc `null`	Không	`null`	TCP_NODELAY
`tcp_keepalive_seconds`	Int hoặc `null`	Không	`null`	TCP keepalive (giây)
`http2_keep_alive_interval_secs`	Int hoặc `null`	Không	`null`	Chu kỳ HTTP/2 keep-alive
`http2_keep_alive_timeout_secs`	Int hoặc `null`	Không	`null`	Timeout HTTP/2 keep-alive
`http2_keep_alive_while_idle`	Bool hoặc `null`	Không	`null`	Keep-alive khi không có stream
`initial_stream_window_size`	Int hoặc `null`	Không	`null`	Window size cho stream (bytes)
`initial_connection_window_size`	Int hoặc `null`	Không	`null`	Window size cho connection (bytes)
`max_concurrent_streams`	Int hoặc `null`	Không	`null`	Số stream đồng thời mỗi connection
`max_frame_size`	Int hoặc `null`	Không	`null`	Kích thước frame HTTP/2 tối đa
`http2_adaptive_window`	Bool hoặc `null`	Không	`null`	Adaptive flow control
`concurrency_limit_per_connection`	Int hoặc `null`	Không	`null`	Số request đồng thời mỗi connection
`max_connection_age_secs`	Int hoặc `null`	Không	`null`	Tuổi tối đa của connection (giây)
`max_connection_age_grace_secs`	Int hoặc `null`	Không	`null`	Grace period khi đóng connection
`max_decoding_message_size`	Int hoặc `null`	Không	`null`	Kích thước message decode tối đa (bytes)
`max_encoding_message_size`	Int hoặc `null`	Không	`null`	Kích thước message encode tối đa (bytes)
`timeout_secs`	Int hoặc `null`	Không	`null`	Timeout mềm mỗi request (giây)
`extra`	Map String→String	Không	`{}`	Tùy chọn transport nâng cao

Ví dụ

Cơ bản:

yaml

grpc_server:
  listen_addr: "0.0.0.0:50051"

Nâng cao (TLS + API key, tắt reflection):

yaml

grpc_server:
  listen_addr: "0.0.0.0:50051"
  tls:
    ca_cert_path: "/etc/ssl/grpc/ca-cert.pem"
    cert_path: "/etc/ssl/grpc/server-cert.pem"
    key_path: "/etc/ssl/grpc/server-key.pem"
    require_client_cert: false
    insecure_skip_verify: false
  auth:
    header_name: "x-api-key"
    api_keys:
      - "key-one"
      - "key-two"
  reflection:
    enabled: false
    include_health_service: false
  max_concurrent_streams: 100
  max_decoding_message_size: 4194304

Lưu ý

⚠️

- Server chỉ chạy khi có **ít nhất một source `type: "grpc"`** trong `sources`. - `api_keys: []` (mặc định) nghĩa là **không kiểm tra API key** — chỉ bật auth bằng cách điền key vào danh sách. - `reflection.enabled` và `reflection.include_health_service` đều mặc định `true`, tức là **gRPC health service được expose** sẵn để load balancer/monitoring dò trạng thái.

TLS có sẵn nhưng tắt mặc định — muốn bật phải khai báo tls.cert_path + tls.key_path.
Section không hot-reload — đổi phải khởi động lại.

Transforms

Tổng quan transforms

transforms là map đặt tên: khóa là identifier, giá trị là object có trường type chọn loại transform. Pipeline dạng graph: source → transform → transform → sink.

yaml

transforms:
  mask:
    type: "filter"
    identifier: "mask"
    inputs: ["syslog_in"]
    condition: 'exists(.msg)'

Mọi transform đều có ba trường phẳng ở top level (không có block bọc ngoài):

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`type`	String	Có	—	`filter`, `remap`, `route`, `dedupe`, `reduce`, `throttle`, `ingestion_rule`
`identifier`	String	Không	lấy từ khóa map	Định danh; nên để trống/lấy từ khóa, không cần khai báo
`inputs`	Danh sách String	Không	`null`	Nguồn vào: id source, id transform khác, hoặc tên route
`enabled`	Bool	Không	`true`	`false` = transform bị bỏ qua, sự kiện đi qua nguyên trạng

⚠️

Ba trường `identifier` / `inputs` / `enabled` được **flatten ra top level** của transform. Đừng bọc chúng trong block `common:` — YAML kiểu đó bị bỏ qua lặng lẽ và `inputs` sẽ mất.

Kiểm tra graph lúc khởi động và mỗi lần reload:

inputs trỏ tới id không tồn tại → bị từ chối: Transform '<id>' inputs unknown id '<input>'. Valid: [...].
Graph có vòng → bị từ chối: Cycle detected in transforms graph!.
Khóa transform không được là * hay chuỗi rỗng.

filter

Giữ lại (hoặc loại bỏ) sự kiện theo một biểu kiện VRL.

Cú pháp

yaml

transforms:
  only_errors:
    type: "filter"
    inputs: ["syslog_in"]
    condition: 'includes(to_string(.level), "error")'

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`condition`	String (VRL)	Có	—	Biểu kiện VRL; sự kiện thỏa mãn thì được giữ
`inputs`	Danh sách String	Không	`null`	Nguồn vào
`enabled`	Bool	Không	`true`	Bật/tắt transform

Ví dụ

Cơ bản:

yaml

transforms:
  drop_debug:
    type: "filter"
    inputs: ["syslog_in"]
    condition: '.level != "debug"'

Nâng cao (chỉ giữ sự kiện từ dải mạng cụ thể):

yaml

transforms:
  internal_only:
    type: "filter"
    inputs: ["syslog_in", "grpc_in"]
    condition: 'match(to_string(.ip), r"^10\.|^192\.168\.")'
    enabled: true

Lưu ý

⚠️

Nếu `condition` **không biên dịch được**, **mọi sự kiện đều bị drop** (không phải đi qua nguyên trạng). Hãy kiểm tra kỹ cú pháp VRL trước khi deploy.

remap

Chuyển đổi trường của sự kiện bằng VRL.

Cú pháp

yaml

transforms:
  normalize:
    type: "remap"
    inputs: ["syslog_in"]
    source: |
      .sourcetype = "syslog_rfc5424"
      .env = "production"

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`source`	String (VRL)	Một trong ba	`null`	Program VRL inline
`file`	String (đường dẫn)	Một trong ba	`null`	Đường dẫn một file VRL
`files`	Danh sách String	Một trong ba	`null`	Nhiều file VRL, chạy tuần tự
`drop_on_abort`	Bool	Không	`true`	Drop sự kiện bị abort lúc xử lý
`drop_on_error`	Bool	Không	`false`	Drop sự kiện gây lỗi runtime
`inputs`	Danh sách String	Không	`null`	Nguồn vào
`enabled`	Bool	Không	`true`	Bật/tắt transform

Ví dụ

Cơ bản:

yaml

transforms:
  add_fields:
    type: "remap"
    inputs: ["syslog_in"]
    source: |
      .collector = "fwd-edge-01"
      .datacenter = "dc1"

Nâng cao (parse JSON, giữ file VRL riêng, drop khi lỗi):

yaml

transforms:
  parse_body:
    type: "remap"
    inputs: ["kafka_in"]
    files:
      - "./vrl/parse_json.vrl"
      - "./vrl/enrich.vrl"
    drop_on_error: true
    drop_on_abort: true

Lưu ý

Nếu program VRL không biên dịch được, sự kiện đi qua nguyên trạng (không bị drop). Lỗi biên dịch được ghi nhận nhưng pipeline vẫn chạy.
drop_on_error / drop_on_abort chỉ điều khiển drop ở runtime (lỗi/abort khi thực thi).
Đúng một trong source, file, files.

route

Phân loại sự kiện theo điều kiện VRL; mỗi route là một kênh có tên mà transform/sink phía sau có thể tham chiếu.

Cú pháp

yaml

transforms:
  split_by_level:
    type: "route"
    inputs: ["syslog_in"]
    routes:
      errors: '.level == "error"'
      others: 'true'

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`routes`	Map tên → String (VRL)	Có	—	Tên route và điều kiện của nó
`inputs`	Danh sách String	Không	`null`	Nguồn vào
`enabled`	Bool	Không	`true`	Bật/tắt transform

Ví dụ

Cơ bản như trên. Nâng cao (route rồi tách nhánh xử lý riêng):

yaml

transforms:
  classify:
    type: "route"
    inputs: ["rsyslog_net"]
    routes:
      auth_failures: 'match(.msg, r"(?i)failed password|authentication failure")'
      normal: 'true'

Sử dụng tên route làm inputs của transform/sink phía sau:

yaml

transforms:
  tag_failures:
    type: "remap"
    inputs: ["auth_failures"]
    source: |
      .alert_priority = "high"

Lưu ý

Tên route là node hợp lệ trong graph: transform/sink phía sau dùng tên đó làm inputs.
Điều kiện chạy theo thứ tự — đặt route "bắt tất cả" (vd true) ở cuối.
Tên route không được trùng id source/transform khác.

dedupe

Loại bỏ sự kiện trùng lặp trong một khoảng thời gian.

Cú pháp

yaml

transforms:
  dedupe_raw:
    type: "dedupe"
    inputs: ["syslog_in"]
    fields:
      match:
        - "raw"
    window:
      window_ms: 60000
      refresh_on_drop: false

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`fields.match`	Danh sách String	Không	`["raw"]`	Các trường dùng để so trùng. Loại trừ `ignore`
`fields.ignore`	Danh sách String	Không	`null`	Các trường bị bỏ qua khi so trùng (loại trừ `match`)
`window.window_ms`	Int (ms)	Không	`60000`	Cửa sổ thời gian dedupe
`window.refresh_on_drop`	Bool	Không	`false`	Reset bộ đếm tuổi khi drop bản trùng
`inputs`	Danh sách String	Không	`null`	Nguồn vào
`enabled`	Bool	Không	`true`	Bật/tắt transform

Giá trị hợp lệ của fields.match: raw, index, source_identifier, ip, log_source, device_id, published_at, sourcetype, tenant, tenant_prefix, và extra.<key> (vd extra.host).

Ví dụ

Cơ bản (dùng mặc định ["raw"]):

yaml

transforms:
  dedupe_raw:
    type: "dedupe"
    inputs: ["syslog_in"]

Nâng cao (so theo IP + sourcetype, cửa sổ 5 phút, bỏ qua host):

yaml

transforms:
  dedupe_conn:
    type: "dedupe"
    inputs: ["rsyslog_net"]
    fields:
      match:
        - "ip"
        - "sourcetype"
    window:
      window_ms: 300000
      refresh_on_drop: true

Lưu ý

⚠️

Mặc định của `fields.match` là **`["raw"]`** (không phải `["_raw"]`). Nếu khai sai tên trường, **mọi sự kiện đều bị coi là trùng** và bị drop hàng loạt.

match và ignore loại trừ nhau — chỉ dùng một trong hai.

reduce

Gộp nhiều sự kiện thành một sự kiện theo transaction.

Cú pháp

yaml

transforms:
  reduce_session:
    type: "reduce"
    inputs: ["rsyslog_net"]
    condition: 'exists(.event_end)'
    group_by: .session_id
    time: 30000

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`condition`	String (VRL)	Có	—	Biểu kiện nhận diện sự kiện cuối của một transaction
`group_by`	String (VRL)	Không	`null`	Biểu thức VRL sinh khóa gộp. Nếu thiếu, dùng tổ hợp `source_identifier` + `ip` (khóa mặc định: `source_identifier:ip`)
`time`	Int (ms)	Không	`null`	Flush theo thời gian: nhóm bị đẩy ra sau khoảng này
`merge_strategies`	Map tên trường → String	Không	`{}`	Cách gộp từng trường
`inputs`	Danh sách String	Không	`null`	Nguồn vào
`enabled`	Bool	Không	`true`	Bật/tắt transform

Giá trị hợp lệ của merge_strategies: array, concat, concat_newline, concat_raw, discard, flat_unique, longest_array, max, min, retain, shortest_array, sum, first, last.

Ví dụ

Cơ bản:

yaml

transforms:
  reduce_session:
    type: "reduce"
    inputs: ["rsyslog_net"]
    condition: 'exists(.session_end)'

Nâng cao (gộp theo session_id, timeout 30s, merge nhiều trường):

yaml

transforms:
  reduce_session:
    type: "reduce"
    inputs: ["rsyslog_net"]
    condition: 'exists(.session_end)'
    group_by: .session_id
    time: 30000
    merge_strategies:
      bytes_out: "sum"
      msg: "concat_newline"
      first_seen: "first"
      last_seen: "last"

Lưu ý

⚠️

`group_by` là **biểu thức VRL thô**, **không** phải template `{{ … }}`. Viết `group_by: "{{ session_id }}"` sẽ không biên dịch được và hệ thống **âm thầm rơi về khóa mặc định `source_identifier:ip`** — mọi gộp session sẽ sai mà không báo lỗi cấu hình.

time tính bằng ms; đặt hợp lý để tránh giữ nhóm quá lâu.

throttle

Giới hạn số sự kiện theo thời gian (rate limiting).

Cú pháp

yaml

transforms:
  limit_ip:
    type: "throttle"
    inputs: ["rsyslog_net"]
    threshold: 100
    window_ms: 60000
    key_field: "{{ ip }}"

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`threshold`	Int	Có	—	Số sự kiện tối đa được phép trong một cửa sổ
`window_ms`	Int (ms)	Có	—	Cửa sổ thời gian
`key_field`	String (template)	Không	`null`	Template `{{ <expr> }}` để gom bucket riêng, vd `{{ ip }}` → field `.ip`
`exclude`	String (VRL)	Không	`null`	Biểu kiện VRL thô: sự kiện thỏa mãn được bỏ qua rate limit
`inputs`	Danh sách String	Không	`null`	Nguồn vào
`enabled`	Bool	Không	`true`	Bật/tắt transform

Ví dụ

Cơ bản (toàn bộ stream tối đa 100 sự kiện/phút):

yaml

transforms:
  limit_global:
    type: "throttle"
    inputs: ["syslog_in"]
    threshold: 100
    window_ms: 60000

Nâng cao (giới hạn theo IP, không chặn traffic khẩn cấp):

yaml

transforms:
  limit_ip:
    type: "throttle"
    inputs: ["rsyslog_net"]
    threshold: 50
    window_ms: 10000
    key_field: "{{ ip }}"
    exclude: '.severity == "critical"'

Lưu ý

key_field dùng template {{ … }}, còn exclude là VRL thô — hai kiểu cú pháp khác nhau, đừng hoán đổi.
Không có threshold/window_ms thì cấu hình không parse được.

ingestion rule

Đánh giá sự kiện theo tập điều kiện; sự kiện khớp có thể được chuyển vào alert sink để phát cảnh báo.

Cú pháp

yaml

transforms:
  brute_force_rule:
    type: "ingestion_rule"
    inputs: ["rsyslog_net"]
    conditions:
      - field: "msg"
        operator: "contains"
        value: "Failed password"
    match_logic: "any"
    alert:
      rule_id: "6ba7b810-9dad-11d1-80b4-00c04fd430c8"
      severity: "high"
      title: "Brute force detected"
      throttle:
        window: "1m"
        group_by: "ip"

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`conditions`	Danh sách object	Có	—	Các điều kiện so sánh (xem bảng dưới)
`match_logic`	String	Không	`"any"`	`any` (một điều kiện đúng là đủ) hoặc `all`
`alert`	Object	Không	`null`	Khối phát cảnh báo cho sự kiện khớp
`alert.rule_id`	String	Khi có `alert`	—	UUID của rule phía backend, không được rỗng
`alert.severity`	String	Không	`null`	Mức độ cảnh báo (backend tự ánh xạ)
`alert.title`	String	Không	`null`	Tiêu đề cảnh báo
`alert.throttle.window`	String (duration)	Không	`null`	Cửa sổ chặn lặp, vd `"1m"`, `"30s"` — phải > 0
`alert.throttle.group_by`	String (tên field)	Không	`null`	Gom bucket theo field; thiếu thì mọi event chung một bucket
`inputs`	Danh sách String	Không	`null`	Nguồn vào
`enabled`	Bool	Không	`true`	Bật/tắt transform

Mỗi điều kiện (FieldCondition):

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`field`	String	Có	—	Tên field cần so
`operator`	String	Không	`"contains"`	`eq`, `ne`, `gt`, `lt`, `contains`, `regex`
`value`	String	Với `eq`/`ne`/`gt`/`lt`/`contains`	`null`	Giá trị literal. Với `gt`/`lt` bắt buộc là số
`pattern`	String (regex)	Với `regex`	`null`	Regex thô. Khi có `pattern`, `value` bị bỏ qua

Ví dụ

Cơ bản (không phát alert, chỉ đánh dấu sự kiện):

yaml

transforms:
  mark_ssh_fail:
    type: "ingestion_rule"
    inputs: ["rsyslog_net"]
    conditions:
      - field: "msg"
        value: "Failed password"
    match_logic: "any"

Nâng cao (regex + điều kiện số + alert có throttle):

yaml

transforms:
  external_scan:
    type: "ingestion_rule"
    inputs: ["rsyslog_net", "grpc_in"]
    conditions:
      - field: "ip"
        operator: "regex"
        pattern: '^(?!10\.|192\.168\.)\d+\.\d+\.\d+\.\d+$'
      - field: "count"
        operator: "gt"
        value: "100"
    match_logic: "all"
    alert:
      rule_id: "6ba7b810-9dad-11d1-80b4-00c04fd430c8"
      severity: "critical"
      title: "Scan từ mạng ngoài"
      throttle:
        window: "5m"
        group_by: "ip"

Lưu ý

⚠️

Khai báo khối `alert` nghĩa là transform này **cần một sink `type: "alert"`** để phát cảnh báo ra ngoài. Kiểm tra này chạy lúc push cấu hình lên Registry — thiếu alert sink sẽ bị từ chối.

alert không có nghĩa là drop: sự kiện vẫn đi tiếp trong pipeline, chỉ được phát thêm một bản sao vào kênh alert.
Điều kiện gt/lt với value không phải số bị từ chối ngay lúc validate; regex hỏng cũng bị từ chối kèm thông báo lỗi.
Block alert yêu cầu ít nhất một conditions (chống alert storm).

Sinks

Tổng quan sinks

sinks là map đặt tên: khóa là identifier của sink, giá trị là object có trường type.

`type`	Sink	Ghi chú
`grpc`	gRPC shipper	Alias chấp nhận được: `forwarder_grpc`
`kafka`	Kafka producer	Alias: `forwarder_kafka`
`mqtt`	MQTT publisher	Alias: `forwarder_mqtt`
`blackhole`	BlackHole / OpenSearch index	Triển khai là OpenSearch-compatible
`file`	Ghi ra file local	Gỡ lỗi / lưu trữ cục bộ
`alert`	Phát cảnh báo lên Registry	Không có web UI đi kèm

⚠️

**`rate_limit` không phải block lồng nhau.** Ba trường `batch_size`, `batch_interval`, `batch_max_bytes` được **flatten ra top level** của sink. Một số tài liệu/ví dụ cũ viết chúng trong một block `rate_limit:` lồng dưới sink — block đó **bị bỏ qua lặng lẽ** và sink chạy với giá trị mặc định. Cú pháp đúng là đặt thẳng ba trường bên cạnh `type`, `url`, `inputs`:

yaml

sinks:
  my_sink:
    type: "mqtt"
    url: "mqtt://broker:1883"
    auth: { type: "none" }
    inputs: ["*"]
    batch_size: 100
    batch_interval: 1000
    batch_max_bytes: 10485760

Mặc định của ba trường batch: batch_size 500, batch_interval 1000 (ms), batch_max_bytes 10485760 (10 MB). Áp dụng cho sink MQTT, gRPC, BlackHole/OpenSearch và File.

Một số trường dùng chung:

Trường	Kiểu	Mặc định	Mô tả
`inputs`	Danh sách String	`["*"]`	Nguồn vào (id source/transform). `` = nhận tất cả. Alias: `includes`*
`batch_size`	Int	`500`	Số message tối đa mỗi lô
`batch_interval`	Int (ms)	`1000`	Thời gian chờ gửi lô
`batch_max_bytes`	Int (bytes)	`10485760`	Kích thước lô tối đa

grpc sink

Gửi sự kiện tới một endpoint gRPC (thường là BlackHole server).

Cú pháp

yaml

sinks:
  upstream:
    type: "grpc"
    url: "http://blackhole-server:50051"
    inputs: ["*"]
    batch_size: 500
    batch_interval: 1000
    batch_max_bytes: 10485760

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`url`	String (URL)	Có	—	Endpoint gRPC đầy đủ scheme/host/port. Alias: `endpoint`
`inputs`	Danh sách String	Không	`["*"]`	Nguồn vào (alias `includes`)
`tls.ca_cert_path`	String	Không	`null`	CA bundle (PEM)
`tls.client_cert_path`	String	Không	`null`	Client certificate cho mTLS
`tls.client_key_path`	String	Không	`null`	Client key cho mTLS
`tls.domain_name`	String	Không	`null`	Ghi đè SNI/hostname verification
`tls.insecure_skip_verify`	Bool	Không	`false`	Bỏ qua xác thực (không khuyến nghị)
`connect_timeout_secs`	Int (giây)	Không	`null`	Timeout kết nối TCP
`request_timeout_secs`	Int (giây)	Không	`null`	Timeout mỗi RPC. Alias: `timeout_secs`
`tcp_keepalive_secs`	Int (giây)	Không	`null`	TCP keepalive
`tcp_nodelay`	Bool	Không	`null`	TCP_NODELAY
`http2_keep_alive_interval_secs`	Int	Không	`null`	HTTP/2 keep-alive interval
`http2_keep_alive_timeout_secs`	Int	Không	`null`	HTTP/2 keep-alive timeout
`http2_keep_alive_while_idle`	Bool	Không	`null`	Keep-alive khi idle
`initial_stream_window_size`	Int	Không	`null`	Window size stream
`initial_connection_window_size`	Int	Không	`null`	Window size connection
`http2_adaptive_window`	Bool	Không	`null`	Adaptive flow control
`concurrency_limit`	Int	Không	`null`	Giới hạn đồng thời của client
`max_concurrent_requests`	Int	Không	`null`	Số request Publish tối đa mỗi batch (mặc định: 1 request bay tại một thời điểm)
`buffer_size`	Int	Không	`null`	Bộ đệm nội bộ của service
`user_agent`	String	Không	`null`	Header User-Agent
`message_limits.max_decoding_message_size`	Int	Không	`null`	Giới hạn decode (bytes)
`message_limits.max_encoding_message_size`	Int	Không	`null`	Giới hạn encode (bytes)
`use_streaming`	Bool	Không	`true`	Dùng `PublishStream` (true) hay `Publish` đơn (false)
`proxy`	Object	Không	`null`	`{ enable, http, https }`
`headers`	Map String→String	Không	`{}`	Header tùy chọn gửi kèm (thành gRPC metadata)
`reflection.enabled`	Bool	Không	`false`	Dùng server reflection của endpoint
`reflection.timeout_secs`	Int	Không	`null`	Timeout cho reflection RPC
`proto.path`	String	Không	`null`	Đường dẫn `.proto`/descriptor set
`proto.include_paths`	Danh sách String	Không	`[]`	Include path khi resolve import
`proto.service`	String	Không	`null`	Tên service đầy đủ, vd `mypkg.LogService`
`proto.method`	String	Không	`null`	Tên method, vd `SendLog`
`batch_size` / `batch_interval` / `batch_max_bytes`	Int	Không	`500` / `1000` / `10485760`	Batch (phẳng, không lồng block)

Ví dụ

Cơ bản:

yaml

sinks:
  upstream:
    type: "grpc"
    url: "http://blackhole-server:50051"

Nâng cao (mTLS, streaming tắt, giới hạn đồng thời):

yaml

sinks:
  upstream_secure:
    type: "grpc"
    endpoint: "https://blackhole-server:443"
    inputs: ["syslog_in", "grpc_in"]
    tls:
      ca_cert_path: "/etc/ssl/grpc/ca-cert.pem"
      client_cert_path: "/etc/ssl/grpc/client-cert.pem"
      client_key_path: "/etc/ssl/grpc/client-key.pem"
      insecure_skip_verify: false
    connect_timeout_secs: 10
    request_timeout_secs: 30
    use_streaming: true
    max_concurrent_requests: 4
    batch_size: 1000
    batch_interval: 500
    batch_max_bytes: 10485760
    headers:
      x-environment: "production"

Lưu ý

url bắt buộc lúc parse — http://localhost:50051 chỉ là giá trị mẫu trong code, không phải default đọc từ YAML. Thiếu url thì file cấu hình không load được.
Chỉ định endpoint hay url đều được (hai tên cho cùng một trường).

kafka sink

Publish sự kiện vào Kafka.

Cú pháp

yaml

sinks:
  kafka_out:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9092"
    inputs: ["*"]

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`bootstrap_servers`	Danh sách String	Có	—	Danh sách broker (không có default đọc từ YAML)
`inputs`	Danh sách String	Không	`["*"]`	Nguồn vào (alias `includes`)
`client_id`	String	Không	`null`	Client id producer
`security`	Object	Không	`type: "none"`	`none`, `tls`, `sasl` (xem bảng dưới)
`acks`	String	Không	`"leader"`	`none`, `leader`, `all`
`compression`	String	Không	`"none"`	`none`, `gzip`, `snappy`, `lz4`, `zstd`
`enable_idempotence`	Bool	Không	`false`	Producer idempotent (cần `acks: "all"`)
`transactional_id`	String	Không	`null`	Transaction id (exactly-once)
`batching.linger_ms`	Int (ms)	Không	`0`	Chờ trước khi gửi batch
`batching.batch_num_messages`	Int	Không	`100000`	Số message tối đa mỗi batch
`batching.batch_kbytes`	Int (KiB)	Không	`1048576`	Ngưỡng size queue (KiB) gửi batch
`retries.max_retries`	Int	Không	`10`	Số lần retry
`retries.backoff_ms`	Int (ms)	Không	`100`	Khoảng nghỉ giữa các lần retry
`timeouts.socket_timeout_ms`	Int (ms)	Không	`60000`	Socket timeout
`timeouts.request_timeout_ms`	Int (ms)	Không	`30000`	Request timeout
`timeouts.message_timeout_ms`	Int (ms)	Không	`300000`	Delivery timeout
`timeouts.connections_max_idle_ms`	Int (ms)	Không	`300000`	Idle timeout kết nối
`proxy`	Object	Không	`null`	`{ enable, http, https }`
`producer_pool_size`	Int (1–64)	Không	`4`	Số producer trong pool
`extra`	Map String→String	Không	`{}`	Thuộc tính librdkafka truyền thêm
`batch_size` / `batch_interval` / `batch_max_bytes`	Int	Không	`500` / `1000` / `10485760`	Batch (phẳng)

Block security:

`type`	Trường con	Nội dung
`none`	—	Không TLS, không SASL
`tls`	`tls:`	Chỉ mã hóa; block `tls` xem bảng dưới
`sasl`	`sasl:`	Xác thực SASL; `mechanism`: `PLAIN`, `SCRAM_SHA256`, `SCRAM_SHA512`, `GSSAPI`, `OAUTHBEARER`

Trường `tls`	Kiểu	Mặc định	Mô tả
`ca_location`	String	`null`	CA bundle (PEM)
`certificate_location`	String	`null`	Client certificate (mTLS)
`key_location`	String	`null`	Client key (mTLS)
`key_password`	String	`null`	Mật khẩu key đã mã hóa
`verify_certificate`	Bool	`true`	Kiểm tra chuỗi chứng chỉ

Trường `sasl`	Kiểu	Mặc định	Mô tả
`mechanism`	String	`"PLAIN"`	Xem danh sách mechanism ở trên
`username` / `password`	String	`null`	Dùng cho `PLAIN`/`SCRAM`
`gssapi.principal`	String	—	Bắt buộc khi `mechanism: "GSSAPI"`
`gssapi.keytab`	String	—	Đường dẫn keytab
`gssapi.service_name`	String	`"kafka"`	Service name
`gssapi.krb5_config`	String	`null`	Đường dẫn krb5.conf
`oauthbearer.token`	String	`null`	Token tĩnh
`oauthbearer.oidc_config`	String	`null`	Chuỗi cấu hình OIDC của librdkafka
`tls`	Object	`null`	Có mặt = SASL over TLS (SASL_SSL); không có = SASL_PLAINTEXT

Ví dụ

Cơ bản:

yaml

sinks:
  kafka_out:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9092"
      - "kafka2.company.com:9092"
    inputs: ["*"]

Nâng cao (SASL + TLS, idempotent, batch lớn):

yaml

sinks:
  kafka_secure:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9093"
    inputs: ["syslog_in"]
    security:
      type: "sasl"
      sasl:
        mechanism: "SCRAM_SHA512"
        username: "fwd"
        password: "secret"
        tls:
          ca_location: "/etc/ssl/certs/kafka-ca.pem"
          verify_certificate: true
    acks: "all"
    compression: "lz4"
    enable_idempotence: true
    batching:
      linger_ms: 5
      batch_num_messages: 50000
    retries:
      max_retries: 20
      backoff_ms: 200
    producer_pool_size: 8
    batch_size: 1000
    batch_interval: 500
    batch_max_bytes: 10485760

Lưu ý

bootstrap_servers bắt buộc lúc parse; localhost:9092 chỉ là giá trị mẫu trong Default, không đọc được từ YAML khi thiếu trường.
Giá trị SASL dùng dạng SCREAMING_SNAKE_CASE: PLAIN, SCRAM_SHA256, SCRAM_SHA512, GSSAPI, OAUTHBEARER.

mqtt sink

Publish sự kiện tới broker MQTT.

Cú pháp

yaml

sinks:
  mqtt_out:
    type: "mqtt"
    url: "mqtt://broker.company.com:1883"
    auth:
      type: "none"
    inputs: ["*"]

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`url`	String (URL)	Có	—	URL broker, `mqtt://` hoặc `mqtts://`
`auth`	Object	Có	—	`{"type":"none"}` hoặc `{"type":"basic","username":...,"password":...}`
`inputs`	Danh sách String	Không	`["*"]`	Nguồn vào (alias `includes`)
`proxy`	Object	Không	`null`	`{ enable, http, https }`
`tls.ca_cert_path`	String	Không	`null`	CA certificate (PEM). Alias field: `capath`
`tls.client_cert_path`	String	Không	`null`	Client certificate (mTLS). Alias: `certpath`
`tls.client_key_path`	String	Không	`null`	Client key (mTLS). Alias: `keypath`
`tls.insecure`	Bool	Không	`false`	Bỏ qua xác thực chứng chỉ (chỉ để test)
`batch_size` / `batch_interval` / `batch_max_bytes`	Int	Không	`500` / `1000` / `10485760`	Batch (phẳng)

Ví dụ

Cơ bản:

yaml

sinks:
  mqtt_out:
    type: "mqtt"
    url: "mqtt://broker.company.com:1883"
    auth:
      type: "basic"
      username: "fwd"
      password: "secret"

Nâng cao (TLS + mTLS, batch nhỏ cho latency thấp):

yaml

sinks:
  mqtt_secure:
    type: "mqtt"
    url: "mqtts://broker.company.com:8883"
    auth:
      type: "none"
    tls:
      ca_cert_path: "/etc/ssl/mqtt/ca-cert.pem"
      client_cert_path: "/etc/ssl/mqtt/client-cert.pem"
      client_key_path: "/etc/ssl/mqtt/client-key.pem"
      insecure: false
    inputs: ["syslog_in"]
    batch_size: 100
    batch_interval: 500
    batch_max_bytes: 1048576

Lưu ý

url và auth là hai trường bắt buộc — thiếu một trong hai thì cấu hình không parse được.
tls field dùng tên ca_cert_path/client_cert_path/client_key_path (đồng nghĩa capath/certpath/keypath).

opensearch (blackhole) sink

Ghi sự kiện vào BlackHole/OpenSearch-compatible index. Đây là sink type: "blackhole".

Cú pháp

yaml

sinks:
  index_out:
    type: "blackhole"
    url: "https://blackhole.company.com:9200"
    inputs: ["*"]

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`url`	String (URL)	Có	—	Endpoint cluster. Alias: `endpoint`
`inputs`	Danh sách String	Không	`["*"]`	Nguồn vào (alias `includes`)
`healthcheck`	Bool	Không	`false`	Gửi health check định kỳ
`request.retry_attempts`	Int	Không	`null`	Số lần retry (tương thích ngược)
`request.timeout_secs`	Int (giây)	Không	`null`	Timeout request (tương thích ngược)
`auth`	Object	Không	`type: "none"`	`none`, `basic`, `clientcert`, `jwt`, `awssigv4` (xem bảng dưới)
`tls.ca_cert_path`	String	Không	`null`	CA bundle (PEM)
`tls.insecure_skip_verify`	Bool	Không	`false`	Bỏ qua xác thực TLS
`proxy`	Object	Không	`null`	`{ enable, http, https }`
`headers`	Map String→String	Không	`{}`	Header HTTP tùy chọn
`bulk_doc_metadata_overhead_bytes`	Int (bytes)	Không	`128`	Ước tính overhead metadata mỗi document trong bulk
`bulk_max_docs_hard_cap`	Int	Không	`5000`	Trần số document mỗi request bulk
`bulk_max_bytes_hard_cap`	Int (bytes)	Không	`67108864` (64 MiB)	Trần byte mỗi request bulk
`timeouts.request_timeout_secs`	Int (giây)	Không	`30`	Timeout request
`timeouts.connect_timeout_secs`	Int (giây)	Không	`10`	Timeout kết nối
`batch_size` / `batch_interval` / `batch_max_bytes`	Int	Không	`500` / `1000` / `10485760`	Batch (phẳng)

Block auth:

`type`	Trường con	Mô tả
`none`	—	Không xác thực
`basic`	`username`, `password`	Basic auth
`clientcert`	`pkcs12_path`, `pkcs12_password`	Client cert PKCS#12/PFX
`jwt`	`token`, `header_name` (mặc định `Authorization`)	Bearer token
`awssigv4`	`region`, `profile?`, `role_arn?`, `service` (`es`/`aoss`)	Ký SigV4 cho AWS

Ví dụ

Cơ bản:

yaml

sinks:
  index_out:
    type: "blackhole"
    url: "https://blackhole.company.com:9200"
    auth:
      type: "basic"
      username: "admin"
      password: "secret"

Nâng cao (basic + custom CA, bulk lớn hơn, header riêng):

yaml

sinks:
  index_bulk:
    type: "blackhole"
    url: "https://opensearch.company.com:9200"
    inputs: ["syslog_in", "grpc_in"]
    healthcheck: true
    auth:
      type: "basic"
      username: "ingest"
      password: "secret"
    tls:
      ca_cert_path: "/etc/ssl/certs/company-ca.pem"
      insecure_skip_verify: false
    headers:
      X-API-Version: "1.0"
    bulk_max_docs_hard_cap: 1000
    timeouts:
      request_timeout_secs: 60
      connect_timeout_secs: 15
    batch_size: 1000
    batch_interval: 1000
    batch_max_bytes: 20971520

Lưu ý

Kiểu sink là blackhole — đó là tên type duy nhất được chấp nhận cho sink OpenSearch-compatible này.
bulk_max_docs_hard_cap / bulk_max_bytes_hard_cap là trần cứng: số effectively dùng là min(batch_size, hard_cap) và min(batch_max_bytes, hard_cap).

file sink

Ghi sự kiện ra file local — tiện cho gỡ lỗi và lưu trữ tạm.

Cú pháp

yaml

sinks:
  debug_out:
    type: "file"
    path: "./debug-events/"
    inputs: ["*"]

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`path`	String	Không	`"./debug-events/"`	Thư mục ghi file (tương đối so với thư mục làm việc)
`max_size_mb`	Int (MB)	Không	`10`	Kích thước file trước khi rotate
`max_files`	Int	Không	`5`	Số file đã rotate giữ lại
`inputs`	Danh sách String	Không	`["*"]`	Nguồn vào (alias `includes`)
`batch_size` / `batch_interval` / `batch_max_bytes`	Int	Không	`500` / `1000` / `10485760`	Batch (phẳng)

Ví dụ

Cơ bản:

yaml

sinks:
  debug_out:
    type: "file"
    path: "./debug-events/"

Nâng cao (thư mục tuyệt đối, giữ nhiều file hơn):

yaml

sinks:
  archive_out:
    type: "file"
    path: "/var/log/blackhole-fwd/events/"
    max_size_mb: 50
    max_files: 20
    inputs: ["syslog_in"]
    batch_size: 2000
    batch_interval: 2000

Lưu ý

Mặc định ghi vào ./debug-events/ — đây là thư mục gỡ lỗi, không phải đích sản xuất.
Tiến trình cần quyền ghi vào path đã khai báo.

alert sink

Phát cảnh báo (sinh bởi ingestion_rule) về Registry qua POST {api_url}/client/alerts.

Cú pháp

yaml

sinks:
  alerts:
    type: "alert"
    inputs: ["alerts"]

Các trường cấu hình

Trường	Kiểu	Bắt buộc	Mặc định	Mô tả
`inputs`	Danh sách String	Không	`null`	Kênh vào. Sự kiện alert đi vào kênh dành riêng `"alerts"` nên giá trị chuẩn là `["alerts"]`; thiếu trường này shipper tự subscribes kênh đó. Alias: `includes`
`max_retry`	Int	Không	`null`	Số lần thử lại tối đa khi gửi lỗi tạm thời (5xx/mạng)
`timeout_ms`	Int (ms)	Không	`null`	Timeout mỗi request

Ví dụ

Cơ bản:

yaml

sinks:
  alerts:
    type: "alert"
    inputs: ["alerts"]

Nâng cao (retry và timeout chủ động):

yaml

sinks:
  alerts:
    type: "alert"
    inputs: ["alerts"]
    max_retry: 5
    timeout_ms: 5000

Lưu ý

Không có URL cấu hình ở đây — đích gửi luôn là registry.api_url (Forwarder tự đọc từ cấu hình Registry và xác thực bằng header x-device-token).
Không tồn tại web UI cho alert sink — đừng tìm giao diện quản trị cho nó.
Lỗi không tạm thời (4xx khác) bị bỏ qua, không retry.
Muốn ingestion_rule phát được alert phải có sink type: "alert" này.

Hướng dẫn tạo chứng chỉ SSL

Tạo chứng chỉ SSL tự ký cho MQTT Broker

Để cấu hình MQTT broker với TLS, bạn cần tạo chứng chỉ SSL. Dưới đây là hướng dẫn tạo chứng chỉ tự ký:

Bước 1: Tạo thư mục lưu trữ chứng chỉ

bash

sudo mkdir -p /etc/ssl/mqtt
sudo chmod 700 /etc/ssl/mqtt

Bước 2: Tạo CA (Certificate Authority) tự ký

bash

# Tạo private key cho CA
sudo openssl genrsa -out /etc/ssl/mqtt/ca-key.pem 4096

# Tạo certificate cho CA
sudo openssl req -new -x509 -days 365 -key /etc/ssl/mqtt/ca-key.pem -out /etc/ssl/mqtt/ca-cert.pem

Trong quá trình tạo CA certificate, bạn sẽ được yêu cầu nhập thông tin:

Country Name: VN
State: Ho Chi Minh
City: Ho Chi Minh City
Organization: Your Company
Organizational Unit: IT Department
Common Name: MQTT CA
Email: admin@yourcompany.com

Bước 3: Tạo server certificate

bash

# Tạo private key cho server
sudo openssl genrsa -out /etc/ssl/mqtt/server-key.pem 4096

# Tạo certificate signing request (CSR)
sudo openssl req -new -key /etc/ssl/mqtt/server-key.pem -out /etc/ssl/mqtt/server.csr

Nhập thông tin tương tự như CA, nhưng Common Name phải là hostname hoặc IP của MQTT broker.

Bước 4: Tạo server certificate từ CSR

bash

# Tạo server certificate từ CSR
sudo openssl x509 -req -in /etc/ssl/mqtt/server.csr -CA /etc/ssl/mqtt/ca-cert.pem -CAkey /etc/ssl/mqtt/ca-key.pem -CAcreateserial -out /etc/ssl/mqtt/server-cert.pem -days 365

Bước 5: Tạo client certificate (tùy chọn)

bash

# Tạo private key cho client
sudo openssl genrsa -out /etc/ssl/mqtt/client-key.pem 4096

# Tạo CSR cho client
sudo openssl req -new -key /etc/ssl/mqtt/client-key.pem -out /etc/ssl/mqtt/client.csr

# Tạo client certificate
sudo openssl x509 -req -in /etc/ssl/mqtt/client.csr -CA /etc/ssl/mqtt/ca-cert.pem -CAkey /etc/ssl/mqtt/ca-key.pem -CAcreateserial -out /etc/ssl/mqtt/client-cert.pem -days 365

Bước 6: Cấu hình MQTT Server với TLS

yaml

mqtt_server:
  server:
    name: "secure-mqtt"
    listen: "0.0.0.0:8883"
    tls:
      capath: "/etc/ssl/mqtt/ca-cert.pem"
      certpath: "/etc/ssl/mqtt/server-cert.pem"
      keypath: "/etc/ssl/mqtt/server-key.pem"
    connections:
      connection_timeout_ms: 60000
      max_payload_size: 104857600
      max_inflight_count: 100
      dynamic_filters: true
  auth:
    - type: "basic"
      username: "device"
      password: "secret"

Cấu hình TLS cho MQTT Client (Source và Sink)

Sau khi đã tạo chứng chỉ SSL cho MQTT broker, bạn cần cấu hình client (agent/forwarder) để kết nối với broker sử dụng TLS. Điều này đặc biệt quan trọng khi sử dụng chứng chỉ tự ký (self-signed certificates).

Cấu hình MQTT Source với TLS

Khi cấu hình MQTT source để subscribe từ broker có TLS, bạn chỉ định đường dẫn CA certificate trong broker.tls:

yaml

sources:
  mqtt_secure_source:
    type: "mqtt"
    broker:
      url: "mqtts://broker.company.com:8883"
      auth:
        type: "basic"
        username: "myuser"
        password: "mypass"
      tls:
        ca_cert_path: "/etc/ssl/mqtt/ca-cert.pem"
    topics:
      - "sensors/+/temperature"
      - "devices/#"
    index: "mqtt_data"

Cấu hình MQTT Sink với TLS

Tương tự, khi cấu hình MQTT sink để publish đến broker có TLS:

yaml

sinks:
  mqtt_secure_sink:
    type: "mqtt"
    url: "mqtts://broker.company.com:8883"
    auth:
      type: "basic"
      username: "myuser"
      password: "mypass"
    tls:
      ca_cert_path: "/etc/ssl/mqtt/ca-cert.pem"
    batch_size: 100
    batch_interval: 1000
    batch_max_bytes: 1048576

Cấu hình Mutual TLS (mTLS)

Nếu broker yêu cầu client certificate authentication (mutual TLS), bạn cần cung cấp cả client certificate và key:

yaml

sources:
  mqtt_mtls_source:
    type: "mqtt"
    broker:
      url: "mqtts://broker.company.com:8883"
      auth:
        type: "none"
      tls:
        ca_cert_path: "/etc/ssl/mqtt/ca-cert.pem"
        client_cert_path: "/etc/ssl/mqtt/client-cert.pem"
        client_key_path: "/etc/ssl/mqtt/client-key.pem"
    topics:
      - "secure/data/#"

ℹ️

Khi sử dụng mTLS, bạn thường không cần username/password vì client certificate đã đóng vai trò xác thực.

Các tùy chọn TLS

ca_cert_path: Đường dẫn CA certificate (PEM). Bắt buộc khi dùng chứng chỉ tự ký hoặc custom CA.
client_cert_path: Đường dẫn client certificate (PEM). Chỉ cần khi broker yêu cầu client authentication.
client_key_path: Đường dẫn client private key (PEM). Phải cung cấp cùng client_cert_path.
insecure: Bỏ qua xác thực certificate (mặc định: false). Không khuyến nghị dùng trong production.

Xử lý lỗi TLS

Nếu bạn gặp lỗi TLS: I/O: tls handshake eof hoặc các lỗi TLS khác, hãy kiểm tra:

CA certificate path đúng chưa? Đảm bảo đường dẫn file CA chính xác và file đọc được.
Certificate format đúng chưa? Certificate phải ở định dạng PEM (bắt đầu với -----BEGIN CERTIFICATE-----).
Quyền truy cập file: Forwarder phải có quyền đọc các file certificate.
URL protocol: Dùng mqtts:// thay vì mqtt:// cho kết nối TLS.
Port đúng chưa? Port mặc định cho MQTT over TLS là 8883, không phải 1883.

Ví dụ cấu hình đầy đủ với TLS:

yaml

sources:
  mqtt_production:
    type: "mqtt"
    broker:
      url: "mqtts://mqtt.company.com:8883"
      auth:
        type: "basic"
        username: "agent_user"
        password: "secure_password"
      tls:
        ca_cert_path: "/etc/ssl/mqtt/ca-cert.pem"
    topics:
      - "logs/#"
      - "metrics/#"
    index: "production_data"

sinks:
  mqtt_production:
    type: "mqtt"
    url: "mqtts://mqtt.company.com:8883"
    auth:
      type: "basic"
      username: "agent_user"
      password: "secure_password"
    tls:
      ca_cert_path: "/etc/ssl/mqtt/ca-cert.pem"
    batch_size: 500
    batch_interval: 2000
    batch_max_bytes: 5242880

Tạo chứng chỉ SSL cho Kafka

Bước 1: Tạo keystore cho Kafka broker

bash

# Tạo keystore
keytool -keystore kafka.server.keystore.jks -alias localhost -validity 365 -genkey -keyalg RSA -keysize 2048 -storepass password -keypass password -dname "CN=localhost, OU=IT, O=YourCompany, L=HCMC, ST=HCMC, C=VN"

# Tạo certificate signing request
keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file -storepass password

# Tạo certificate từ CSR (cần CA)
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:password

# Import CA certificate vào keystore
keytool -keystore kafka.server.keystore.jks -alias CARoot -import -file ca-cert -storepass password

# Import signed certificate vào keystore
keytool -keystore kafka.server.keystore.jks -alias localhost -import -file cert-signed -storepass password

Bước 2: Tạo truststore cho client

bash

# Tạo truststore và import CA certificate
keytool -keystore kafka.client.truststore.jks -alias CARoot -import -file ca-cert -storepass password

Bước 3: Cấu hình Kafka Sink với TLS

yaml

sinks:
  kafka_secure:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9093"
    security:
      type: "tls"
      tls:
        ca_location: "/path/to/ca-cert"
        certificate_location: "/path/to/client-cert"
        key_location: "/path/to/client-key"
        verify_certificate: true

Cấu hình nâng cao

High Availability

Kafka sink với nhiều broker, acks: "all" và idempotent producer:

yaml

sinks:
  kafka_ha:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9092"
      - "kafka2.company.com:9092"
      - "kafka3.company.com:9092"
    acks: "all"
    enable_idempotence: true
    retries:
      max_retries: 2147483647
      backoff_ms: 100

Performance Tuning

yaml

# Tối ưu cho high-throughput
inventory:
  max_messages: 1000000
  max_bytes: 1073741824
  backpressure_on_limit: true
  per_shipper:
    kafka:
      ttl: 86400
      max_retries: 20
      flush_interval_secs: 5

mqtt_server:
  router:
    max_connections: 50000
    max_segment_size: 209715200
    max_segment_count: 20

sinks:
  high_throughput_kafka:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9092"
    batching:
      linger_ms: 0
      batch_num_messages: 10000
      batch_kbytes: 10240
    batch_size: 1000
    batch_interval: 100
    batch_max_bytes: 10485760

📝

Trong ví dụ trên, `mqtt_server` thuộc nhóm section **không hot-reload được** — đổi xong phải khởi động lại service. Nhóm này gồm `logging`, `hot_reload`, `registry`, `resources_threshold` và mọi embedded server; xem mục [Hot-reload](#hot-reload).

Tổng quan cấu hình ​

Vị trí file cấu hình ​

Các section cấp cao nhất ​

Lệnh CLI liên quan tới cấu hình ​

Yêu cầu khi dùng Registry ​

Cấu hình logging ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Channel buffers ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Inventory ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Hot-reload ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Ngưỡng tài nguyên ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Registry ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Cổng REST next-hop ​

Cú pháp ​

Các trường cấu hình ​

Các endpoint ​

Ví dụ ​

Lưu ý ​

Device guard (danh sách chặn token) ​

Hoạt động ​

Header liên quan ​

Lưu ý ​

Sources ​

Tổng quan sources ​

Rsyslog source ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

SNMP source ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

MQTT source ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Kafka source ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

gRPC source ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

Embedded servers ​

Tổng quan và cổng lắng nghe ​

rsyslog server ​

Cú pháp ​

Các trường cấu hình ​

Ví dụ ​

Lưu ý ​

SNMP server ​

Tổng quan cấu hình

Vị trí file cấu hình

Các section cấp cao nhất

Lệnh CLI liên quan tới cấu hình

Yêu cầu khi dùng Registry

Cấu hình logging

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Channel buffers

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Inventory

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Hot-reload

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Ngưỡng tài nguyên

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Registry

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Cổng REST next-hop

Cú pháp

Các trường cấu hình

Các endpoint

Ví dụ

Lưu ý

Device guard (danh sách chặn token)

Hoạt động

Header liên quan

Lưu ý

Sources

Tổng quan sources

Rsyslog source

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

SNMP source

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

MQTT source

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Kafka source

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

gRPC source

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

Embedded servers

Tổng quan và cổng lắng nghe

rsyslog server

Cú pháp

Các trường cấu hình

Ví dụ

Lưu ý

SNMP server