Log Pipeline Processors

The idea in one sentence

A log line is transformed in three separate places — the Agent (before it leaves the host), JSON preprocessing (on arrival), and your pipelines (an ordered list of filters and processors) — and inside a pipeline the processors run top to bottom on the output of the previous one, so the meaning of a config depends on order, on which duplicate “wins”, and on facts the JSON schema cannot see.

🐶 What is run here and what is quoted

Datadog’s pipeline engine is a hosted service: there is no local runtime for Grok, remappers or pipelines, and this lesson does not fake one. Three kinds of evidence are used and each claim says which. (1) Run for real: the Agent-side stage, on a real datadog/agent:7 container (Agent 7.83.3, container dd-internals, fake key, nothing sent) with agent analyze-logs; and pipeline JSON checked with datadog-api-client 2.61.0 from the harness venv, including a stricter per-processor check written for this lesson. (2) Docs, quoted with the URL, fetched 2026-09-30 from docs.datadoghq.com/logs/log_configuration/ (pipelines, parsing, and the log_date_remapper / log_status_remapper pages). (3) My own transcription of a documented rule, labelled as such, used only to compute examples. Pricing and UI behaviour are not covered. The Phase 3 lesson Log Collection and Processing Pipelines introduces the pipeline JSON and the “remapper before parser” bug; this lesson opens the box.

The three places, once

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"]

Where a log line is changed

Stage 1: the Agent, proven with agent analyze-logs

Rules in a logs: block run on the host, before the line is sent, and nothing downstream can undo them. agent analyze-logs <conf.yaml> runs a logs config against a running Agent and prints what would be sent. The input file 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

with an exclude_at_match on healthcheck, a mask_sequences for a 16-digit card number (replace_placeholder: "[masked_card]") and a multi_line rule (new event starts with a date). Real 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

Three effects, all on the host: the DEBUG healthcheck line is gone, the card number is masked before it can reach any index, and the four-line Python traceback became one event (\n joined). Two operational notes from the same runs. With the default -t 1s the last event was missing: a multi-line group is only emitted after it is closed by the next start line or a timeout, so give the command a longer inactivity timeout. And the traceback was also folded when I removed the explicit multi_line rule, because agent config --all shows auto_multi_line_detection: true on this Agent — do not rely on that for your own formats. A broken rule fails loudly and early, with a real error, before anything is sent:

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

The contrast with the backend is the point: a pipeline processor can only rewrite a log that has already arrived. Data you must never store belongs here (the logs: block and log_processing_rules are introduced in Log Collection and Processing Pipelines), not in a pipeline.

Stage 2: preprocessing, then the pipeline list

Two facts from the pipelines page (docs.datadoghq.com/logs/log_configuration/pipelines) frame everything after the Agent. First, “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.” A JSON log already carrying level, severity, status or syslog.severity gets its status before any pipeline exists to help — the docs list those four names for the status and @timestamp, timestamp, _timestamp, Timestamp, eventTime, date, published_date, syslog.timestamp for the date. Second, the traversal rule: “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.” So pipelines are not exclusive: a log can pass through several, and pipeline 2 sees what pipeline 1 wrote.

Two limits and one trap in the same page: the recommendation is “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”; and the trap — “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: first match wins, and rules match the whole line

From the parsing docs (docs.datadoghq.com/logs/log_configuration/parsing), verbatim: “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.” And when nothing matches the log passes through unchanged — no error, no tag — which is why a Grok that stops matching after a log-format change fails silently and the downstream remappers silently do nothing.

The date("pattern"[, "timezoneId"[, "localeId"]]) matcher is where most real damage happens, because the pattern has no zone unless you give one. The docs table pairs inputs with epoch-millisecond outputs; I recomputed the rows with Python’s zoneinfo:

'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

Three rows reproduce exactly with “no zone means UTC”. The fourth (11/10/2014 to 1412978400000) is 2014-10-10T22:00:00Z, i.e. midnight at UTC+2 — it only matches if the docs example was produced in a UTC+2 zone. I could not run Datadog’s parser, so I cannot say which behaviour it actually has for that row; the safe reading is to never rely on the default and always pass the zone. What a missing zone costs, for an application that logs local time in 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

