Processor ใน Log Pipeline

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

บรรทัด log ถูกแปลงใน สามที่แยกกัน — Agent (ก่อนออกจาก host), JSON preprocessing (ตอนมาถึง) และ pipeline ของคุณ (รายการ filter กับ processor ที่เรียงลำดับ) — และภายใน pipeline processor รันจากบนลงล่างบน ผลลัพธ์ของตัวก่อนหน้า ความหมายของ config จึงขึ้นกับลำดับ ขึ้นกับว่าตัวซ้ำตัวไหน “ชนะ” และขึ้นกับข้อเท็จจริงที่ JSON schema มองไม่เห็น

🐶 อะไรรันจริง อะไรอ้างจาก docs

pipeline engine ของ Datadog เป็น service ที่ host ให้ ไม่มี runtime ในเครื่องสำหรับ Grok, remapper หรือ pipeline และบทนี้ไม่แกล้งสร้างขึ้นมา หลักฐานสามแบบ และทุก claim ระบุว่าใช้แบบไหน (1) รันจริง: ขั้นตอนฝั่ง Agent บน datadog/agent:7 container จริง (Agent 7.83.3, container dd-internals, key ปลอม ไม่ส่งอะไรออก) ด้วย agent analyze-logs; และ pipeline JSON ที่ตรวจด้วย datadog-api-client 2.61.0 จาก venv ของ harness รวมถึงการตรวจแบบเข้มงวดต่อ processor ที่เขียนขึ้นสำหรับบทนี้ (2) docs อ้างพร้อม URL ดึงเมื่อ 2026-09-30 จาก docs.datadoghq.com/logs/log_configuration/ (pipelines, parsing และหน้า log_date_remapper / log_status_remapper) (3) การถอดกฎที่ระบุไว้ใน docs เป็นโค้ดโดยผมเอง ระบุว่าเป็นเช่นนั้น ใช้เพื่อคำนวณตัวอย่างเท่านั้น ราคาและพฤติกรรมของ UI ไม่ครอบคลุม บท Phase 3 Log Collection and Processing Pipelines แนะนำ pipeline JSON กับ bug “remapper ก่อน parser” บทนี้เปิดกล่องดู

สามที่ ในภาพเดียว

flowchart LR
  F["Log file line"] --> A["Agent: log_processing_rules (exclude_at_match, mask_sequences, multi_line)"]
  A --> I["Intake"]
  I --> P["JSON preprocessing: reserved attributes"]
  P --> L1["Pipeline 1: filter is tested first, then processors in order"]
  L1 --> L2["Pipeline 2: same, on the output of pipeline 1"]
  L2 --> X["Indexes, exclusion filters, archives, log-based metrics"]

บรรทัด log ถูกเปลี่ยนที่ไหนบ้าง

ขั้นที่ 1: Agent พิสูจน์ด้วย agent analyze-logs

rule ใน block logs: รันบน host ก่อนส่งบรรทัดออกไป และไม่มีอะไรฝั่งปลายทางย้อนได้ agent analyze-logs <conf.yaml> รัน logs config กับ Agent ที่กำลังรันแล้วพิมพ์สิ่งที่จะถูกส่ง ไฟล์ input app.log:

2026-09-30 10:00:01.123 INFO order=1001 user=alice card=4111-1111-1111-1111 paid
2026-09-30 10:00:02.456 DEBUG healthcheck ok
2026-09-30 10:00:03.789 ERROR order=1002 user=bob payment declined
Traceback (most recent call last):
  File "app.py", line 3, in <module>
ValueError: boom
2026-09-30 10:00:04.000 INFO order=1003 user=carol paid

โดยมี exclude_at_match บน healthcheck, mask_sequences สำหรับเลขบัตร 16 หลัก (replace_placeholder: "[masked_card]") และ rule multi_line (event ใหม่ขึ้นต้นด้วยวันที่) output จริง:

$ agent analyze-logs -t 5s /tmp/lab/conf.yaml
2026-09-30 10:00:01.123 INFO order=1001 user=alice card=[masked_card] paid
2026-09-30 10:00:03.789 ERROR order=1002 user=bob payment declined\nTraceback (most recent call last):\n  File "app.py", line 3, in <module>\nValueError: boom
2026-09-30 10:00:04.000 INFO order=1003 user=carol paid

