Log Collection and Processing Pipelines

ไอเดียหลักในหนึ่งประโยค

log จะเข้า Datadog ได้เพราะ Agent ถูกบอกผ่าน logs: config block ว่าอ่านจากไหน แล้วติด label service/source อะไร และพอ log มาถึง Processing Pipeline จะ parse กับ enrich log แต่ละตัวก่อนที่จะไปถึง index

บอก Agent ว่าให้มองที่ไหน

Agent ไม่ได้ไปตามหา log เองลอย ๆ — ทุก log source ต้องระบุชัดเจน คุณเพิ่ม logs: block เข้าไป โดยปกติจะอยู่ใน conf.d integration config หรือ logs config ของ Agent เอง แล้วบอกว่า source เป็นประเภทไหน, path (หรือ port สำหรับ tcp) อยู่ที่ไหน และ label สองตัวที่สำคัญมากในขั้นตอนถัดไป คือ service กับ source

logs:
  - type: file
    path: /my/app/file.log
    service: payments-api
    source: java

• type — Agent อ่าน log แบบไหน file คือ tail ไฟล์บน disk ส่วนค่าที่ใช้บ่อยอื่น ๆ คือ tcp, udp และ docker สำหรับเก็บ log จาก container
• path — log อยู่ที่ไหน สำหรับ type file เป็น path ของไฟล์ตรง ๆ ไม่ใช่ query
• service — ชื่อ service ที่สร้าง log นี้ขึ้นมา เป็น tag service ตัวเดียวกับที่ใช้ทุกที่ใน Datadog (metric, trace) ที่เป็นตัวที่ทำให้ log line, trace และ metric ผูกกลับไปหา deployable unit เดียวกันได้
• source — เป็น technology identifier ไม่ใช่ label แบบข้อความอิสระ การตั้ง source: java (หรือ nginx, postgresql ฯลฯ) จะบอก Datadog ว่าให้แนบ pipeline ที่ built-in เฉพาะ source นั้น ให้อัตโนมัติ เพราะ Datadog มี Grok parsing rule สำเร็จรูปให้ framework/service ยอดนิยมหลายสิบตัว

พอ Agent อ่าน config นี้ในรอบ reload ถัดไป จะเริ่ม tail /my/app/file.log แล้วส่งแต่ละบรรทัดไปที่ intake ของ Datadog พร้อม tag service:payments-api และ source:java

Processing Pipeline: parse และ enrich หลัง intake

Processing Pipeline คือ list ของ processor ที่เรียงลำดับไว้ รันกับ log ทุกตัวที่มาถึง intake และ match query filter ของ pipeline นั้น ขั้นตอนนี้เกิดหลังจาก log ออกจาก host มาถึง Datadog แล้ว และก่อนที่จะถูกเขียนลง index ไหน ๆ — หน้าที่ของตัวเองคือ parse ข้อความที่ไม่มีโครงสร้างให้เป็น attribute ที่มีโครงสร้าง แล้ว enrich log เท่านั้น ไม่ใช่การตัดสินว่า log จะถูกเก็บไว้หรือไม่

pipeline ถูก scope ด้วย filter เช่น service:my-service เพื่อให้มีแค่ log จาก service หรือ source ที่ตั้งใจไว้เท่านั้นที่ผ่านเข้ามา คุณมี pipeline หลายตัวทำงานพร้อมกันได้ แต่ละตัว scope ไปคนละ service หรือ source รวมถึง pipeline built-in ที่ source: java, source: nginx ฯลฯ แนบให้อัตโนมัติด้วย

processor ที่พบบ่อยใน pipeline เรียงตามลำดับที่รัน:

