Logs pipeline: Logpush, R2, SIEM và cross-layer correlation

TL;DR

Logs là nền tảng observability của Cloudflare One. Dashboard có sẵn chỉ giữ 30 ngày, rất khó tương quan đa tầng. Production cần:

• Logpush: xuất log dạng streaming tới đích nhận (R2, S3, Splunk, Sentinel, Datadog, …).
• Retention phân tầng: hot 30d (CF), warm 1y (R2/S3), cold 7y (Glacier).
• Tương quan đa tầng: join DNS + Network + HTTP + Access logs theo UserID để bắt tấn công nhiều giai đoạn.
• Rule phát hiện trong SIEM.
• Kiểm soát chi phí: lấy mẫu, nén, lifecycle policy.

Bài này đi qua:

• Dataset Cloudflare One phơi ra.
• Cơ chế Logpush: gom batch, định dạng, đích nhận.
• R2 làm data lake: thiết lập và truy vấn với Athena/DuckDB.
• Mẫu tích hợp SIEM: Splunk, Sentinel, Elastic.
• Rule tương quan bắt DoH bypass, credential stuffing, session hijack.
• Chi phí và retention.

Luận điểm chính:

Logs không phải thứ nghĩ sau, nó là xương sống của Zero Trust. Nếu không đưa vào SIEM và không tương quan đa tầng, bạn chỉ có phòng ngừa mà không có phát hiện. Kẻ tấn công sẽ thử đủ hướng, và bạn cần thấy hết để nhận ra pattern.

Bài này là Part 14 của Cloudflare One Handbook, mở khối Observability & Ops (Part 14-16).

---

Dành cho ai

• Kỹ sư bảo mật xây tích hợp SIEM.
• SOC analyst viết rule phát hiện đa tầng.
• Kỹ sư nền tảng thiết lập data lake observability.

Bạn nên đã đọc:

• Bất kỳ Part 11, 12, 13: đã đề cập Logpush nhưng chỉ ở bề mặt.
• Part 9, WARP (làm giàu identity cho logs).

Sau bài này bạn sẽ:

• Biết tất cả dataset Cloudflare One và field quan trọng.
• Thiết lập Logpush tới R2 và SIEM với đúng định dạng.
• Viết rule tương quan đa tầng phát hiện chuỗi tấn công.
• Lên kế hoạch retention và chi phí không vỡ ngân sách.

---

Bài này không nói về gì

• Logs Cloudflare chung ngoài Zero Trust (CDN, WAF, Workers: bài khác).
• Tinh chỉnh SIEM chi tiết (Splunk SPL, tối ưu KQL): chỉ nhắc qua.
• Báo cáo đặc thù tuân thủ (PCI, HIPAA): chỉ nhắc retention cần, chi tiết để bài khác.
• Analytics Engine: bài riêng.

---

Khái niệm

• Dataset: dòng log Cloudflare phơi ra; mỗi dataset có schema cố định (gateway_dns, gateway_network, gateway_http, access_requests, …).
• Logpush: dịch vụ CF đẩy log theo batch tới đích nhận qua giao thức HTTP/S3.
• Đích nhận: nơi tiếp nhận: R2, S3, GCS, Azure Blob, Splunk HEC, Sumo Logic HTTP, Datadog, New Relic.
• NDJSON: JSON phân tách bằng dòng mới, định dạng mặc định của Logpush.
• Logpull: API kéo cũ (đã ngừng cho phần lớn dataset).
• SIEM: Security Information and Event Management: Splunk, Sentinel, Elastic, Sumo, Datadog.
• Data lake: lưu trữ khối (R2/S3) + engine truy vấn (Athena, DuckDB, BigQuery).

---

Dataset của Cloudflare One

Dataset	Content	Volume (enterprise, typical)
access_requests	Every ZTNA auth attempt, user, app, decision, identity	10K-100K/day
gateway_dns	Every DNS query through Gateway	10M-500M/day
gateway_network	L4 TCP/UDP connection events	50M-1B/day
gateway_http	HTTP request (if decrypt được bật)	10M-500M/day
audit_logs	Admin/config changes	100-10K/day
device_posture_results	Device posture check results	1M-50M/day
zero_trust_dex_test_results	DEX (Digital Experience Monitoring) tests	10K-1M/day
access_login	ZTNA login event (session)	1K-100K/day
casb_findings	CASB scan findings on connected SaaS	100-10K/day

