G-BlackHole Documentation

Tổng quan cấu hình

Agent sử dụng file YAML để định nghĩa sources, sinks và các thiết lập khác.

Tạo cấu hình mẫu

bash

blackhole-agt configure sample --path config.yml

Lệnh này tạo file cấu hình mẫu với tất cả sources và sinks có sẵn.

Pull/push cấu hình từ Registry

bash

# Pull cấu hình từ Registry
blackhole-agt configure pull --path config.yml

# Push cấu hình local lên Registry
blackhole-agt configure push --path config.yml

Cấu hình chi tiết

Logging Configuration

Cấu hình logging cho Agent (tùy chọn):

yaml

logging:
  level: "info" # Mức độ log: debug, info, warn, error
  dir: "./logs" # Thư mục chứa log files
  file: "default.log" # File log chính
  error_file: "default-error.log" # File log lỗi (tùy chọn)
  max_size_mb: 10 # Kích thước tối đa toàn bộ file log (MB)

Chi tiết các tham số:

level (bắt buộc): Mức độ logging
- debug: Hiển thị tất cả thông tin chi tiết, bao gồm debug messages
- info: Hiển thị thông tin chung và các sự kiện quan trọng (mặc định)
- warn: Chỉ hiển thị cảnh báo và lỗi
- error: Chỉ hiển thị lỗi
dir (bắt buộc): Đường dẫn thư mục lưu trữ log files (mặc định: ./logs)
file (bắt buộc): Tên file log chính (mặc định: default.log)
error_file (tùy chọn): Tên file log riêng cho lỗi (mặc định: default-error.log)
max_size_mb (bắt buộc): Kích thước tối đa của tất cả log files tính bằng MB (mặc định: 10MB)

Channel Buffers Configuration

Cấu hình buffer cho performance tuning và memory management:

yaml

channel_buffers:
  orchestrator_capacity: 10000000 # Buffer capacity cho orchestrator collector channel
  shipper_capacity: 10000000 # Buffer capacity cho shipper channels
  router_max_in_flight_routes: 4 # Maximum concurrent routing tasks

Chi tiết các tham số:

orchestrator_capacity (bắt buộc): Buffer capacity cho orchestrator collector channel (mặc định: 10000000)
- Controls how many events can be queued between sources and orchestrator
- Higher values improve burst handling but use more memory
shipper_capacity (bắt buộc): Buffer capacity cho shipper channels (mặc định: 10000000)
- Controls how many events can be queued between orchestrator and sinks
- Higher values improve throughput but use more memory
router_max_in_flight_routes (bắt buộc): Maximum concurrent routing tasks in orchestrator (mặc định: 4)
- Controls parallelism when routing events to multiple sinks
- Higher values improve throughput but increase CPU usage

Registry Configuration

Registry là điểm điều phối trung tâm (tùy chọn) để xác thực và đồng bộ cấu hình. Nếu không sử dụng Registry, bạn có thể bỏ qua phần này:

(Từ v0.5.5+)

yaml

registry:
  api_url: "https://blackhole.glabs.one" # API endpoint của Registry
  config_update_interval: "15s" # Khoảng thời gian cập nhật cấu hình (mặc định: 15 giây)
  proxy: # Cấu hình proxy (tùy chọn)
    enable: true # Bật/tắt proxy
    http: "http://proxy.company.com:8080" # HTTP proxy URL
    https: "https://proxy.company.com:8080" # HTTPS proxy URL
  tls: # TLS settings cho kết nối tới `api_url`
    ca_cert_path: null # Optional: custom CA bundle (PEM)
    insecure_skip_verify_https: false # Optional: skip HTTPS certificate verification (NOT recommended)

Chi tiết các tham số:

api_url (bắt buộc): URL của API endpoint Registry (mặc định: https://blackhole.glabs.one)
proxy (tùy chọn): Cấu hình proxy
- enable (bắt buộc): Bật/tắt proxy (mặc định: true)
- http (tùy chọn): URL của HTTP proxy
- https (tùy chọn): URL của HTTPS proxy
tls (tùy chọn): TLS settings cho kết nối tới api_url (chỉ áp dụng khi api_url dùng https://)
- ca_cert_path (tùy chọn): Đường dẫn file CA bundle (PEM) để validate certificate của Registry
- insecure_skip_verify_https (tùy chọn): Skip HTTPS certificate verification (không khuyến nghị)
config_update_interval (bắt buộc): Khoảng thời gian cập nhật cấu hình (mặc định: 15 giây)

Lưu ý quan trọng:

Registry là tùy chọn. Agent có thể hoạt động độc lập mà không cần Registry
Nếu sử dụng Registry, bạn phải xác thực trước bằng lệnh auth --key

(Từ v0.5.4-)

yaml

registry:
  mqtt_url: "mqtts://blackhole.glabs.one:8883" # MQTT broker URL của Registry
  api_url: "https://blackhole.glabs.one" # API endpoint của Registry
  identifier: "registry" # Tên định danh cho registry sink
  proxy: # Cấu hình proxy (tùy chọn)
    enable: true # Bật/tắt proxy
    http: "http://proxy.company.com:8080" # HTTP proxy URL
    https: "https://proxy.company.com:8080" # HTTPS proxy URL
  rate_limit: # Giới hạn tốc độ gửi dữ liệu
    batch_size: 500 # Số message trong mỗi batch
    batch_interval: 1000 # Khoảng thời gian giữa các batch (ms)
    batch_max_bytes: 10485760 # Kích thước tối đa mỗi batch (bytes)

Chi tiết các tham số:

mqtt_url (bắt buộc): URL của MQTT broker Registry (mặc định: mqtts://blackhole.glabs.one:8883)
api_url (bắt buộc): URL của API endpoint Registry (mặc định: https://blackhole.glabs.one)
identifier (bắt buộc): Tên định danh cho registry sink (mặc định: registry)
proxy (tùy chọn): Cấu hình proxy
- enable (bắt buộc): Bật/tắt proxy (mặc định: true)
- http (tùy chọn): URL của HTTP proxy
- https (tùy chọn): URL của HTTPS proxy
rate_limit (bắt buộc): Cấu hình giới hạn tốc độ gửi dữ liệu
- batch_size (bắt buộc): Số message trong mỗi batch (mặc định: 500)
- batch_interval (bắt buộc): Khoảng thời gian giữa các batch tính bằng ms (mặc định: 1000ms)
- batch_max_bytes (bắt buộc): Kích thước tối đa mỗi batch tính bằng bytes (mặc định: 10485760 bytes = 10MB)

Lưu ý quan trọng:

Registry là tùy chọn. Agent có thể hoạt động độc lập mà không cần Registry
Nếu sử dụng Registry, bạn phải xác thực trước bằng lệnh auth --key hoặc Forwarder sẽ tự động xác thực thay cho Agent

Sources Configuration

APM Source

Thu thập metrics hệ thống:

yaml

sources:
  apm_system:
    type: "apm"
    interval: 30 # Chu kỳ thu thập (giây)
    index: "apm" # Index cho dữ liệu
    sourcetype: "apm_metrics" # Loại dữ liệu

Chi tiết các tham số:

type (bắt buộc): Loại source, phải là "apm"
interval (bắt buộc): Chu kỳ thu thập metrics tính bằng giây (mặc định: 30 giây)
- Giá trị nhỏ hơn = thu thập thường xuyên hơn nhưng tốn tài nguyên
- Giá trị lớn hơn = tiết kiệm tài nguyên nhưng ít chi tiết
index (tùy chọn): Index để lưu trữ dữ liệu APM (mặc định: null)
sourcetype (tùy chọn): Loại dữ liệu để phân loại (mặc định: null)

Metrics được thu thập:

CPU usage (global và per-core)
Memory usage (total, used, available)
Disk I/O và usage
Network interface statistics
Process information
File descriptor usage (Linux)
Load average (Linux/macOS)

Ví dụ cấu hình APM:

yaml

sources:
  # Thu thập metrics mỗi 10 giây cho monitoring real-time
  apm_realtime:
    type: "apm"
    interval: 10
    index: "system_metrics"
    sourcetype: "apm_realtime"

  # Thu thập metrics mỗi 5 phút cho báo cáo dài hạn
  apm_summary:
    type: "apm"
    interval: 300
    index: "system_summary"
    sourcetype: "apm_summary"

File Source

Theo dõi và thu thập log từ files:

yaml

sources:
  security_logs:
    type: "file"
    includes: # Danh sách file patterns
      - "/var/log/auth.log"
      - "/var/log/secure.log"
      - "/var/log/audit/audit.log"
    index: "security" # Index cho dữ liệu
    sourcetype: "auth_logs" # Loại dữ liệu
    tail: true # Bắt đầu từ cuối file
    debounce_ms: 1000 # Thời gian chờ để nhóm events

  application_logs:
    type: "file"
    includes:
      - "/var/log/application/*.log"
      - "/opt/app/logs/**/*.log"
    index: "application"
    sourcetype: "app_logs"
    tail: false # Đọc từ đầu file
    debounce_ms: 500

Chi tiết các tham số:

type (bắt buộc): Loại source, phải là "file"
includes (bắt buộc): Danh sách file patterns để theo dõi (mặc định: ["/var/log/auth.log", "/var/log/secure.log", "/var/log/commands.log"])
- Hỗ trợ glob patterns: *, **, ?
- Ví dụ: "/var/log/*.log", "/opt/app/**/*.log"
index (tùy chọn): Index để lưu trữ dữ liệu (mặc định: null)
sourcetype (tùy chọn): Loại dữ liệu để phân loại (mặc định: null)
tail (bắt buộc): Có bắt đầu đọc từ cuối file không (mặc định: true)
- true: Chỉ đọc dữ liệu mới (như lệnh tail -f)
- false: Đọc toàn bộ file từ đầu
debounce_ms (bắt buộc): Thời gian chờ để nhóm các sự kiện thay đổi file tính bằng ms (mặc định: 1000ms)
- Giá trị nhỏ = phản hồi nhanh nhưng có thể tạo nhiều events
- Giá trị lớn = giảm số lượng events nhưng phản hồi chậm hơn

Tính năng File Source:

Hỗ trợ glob patterns
Tự động phát hiện log rotation
Debouncing để tối ưu hiệu suất
Theo dõi real-time

Ví dụ cấu hình File Source:

yaml

sources:
  # Theo dõi security logs với debounce ngắn
  security_logs:
    type: "file"
    includes:
      - "/var/log/auth.log"
      - "/var/log/secure.log"
    index: "security"
    sourcetype: "auth_logs"
    tail: true
    debounce_ms: 500

  # Theo dõi application logs với debounce dài hơn
  app_logs:
    type: "file"
    includes:
      - "/var/log/application/*.log"
      - "/opt/app/logs/**/*.log"
    index: "application"
    sourcetype: "app_logs"
    tail: true
    debounce_ms: 2000

SCA Source (Security Configuration Assessment) [v0.5.4+]

Thu thập và đánh giá cấu hình bảo mật theo các policy tương thích Wazuh:

yaml

sources:
  sca_default:
    type: "sca"
    policy_directory: "/etc/agent/sca/policies" # Thư mục chứa file policy YAML
    scan_interval: "1h" # Chu kỳ quét (humantime format: 1h, 30m, 2d)
    scan_cron: "0 */15 * * * * *" # Cron schedule (thay thế scan_interval)
    run_on_start: true # Quét ngay khi khởi động
    allow_commands: false # Cho phép thực thi lệnh (c: rules)
    enabled_policies: [] # Danh sách policy files (empty = tất cả)
    index: "sca"
    sourcetype: "sca_results"

Chi tiết các tham số:

type (bắt buộc): Loại source, phải là "sca"
policy_directory (bắt buộc): Đường dẫn thư mục chứa file policy YAML
- Linux/macOS mặc định: /etc/agent/sca/policies
- Windows mặc định: C:\ProgramData\agent\sca\policies
- Collector sẽ quét đệ quy các file .yml hoặc .yaml
scan_interval (tùy chọn): Chu kỳ quét theo humantime format (mặc định: "1h")
- Ví dụ: "1h", "30m", "2d", "90s"
- Đặt "0s" để tắt quét định kỳ
scan_cron (tùy chọn): Cron expression cho lịch quét
- Khi sử dụng scan_cron, không được đặt scan_interval
- Ví dụ: "0 */15 * * * * *" (mỗi 15 phút)
run_on_start (bắt buộc): Có quét ngay khi collector khởi động không (mặc định: true)
allow_commands (bắt buộc): Cho phép thực thi lệnh trong rules c: (mặc định: false)
- Lưu ý bảo mật: Mặc định tắt để ngăn chặn rủi ro bảo mật
run_on_start (bắt buộc): Có quét ngay khi collector khởi động không (mặc định: true)
- true: Thực hiện quét ngay lập tức khi khởi động
- false: Đợi đến chu kỳ quét đầu tiên
allow_commands (bắt buộc): Cho phép thực thi lệnh trong rules c: (mặc định: false)
- true: Cho phép thực thi lệnh shell/command
- false: Các check yêu cầu command sẽ được đánh dấu "not applicable"
- Lưu ý bảo mật: Mặc định tắt để ngăn chặn rủi ro bảo mật
enabled_policies (bắt buộc): Danh sách file policy cụ thể để bật (mặc định: [])
- []: Xử lý tất cả policy files trong thư mục
- ["cis_ubuntu22.yml", "pci_dss.yml"]: Chỉ xử lý các file được chỉ định
index (tùy chọn): Index để lưu trữ kết quả SCA (mặc định: null)
sourcetype (tùy chọn): Loại dữ liệu để phân loại (mặc định: null)

Các loại rule được hỗ trợ:

Prefix	Mô tả	Ví dụ
`f:`	File operations (tồn tại, nội dung, regex)	`f:/etc/passwd -> exists`
`d:`	Directory operations	`d:/etc/ssh`
`p:`	Process existence	`p:sshd`
`c:`	Command execution (yêu cầu `allow_commands: true`)	`c:systemctl is-enabled sshd -> r:enabled`
`r:`	Windows registry (chỉ Windows)	`r:HKLM\SOFTWARE\Policies -> ValueName`

Tính năng nâng cao:

Hỗ trợ negation với prefix not (ví dụ: not f:/etc/insecure -> exists)
Variable substitution trong rules (ví dụ: $path)
Conditions: all, any, none để aggregate kết quả rules
State tracking: Chỉ báo cáo các check có thay đổi trạng thái
Policy hash tracking: Tự động phát hiện thay đổi file policy

Kết quả quét:

Mỗi check sẽ tạo ra một LogEvent với các thông tin:

policy_id: ID của policy
check_id: ID của check
check_title: Tiêu đề check
result: Kết quả (pass, fail, not_applicable, error)
evidence: Bằng chứng thu thập được
description, rationale, remediation: Thông tin bổ sung (nếu có)
compliance: Frameworks tuân thủ (CIS, PCI-DSS, HIPAA, ...)
references: Tài liệu tham khảo

Ví dụ cấu hình SCA:

yaml

sources:
  # Quét CIS benchmark hàng ngày
  sca_cis:
    type: "sca"
    policy_directory: "/etc/agent/sca/policies"
    scan_interval_secs: 86400 # 24 giờ
    run_on_start: true
    allow_commands: false
    enabled_policies:
      - "cis_ubuntu22.yml"
      - "cis_debian11.yml"
    index: "compliance"
    sourcetype: "sca_cis"

  # Quét PCI-DSS với command execution cho compliance đầy đủ
  sca_pci:
    type: "sca"
    policy_directory: "/etc/agent/sca/policies"
    scan_interval_secs: 3600 # 1 giờ
    run_on_start: true
    allow_commands: true # Bật command execution cho PCI-DSS checks
    enabled_policies:
      - "pci_dss.yml"
    index: "compliance"
    sourcetype: "sca_pci"

Ví dụ cấu hình SCA trên Windows:

yaml

sources:
  sca_windows:
    type: "sca"
    policy_directory: "C:\\ProgramData\\agent\\sca\\policies"
    scan_interval_secs: 3600
    run_on_start: true
    allow_commands: false
    enabled_policies:
      - "cis_windows_server_2022.yml"
    index: "windows_compliance"
    sourcetype: "sca_windows"

Cấu trúc file Policy YAML:

File policy YAML tương thích Wazuh có cấu trúc như sau:

yaml

policy:
  id: "example_policy"
  name: "Example Security Policy"
  description: "Sample policy for demonstration"
  references:
    - "https://example.com/docs"

variables:
  sshd_config: "/etc/ssh/sshd_config"

requirements:
  title: "System requirements"
  description: "Check if this policy applies"
  condition: all
  rules:
    - "f:/etc/os-release -> r:Ubuntu"

checks:
  - id: 1001
    title: "Ensure SSH root login is disabled"
    description: "Root login via SSH should be disabled"
    rationale: "Prevents direct root access via SSH"
    remediation: "Set PermitRootLogin no in /etc/ssh/sshd_config"
    compliance:
      - cis: ["5.2.10"]
      - pci_dss: ["2.2.4"]
    condition: all
    rules:
      - 'f:$sshd_config -> r:^\s*PermitRootLogin\s+no'

  - id: 1002
    title: "Ensure password authentication is disabled"
    description: "SSH should use key-based authentication only"
    condition: all
    rules:
      - 'f:$sshd_config -> r:^\s*PasswordAuthentication\s+no'

Windows Event Log Source

Chỉ có trên Windows:

yaml

sources:
  windows_events:
    type: "windows_event"
    channels: # Danh sách event channels
      - name: "Application"
        ids: [] # Event IDs cụ thể (empty = tất cả)
      - name: "Security"
        ids: [4624, 4625, 4634] # Login/logout events
      - name: "System"
        ids: []
    index: "windows_events"
    sourcetype: "windows_events"

Chi tiết các tham số:

type (bắt buộc): Loại source, phải là "windows_event"
channels (bắt buộc): Danh sách Windows Event channels để theo dõi (mặc định: [{"name": "Application", "ids": []}])
- name (bắt buộc): Tên event channel (ví dụ: "Application", "Security", "System")
- ids (bắt buộc): Danh sách Event IDs cụ thể (mặc định: [] = tất cả events)
index (tùy chọn): Index để lưu trữ dữ liệu (mặc định: null)
sourcetype (tùy chọn): Loại dữ liệu để phân loại (mặc định: null)

Các Event Channels phổ biến:

Application: Events từ applications
Security: Security-related events (login, logout, access)
System: System events (startup, shutdown, errors)

Ví dụ cấu hình Windows Event Log:

yaml

sources:
  # Theo dõi security events quan trọng
  security_events:
    type: "windows_event"
    channels:
      - name: "Security"
        ids: [4624, 4625, 4634, 4648, 4768, 4769] # Login/logout events
    index: "security"
    sourcetype: "windows_security"

  # Theo dõi tất cả system events
  system_events:
    type: "windows_event"
    channels:
      - name: "System"
        ids: [] # Tất cả system events
    index: "system"
    sourcetype: "windows_system"

Sinks Configuration

MQTT Sink

yaml

sinks:
  mqtt_production:
    type: "mqtt"
    url: "mqtts://broker.company.com:8883?client_id=agent-001"
    inputs: ["*"] # Optional: specific source identifiers
    auth:
      type: "basic"
      username: "agent"
      password: "secure-password"
    proxy: # Proxy configuration (tùy chọn)
      enable: true
      http: "http://proxy.company.com:8080"
    rate_limit:
      batch_size: 100
      batch_interval: 500
      batch_max_bytes: 1048576

Chi tiết các tham số:

type (bắt buộc): Loại sink, phải là "mqtt"
url (bắt buộc): URL của MQTT broker (mặc định: mqtt://localhost:1883?client_id=siem-agent)
- Hỗ trợ protocols: mqtt://, mqtts://, ws://, wss://
- Có thể thêm query parameters: ?client_id=agent-001
inputs (tùy chọn): Danh sách source identifiers để consume (mặc định: ["*"])
- ["*"]: Consume tất cả sources
- ["file_logs", "apm_metrics"]: Chỉ consume từ specific sources
auth (bắt buộc): Cấu hình xác thực
- type (bắt buộc): Loại xác thực
  - "none": Không xác thực (mặc định)
  - "basic": Username/password authentication
- username (bắt buộc khi type="basic"): Tên người dùng
- password (bắt buộc khi type="basic"): Mật khẩu
proxy (tùy chọn): Cấu hình proxy
- enable (bắt buộc): Bật/tắt proxy (mặc định: true)
- http (tùy chọn): URL của HTTP proxy
- https (tùy chọn): URL của HTTPS proxy
rate_limit (bắt buộc): Cấu hình giới hạn tốc độ
- batch_size (bắt buộc): Số message trong mỗi batch (mặc định: 500)
- batch_interval (bắt buộc): Khoảng thời gian giữa các batch tính bằng ms (mặc định: 1000ms)
- batch_max_bytes (bắt buộc): Kích thước tối đa mỗi batch tính bằng bytes (mặc định: 10485760 bytes = 10MB)

Ví dụ cấu hình MQTT Sink:

yaml

sinks:
  # MQTT broker không bảo mật
  mqtt_local:
    type: "mqtt"
    url: "mqtt://localhost:1883?client_id=agent-local"
    auth:
      type: "none"
    rate_limit:
      batch_size: 50
      batch_interval: 1000
      batch_max_bytes: 1048576

  # MQTT broker bảo mật với TLS
  mqtt_secure:
    type: "mqtt"
    url: "mqtts://broker.company.com:8883?client_id=agent-001"
    auth:
      type: "basic"
      username: "agent"
      password: "secure-password"
    proxy:
      enable: true
      https: "https://proxy.company.com:8080"
    rate_limit:
      batch_size: 200
      batch_interval: 500
      batch_max_bytes: 2097152

Kafka Sink

yaml

sinks:
  kafka_production:
    type: "kafka"
    inputs: ["*"] # Optional: specific source identifiers
    bootstrap_servers:
      - "kafka1.company.com:9092"
      - "kafka2.company.com:9092"
    client_id: "blackhole-agent" # Optional: auto-generated if not specified
    security:
      type: "sasl"
      sasl:
        mechanism: "PLAIN"
        username: "agent"
        password: "secure-password"
        tls:
          verify_certificate: true
    acks: "leader" # none, leader, all
    compression: "gzip" # none, gzip, snappy, lz4, zstd
    enable_idempotence: true
    transactional_id: null # Optional: for exactly-once semantics
    batching:
      linger_ms: 100
      batch_num_messages: 1000
      batch_kbytes: 1024
    retries:
      max_retries: 10
      backoff_ms: 100
    timeouts:
      socket_timeout_ms: 60000
      request_timeout_ms: 30000
      message_timeout_ms: 300000
      connections_max_idle_ms: 300000
    producer_pool_size: 4 # Number of producer instances (1-64)
    extra: {} # Additional librdkafka properties

Chi tiết các tham số:

type (bắt buộc): Loại sink, phải là "kafka"
inputs (tùy chọn): Danh sách source identifiers để consume (mặc định: ["*"])
- ["*"]: Consume tất cả sources
- ["file_logs", "apm_metrics"]: Chỉ consume từ specific sources
bootstrap_servers (bắt buộc): Danh sách Kafka brokers (mặc định: ["localhost:9092"])
client_id (tùy chọn): Client ID để nhận diện producer (mặc định: null - auto-generated)
security (bắt buộc): Cấu hình bảo mật (mặc định: "none")
- type (bắt buộc): Loại bảo mật
  - "none": Không bảo mật (mặc định)
  - "tls": Chỉ TLS
  - "sasl": SASL authentication
- sasl (bắt buộc khi type="sasl"): Cấu hình SASL
  - mechanism (bắt buộc): Cơ chế SASL (mặc định: "PLAIN")
    - "PLAIN": Username/password
    - "SCRAM-SHA-256": SCRAM-SHA-256
    - "SCRAM-SHA-512": SCRAM-SHA-512
    - "GSSAPI": Kerberos
    - "OAUTHBEARER": OAuth 2.0
  - username (tùy chọn): Tên người dùng
  - password (tùy chọn): Mật khẩu
  - tls (tùy chọn): Cấu hình TLS cho SASL
    - verify_certificate (bắt buộc): Xác minh chứng chỉ (mặc định: true)
acks (bắt buộc): Mức độ xác nhận (mặc định: "leader")
- "none": Không yêu cầu xác nhận (acks=0) - nhanh nhất nhưng có thể mất dữ liệu
- "leader": Chỉ yêu cầu xác nhận từ leader (acks=1) - cân bằng tốc độ và độ tin cậy
- "all": Yêu cầu xác nhận từ tất cả replicas (acks=all) - chậm nhất nhưng đảm bảo nhất
compression (bắt buộc): Loại nén dữ liệu (mặc định: "none")
- "none": Không nén - nhanh nhất
- "gzip": Gzip compression - tốt cho text data
- "snappy": Snappy compression - cân bằng tốc độ và tỷ lệ nén
- "lz4": LZ4 compression - nhanh nhất trong các loại nén
- "zstd": Zstandard compression - tỷ lệ nén tốt nhất
enable_idempotence (bắt buộc): Bật idempotent producer (mặc định: false)
- true: Đảm bảo không có duplicate messages
- false: Có thể có duplicate messages
transactional_id (tùy chọn): Transactional ID cho exactly-once semantics (mặc định: null)
batching (bắt buộc): Cấu hình batching
- linger_ms (bắt buộc): Thời gian chờ để tạo batch tính bằng ms (mặc định: 0)
- batch_num_messages (bắt buộc): Số message tối đa trong batch (mặc định: 100000)
- batch_kbytes (bắt buộc): Kích thước tối đa batch tính bằng KB (mặc định: 1048576 KB = 1GB)
retries (bắt buộc): Cấu hình retry
- max_retries (bắt buộc): Số lần retry tối đa (mặc định: 10)
- backoff_ms (bắt buộc): Thời gian chờ giữa các retry tính bằng ms (mặc định: 100)
timeouts (bắt buộc): Cấu hình timeout
- socket_timeout_ms (bắt buộc): Socket timeout tính bằng ms (mặc định: 60000)
- request_timeout_ms (bắt buộc): Request timeout tính bằng ms (mặc định: 30000)
- message_timeout_ms (bắt buộc): Message timeout tính bằng ms (mặc định: 300000)
- connections_max_idle_ms (bắt buộc): Connection idle timeout tính bằng ms (mặc định: 300000)
producer_pool_size (bắt buộc): Số producer instances trong pool (mặc định: 4, range: 1-64)
extra (bắt buộc): Additional librdkafka configuration properties (mặc định: {})

Ví dụ cấu hình Kafka Sink:

yaml

sinks:
  # Kafka cluster đơn giản không bảo mật
  kafka_local:
    type: "kafka"
    bootstrap_servers:
      - "localhost:9092"
    acks: "leader"
    compression: "none"
    enable_idempotence: false

  # Kafka cluster production với SASL/TLS
  kafka_production:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9092"
      - "kafka2.company.com:9092"
      - "kafka3.company.com:9092"
    client_id: "blackhole-agent"
    security:
      type: "sasl"
      sasl:
        mechanism: "SCRAM-SHA-256"
        username: "agent"
        password: "secure-password"
        tls:
          verify_certificate: true
    acks: "all"
    compression: "gzip"
    enable_idempotence: true
    batching:
      linger_ms: 100
      batch_num_messages: 1000
      batch_kbytes: 1024
    retries:
      max_retries: 10
      backoff_ms: 100
    timeouts:
      socket_timeout_ms: 60000
      request_timeout_ms: 30000
      message_timeout_ms: 300000

gRPC Sink

Gửi sự kiện tới một gRPC endpoint (ví dụ Forwarder chạy embedded gRPC server). gRPC sink sẽ serialize mỗi LogEvent thành JSON và gửi theo RPC Publish (unary) hoặc PublishStream (client-streaming). Trường topic của gRPC request mặc định là index của event.

yaml

sinks:
  grpc_forwarder:
    type: "grpc"
    url: "http://forwarder.company.com:50051"
    inputs: ["*"] # Optional: specific source identifiers
    # Custom headers sẽ được map thành gRPC metadata (ví dụ để gửi API key)
    headers:
      x-api-key: "forwarder-secret"
    use_streaming: true # Mặc định: true (PublishStream)
    max_concurrent_requests: 4 # Số request gRPC đồng thời (mặc định: 1)
    message_limits:
      max_encoding_message_size: 4194304
      max_decoding_message_size: 4194304
    rate_limit:
      batch_size: 500
      batch_interval: 1000
      batch_max_bytes: 10485760

Chi tiết các tham số:

type (bắt buộc): Loại sink, phải là "grpc".
url (bắt buộc): URL đầy đủ của gRPC endpoint (mặc định: http://localhost:50051).
- Hỗ trợ http:// và https://.
- Alias: endpoint.
inputs (tùy chọn): Danh sách source identifiers để consume (mặc định: ["*"]).
- ["*"]: Consume tất cả sources.
headers (tùy chọn): Map {header: value} được gửi kèm mọi request dưới dạng gRPC metadata (mặc định: {}).
- Ví dụ: x-api-key: "forwarder-secret".
use_streaming (tùy chọn): Chọn kiểu RPC gửi dữ liệu (mặc định: true).
- true: dùng PublishStream để tăng throughput.
- false: dùng Publish cho từng event (unary).
max_concurrent_requests (tùy chọn): Số request gRPC đồng thời mà shipper có thể gửi (mặc định: 1).
rate_limit (bắt buộc): Cấu hình batching/throughput (giống các sinks khác).
- batch_size, batch_interval, batch_max_bytes.
message_limits (tùy chọn): Giới hạn kích thước message gRPC (bytes).
- max_decoding_message_size: giới hạn decode response.
- max_encoding_message_size: giới hạn encode request.

Các tùy chọn nâng cao (tùy chọn):

connect_timeout_secs: Timeout cho TCP connect (seconds).
request_timeout_secs (alias timeout_secs): Timeout per-RPC (seconds). Hiện tại shipper chưa áp dụng timeout này.
tcp_keepalive_secs: TCP keepalive (seconds).
tcp_nodelay: Bật/tắt TCP_NODELAY.
http2_keep_alive_interval_secs, http2_keep_alive_timeout_secs, http2_keep_alive_while_idle: HTTP/2 keepalive tuning.
initial_stream_window_size, initial_connection_window_size, http2_adaptive_window: HTTP/2 flow control tuning.
concurrency_limit: Giới hạn concurrency ở mức tonic::transport::Endpoint.
buffer_size: Buffer nội bộ của endpoint/service.
user_agent: User-Agent header cho outbound.
tls: Cấu hình TLS client.
- domain_name hiện được dùng để override SNI/hostname verify.
- ca_cert_path, client_cert_path, client_key_path, insecure_skip_verify hiện chưa được áp dụng trong shipper.
proxy: Cấu hình proxy. Hiện tại gRPC shipper chưa hỗ trợ proxy cho Endpoint.
reflection, proto: Hook forward-compatible (hiện tại runtime có thể bỏ qua).

BlackHole Sink

yaml

sinks:
  blackhole_production:
    type: "blackhole"
    inputs: ["*"] # Optional: specific source identifiers
    url: "https://blackhole.company.com:9200"
    healthcheck: false # Enable health check requests
    request: null # Optional: compatibility request config
    auth:
      type: "basic"
      username: "agent"
      password: "secure-password"
    tls:
      ca_cert_path: null # Optional: custom CA bundle
      insecure_skip_verify: false # Skip certificate verification
    headers: # Custom headers
      X-Custom-Header: "value"
    timeouts:
      request_timeout_secs: 30
      connect_timeout_secs: 10
    rate_limit:
      batch_size: 200
      batch_interval: 2000
      batch_max_bytes: 2097152

Chi tiết các tham số:

type (bắt buộc): Loại sink, phải là "blackhole"
inputs (tùy chọn): Danh sách source identifiers để consume (mặc định: ["*"])
- ["*"]: Consume tất cả sources
- ["file_logs", "apm_metrics"]: Chỉ consume từ specific sources
url (bắt buộc): URL của BlackHole endpoint (mặc định: http://localhost:9200)
healthcheck (bắt buộc): Enable health check requests (mặc định: false)
request (tùy chọn): Compatibility request configuration (mặc định: null)
- retry_attempts (tùy chọn): Số lần retry attempts
- timeout_secs (tùy chọn): Request timeout in seconds
auth (bắt buộc): Cấu hình xác thực (mặc định: "none")
- type (bắt buộc): Loại xác thực
  - "none": Không xác thực (mặc định)
  - "basic": Username/password authentication
  - "client_cert": Client certificate (mTLS)
  - "jwt": JWT/OIDC bearer token
  - "aws_sigv4": AWS SigV4 signing
- username (bắt buộc khi type="basic"): Tên người dùng
- password (bắt buộc khi type="basic"): Mật khẩu
- pkcs12_path (bắt buộc khi type="client_cert"): Đường dẫn file PKCS#12
- pkcs12_password (bắt buộc khi type="client_cert"): Mật khẩu PKCS#12
- token (bắt buộc khi type="jwt"): JWT token
- header_name (bắt buộc khi type="jwt"): Tên header (mặc định: "Authorization")
- region (bắt buộc khi type="aws_sigv4"): AWS region
- profile (tùy chọn khi type="aws_sigv4"): AWS profile
- role_arn (tùy chọn khi type="aws_sigv4"): Role ARN để assume
- service (bắt buộc khi type="aws_sigv4"): AWS service name
  - "es": Amazon BlackHole Service
  - "aoss": BlackHole Serverless
tls (bắt buộc): TLS configuration (mặc định: {ca_cert_path: null, insecure_skip_verify: false})
- ca_cert_path (tùy chọn): Custom CA bundle file path
- insecure_skip_verify (bắt buộc): Skip certificate verification (mặc định: false)
headers (bắt buộc): Custom headers (mặc định: {})
timeouts (bắt buộc): Cấu hình timeout
- request_timeout_secs (bắt buộc): Request timeout tính bằng giây (mặc định: 30)
- connect_timeout_secs (bắt buộc): Connect timeout tính bằng giây (mặc định: 10)
rate_limit (bắt buộc): Cấu hình giới hạn tốc độ
- batch_size (bắt buộc): Số message trong mỗi batch (mặc định: 500)
- batch_interval (bắt buộc): Khoảng thời gian giữa các batch tính bằng ms (mặc định: 1000ms)
- batch_max_bytes (bắt buộc): Kích thước tối đa mỗi batch tính bằng bytes (mặc định: 10485760 bytes = 10MB)
bulk_doc_metadata_overhead_bytes (tùy chọn): Ước tính overhead bytes cho metadata mỗi document trong bulk request (mặc định: 128)
- Bao gồm action line và JSON framing overhead trong _bulk requests
bulk_max_docs_hard_cap (tùy chọn): Giới hạn cứng số document tối đa trong một bulk request (mặc định: 5000)
- Giá trị hiệu quả: min(rate_limit.batch_size, bulk_max_docs_hard_cap)
bulk_max_bytes_hard_cap (tùy chọn): Giới hạn cứng kích thước payload tối đa trong một bulk request (mặc định: 67108864 = 64MB)
- Giá trị hiệu quả: min(rate_limit.batch_max_bytes, bulk_max_bytes_hard_cap)

Ví dụ cấu hình BlackHole Sink:

yaml

sinks:
  # BlackHole không bảo mật
  blackhole_local:
    type: "blackhole"
    url: "http://localhost:9200"
    auth:
      type: "none"
    timeouts:
      request_timeout_secs: 30
      connect_timeout_secs: 10

  # BlackHole với basic auth
  blackhole_basic:
    type: "blackhole"
    url: "https://blackhole.company.com:9200"
    auth:
      type: "basic"
      username: "agent"
      password: "secure-password"
    headers:
      X-Custom-Header: "value"
    timeouts:
      request_timeout_secs: 30
      connect_timeout_secs: 10

  # BlackHole với AWS SigV4
  blackhole_aws:
    type: "blackhole"
    url: "https://search-domain.us-east-1.es.amazonaws.com"
    auth:
      type: "aws_sigv4"
      region: "us-east-1"
      service: "es"
    timeouts:
      request_timeout_secs: 30
      connect_timeout_secs: 10

File Sink

Ghi sự kiện ra file local để debug hoặc lưu trữ:

yaml

sinks:
  file_debug:
    type: "file"
    path: "./debug-events/" # Thư mục chứa log files
    max_size_mb: 10 # Kích thước tối đa mỗi file (MB)
    max_files: 5 # Số file tối đa giữ lại
    inputs: ["*"] # Optional: specific source identifiers
    rate_limit:
      batch_size: 100
      batch_interval: 1000
      batch_max_bytes: 1048576

Chi tiết các tham số:

type (bắt buộc): Loại sink, phải là "file"
path (bắt buộc): Thư mục chứa log files (mặc định: "./debug-events/")
- Đường dẫn tương đối sẽ được resolve theo working directory
max_size_mb (bắt buộc): Kích thước tối đa mỗi file tính bằng MB (mặc định: 10)
- Khi file vượt quá kích thước này, file mới sẽ được tạo
max_files (bắt buộc): Số file tối đa giữ lại (mặc định: 5)
- Các file cũ hơn sẽ bị xóa
inputs (tùy chọn): Danh sách source identifiers (mặc định: ["*"])
rate_limit (bắt buộc): Cấu hình batching

Ví dụ:

yaml

sinks:
  # Debug sink cho development
  file_debug:
    type: "file"
    path: "./debug-events/"
    max_size_mb: 50
    max_files: 10

  # Lưu security logs riêng
  file_security:
    type: "file"
    path: "/var/log/siem/security/"
    max_size_mb: 100
    max_files: 20
    inputs: ["security_logs"]

Inventory Configuration

Cấu hình queue persistent cho dữ liệu:

yaml

inventory:
  max_messages: 100000 # Số message tối đa (None = unlimited)
  max_bytes: 268435456 # Dung lượng tối đa (256MB, None = unlimited)
  backpressure_on_limit: false # Áp dụng backpressure khi đạt giới hạn
  per_shipper: # Cấu hình cho từng shipper
    mqtt:
      ttl: 86400 # Time-to-live (giây)
      max_retries: 20 # Số lần retry tối đa
      flush_interval_secs: 5 # Chu kỳ flush
    kafka:
      ttl: 3600
      max_retries: 10
      flush_interval_secs: 10
    blackhole:
      ttl: 7200
      max_retries: 15
      flush_interval_secs: 5

Chi tiết các tham số:

max_messages (bắt buộc): Số message tối đa được lưu trữ (mặc định: 100000)
- null: Không giới hạn số message
- Số nguyên: Giới hạn số message
max_bytes (bắt buộc): Dung lượng tối đa tính bằng bytes (mặc định: 268435456 bytes = 256MB)
- null: Không giới hạn dung lượng
- Số nguyên: Giới hạn dung lượng
backpressure_on_limit (bắt buộc): Có áp dụng backpressure khi đạt giới hạn không (mặc định: false)
- true: Dừng nhận dữ liệu mới khi đạt giới hạn
- false: Xóa dữ liệu cũ nhất khi đạt giới hạn
per_shipper (bắt buộc): Cấu hình cho từng loại shipper
- mqtt: Cấu hình cho MQTT shipper
- kafka: Cấu hình cho Kafka shipper
- blackhole: Cấu hình cho BlackHole shipper
- Mỗi shipper có các tham số:
  - ttl (bắt buộc): Time-to-live tính bằng giây (mặc định: 86400 = 24 giờ)
  - max_retries (bắt buộc): Số lần retry tối đa (mặc định: 20)
  - flush_interval_secs (bắt buộc): Chu kỳ flush tính bằng giây (mặc định: 5)

Ví dụ cấu hình Inventory:

yaml

inventory:
  # Cấu hình cho môi trường development
  max_messages: 10000
  max_bytes: 67108864 # 64MB
  backpressure_on_limit: false
  per_shipper:
    mqtt:
      ttl: 3600 # 1 giờ
      max_retries: 5
      flush_interval_secs: 10
    kafka:
      ttl: 1800 # 30 phút
      max_retries: 3
      flush_interval_secs: 15
    blackhole:
      ttl: 3600 # 1 giờ
      max_retries: 5
      flush_interval_secs: 10

  # Cấu hình cho môi trường production
  max_messages: 1000000
  max_bytes: 1073741824 # 1GB
  backpressure_on_limit: true
  per_shipper:
    mqtt:
      ttl: 86400 # 24 giờ
      max_retries: 20
      flush_interval_secs: 5
    kafka:
      ttl: 3600 # 1 giờ
      max_retries: 10
      flush_interval_secs: 10
    blackhole:
      ttl: 7200 # 2 giờ
      max_retries: 15
      flush_interval_secs: 5

Hot Reload Configuration

Cấu hình tự động cập nhật cấu hình khi file thay đổi:

yaml

hot_reload:
  enabled: true # Bật/tắt hot-reload
  poll_interval_ms: 5000 # Chu kỳ kiểm tra thay đổi (ms)
  debounce_ms: 1000 # Thời gian chờ sau khi phát hiện thay đổi (ms)

Chi tiết các tham số:

enabled (bắt buộc): Bật/tắt hot-reload (mặc định: true)
- Khi bật, watcher sẽ kiểm tra file config định kỳ
- Khi có thay đổi, config sẽ được reload tự động
poll_interval_ms (bắt buộc): Chu kỳ kiểm tra thay đổi tính bằng ms (mặc định: 5000)
- Giá trị nhỏ hơn = phát hiện thay đổi nhanh hơn nhưng tốn tài nguyên hơn
debounce_ms (bắt buộc): Thời gian chờ sau khi phát hiện thay đổi (mặc định: 1000)
- Ngăn reload liên tục khi file thay đổi nhiều lần

Resource Threshold Configuration

Cấu hình giới hạn tài nguyên để tạm dừng pipeline khi quá tải:

yaml

resources_threshold:
  cpu:
    threshold_percentage: 80.0 # Ngưỡng CPU (%)
    sustained_secs: 60 # Thời gian duy trì ngưỡng (giây)
  memory:
    threshold_percentage: 80.0 # Ngưỡng Memory (%)
    sustained_secs: 60
  disk:
    threshold_percentage: 101.0 # Ngưỡng Disk (%)
    sustained_secs: 60
  check_interval: 10 # Chu kỳ kiểm tra (giây)

Chi tiết các tham số:

cpu (bắt buộc): Cấu hình ngưỡng CPU
- threshold_percentage (mặc định: 80.0): Ngưỡng CPU (%). Đặt > 100 để tắt monitoring
- sustained_secs (mặc định: 60): Thời gian CPU phải duy trì ở ngưỡng để trigger pause
memory (bắt buộc): Cấu hình ngưỡng Memory
- threshold_percentage (mặc định: 80.0): Ngưỡng Memory (%)
- sustained_secs (mặc định: 60): Thời gian Memory phải duy trì ở ngưỡng
disk (bắt buộc): Cấu hình ngưỡng Disk
- threshold_percentage (mặc định: 101.0): Ngưỡng Disk (%). Mặc định 101% = tắt monitoring
- sustained_secs (mặc định: 60): Thời gian Disk phải duy trì ở ngưỡng
check_interval (bắt buộc): Chu kỳ kiểm tra tài nguyên tính bằng giây (mặc định: 10)

Hành vi:

Khi CPU hoặc Memory vượt ngưỡng trong thời gian sustained_secs, pipeline sẽ tạm dừng
Khi tài nguyên giảm xuống dưới ngưỡng, pipeline sẽ tự động tiếp tục
Disk monitoring mặc định tắt (threshold 101%)

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 Sink với TLS

yaml

sinks:
  mqtt_secure:
    type: "mqtt"
    url: "mqtts://broker.company.com:8883?client_id=agent-001"
    auth:
      type: "basic"
      username: "agent"
      password: "secure-password"
    # Thêm cấu hình TLS nếu cần
    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"

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 cần chỉ định đường dẫn đến CA certificate:

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"
    rate_limit:
      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/#"

Lưu ý

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 đến CA certificate file (PEM format). Bắt buộc khi sử dụng chứng chỉ tự ký hoặc custom CA.
client_cert_path: Đường dẫn đến client certificate file (PEM format). Chỉ cần khi broker yêu cầu client authentication.
client_key_path: Đường dẫn đến client private key file (PEM format). Phải được cung cấp cùng với client_cert_path.
insecure: Bỏ qua xác thực certificate (mặc định: false). Không khuyến nghị sử dụng trong môi trường production vì làm giảm bảo mật.

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 đến file CA certificate là chính xác và file có thể đọc được.
Certificate format đúng chưa? Các certificate phải ở định dạng PEM (text format, bắt đầu với -----BEGIN CERTIFICATE-----).
Quyền truy cập file: Đảm bảo agent/forwarder có quyền đọc các file certificate.
URL protocol: Sử 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"
    rate_limit:
      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 Configuration

yaml

sinks:
  kafka_ha:
    type: "kafka"
    bootstrap_servers:
      - "kafka1.company.com:9092"
      - "kafka2.company.com:9092"
      - "kafka3.company.com:9092"
    acks: "all" # Đảm bảo replication
    enable_idempotence: true # Tránh duplicate
    retries:
      max_retries: 2147483647 # Retry vô hạn
      backoff_ms: 100

Performance Tuning

yaml

# Tối ưu cho high-throughput
inventory:
  max_messages: 1000000
  max_bytes: 1073741824 # 1GB
  backpressure_on_limit: true

sources:
  high_freq_apm:
    type: "apm"
    interval: 5 # Thu thập mỗi 5 giây

sinks:
  high_throughput_kafka:
    type: "kafka"
    batching:
      linger_ms: 0 # Không delay
      batch_num_messages: 10000 # Batch lớn
      batch_kbytes: 10240 # 10MB batches
    rate_limit:
      batch_size: 1000
      batch_interval: 100
      batch_max_bytes: 10485760

Tổng quan cấu hình ​

Tạo cấu hình mẫu ​

Pull/push cấu hình từ Registry ​

Cấu hình chi tiết ​

Logging Configuration ​

Channel Buffers Configuration ​

Registry Configuration ​

Sources Configuration ​

APM Source ​

File Source ​

SCA Source (Security Configuration Assessment) [v0.5.4+] ​

Windows Event Log Source ​

Sinks Configuration ​

MQTT Sink ​

Kafka Sink ​

gRPC Sink ​

BlackHole Sink ​

File Sink ​

Inventory Configuration ​

Hot Reload Configuration ​

Resource Threshold Configuration ​

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

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

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

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

Bước 3: Tạo server certificate ​

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

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

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

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

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

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

Cấu hình Mutual TLS (mTLS) ​

Các tùy chọn TLS ​

Xử lý lỗi TLS ​

Tạo chứng chỉ SSL cho Kafka ​

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

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

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

Cấu hình nâng cao ​

High Availability Configuration ​

Performance Tuning ​

Tổng quan cấu hình

Tạo cấu hình mẫu

Pull/push cấu hình từ Registry

Cấu hình chi tiết

Logging Configuration

Channel Buffers Configuration

Registry Configuration

Sources Configuration

APM Source

File Source

SCA Source (Security Configuration Assessment) [v0.5.4+]

Windows Event Log Source

Sinks Configuration

MQTT Sink

Kafka Sink

gRPC Sink

BlackHole Sink

File Sink

Inventory Configuration

Hot Reload Configuration

Resource Threshold Configuration

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

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

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

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

Bước 3: Tạo server certificate

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

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

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

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

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

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

Cấu hình Mutual TLS (mTLS)

Các tùy chọn TLS

Xử lý lỗi TLS

Tạo chứng chỉ SSL cho Kafka

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

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

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

Cấu hình nâng cao

High Availability Configuration

Performance Tuning