สามผล ทั้งหมดเกิดบน host: บรรทัด DEBUG healthcheck หายไป, เลขบัตรถูก mask ก่อนไปถึง index ใด ๆ และ traceback Python สี่บรรทัดกลายเป็น event เดียว (เชื่อมด้วย \n) สองข้อสังเกตด้านปฏิบัติการจากการรันเดียวกัน ด้วย -t 1s ที่เป็น default event สุดท้ายหายไป: กลุ่ม multi-line จะถูก emit หลังถูกปิดโดยบรรทัดเริ่มถัดไปหรือ timeout เท่านั้น จึงควรให้ inactivity timeout ของคำสั่งนานขึ้น และ traceback ก็ถูกพับด้วยเมื่อผมเอา rule multi_line ที่ระบุชัดออก เพราะ agent config --all แสดง auto_multi_line_detection: true บน Agent นี้ — อย่าพึ่งพามันกับ format ของคุณเอง rule ที่พังจะล้มดังและเร็ว พร้อม error จริง ก่อนที่จะส่งอะไรออกไป:

Error with config:  no pattern provided for processing rule: mask_card
Error: no pattern provided for processing rule: mask_card

ประเด็นคือความต่างกับ backend: processor ของ pipeline เขียนทับได้เฉพาะ log ที่มาถึงแล้ว ข้อมูลที่ห้ามเก็บเด็ดขาดต้องจัดการที่นี่ (logs: block กับ log_processing_rules แนะนำใน Log Collection and Processing Pipelines) ไม่ใช่ใน pipeline

ขั้นที่ 2: preprocessing แล้วจึงรายการ pipeline

สองข้อเท็จจริงจากหน้า pipelines (docs.datadoghq.com/logs/log_configuration/pipelines) กำหนดกรอบทุกอย่างหลัง Agent ข้อแรก “Preprocessing of JSON logs occurs before logs enter pipeline processing. Preprocessing runs a series of operations based on reserved attributes, such as timestamp, status, host, service, and message.” JSON log ที่มี level, severity, status หรือ syslog.severity อยู่แล้วจะได้ status ก่อนที่จะมี pipeline ไหนมาช่วย — docs ลิสต์ชื่อทั้งสี่นั้นสำหรับ status และ @timestamp, timestamp, _timestamp, Timestamp, eventTime, date, published_date, syslog.timestamp สำหรับวันที่ ข้อสอง กฎการเดินผ่าน: “Each log that comes through the pipelines is tested against every pipeline filter. If it matches a filter, then all the processors are applied sequentially before moving to the next pipeline.” pipeline จึงไม่ได้แยกขาดจากกัน: log หนึ่งผ่านได้หลาย pipeline และ pipeline 2 เห็นสิ่งที่ pipeline 1 เขียนไว้

สองขีดจำกัดกับหนึ่งกับดักในหน้าเดียวกัน: คำแนะนำคือ “at most 20 processors per pipeline and 10 parsing rules within a Grok processor”; “A pipeline can contain nested pipelines and processors whereas a nested pipeline can only contain processors”; และกับดัก — “The pipeline filtering is applied before any of the pipeline’s processors. For this reason, you cannot filter on an attribute that is extracted in the pipeline itself.”

Grok: ตัวแรกที่ match ชนะ และ rule ต้อง match ทั้งบรรทัด

จาก parsing docs (docs.datadoghq.com/logs/log_configuration/parsing) แบบคำต่อคำ: “Only one can match any given log. The first one that matches, from top to bottom, is the one that does the parsing.” “You must have unique rule names within the same Grok parser.” “You must define your parsing rule to match the entire log entry, as each rule applies from the beginning to the end of the log.” และเมื่อไม่มี rule ไหน match log ผ่านไปโดยไม่เปลี่ยน — ไม่มี error ไม่มี tag — นี่คือเหตุผลที่ Grok ที่หยุด match หลัง format ของ log เปลี่ยนจะล้มเงียบ ๆ และ remapper ที่ตามมาไม่ทำอะไรเงียบ ๆ

matcher date("pattern"[, "timezoneId"[, "localeId"]]) คือจุดที่ความเสียหายจริงส่วนใหญ่เกิด เพราะ pattern ไม่มี zone ถ้าไม่ระบุ ตารางใน docs จับคู่ input กับ epoch มิลลิวินาที ผมคำนวณแถวเหล่านั้นใหม่ด้วย zoneinfo ของ Python:

'Thu Jun 16 08:29:03 2016'  tz=None          python=1466065743000 docs=1466065743000 match=True
'Thu Jun 16 08:29:03 2016'  tz=Europe/Paris  python=1466058543000 docs=1466058543000 match=True
'Thu Jun 16 08:29:03 2016'  tz=+3            python=1466054943000 docs=1466054943000 match=True
'11/10/2014'                tz=None          python=1412985600000 docs=1412978400000 match=False

สามแถว reproduce ได้พอดีด้วย “ไม่มี zone แปลว่า UTC” แถวที่สี่ (11/10/2014 เป็น 1412978400000) คือ 2014-10-10T22:00:00Z หรือเที่ยงคืนที่ UTC+2 — ตรงก็ต่อเมื่อตัวอย่างใน docs สร้างในโซน UTC+2 ผมรัน parser ของ Datadog ไม่ได้ จึงบอกไม่ได้ว่าจริง ๆ มันทำอย่างไรกับแถวนั้น การอ่านที่ปลอดภัยคือ ห้ามพึ่ง default และส่ง zone ทุกครั้ง ราคาของการไม่ใส่ zone สำหรับแอปที่ log เวลา local ใน Asia/Bangkok (UTC+7):

true instant (Bangkok): 2026-09-30T03:00:01.123+00:00   parsed as UTC: 2026-09-30T10:00:01.123+00:00   shift ahead = 7:00:00

เร็วกว่าจริงเจ็ดชั่วโมงในอนาคต — และหน้า Date remapper ระบุว่า “Log events can be submitted up to 18 hours in the past and two hours in the future.” (เครื่องที่อยู่ทางตะวันตกของ UTC ได้ผลตรงข้าม: เวลา local ที่อ่านเป็น UTC ตกไปอยู่ในอดีต ภายในหน้าต่าง 18 ชั่วโมง จึงถูกรับด้วย timestamp ที่ผิดโดยไม่มีใครสังเกต)

Date remapper กับ Status remapper: กฎตัดสินเสมอกันตรงข้ามกัน

ทั้งสอง processor รับ sources (รายชื่อ attribute) แล้วตั้ง reserved attribute; ทั้งคู่ล้มเงียบ ๆ ถ้า attribute ต้นทางยังไม่มี ต่างกันตรงตัวซ้ำ และ docs ระบุชัดทั้งคู่:

• Date: “If multiple log date remapper processors are applied to a given log within the pipeline, the last one (according to the pipeline’s order) is taken into account.” format ที่รู้จักคือ “ISO8601, UNIX (the milliseconds EPOCH format), and RFC3164”; หากไม่มีวันที่ที่ใช้ได้ “Datadog timestamps the logs with the date it received them” (log_date_remapper)
• Status: “If multiple log status remapper processors are applied to a log within a pipeline, only the first one in the pipeline’s order is considered. Additionally, for all pipelines that match the log, only the first status remapper encountered (from all applicable pipelines) is applied.” (log_status_remapper)

date ตัวสุดท้ายชนะ, status ตัวแรกชนะ status remapper ตัวที่สองที่ copy-paste ไว้ใน pipeline ถัดไปเป็น dead code; date remapper ตัวที่สองที่ copy-paste จะ override ตัวที่ดีเงียบ ๆ

การ map status ทำตาม prefix ของ string อ้างจากหน้าเดียวกัน: “Integers from 0 to 7 map to the Syslog severity standards”; emerg หรือ f -> emerg; a -> alert; c -> critical; err -> error; w -> warning; n -> notice; i -> info; d, t, v, trace, verbose -> debug; “Strings beginning with o or s, or matching OK or Success … map to OK”; “All others map to info (6).” ผมถอดกฎเหล่านั้นเป็นฟังก์ชันเล็ก ๆ (status_map.py, ไม่ใช่ โค้ดของ Datadog) แล้วรันด้วยชื่อ level ที่ logger จริงปล่อยออกมา:

'DEBUG' -> debug      'WARN' -> warning     'ERROR' -> error       'CRITICAL' -> critical
'FATAL' -> emerg      'TRACE' -> debug      'W' -> warning
'SEVERE' -> OK        'FINE' -> emerg       'CONFIG' -> critical   'E' -> info
'Exception' -> info   'panic' -> info       'dpanic' -> debug      'Started' -> OK