Đặc điểm dataset

• Access logs: lượng thấp, giá trị cao. Thiết yếu cho điều tra số.
• Gateway DNS: độ phủ rộng, lượng trung bình. Giám sát baseline.
• Gateway Network: lượng cao nhất. Tốn chi phí. Lấy mẫu mạnh tay.
• Gateway HTTP: lượng trung bình-cao. Phụ thuộc phạm vi giải mã.
• Audit: phải đẩy 100% (tuân thủ).

---

Cơ chế Logpush

Hành vi batch

• Kích hoạt batch: tick 5 phút HOẶC kích thước 5MB (cái nào tới trước).
• Batch tối đa: 5MB đã nén → ~50MB chưa nén.
• Giao nhận: tối thiểu một lần (hiếm khi trùng, SIEM cần khử trùng theo event ID).
• Thứ tự: KHÔNG đảm bảo trong batch; dùng trường timestamp.

Tuỳ chọn format

{
  "output_options": {
    "output_type": "ndjson",           // or "csv"
    "timestamp_format": "rfc3339",     // or "unix"
    "field_names": [],                 // empty = all
    "field_delimiter": ",",            // csv only
    "record_delimiter": "\n"
  }
}

Loại đích nhận

• Object storage: R2, S3, GCS, Azure Blob. Mẫu đường dẫn: dataset/{DATE}/{TIME}.
• HTTP webhook chung: Splunk HEC, Datadog, Sumo, tuỳ biến.
• Tích hợp SIEM sẵn có: connector Sentinel (hệ sinh thái Microsoft), app Splunk chính thức.

Template đường dẫn

r2://gateway-logs/{dataset}/year={YEAR}/month={MONTH}/day={DAY}/hour={HOUR}/{UUID}.json.gz

Bố cục hive-partitioned → Athena/DuckDB truy vấn hiệu quả. Partition theo year/month/day là chuẩn.

Lọc tại nguồn

{
  "filter": "{\"where\":{\"key\":\"Action\",\"operator\":\"eq\",\"value\":\"block\"}}"
}

Ví dụ: chỉ đẩy event block, giảm 80-90% lượng cho dataset DNS.

---

R2 làm data lake

Vì sao chọn R2 thay vì S3

• Egress miễn phí: truy vấn dữ liệu R2 không tốn phí egress (S3 tốn).
• Tích hợp Cloudflare sẵn có: Logpush trực tiếp, không phiền IAM.
• Storage rẻ ~$15/TB/tháng, tương đương S3 nhưng egress miễn phí là lợi thế lớn.

Thiết lập

1. Tạo bucket R2:

wrangler r2 bucket create gateway-logs

2. Tạo API token cho Logpush:

Dashboard → R2 → Manage R2 API Tokens → tạo token với Object Read & Write.

3. Cấu hình Logpush:

curl -X POST \
  "https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/logpush/jobs" \
  -H "Authorization: Bearer ${CF_API_TOKEN}" \
  -H "Content-Type: application/json" \
  --data @- <<'EOF'
{
  "name": "gateway-dns-to-r2",
  "dataset": "gateway_dns",
  "destination_conf": "r2://gateway-logs/dns/year={YEAR}/month={MONTH}/day={DAY}/?account-id=${ACCOUNT_ID}&access-key-id=${R2_KEY}&secret-access-key=${R2_SECRET}",
  "output_options": {
    "output_type": "ndjson",
    "timestamp_format": "rfc3339"
  },
  "enabled": true
}
EOF

4. Xác minh:

wrangler r2 object list gateway-logs --prefix dns/

Batch xuất hiện sau 5-10 phút.

Truy vấn với DuckDB (local)

-- Install httpfs extension
INSTALL httpfs;
LOAD httpfs;

-- Configure R2 credentials
SET s3_endpoint='<account-id>.r2.cloudflarestorage.com';
SET s3_access_key_id='<r2-key>';
SET s3_secret_access_key='<r2-secret>';
SET s3_url_style='path';

-- Query
SELECT
  Email,
  COUNT(*) as blocked_count,
  ARRAY_AGG(DISTINCT DNSQuestion) as domains