• Grok Parser — apply Grok pattern กับ raw log message เพื่อดึง attribute ที่มีโครงสร้างออกจากข้อความที่ไม่มีโครงสร้าง เช่น ดึง duration, http.status_code, หรือ user.id ออกจากบรรทัดที่เดิมเป็น string ยาว ๆ ก้อนเดียว
• Date Remapper — บอก Datadog ว่า attribute ไหนที่ parse ออกมาแล้วคือ timestamp ทางการของ log นี้ (แทนที่จะ fallback ไปใช้เวลา intake) ซึ่งสำคัญกับการวาง log ให้ถูกตำแหน่งบน timeline
• Status Remapper — map attribute ที่ parse ออกมา (เช่น field ข้อความที่มีคำว่า ERROR หรือ WARN) เข้ากับ field status (severity) มาตรฐานของ Datadog เพื่อให้ view และ facet ที่อิงตาม severity ทำงานสอดคล้องกันทุก service

Pipeline: "payments-api logs"
filter: service:payments-api

1. Grok Parser    -> extract duration, http.status_code, user.id from message
2. Date Remapper  -> use parsed `timestamp` attribute as the log's official time
3. Status Remapper -> map parsed `level` attribute to Datadog's status field

flowchart LR
  A["logs: block in Agent config\ntype/path/service/source"] --> B[Agent tails the file]
  B --> C[Intake]
  C --> D{Pipeline filter matches?}
  D -->|yes| E[Grok Parser]
  E --> F[Date Remapper]
  F --> G[Status Remapper]
  G --> H["Parsed, enriched log\n(not yet indexed)"]
  D -->|no match| H

จาก Agent config ไปเป็น log ที่ parse แล้ว

🐶 สรุปในหนึ่งบรรทัด

logs: config block ของ Agent (type, path, service, source) คือตัวที่ทำให้ log ถูกเก็บได้ตั้งแต่แรก และ Processing Pipeline — ชุด processor ที่เรียงลำดับและ scope ด้วย filter อย่าง Grok Parser, Date Remapper, Status Remapper — parse และ enrich log ทุกตัวที่ match หลัง intake ก่อนถึง index ไม่ได้ตัดสินว่าอะไรจะถูกเก็บไว้

พิสูจน์: Agent จริงที่ tail ไฟล์จริงจริง ๆ

mount config ด้านบน (เป็น conf.d/mylogs.d/conf.yaml บวก block type: tcp อีกตัว) เข้า container datadog/agent:7 ที่ pull มาสด ๆ จริง (DD_API_KEY ปลอม DD_DD_URL ชี้ไป local port ที่ปิดอยู่ — ไม่มีอะไรส่งไปไหนเลย) แล้วรัน agent configcheck

conf.d/mylogs.d/conf.yaml

logs:
  - type: file
    path: /var/log/lesson/app.log
    service: payments-api
    source: java
    log_processing_rules:
      - type: exclude_at_match
        name: exclude_healthcheck_lines
        pattern: 'GET /healthz'

$ docker exec dd-logs agent configcheck
=== mylogs check ===
Configuration provider: file
Configuration source: file:/etc/datadog-agent/conf.d/mylogs.d/conf.yaml
Log Config:
logs:
  - log_processing_rules:
      - name: exclude_healthcheck_lines
        pattern: GET /healthz
        type: exclude_at_match
    path: /var/log/lesson/app.log
    service: payments-api
    source: java
    type: file

config ถูกยอมรับโดยไม่มี error เลย เพื่อพิสูจน์ว่า Agent กำลัง tail จริง ไม่ใช่แค่ parse config เฉย ๆ append บรรทัดจริงเข้าไปในไฟล์ที่ mount ไว้ แล้วเช็ค section Logs Agent ของ agent status ก่อนและหลัง

$ echo "2026-09-29 10:00:05 ERROR duration=812ms user=42 checkout failed" >> app.log
$ docker exec dd-logs agent status | grep -A9 'mylogs$'
  mylogs
  ------
    - Type: file
      Path: /var/log/lesson/app.log
      Service: payments-api
      Source: java
      Status: OK
      Bytes Read: 65
      Pipeline Latency:
        Average Latency: 1.001441792s

Bytes Read: 65 คือจำนวน byte จริงเป๊ะของบรรทัดที่เพิ่งเพิ่มเข้าไป — หลักฐานจริงที่วัดได้ว่า Agent อ่าน content ใหม่ ไม่ใช่แค่ echo config นิ่ง ๆ config block ตัวที่สองสำหรับ type: tcp (ประเภทที่สามของการเก็บ log นอกจาก file/docker) ก็ configcheck ผ่านสะอาดกับ Agent ตัวเดียวกันที่รันอยู่