หากใช้กฎที่ระบุใน docs ตามตัวอักษร SEVERE ของ java.util.logging กลายเป็น OK, FINE กลายเป็น emerg, E ตัวอักษรเดียวของ Ruby กลายเป็น info เพราะมีแต่ err ที่ map เป็น error และ dpanic ของ zap กลายเป็น debug ไม่มีอะไรเป็น error; นี่คือสิ่งที่ตาราง prefix ทำ ทางแก้คือ category-processor (หรือ normalise level ในแอป) ก่อน status remapper

Model JSON จับอะไรได้และไม่ได้

pipeline JSON ที่ส่งไป API คือรายการ processor object ที่เรียงลำดับ LogsPipeline(**payload, _check_type=True) คือสิ่งที่ harness ใช้กับทุก fence // logs-pipelines/... การตรวจของบทนี้ไปไกลกว่านั้น เพราะช่องโหว่ที่บท Phase 3 พบ (processor ที่ซ้อนใน processors เป็น oneOf ที่ตกไปเป็น object ทั่วไปเงียบ ๆ): สคริปต์ที่เลือก class จริงของแต่ละ type แล้ว validate โดยไล่เข้า nested pipeline, Grok rule และ category item

app/validate_pipeline.py

import json, sys
from datadog_api_client.v1.model.logs_pipeline import LogsPipeline
from datadog_api_client.v1.model import (logs_grok_parser, logs_date_remapper, logs_status_remapper,
    logs_message_remapper, logs_category_processor, logs_pipeline_processor,
    logs_grok_parser_rules, logs_category_processor_category)

CLASSES = {
    'grok-parser': logs_grok_parser.LogsGrokParser,
    'date-remapper': logs_date_remapper.LogsDateRemapper,
    'status-remapper': logs_status_remapper.LogsStatusRemapper,
    'message-remapper': logs_message_remapper.LogsMessageRemapper,
    'category-processor': logs_category_processor.LogsCategoryProcessor,
    'pipeline': logs_pipeline_processor.LogsPipelineProcessor,
}

def strict(processors, path='processors'):
    for i, p in enumerate(processors):
        cls = CLASSES.get(p.get('type'))
        where = f'{path}[{i}] ({p.get("type")})'
        if cls is None:
            raise SystemExit(f'{where}: unknown processor type')
        cls(**p, _check_type=True)
        if p['type'] == 'grok-parser':
            logs_grok_parser_rules.LogsGrokParserRules(**p['grok'], _check_type=True)
        if p['type'] == 'category-processor':
            for c in p['categories']:
                logs_category_processor_category.LogsCategoryProcessorCategory(**c, _check_type=True)
        if p['type'] == 'pipeline':
            strict(p.get('processors', []), where + '.processors')

data = json.load(open(sys.argv[1]))
LogsPipeline(**data, _check_type=True)
strict(data['processors'])
print('OK')

pipeline ที่ parse, remap วันที่และ status, ยก message ขึ้น และแบ่ง bucket ผลลัพธ์ (fence นี้ถูก collect ดังนั้น harness ของคอร์สรัน model LogsPipeline ระดับบนกับมัน และ validate_pipeline.py ข้างบนก็ผ่านมันเช่นกัน: พิมพ์ OK):

logs-pipelines/orders-api.json

{
  "name": "orders-api logs",
  "filter": { "query": "service:orders-api source:python" },
  "is_enabled": true,
  "processors": [
    {
      "type": "grok-parser",
      "name": "Parse order log line",
      "is_enabled": true,
      "source": "message",
      "grok": {
        "support_rules": "",
        "match_rules": "OrderRule %{date(\"yyyy-MM-dd HH:mm:ss.SSS\",\"Asia/Bangkok\"):timestamp} %{word:level} order=%{integer:order.id} user=%{word:usr.name} %{data:msg}"
      }
    },
    { "type": "date-remapper", "name": "Use the parsed timestamp", "is_enabled": true, "sources": ["timestamp"] },
    { "type": "status-remapper", "name": "Use level as status", "is_enabled": true, "sources": ["level"] },
    { "type": "message-remapper", "name": "Use msg as message", "is_enabled": true, "sources": ["msg"] },
    {
      "type": "category-processor",
      "name": "Bucket by payment outcome",
      "is_enabled": true,
      "target": "payment.outcome",
      "categories": [
        { "filter": { "query": "@msg:*declined*" }, "name": "declined" },
        { "filter": { "query": "@msg:*paid*" }, "name": "paid" }
      ]
    }
  ]
}