Seven hours in the future — and the Date remapper page states “Log events can be submitted up to 18 hours in the past and two hours in the future.” (A machine west of UTC gets the opposite: local time read as UTC lands in the past, inside the 18-hour window, so it is accepted with a wrong timestamp and nobody notices.)

Date remapper and Status remapper: opposite tie-breaks

Both processors take sources (a list of attribute names) and set a reserved attribute; both fail silently if the source attribute does not exist yet. They differ on duplicates, and the docs are explicit on both:

• 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.” The recognised formats are “ISO8601, UNIX (the milliseconds EPOCH format), and RFC3164”; without a usable date “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)

Last one wins for the date, first one wins for the status. A copy-pasted second status remapper in a later pipeline is dead code; a copy-pasted second date remapper silently overrides the good one.

The status mapping is by string prefix, quoted from the same page: “Integers from 0 to 7 map to the Syslog severity standards”; emerg or 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).” I transcribed those rules into a small function (status_map.py, not Datadog’s code) and ran the level names real loggers emit:

'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

If the documented rules are applied as written, java.util.logging’s SEVERE becomes OK, its FINE becomes emerg, a one-letter Ruby E becomes info because only err maps to error, and zap’s dpanic becomes debug. None of that is an error; it is what a prefix table does. The fix is a category-processor (or normalising the level in the app) before the status remapper.

What the JSON model does and does not catch

Pipeline JSON sent to the API is an ordered list of processor objects. LogsPipeline(**payload, _check_type=True) is what the harness uses on every // logs-pipelines/... fence. This lesson’s check goes further, because of the gap the Phase 3 lesson found (a processor nested in processors is a oneOf that silently falls back to a generic object): a script that picks the concrete class for each processor’s type and validates it, recursing into nested pipelines, Grok rules and category items.

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')

A pipeline that parses, remaps the date and status, promotes a message and buckets the outcome (this fence is collected, so the course harness runs the top-level LogsPipeline model on it, and validate_pipeline.py above passed it too: it printed 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" }
      ]
    }
  ]
}

Now six broken variants of that file, run through both checks. The first column is what the top-level model said; the second is the strict script’s last line (exception module paths shortened):

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']

The top-level model accepted all six. I also ran the client’s own deserializer (validate_and_convert_types) and it too turned every broken processor into a generic LogsProcessor without complaint. So a // logs-pipelines/ fence passing npm run verify proves the outer shape only; the concrete processor classes are what catch a field-level mistake. Two things no schema check can catch, both run above: an empty match_rules ({"match_rules": ""}) and a misspelled matcher (%{wrod:level}) both passed the strict script too. Grok syntax and semantics are validated only by Datadog, in the Grok parser UI or by the API at apply time.

Mental Model

Treat a pipeline as a function from a log to a log, composed in file order, with three pieces of hidden state. (1) Which attributes exist right now: a remapper needs its source to exist already, so a parser must precede it. (2) Tie-break rules: first wins (Grok rule, status remapper) versus last wins (date remapper). (3) What was decided before you ran: the Agent’s rules, JSON preprocessing, an earlier pipeline, and the pipeline’s own filter, which is evaluated against the log as it arrived and never sees what your processors extract. Everything else is data flow.

Trade-offs

Choice	Benefit	Cost
Mask or drop in the Agent (log_processing_rules)	Never leaves the host; proven above with analyze-logs	Irreversible; a wrong regex loses data; config lives on every host
Parse in a pipeline (Grok)	Change parsing without redeploying the app; works on logs you did not write	A format change silently stops matching; only 10 rules per Grok processor are recommended
Structured JSON logs + preprocessing	No Grok to break; reserved attributes handled before pipelines	The app must emit the names Datadog expects, or you add remappers
One big pipeline	Easy to read top to bottom	The 20-processor recommendation; every log is tested against its filter
Several small pipelines	Each owned by a team; nested pipelines split by service	A duplicate status remapper in a later pipeline is dead; order between pipelines matters
Explicit timezone in date()	Deterministic timestamps	You must know the zone of the writer; DST needs a named zone, not a fixed offset