conf.d/mytcplogs.d/conf.yaml

logs:
  - type: tcp
    port: 10518
    service: checkout-web
    source: nginx

$ docker exec dd-logs agent configcheck
=== mytcplogs check ===
Log Config:
logs:
  - port: 10518
    service: checkout-web
    source: nginx
    type: tcp

ประเภทที่สี่ type: docker (container log) ไม่ขอพูดซ้ำที่นี่ — The Datadog Agent and Integrations พูดถึง container-log Autodiscovery (annotation ad.datadoghq.com/<container>.logs บน Pod template) ไว้ลึกแล้ว concept logs: เดียวกันนี้ใช้ได้เหมือนกัน แค่ขับเคลื่อนด้วย annotation แทนไฟล์ conf.d แบบ static

Pipeline JSON ที่ validate แล้ว — และช่องว่างจริงใน harness ของคอร์สนี้เอง

สร้าง pipeline สามตัว processor ตามที่ร่างไว้ด้านบนเป็น JSON จริง แล้ว validate กับ datadog_api_client==2.61.0 ที่ติดตั้งจริง โดย model LogsPipeline

logs-pipelines/payments-api.json

{
  "name": "payments-api logs",
  "filter": { "query": "service:payments-api" },
  "is_enabled": true,
  "processors": [
    {
      "type": "grok-parser",
      "name": "Parse payments log line",
      "is_enabled": true,
      "grok": {
        "match_rules": "PaymentsRule %{date(\"yyyy-MM-dd HH:mm:ss\"):timestamp} %{word:level} duration=%{number:duration}ms user=%{number:user.id}"
      }
    },
    { "type": "date-remapper", "name": "Set official timestamp", "is_enabled": true, "sources": ["timestamp"] },
    { "type": "status-remapper", "name": "Map level to status", "is_enabled": true, "sources": ["level"] },
    {
      "type": "attribute-remapper",
      "name": "Rename duration",
      "is_enabled": true,
      "sources": ["duration"],
      "target": "http.response_time_ms",
      "source_type": "attribute",
      "target_type": "attribute"
    }
  ]
}

$ python -c "
from datadog_api_client.v1.model.logs_pipeline import LogsPipeline
import json
payload = json.load(open('payments-api.json'))
LogsPipeline(**payload, _check_type=True)
print('validated OK')
"
validated OK

🐶 ช่องว่างจริงใน harness ที่น่าแปลกใจ: processor ที่พังซ้อนอยู่ใน processors จับไม่ได้

สร้าง pipeline ที่พังโดยตั้งใจ — category-processor ที่ขาด field ที่จำเป็นสองตัว categories กับ target — แล้ว validate ทั้ง LogsPipeline แม้จะมี _check_type=True ก็ ไม่ raise error

$ python -c "
from datadog_api_client.v1.model.logs_pipeline import LogsPipeline
payload = {'name': 'broken', 'processors': [{'type': 'category-processor', 'name': 'x', 'is_enabled': True}]}
m = LogsPipeline(**payload, _check_type=True)
print(type(m.processors[0]))
"
<class 'datadog_api_client.v1.model.logs_processor.LogsProcessor'>

processors มี type เป็น Union ยักษ์ของ processor model เฉพาะทาง 20 กว่าตัว พอ dict ไม่ match กับตัวไหนแบบเป๊ะ ๆ client จะ fallback ไปใช้ type LogsProcessor ทั่วไปแบบเงียบ ๆ แทนที่จะ raise — มันไม่บอกด้วยซ้ำว่าคุณตั้งใจหมายถึง processor type ไหน หรือว่ามัน field ขาด การสร้าง model เฉพาะทาง ตรง ๆ จับได้ทันที

# @expect-error LogsCategoryProcessor.__init__() missing 2 required positional arguments: 'categories' and 'target'
from datadog_api_client.v1.model.logs_category_processor import LogsCategoryProcessor
LogsCategoryProcessor(type='category-processor', name='x', is_enabled=True)