ทีนี้หกตัวแปรที่พังของไฟล์นั้น รันผ่านทั้งสองการตรวจ คอลัมน์แรกคือสิ่งที่ model ระดับบนบอก; คอลัมน์ที่สองคือบรรทัดสุดท้ายของสคริปต์เข้มงวด (path ของ module ใน exception ถูกย่อ):

variant                           LogsPipeline model   strict script
date-remapper sources: "timestamp"   OK                ApiTypeError: Invalid type for variable 'sources'. Required value type is list and passed type was str at ['sources']
grok-parser without "grok"           OK                TypeError: LogsGrokParser.__init__() missing 1 required positional argument: 'grok'
date-remapper without "sources"      OK                TypeError: LogsDateRemapper.__init__() missing 1 required positional argument: 'sources'
type "status_remapper" (underscore)  OK                processors[2] (status_remapper): unknown processor type
category filter as a string          OK                ApiTypeError: Invalid type for variable 'filter'. Required value type is LogsFilter and passed type was str at ['filter']
broken date-remapper inside a nested pipeline   OK    ApiTypeError: Invalid type for variable 'sources'. Required value type is list and passed type was str at ['sources']

model ระดับบน รับทั้งหกตัว ผมยังรัน deserializer ของ client เอง (validate_and_convert_types) และมันก็แปลง processor ที่พังทุกตัวเป็น LogsProcessor ทั่วไปโดยไม่บ่น ดังนั้น fence // logs-pipelines/ ที่ผ่าน npm run verify พิสูจน์แค่รูปร่างชั้นนอก; class จริงของแต่ละ processor คือสิ่งที่จับความผิดระดับ field ได้ สองสิ่งที่ schema check ไหนก็จับไม่ได้ ทั้งสองรันแล้วข้างบน: match_rules ว่าง ({"match_rules": ""}) และ matcher ที่สะกดผิด (%{wrod:level}) ผ่านสคริปต์เข้มงวดด้วย ไวยากรณ์และความหมายของ Grok มีแต่ Datadog ที่ validate ได้ ใน Grok parser UI หรือโดย API ตอน apply

Mental Model

มอง pipeline เป็น ฟังก์ชันจาก log ไป log ที่ประกอบกันตามลำดับไฟล์ พร้อมสถานะซ่อนสามอย่าง (1) ตอนนี้ attribute ไหนมีอยู่: remapper ต้องการให้ source มีอยู่แล้ว ดังนั้น parser ต้องมาก่อน (2) กฎตัดสินเสมอกัน: ตัวแรกชนะ (rule ของ Grok, status remapper) เทียบกับตัวสุดท้ายชนะ (date remapper) (3) สิ่งที่ตัดสินไปแล้วก่อนคุณรัน: rule ของ Agent, JSON preprocessing, pipeline ก่อนหน้า และ filter ของ pipeline เอง ซึ่งถูกประเมินกับ log ตามที่มาถึงและไม่เคยเห็นสิ่งที่ processor ของคุณสกัดออกมา ที่เหลือคือ data flow

Trade-offs

ทางเลือก	ข้อดี	ข้อเสีย
mask หรือ drop ใน Agent (log_processing_rules)	ไม่ออกจาก host; พิสูจน์ข้างบนด้วย analyze-logs	ย้อนไม่ได้; regex ผิดคือข้อมูลหาย; config อยู่ทุก host
parse ใน pipeline (Grok)	เปลี่ยนการ parse โดยไม่ต้อง redeploy แอป; ใช้กับ log ที่คุณไม่ได้เขียนได้	การเปลี่ยน format ทำให้หยุด match เงียบ ๆ; แนะนำแค่ 10 rule ต่อ Grok processor
JSON log แบบมีโครงสร้าง + preprocessing	ไม่มี Grok ให้พัง; reserved attribute จัดการก่อน pipeline	แอปต้อง emit ชื่อที่ Datadog คาดหวัง ไม่งั้นต้องเพิ่ม remapper
pipeline ใหญ่อันเดียว	อ่านบนลงล่างง่าย	คำแนะนำ 20 processor; ทุก log ถูกทดสอบกับ filter ของมัน
pipeline เล็กหลายอัน	แต่ละอันมีทีมเป็นเจ้าของ; nested pipeline แยกตาม service	status remapper ซ้ำใน pipeline หลังเป็น dead; ลำดับระหว่าง pipeline สำคัญ
ระบุ timezone ใน date()	timestamp แน่นอน	ต้องรู้ zone ของผู้เขียน; DST ต้องใช้ชื่อ zone ไม่ใช่ offset คงที่