FROM read_json('s3://gateway-logs/dns/year=2026/month=05/day=15/*.json.gz')
WHERE Action = 'block'
GROUP BY Email
ORDER BY blocked_count DESC
LIMIT 20;

Chạy local, không cần cluster.

Truy vấn với Athena (AWS)

R2 tương thích S3, Athena truy vấn trực tiếp nếu bucket public hoặc có role cross-account. Pattern:

1. Glue Crawler quét các partition R2 → bảng catalog.
2. Athena SQL chạy trên catalog.
3. Dùng CTAS (Create Table As Select) cho tầng tổng hợp.

Chi phí: Athena $5/TB quét, partition pruning quan trọng.

---

Mẫu tích hợp SIEM

Splunk

Phương án A, HEC trực tiếp:

{
  "destination_conf": "splunk://hec.splunk.company.com/services/collector/raw?header_Authorization=Splunk%20<token>&channel=<uuid>&header_Content-Type=application%2Fjson&insecure-skip-verify=false",
  "dataset": "gateway_dns"
}

Phương án B, S3 → Splunk pull (tiết kiệm chi phí khi lượng cao):

Logpush → R2 → Splunk SmartStore hoặc script kéo tuỳ biến.

Ví dụ SPL, phát hiện né DoH:

index=cloudflare sourcetype=gateway_network
| where match(SNI, "dns\\.google|dns\\.quad9\\.net|doh\\.opendns\\.com")
| where Action="block"
| stats count by UserID, Email, SNI
| where count > 5
| sort -count

Microsoft Sentinel

Connector Cloudflare sẵn có (trong Content Hub):

1. Cài đặt connector “Cloudflare (using Azure Function)”.
2. Cung cấp Cloudflare API token + account ID.
3. Luồng dữ liệu vào workspace Log Analytics.

Phương án khác: Logpush → Sentinel qua Event Hub.

Ví dụ KQL, tương quan đa tầng:

let timeframe = 1h;
Cloudflare_Gateway_DNS_CL
| where TimeGenerated > ago(timeframe)
| where Action_s == "block"
| where Categories_s contains "Malware"
| project UserID=UserID_s, blockedDomain=DNSQuestion_s, t_dns=TimeGenerated
| join kind=inner (
    Cloudflare_Gateway_Network_CL
    | where TimeGenerated > ago(timeframe)
    | where Action_s == "block"
    | project UserID=UserID_s, blockedIP=DestinationIP_s, t_net=TimeGenerated
) on UserID
| where t_net between (t_dns .. t_dns + 5m)
| project UserID, blockedDomain, blockedIP, t_dns, t_net

Phát hiện: người dùng bị chặn DNS, 5 phút sau thử IP trực tiếp → ý đồ đa tầng.

Elastic / OpenSearch

Logstash pipeline:

input {
  s3 {
    bucket => "gateway-logs"
    endpoint => "<account-id>.r2.cloudflarestorage.com"
    access_key_id => "<r2-key>"
    secret_access_key => "<r2-secret>"
    codec => "json_lines"
  }
}

filter {
  if [dataset] == "gateway_dns" {
    mutate { add_field => { "[@metadata][index]" => "cf-gateway-dns-%{+YYYY.MM.dd}" } }
  }
}

output {
  elasticsearch {
    hosts => ["https://es.company.com:9200"]
    index => "%{[@metadata][index]}"
  }
}

Datadog

Đích nhận HTTP webhook:

datadog://http-intake.logs.datadoghq.com/api/v2/logs?header_DD-API-KEY=<key>&ddsource=cloudflare&service=gateway

Datadog phân tích JSON tự động, áp dụng pipeline xử lý tag.

---

Rule tương quan đa tầng

Rule 1: Thử né DoH

Tín hiệu: Người dùng có DNS block + Network DoH block trong cùng 15 phút.

Mô hình mối đe doạ: malware thử resolver hệ thống → bị chặn → dự phòng DoH → cũng bị chặn. Xác nhận ý đồ xâm nhập.

let win = 15m;
let dns_blocks = Cloudflare_Gateway_DNS_CL
  | where Action_s == "block" and Categories_s has "Malware"
  | project UserID, t1=TimeGenerated, dns_q=DNSQuestion_s;