บทเรียน: validate pipeline JSON ทั้งไฟล์ผ่าน LogsPipeline อย่างเดียวไม่ใช่หลักฐานเพียงพอว่า processor ที่ซ้อนอยู่ข้างในทุกตัว well-formed — README ของ harness คอร์สนี้เองมี caveat ใกล้เคียงกัน (ค่า enum/query string ไม่ถูก reject แบบเข้มงวด) นี่คือช่องว่างรูปแบบเดียวกัน แค่ลึกไปอีกชั้น อยู่ใน list field ที่ type เป็น Union

ลำดับสำคัญ: remapper ที่มาก่อน parser ไม่มีอะไรให้ remap

processor สามตัวด้านบนเรียงตามลำดับที่ต้องรัน: Grok Parser ก่อน (เป็นตัวเดียวที่แปลง string message ดิบให้เป็น attribute ที่มีชื่ออย่าง timestamp กับ level ขึ้นมาเลย) แล้วค่อย Date Remapper กับ Status Remapper (มีอะไรให้ remap ก็ต่อเมื่อ attribute พวกนั้นมีอยู่แล้วเท่านั้น) สลับลำดับ — Date Remapper มาก่อน Grok Parser — pipeline ก็ยัง validate ผ่านเป็น LogsPipeline ที่ well-formed สมบูรณ์ (ทั้งคู่เป็น processor object ที่ถูกต้องตามรูปแบบ ไม่มีอะไรใน model ที่รู้ได้ว่าต้องรันตามลำดับ) แต่ตอน runtime Date Remapper จะหา attribute timestamp ที่ยังไม่มีอยู่ เพราะ Grok Parser ยังไม่ได้รันสร้างมันขึ้นมา

Pipeline processors, order swapped (broken):
1. Date Remapper  -> looks for `timestamp` attribute — does not exist yet
2. Grok Parser    -> creates `timestamp` — too late, remapper already ran
3. Status Remapper -> looks for `level` — also does not exist yet

นี่คือ bug เชิง semantic ล้วน ๆ — processor แต่ละตัวเป็น JSON ที่ valid เอง LogsPipeline validate ทั้งไฟล์โดยไม่บ่นอะไรเลย วิธีเดียวที่จะจับได้คือต้องรู้ (หรือทดสอบ) ว่าลำดับ processor มีความหมาย ไม่ใช่แค่ list เฉย ๆ

🐶 Grok syntax: สิ่งที่บทนี้ verify ได้และไม่ได้แบบ offline

harness นี้ไม่มี Grok matching engine ติดตั้งไว้ในเครื่อง — syntax ของ Grok pattern (%{MATCHER:attribute}, named capture group, matcher library ที่มีให้ในตัว) ทดสอบกับ sample text แบบที่ terraform validate หรือ Python model ทำได้ไม่ได้ สิ่งที่ model LogsGrokParserRules เช็คได้จริง คือระดับ structure: match_rules ต้องเป็น string ที่ไม่ว่างเปล่า — มันยอมรับ match_rules: "not even valid grok syntax {{{" ได้สบาย ๆ เพราะ model รู้แค่ว่ามันควรเป็น string ไม่รู้ว่าควร parse เป็น Grok ได้ ค่า match_rules ของบทนี้ด้านบนเขียนด้วยมือตาม pattern syntax ที่ Grok มีเอกสารรองรับ ไม่ได้ verify ด้วยการรันจริง

Reading AI-generated code

🐛 หา bug ในโค้ดนี้

logs-pipelines/broken-order.json

{
"name": "orders logs",
"filter": { "query": "service:orders-api" },
"is_enabled": true,
"processors": [
  { "type": "date-remapper", "name": "Set timestamp", "is_enabled": true, "sources": ["timestamp"] },
  { "type": "status-remapper", "name": "Map level", "is_enabled": true, "sources": ["level"] },
  { "type": "grok-parser", "name": "Parse message", "is_enabled": true,
    "grok": { "match_rules": "OrdersRule %{date(\"yyyy-MM-dd HH:mm:ss\"):timestamp} %{word:level} order=%{number:order.id}" } }
]
}

