Tài liệu này hướng dẫn cách cấu hình parser cho hệ thống SecOps để xử lý dữ liệu log thô thành định dạng tuân thủ ECS. Parser sử dụng ba block: #regex, #conditional,#normalize và #enrich . Cấu hình được viết dưới dạng văn bản thuần và dữ liệu được xử lý tuần tự.

Cấu trúc cấu hình

Cấu hình parser gồm bốn block chạy tuần tự:

`#regex` → `#conditional` → `#normalize` → `#enrichment`

Chi tiết từng block

#regex: Phân tích dữ liệu thô thành JSON

• Mục đích: Dùng regex để tách các trường từ log thô sang JSON (field - value) dựa trên các mẫu định dạng cố định (pattern).
• Khi nào cần dùng #regex?
  • Dữ liệu log là chuỗi thô và chưa có parser dựng sẵn.
  • Bạn cần tách các phần như IP, thời gian, method, URL, status, bytes.
  • Định dạng log cố định theo từng dòng.
• Một số regex thường dùng
  • \d+: số nguyên
  • \S+: chuỗi không chứa khoảng trắng
  • [^\"]+: chuỗi cho đến trước dấu "
  • \[(.*?)\]: lấy nội dung trong [...]
  • \s+: một hoặc nhiều khoảng trắng
• Cú pháp: field_name = regex_pattern
• Ví dụ thực tế:

Log mẫu:

10.10.10.1 - - [12/Jul/2026:10:30:15 +0700] "GET /api/v1/users?id=1 HTTP/1.1" 200 1234

Regex minh họa để bóc tách các phần chính:

(\S+)\s+-\s+-\s+\[(.*?)\]\s+"(\S+)\s+(\S+)\s+(.*?)"\s+(\d+)\s+(\d+)

Ý nghĩa từng group:

• Group 1: IP client (10.10.10.1)
• Group 2: Thời gian (12/Jul/2026:10:30:15 +0700)
• Group 3: HTTP method (GET)
• Group 4: URL (/api/v1/users?id=1)
• Group 5: HTTP version (HTTP/1.1)
• Group 6: Status code (200)
• Group 7: Bytes (1234)

#conditional: Xử lý logic và biến đổi dữ liệu

• Mục đích: Thực hiện các phép xử lý dữ liệu sau khi parse như chuyển đổi định dạng, tách dữ liệu, tính toán giá trị mới và xử lý theo điều kiện trước khi đưa vào #normalize.
• Khi nào cần dùng #conditional?
  • Khi cần tách thêm dữ liệu từ một field đã có (ví dụ tách method và URL từ .request).
  • Khi cần rẽ nhánh theo điều kiện (ví dụ parse thành công hoặc parse lỗi).
  • Khi cần chuẩn hóa kiểu dữ liệu (đưa về chuỗi/số) trước khi map sang ECS.
  • Khi cần gán giá trị mặc định khi thiếu dữ liệu.
• Hàm thường dùng trong #conditional
  • Hàm xử lý chuỗi: split!, replace, trim, lowercase, uppercase, ...
  • Hàm chuyển đổi kiểu dữ liệu: to_string, to_int, to_float, ...
  • Hàm parser dựng sẵn: parse_nginx_log, parse_json, parse_key_value, ...
  • Hàm kiểm tra điều kiện: exists, contains, matches, ...

• Ví dụ thực tế:

#conditional

parsed, err = parse_nginx_log(.message)

if err == null {
    .status = to_int(parsed.status)
    .bytes = to_int(parsed.bytes)
    .path = split!(parsed.request, "?")[0]

    if .status >= 400 {
        .outcome = "FAIL"
    } else {
        .outcome = "SUCCESS"
    }
} else {
    .path = .message
    .status = 0
    .bytes = 0
    .outcome = "UNKNOWN"
}

Giải thích:

• parsed, err = parse_nginx_log(.message): Thử parse nội dung log trong .message.
• if err == null: Parse thành công, tiếp tục:
  • Ép status và bytes sang kiểu số.
  • Tách URL để lấy path, loại bỏ query string (?id=1).
  • Xác định outcome:
    • status >= 400 → FAIL
    • status < 400 → SUCCESS
  • else: Parse thất bại:
    • Fallback về .message để vẫn giữ lại dữ liệu gốc.
    • Gán giá trị mặc định (0, UNKNOWN) để các bước #normalize và #enrichment không bị lỗi do thiếu field.
• Lỗi thường gặp và cách xử lý:
  • Truy cập index không tồn tại sau split!: Kiểm tra dữ liệu đầu vào hoặc thêm fallback.
  • Không kiểm tra err: Dễ làm pipeline lỗi dây chuyền ở bước normalize.
  • Trộn kiểu dữ liệu: Dùng to_string trước khi thao tác chuỗi.
  • Biến chưa khởi tạo: Luôn khai báo biến ở đầu block để tránh giá trị rỗng ngoài ý muốn.

#normalize: Ánh xạ sang ECS Schema

• Mục đích: Ánh xạ dữ liệu đã xử lý vào các trường chuẩn của hệ thống CS (Common Schema).
• Khi nào cần dùng normalize:
  • Khi cần ánh xạ các field từ log nguồn sang schema chuẩn.
  • Khi cần thống nhất tên field giữa nhiều loại log khác nhau.
  • Khi cần chuẩn hóa cấu trúc dữ liệu để phục vụ Search, Detection, Dashboard và Enrichment.
  • Khi muốn các parser từ nhiều vendor khác nhau đều tạo ra cùng một output.
• Cú pháp: cs_field: source_var_or_field

Ví dụ:

#normalize
http.request.referrer: .referer
client.ip: .client
http.request.method: method
url.full: url_full
url.path: url_path
url.query: url_query
http.response.status: .status
agent.name: .agent
http.request.body.bytes: .size
error.message: err

#enrichment: Bổ sung thêm thông tin

• Mục đích: giúp bổ sung thêm thông tin (context) vào log bằng cách tra cứu (lookup) từ các nguồn dữ liệu có sẵn.
• Khi nào cần dùng enrichment: Sử dụng Enrichment khi muốn gắn thêm metadata vào log để phục vụ tìm kiếm, điều tra và xây dựng Detection Rule. Các trường hợp phổ biến:
  • Xác định vị trí địa lý từ IP.
  • Xác định ISP hoặc ASN của IP.
  • Ánh xạ username/email thành thông tin người dùng.
  • Ánh xạ device ID thành thông tin thiết bị.
  • Bổ sung thông tin từ danh sách nội bộ như: Asset Inventory; VIP Users; Critical Servers; Blacklist / Whitelist,...
• Các loại Lookup hỗ trợ:

• Cú pháp:

#enrichment

lookup_<table_name>: <lookup_key> = <log_field>
  .<target_field_1> = <source_column_1>
  .<target_field_2> = <source_column_2>

• Ví dụ thực tế:

VD1: Lookup thông tin vị trí địa lý

• geoip_city là bảng dùng để tra cứu vị trí địa lý của 1 địa chỉ IP (quốc gia, thành phố, tọa độ).
• geoip_asn là bảng để tra cứu thông tin mạng (ISP) và Autonomous System (AS) của IP.

#enrichment

lookup_geoip_city: ip = .src_ip
  .customs.country = country_name
  .customs.city = city_name

=> Giải thích: Nếu giá trị của cột ip trong lookup_geoip_city khớp với giá trị của trường .src_ip trong log, hệ thống sẽ lấy các cột được chỉ định (như country_name, city_name) và gán chúng vào các trường đích (.customs.country, .customs.city) trong event.

#enrichment

lookup_geoip_asn: ip = .src_ip
  .customs.asn = autonomous_system_number
  .customs.organization = autonomous_system_organization

VD2: Lookup từ bảng do người dùng tự tạo

• Người dùng tự upload file trong tính năng Import file với tên file là asset_inventory

#enrichment

lookup_asset_inventory: ip = .src_ip
  .customs.hostname = hostname
  .customs.owner = owner
  .customs.department = department

VD3: Lookup thông tin User hoặc Device đã được đồng bộ từ các hệ thống quản lý khác

• Người dùng có thể xem danh sách Entities ở đây.
• entities: là bảng tra cứu thông tin User/Device đã được hệ thống đồng bộ từ các nguồn như Entra ID, SentinelOne...

#enrichment

lookup_entities: username = .username
  .customs.display_name = display_name
  .customs.department = department
  .customs.job_title = job_title

Cấu hình mẫu (Nginx Combined Log)

Đây là ví dụ đầy đủ về phân tích định dạng log Nginx kiểu combined:

#regex
([^"]+)

#conditional
method = ""
url_full = ""
url_path = ""
url_query = ""
if (., err = parse_nginx_log(.message, "combined"); err == null) {
  method = split!(.request, " ")[0]
  url_full = split!(.request, " ")[1]
  url_path = split(to_string(url_full), "?")[0]
  url_query = split(to_string(url_full), "?")[1]
} else {
  . = .message
}

#normalize
http.request.referrer: .referer
client.ip: .client
http.request.method: method
url.full: url_full
url.path: url_path
url.query: url_query
http.response.status: .status
agent.name: .agent
http.request.body.bytes: .size
error.message: err

#enrichment

lookup_geoip_city: ip = .client
  .customs.country = country_name
  .customs.city = city_name

lookup_geoip_asn: ip = .client
  .customs.asn = autonomous_system_number
  .customs.organization = autonomous_system_organization

Các điểm cần ghi nhớ

• Xử lý tuần tự: Dữ liệu đi theo thứ tự #regex → #conditional → #normalize → #enrichment.
• Xử lý lỗi: Luôn kiểm tra lỗi bằng biến err và có logic fallback
• Tuân thủ ECS: Đảm bảo mọi trường ánh xạ đúng theo ECS schema
• Hàm dựng sẵn: Tận dụng parser dựng sẵn như parse_nginx_log khi có để đơn giản cấu hình
• Kiểm thử: Luôn kiểm thử parser với log mẫu trước khi triển khai môi trường thực tế