let doh_blocks = Cloudflare_Gateway_Network_CL
  | where Action_s == "block" and PolicyName_s has "DoH"
  | project UserID, t2=TimeGenerated, sni=SNI_s;
dns_blocks
| join kind=inner doh_blocks on UserID
| where t2 between (t1 .. t1 + win)
| project UserID, dns_q, sni, t1, t2

Mức độ: cao, mở ticket, cô lập thiết bị.

Rule 2: Credential stuffing vào Access

Tín hiệu: N lần Access login thất bại cho cùng một app trong 10 phút từ các IP khác nhau.

Cloudflare_Access_CL
| where TimeGenerated > ago(10m)
| where Result_s == "blocked"
| summarize ip_count=dcount(IP_s), attempts=count() by App_s
| where attempts > 20 and ip_count > 5

Rule 3: Dấu hiệu chiếm phiên

Tín hiệu: Cùng UserID, cùng SessionID, IP khác và quốc gia khác trong 5 phút.

Cloudflare_Access_CL
| where TimeGenerated > ago(5m)
| summarize ip_list=make_set(IP_s), country_list=make_set(Country_s) by UserID, SessionID
| where array_length(ip_list) > 1 and array_length(country_list) > 1

Rule 4: Bùng nổ rò rỉ dữ liệu

Tín hiệu: Người dùng tải lên > 1GB qua HTTP POST trong 30 phút, đích không phải SaaS của công ty.

Cloudflare_Gateway_HTTP_CL
| where TimeGenerated > ago(30m)
| where Method_s == "POST" and ContentLength_d > 0
| where Host_s !in ("drive.google.com", "onedrive.live.com", "s3.company.com")
| summarize total_bytes=sum(ContentLength_d) by UserID, Email
| where total_bytes > 1073741824  // 1 GB

Rule 5: Di chuyển bất khả thi

Tín hiệu: Cùng một người dùng login từ hai quốc gia cách nhau quá xa trong thời gian ngắn (khoảng cách / thời gian > 1000 km/h).

Cloudflare_Access_CL
| where Result_s == "allowed"
| project UserID, Country_s, Latitude_d, Longitude_d, TimeGenerated
| sort by UserID, TimeGenerated asc
| extend prev_country=prev(Country_s), prev_time=prev(TimeGenerated),
         prev_lat=prev(Latitude_d), prev_lon=prev(Longitude_d)
| extend time_diff_h = (TimeGenerated - prev_time) / 1h
| extend dist_km = geo_distance_2points(prev_lon, prev_lat, Longitude_d, Latitude_d) / 1000
| where time_diff_h > 0 and time_diff_h < 12 and dist_km / time_diff_h > 1000

---

Chiến lược retention

Storage theo tier

Tier	Vị trí	Retention	Chi phí (/TB/tháng)	Dùng cho
Hot	Dashboard Cloudflare	30d	đã gồm	truy vấn dashboard, debug
Warm	R2	1y	~$15	điều tra số, nguồn SIEM
Cold	R2 archive / Glacier	7y	~$1-4	tuân thủ, tranh chấp

R2 lifecycle policy

{
  "rules": [
    {
      "id": "archive-after-90d",
      "prefix": "gateway-logs/",
      "transitions": [
        { "days": 90, "storage_class": "INFREQUENT_ACCESS" }
      ]
    },
    {
      "id": "delete-after-7y",
      "prefix": "gateway-logs/",
      "expiration": { "days": 2555 }
    }
  ]
}

Mức tối thiểu tuân thủ

• PCI DSS 4.0: 1 năm online, tối thiểu 1 năm cold cho audit log liên quan cardholder.
• HIPAA: 6 năm.
• SOC 2: 1 năm audit, 3-7 năm sự kiện bảo mật.
• GDPR: tối thiểu cần, xoá khi mục đích đã hoàn tất (thường < 1 năm cho chi tiết, lâu hơn cho dữ liệu tổng hợp).

---

Kiểm soát chi phí

Giảm khối lượng

1. Lọc tại Logpush, chỉ đẩy event quan trọng:

"filter": "{\"where\":{\"key\":\"Action\",\"operator\":\"ne\",\"value\":\"allow\"}}"

Bỏ event allowed → khối lượng giảm 80-90% cho DNS/Network.