เฉลย

JSON นี้ well-formed สมบูรณ์ — processor object ทุกตัวมี field ที่จำเป็นครบ และ LogsPipeline(**payload, _check_type=True) validate ผ่านไม่มีคำบ่นเลย (พิสูจน์ด้านบน model เช็ครูปร่าง ไม่เช็คลำดับ) bug ที่พิสูจน์ด้านบนด้วยการไล่ดูว่าแต่ละ processor ต้องการอะไรจริง ๆ: Date Remapper กับ Status Remapper รันก่อน Grok Parser ที่จะสร้าง attribute timestamp กับ level ที่มันกำลังมองหา remapper ทั้งสองตัวไม่มีผลอะไรเลยแบบเงียบ ๆ กับทุก log ที่ pipeline นี้ประมวลผล — ไม่มี error ที่ไหนเลย มีแค่ log ที่ timestamp ทางการยังเป็น “เวลา intake” แทนที่จะเป็นเวลาที่ฝังอยู่ในข้อความ และ facet status ไม่เคยถูกตั้งจาก level ที่ parse ได้เลย ทางแก้คือสลับลำดับ array: Grok Parser ก่อน แล้วค่อยสอง remapper

Mental Model

Processing Pipeline ไม่ใช่ชุด rule อิสระที่เห็น input เดียวกันหมด — มันคือสายพานที่แต่ละสถานีมี input แค่สิ่งที่สถานีก่อนหน้าทิ้งไว้ให้เท่านั้น string message ดิบเป็นสิ่งเดียวที่รับประกันว่ามีอยู่ตอน pipeline เริ่มทำงาน attribute อื่น ๆ ทั้งหมด (timestamp, level, อะไรก็ตามที่ Grok Parser ดึงออกมา) มีอยู่ก็ต่อเมื่อ processor ก่อนหน้าสร้างมันขึ้นมาแล้วเท่านั้น model LogsPipeline — พิสูจน์ด้านบน — validate ว่าแต่ละสถานี well-built เองเท่านั้น มันไม่มีทางรู้ และไม่พยายามจะรู้ ว่าสถานีที่ 3 ต้องพึ่งอะไรที่มีแค่สถานีที่ 5 สร้างขึ้นมา

Trade-offs

• exclude_at_match ฝั่ง Agent เทียบกับ Processing Pipeline — การทิ้งบรรทัดบน host (จาก log_processing_rules ใน logs: block) ฟรีแต่ย้อนกลับไม่ได้ ส่วน Processing Pipeline รันเสมอ (ข้ามไม่ได้) การตัดสินใจกรองจึงควรอยู่ที่ขั้น index/exclusion-filter ในบทถัดไป ไม่ใช่ที่นี่ — pipeline ของบทนี้แค่ parse กับ enrich ไม่เคย drop เลย
• Structural JSON validation เทียบกับความถูกต้องเชิง semantic — พิสูจน์ด้านบนสองครั้ง (ช่องว่างของ processor ที่ type เป็น Union และ bug เรื่อง order dependency): LogsPipeline ที่ validate ผ่านบอกแค่ว่ารูปร่างถูก ไม่ใช่ว่า pipeline ทำสิ่งที่คุณตั้งใจ ทั้งคู่ต้องคิดแยกต่างหากว่าแต่ละ processor ต้องการ input อะไรจริง ๆ
• source: java (pipeline built-in) เทียบกับ pipeline ที่เขียนเอง — pipeline built-in เฉพาะ source ไม่ต้องเขียนเอง Datadog ดูแลให้ตามที่ format ของ framework เปลี่ยนไป แลกกับความ generic ส่วน pipeline ที่เขียนเองแบบด้านบนตรงกับ format log ของคุณเป๊ะ แลกกับที่คุณต้องดูแลมันไปตลอด

Common Misconceptions