Common Misconceptions

• “pipeline drop หรือ mask log ได้” processor เขียนทับ; การเก็บหรือทิ้งเป็นเรื่อง rule ของ Agent และ exclusion filter ฝั่ง backend การ mask ที่ห้ามถึง storage เด็ดขาดรันใน Agent (พิสูจน์ด้วย analyze-logs)
• “filter ของ pipeline ใช้ attribute ที่ Grok ของฉันสกัดได้” docs: การ filter เกิด “before any of the pipeline’s processors.”
• “ถ้า Grok สอง rule match ทั้งสองถูกใช้” “Only one can match any given log. The first one that matches, from top to bottom.”
• “date remapper สองตัว: ตัวแรกชนะเหมือน status” ตรงข้าม: date remapper “the last one … is taken into account”; status remapper “only the first one”
• “SEVERE map เป็น error” ตามตาราง prefix ใน docs มัน map เป็น OK (s) และ FINE เป็น emerg (f)
• “npm run verify บน fence logs-pipelines พิสูจน์ว่าใช้ได้” มันพิสูจน์รูปร่างชั้นนอก; ทั้งหกตัวแปรที่พังข้างบนผ่าน model ระดับบน
• “matcher date() ที่ไม่มี timezone ปลอดภัย” สำหรับ log เวลา Bangkok ผลลัพธ์เร็วกว่าเวลาจริง 7 ชั่วโมง เกินหน้าต่างอนาคต 2 ชั่วโมงที่ docs ระบุ

Reading AI-generated code

ทั้งสองอันนี้ถูกต้องเชิงโครงสร้าง: LogsPipeline และสคริปต์เข้มงวดรับทั้งคู่ แต่ผิดด้วยเหตุผลที่มีแต่ docs บอกได้

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

logs-pipelines/orders-api-errors.json

{
"name": "orders-api errors",
"filter": { "query": "@level:ERROR service:orders-api" },
"is_enabled": true,
"processors": [
  {
    "type": "grok-parser",
    "name": "Parse order log line",
    "is_enabled": true,
    "source": "message",
    "grok": {
      "support_rules": "",
      "match_rules": "OrderRule %{date(\"yyyy-MM-dd HH:mm:ss.SSS\",\"Asia/Bangkok\"):timestamp} %{word:level} %{data:msg}"
    }
  },
  { "type": "date-remapper", "name": "Use the parsed timestamp", "is_enabled": true, "sources": ["timestamp"] },
  { "type": "status-remapper", "name": "Use level as status", "is_enabled": true, "sources": ["level"] }
]
}

เฉลย

filter ของ pipeline คือ @level:ERROR แต่ level ถูกสร้างโดย Grok parser ใน pipeline เดียวกันนี้ docs: “The pipeline filtering is applied before any of the pipeline’s processors. For this reason, you cannot filter on an attribute that is extracted in the pipeline itself.” ตอน filter log ยังไม่มี @level pipeline จึง match ไม่ได้เลย และ Grok, date และ status remapper ไม่เคยรัน: ทุก log ยังไม่ถูก parse โดยมีเวลา intake เป็น timestamp และไม่มี error ที่ไหนเลย model รับมัน (ตรวจรูปร่าง ไม่ตรวจ scope) ให้ filter ด้วยสิ่งที่มีตอน intake แล้วให้ status remapper พา level ไป:

logs-pipelines/orders-api-errors-fixed.json

{
  "name": "orders-api logs",
  "filter": { "query": "service:orders-api" },
  "is_enabled": true,
  "processors": [
    {
      "type": "grok-parser",
      "name": "Parse order log line",
      "is_enabled": true,
      "source": "message",
      "grok": {
        "support_rules": "",
        "match_rules": "OrderRule %{date(\"yyyy-MM-dd HH:mm:ss.SSS\",\"Asia/Bangkok\"):timestamp} %{word:level} %{data:msg}"
      }
    },
    { "type": "date-remapper", "name": "Use the parsed timestamp", "is_enabled": true, "sources": ["timestamp"] },
    { "type": "status-remapper", "name": "Use level as status", "is_enabled": true, "sources": ["level"] }
  ]
}

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

logs-pipelines/orders-api-utc.json