2. Lấy mẫu, lấy mẫu ngẫu nhiên event allowed để baseline:

CF không hỗ trợ lấy mẫu sẵn có ở Logpush → tuỳ biến: đẩy tất cả → lấy mẫu ở pipeline ingest SIEM.

3. Nén, Logpush gzip mặc định. Khi push HTTP webhook, xác minh endpoint hỗ trợ gzip.

4. Chọn dataset, không phải dataset nào cũng cần vào SIEM:

• gateway_dns chỉ blocked → SIEM.
• gateway_dns allowed → R2 raw (cold).
• gateway_network chỉ blocked → SIEM.
• audit_logs 100% → SIEM (tuân thủ).

Tính toán ngân sách storage

Doanh nghiệp 1000 người dùng, 50M event/ngày tổng:

• NDJSON ~500 bytes/event, nén còn ~100 bytes.
• 5 GB/ngày chưa nén → 500 MB/ngày sau nén.
• Một năm: ~180 GB đã nén.
• R2: 180 GB × $0.015 = ~$2.70/tháng. Rẻ.

SIEM ingest (Splunk): $1,800/GB/năm là điển hình. Lọc mạnh tay trước khi đẩy.

Tỉ lệ nén

NDJSON lặp lại (tên field lặp đi lặp lại) → tỉ lệ gzip 5-10x. Định dạng Parquet tốt hơn nhưng Logpush chưa hỗ trợ sẵn → chuyển đổi ở tầng R2 bằng cron job.

---

Vận hành: liên tục

Giám sát sức khoẻ Logpush

# Check job status
curl "https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/logpush/jobs" \
  -H "Authorization: Bearer ${CF_API_TOKEN}" | jq '.result[] | {name, enabled, last_complete, last_error}'

Alert khi last_complete cũ hơn 30 phút = job kẹt.

Mẫu cảnh báo

• Logpush job fail > 3 lần liên tiếp → pager.
• Khối lượng log giảm > 50% baseline → điều tra (job kẹt? CF outage? filter cấu hình sai?).
• Độ trễ ingest SIEM > 15 phút → SOC escalate.
• Lượng dùng storage gần chạm quota → cảnh báo trước khi vượt.

Xác minh pipeline hàng tuần

• Sinh event tổng hợp (query DNS thử tới domain canary bị chặn bởi policy).
• Xác minh nó đi tới: dashboard CF (5 phút) → R2 (10 phút) → SIEM (15 phút).
• Ghi tài liệu: /runbooks/logs-pipeline-health-check.md.

---

Xử lý sự cố

”Logpush job fail thất thường”

• Kiểm tra trường last_error.
• Phổ biến: quota đích nhận, token xác thực hết hạn, vấn đề mạng.
• CF tự động retry 3 lần trong 15 phút; nếu fail hết, batch sau sẽ thử tiếp.

”SIEM không thấy log mới”

1. Dashboard: có event không? → nếu không, vấn đề ở phía trên.
2. Trạng thái job Logpush: active, last_complete gần đây?
3. Đích nhận có kết nối được không? Thử bằng curl.
4. Queue ingest SIEM có tồn đọng?
5. Rule parser từ chối event? Kiểm tra log lỗi SIEM.

”Khối lượng log đột ngột tăng 10x”

• Policy giải mã mới được bật → khối lượng HTTP tăng vọt.
• Location DNS mới có lưu lượng cao.
• Kịch bản tấn công (scan, DDoS).
• Kiểm tra: xu hướng event/ngày ở dashboard Logpush.

”Query Athena tốn kém”

• Partition pruning không chạy → kiểm tra partition tồn tại trong Glue catalog.
• Quét toàn bộ prefix lớn → thêm filter WHERE year/month/day.
• Chuyển NDJSON → Parquet cho khoảng hay truy vấn.

”Event trùng lặp trong SIEM”

• Logpush tối thiểu một lần. CF retry có thể gây trùng.
• Khử trùng theo (Timestamp, EventID) ở SIEM.
• Trường Event ID khác nhau theo dataset: tra tài liệu schema.

---

Danh sách kiểm tra: production logs pipeline

Độ phủ dataset:

• access_requests → SIEM 100%.
• audit_logs → SIEM 100%.
• gateway_dns blocks → SIEM, toàn bộ event → R2.
• gateway_network blocks → SIEM, toàn bộ event → R2.
• gateway_http blocks + đường dẫn nhạy cảm → SIEM, toàn bộ → R2.
• device_posture_results → SIEM tóm tắt hàng ngày.
• zero_trust_dex_test_results → SIEM.

Hạ tầng:

• Bucket R2 đã tạo + lifecycle policy.
• Job Logpush cho mỗi dataset đã cấu hình.
• Giám sát trạng thái job + cảnh báo.
• Retention khớp tuân thủ (PCI 1y+, SOC 3y+, HIPAA 6y).

SIEM:

• Connector sẵn có hoặc HEC/webhook đã cấu hình.
• Parser/dashboard cho mỗi dataset.
• Rule tương quan đa tầng đã triển khai.
• Điều phối cảnh báo tới SOC/PagerDuty.

Phát hiện:

• Rule né DoH.
• Rule credential stuffing.
• Rule chiếm phiên.
• Rule rò rỉ dữ liệu.
• Rule di chuyển bất khả thi.
• Phát hiện đặc thù policy (rule nhận biết tenant).

Vận hành:

• Xác minh event tổng hợp hàng tuần.
• Rà soát khối lượng log hàng tháng.
• Tinh chỉnh rule phát hiện hàng quý.
• Runbook cho “pipeline down”.

---

Bài học thực tế

• Đẩy mọi dataset, bật mọi rule ngay → chi phí ingest vượt ngân sách. Bắt đầu tinh gọn: chỉ block + audit 100%, mở thêm khi có trường hợp sử dụng phát hiện.
• Chuyển đổi NDJSON → Parquet tiết kiệm 60-70% storage + chi phí truy vấn. Đáng tạo cron job sau 30 ngày.
• Tương quan đa tầng đáng giá. Cảnh báo một tầng thì ồn; join nhiều tầng cắt hẳn dương tính giả.
• Kẻ tấn công nhắm vào hệ thống log. Rule phát hiện “khối lượng log giảm” rất quan trọng: kẻ tấn công tắt log = dấu hiệu đầu tiên.
• R2 là điểm ngọt cho khối lượng trung bình. S3 đắt ở egress; BigQuery quá tay; Elastic ingest đắt.
• Tinh chỉnh SIEM hàng quý. App mới, SaaS mới, threat mới → rule phát hiện phải cập nhật. Rule cũ = an toàn giả.
• Nén + partition không phải “nên có”: ở quy mô production, truy vấn dữ liệu không partition tốn vài trăm $ mỗi lần.
• Test tổng hợp hàng tuần. Pipeline hoạt động thầm lặng cho tới khi lỗi. Event canary là cách duy nhất biết pipeline còn sống.

---

Kết luận

Logs pipeline là nền tảng của khối Observability & Ops. Không có nó, Zero Trust chỉ còn phòng ngừa, kẻ tấn công thử đủ hướng mà bạn không thấy pattern.

Công thức production:

• Logpush → R2 data lake 1 năm.
• Ingest SIEM: block + audit + tín hiệu giá trị cao.
• Tương quan đa tầng là siêu năng lực: join UserID giữa các dataset.
• Retention theo tier: hot/warm/cold theo tuân thủ.
• Kiểm soát chi phí: lọc ở nguồn, đừng đẩy tất cả vào SIEM.

Nếu phải nhớ một câu:

Logs không chỉ là dữ liệu, nó là bằng chứng của kiểm soát. Không có tương quan đa tầng, Zero Trust chỉ là zero visibility.

Trong Part 15 chuyển sang DEX, Digital Experience Monitoring: đo độ trễ, sức khoẻ WARP, khả năng tiếp cận app từ góc nhìn người dùng cuối, phát hiện vấn đề trước khi có ticket helpdesk.

---

Tài liệu tham khảo

• Logpush overview
• Logpush dataset schemas
• Zero Trust logs datasets
• Logpush to R2
• Microsoft Sentinel Cloudflare connector
• Splunk Cloudflare app

Trong series này:

• ← Part 13: Network policy L4
• Tiếp theo → Part 15: DEX, Digital Experience Monitoring
• Xem toàn bộ: Series Cloudflare One Handbook