• “Processing Pipeline ตัดสินว่า log จะถูกเก็บไว้หรือไม่” ไม่ใช่ — บทนี้เปิดเรื่องด้วยความต่างนี้ตั้งแต่แรก และยังคงยืนยันได้จากหลักฐาน JSON ด้านบน: pipeline parse และ enrich log ทุกตัวที่ match filter ของมัน ส่วนว่า log จะรอดพอที่จะถูก index หรือไม่เป็นการตัดสินใจคนละเรื่อง อยู่ในบทถัดไป
• “ถ้า JSON ของ pipeline validate ผ่าน pipeline นั้นก็ถูกต้อง” ไม่ใช่ — พิสูจน์ด้านบนสองครั้ง: LogsPipeline ที่มี processor เรียงผิดลำดับ validate ผ่านสะอาดแต่ไม่มีผลอะไรเลยแบบเงียบ ๆ และ processor ที่ซ้อนอยู่ผิดรูปแบบ fallback ไปเป็น type ทั่วไปแบบเงียบ ๆ แทนที่จะ raise
• “Grok syntax ถูกเช็คก่อนสร้าง pipeline” ไม่ใช่จาก tooling ของคอร์สนี้ — พิสูจน์ด้านบน: LogsGrokParserRules ต้องการแค่ match_rules เป็น string ที่ไม่ว่างเปล่า ยอมรับ Grok ที่ syntax ผิดโดยไม่บ่นเลย

ใน Agent logs: block ที่มี type: file field path ระบุอะไร

Grok pattern ที่จะ apply ไฟล์บน disk ที่ Agent ต้อง tail ชื่อของ processing pipeline index ที่ log ควรถูกเขียนลงไป

การตั้ง source: java ใน logs config หลัก ๆ แล้วทำอะไร

ลบ log line ที่ไม่ใช่ Java ก่อนออกจาก host ตั้งค่า retention period ของ log แนบ pipeline built-in เฉพาะ source สำหรับ log ของ Java ให้อัตโนมัติ เข้ารหัส log ระหว่างส่ง

query filter ของ Processing Pipeline (เช่น service:my-service) ควบคุมอะไร

log ไหนถูก drop ถาวรและไม่ถูกคิดเงินเลย log ไหนจาก intake ที่ถูก scope เข้า pipeline นั้นเพื่อ parse และ enrich log ที่ match ถูกเก็บใน index กี่วัน host ไหนรัน Agent

Processing Pipeline parse เกิดขึ้นตรงไหนเทียบกับการ indexing

หลังจาก log ถูกเขียนลง index แล้ว บน host เอง ก่อนที่ log จะออกไปผ่าน Agent หลัง intake และก่อนที่ log จะไปถึง index pipeline กับ indexing เกิดที่ขั้นตอนเดียวกัน ไม่มีลำดับตายตัว

pipeline JSON มี Date Remapper กับ Status Remapper เรียงไว้ก่อน Grok Parser LogsPipeline(**payload, _check_type=True) ทำอะไรกับมัน แล้วเกิดอะไรขึ้นจริงตอน runtime

model raise error ทันที ปฏิเสธ processor ที่เรียงผิดลำดับ model validate ผ่านสะอาด (รูปร่างถูก) พอ runtime ทั้งสอง remapper ไม่มีผลอะไรเลยแบบเงียบ ๆ เพราะ attribute ที่มันมองหายังไม่มีอยู่จนกว่า Grok Parser จะรัน model เรียง processor ใหม่ให้อัตโนมัติตามลำดับที่ถูกต้อง Grok Parser รันก่อนเสมอไม่ว่า array จะเรียงยังไง

category-processor ถูกซ้อนอยู่ใน list processors ของ LogsPipeline โดยขาด field ที่จำเป็น categories กับ target เกิดอะไรขึ้นจริงตอน validate pipeline ทั้งไฟล์

TypeError ระบุ field categories กับ target ที่ขาดหายไป ไม่มี error เลย — dict resolve ไปเป็น type ทั่วไป LogsProcessor แบบเงียบ ๆ แทนที่จะเป็น LogsCategoryProcessor และปัญหา field ขาดไม่เคยถูกเผยออกมาเลย processor ตัวนั้นถูกตัดออกจาก list อัตโนมัติ validation fail เสมอสำหรับ processors list ที่มีมากกว่า 3 รายการ

Submit

Mark as complete