Common Misconceptions

• “A pipeline can drop or mask a log.” Processors rewrite; keeping or dropping is Agent rules and backend exclusion filters. Masking that must never reach storage runs in the Agent (proven with analyze-logs).
• “The pipeline filter can use attributes my Grok extracts.” The docs: filtering happens “before any of the pipeline’s processors.”
• “If two Grok rules match, both apply.” “Only one can match any given log. The first one that matches, from top to bottom.”
• “Two date remappers: the first wins, like status.” The opposite: for the date remapper “the last one … is taken into account”; for the status remapper “only the first one”.
• “SEVERE maps to error.” By the documented prefix table it maps to OK (s), and FINE to emerg (f).
• “npm run verify on a logs-pipelines fence proves it works.” It proves the outer shape; all six broken variants above passed the top-level model.
• “A date() matcher without a timezone is safe.” For Bangkok-local logs the result is 7 hours ahead of the true instant, beyond the documented 2-hour future window.

Reading AI-generated code

Both of these are structurally valid: LogsPipeline and the strict script accept them. They are wrong for reasons only the docs can tell you.

🐛 Spot the 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"] }
]
}

Show the answer

The pipeline filter is @level:ERROR, but level is created by the Grok parser inside this same pipeline. The 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.” At filter time the log has no @level, so the pipeline matches nothing and the parser, date and status remappers never run: every log stays unparsed with the intake time as its timestamp, and there is no error anywhere. The model accepts it (it checks shape, not scoping). Filter on what exists at intake, and let the status remapper carry the 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"] }
  ]
}

🐛 Spot the 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"] }
]
}

Show the answer

The application writes Bangkok local time (UTC+7) but the date() matcher has no timezone argument. Computed above: a line stamped 10:00:01.123 is really 03:00:01.123Z; parsed as UTC it lands 7 hours in the future, and “Log events can be submitted up to 18 hours in the past and two hours in the future.” Whether the docs default is UTC is itself uncertain (one of four docs rows reproduced only at UTC+2), so the only safe answer is to name the zone: %{date("yyyy-MM-dd HH:mm:ss.SSS","Asia/Bangkok"):timestamp}, as in the pipeline JSON fixed earlier. Emitting UTC or ISO 8601 with an offset from the application avoids the guess altogether.

In the real agent analyze-logs run, what happened to the DEBUG healthcheck line and the card number?

Both reached Datadog and were cleaned by a pipeline The healthcheck line was excluded and the card number masked on the host, before anything was sent Only the card number was changed; healthchecks are always kept Nothing; analyze-logs ignores processing rules

A pipeline has the filter @level:ERROR, where level is extracted by a Grok parser inside the same pipeline. What happens?

It works, filters are evaluated lazily The filter is evaluated before the processors, so @level does not exist yet and the pipeline matches nothing -- silently The API rejects the pipeline Only the Grok parser is skipped

Two date remappers and two status remappers appear in the same pipeline, each pair pointing at different attributes. Which processors take effect according to the docs?

The first date remapper and the first status remapper The last date remapper and the first status remapper The last of each Both of each are applied and merged

By the documented status prefix rules (implemented in the lesson's status_map.py), which mapping is correct?

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

You ran the top-level LogsPipeline model on six deliberately broken pipelines (string instead of list for sources, missing grok, misspelled type, ...). How many did it reject?

All six None -- processors are a oneOf that falls back to a generic object; per-processor classes were needed to catch them Three of the six Only the nested one

An app logs Asia/Bangkok local time and the Grok rule uses date("yyyy-MM-dd HH:mm:ss.SSS") with no timezone. What does the computation in this lesson show?

A 7-hour shift into the future if the default is UTC, beyond the documented 2-hour future limit No effect; Datadog detects the zone A 7-hour shift into the past, inside the 18-hour window The log is always dropped by the Agent

Submit

Mark as complete