{
"name": "orders-api logs",
"filter": { "query": "service:orders-api" },
"is_enabled": true,
"processors": [
  {
    "type": "grok-parser",
    "name": "Parse order log line (app writes local time, Asia/Bangkok)",
    "is_enabled": true,
    "source": "message",
    "grok": {
      "support_rules": "",
      "match_rules": "OrderRule %{date(\"yyyy-MM-dd HH:mm:ss.SSS\"):timestamp} %{word:level} %{data:msg}"
    }
  },
  { "type": "date-remapper", "name": "Use the parsed timestamp", "is_enabled": true, "sources": ["timestamp"] }
]
}

เฉลย

แอปเขียนเวลา local ของกรุงเทพ (UTC+7) แต่ matcher date() ไม่มี argument timezone คำนวณข้างบนแล้ว: บรรทัดที่ประทับ 10:00:01.123 จริง ๆ คือ 03:00:01.123Z เมื่ออ่านเป็น UTC มันตกไป 7 ชั่วโมงในอนาคต และ “Log events can be submitted up to 18 hours in the past and two hours in the future.” ตัว default ใน docs เป็น UTC หรือไม่ก็ยังไม่แน่ใจ (หนึ่งจากสี่แถวใน docs reproduce ได้เฉพาะที่ UTC+2) คำตอบที่ปลอดภัยเดียวคือระบุ zone: %{date("yyyy-MM-dd HH:mm:ss.SSS","Asia/Bangkok"):timestamp} เหมือนใน pipeline JSON ที่แก้ไว้ก่อนหน้า การให้แอป emit UTC หรือ ISO 8601 พร้อม offset หลีกเลี่ยงการเดาได้ทั้งหมด

ในการรัน agent analyze-logs จริง เกิดอะไรกับบรรทัด DEBUG healthcheck และเลขบัตร?

ทั้งสองไปถึง Datadog แล้วถูก pipeline ทำความสะอาด บรรทัด healthcheck ถูกตัดทิ้งและเลขบัตรถูก mask บน host ก่อนส่งอะไรออกไป มีแค่เลขบัตรที่ถูกเปลี่ยน; healthcheck ถูกเก็บเสมอ ไม่มีอะไร; analyze-logs เมิน processing rule

pipeline มี filter @level:ERROR โดย level ถูกสกัดโดย Grok parser ใน pipeline เดียวกัน เกิดอะไรขึ้น?

ใช้ได้ filter ประเมินแบบ lazy filter ถูกประเมินก่อน processor ดังนั้น @level ยังไม่มี และ pipeline ไม่ match อะไรเลย — แบบเงียบ ๆ API ปฏิเสธ pipeline ข้ามแค่ Grok parser

date remapper สองตัวและ status remapper สองตัวอยู่ใน pipeline เดียวกัน แต่ละคู่ชี้ไปคนละ attribute ตาม docs processor ใดมีผล?

date remapper ตัวแรกและ status remapper ตัวแรก date remapper ตัวสุดท้ายและ status remapper ตัวแรก ตัวสุดท้ายของทั้งสองแบบ ทั้งสองตัวของแต่ละแบบถูกใช้แล้ว merge กัน

ตามกฎ prefix ของ status ใน docs (ที่ implement ใน status_map.py ของบทเรียน) การ map ข้อใดถูก?

SEVERE -> error SEVERE -> OK, FINE -> emerg, Exception -> info E -> error, Exception -> error FATAL -> critical

คุณรัน model LogsPipeline ระดับบนกับ pipeline ที่พังโดยตั้งใจหกตัว (string แทน list ของ sources, ขาด grok, type สะกดผิด, ...) มันปฏิเสธกี่ตัว?

ทั้งหก ไม่มีเลย — processor เป็น oneOf ที่ตกไปเป็น object ทั่วไป ต้องใช้ class ต่อ processor ถึงจะจับได้ สามในหก เฉพาะตัว nested

แอป log เวลา local Asia/Bangkok และ Grok rule ใช้ date("yyyy-MM-dd HH:mm:ss.SSS") โดยไม่มี timezone การคำนวณในบทนี้แสดงอะไร?

เลื่อนไปอนาคต 7 ชั่วโมงถ้า default เป็น UTC เกินขีดจำกัดอนาคต 2 ชั่วโมงใน docs ไม่มีผล; Datadog ตรวจจับ zone เอง เลื่อนไปอดีต 7 ชั่วโมง อยู่ในหน้าต่าง 18 ชั่วโมง Agent ทิ้ง log เสมอ

Submit

Mark as complete