Log Collection and Processing Pipelines

The idea in one sentence

Logs get into Datadog because the Agent is told, via a logs: config block, where to read them from and what service/source to label them with, and once they arrive a Processing Pipeline parses and enriches each one before it ever reaches an index.

Telling the Agent where to look

The Agent does not go hunting for logs on its own — every log source is explicit. You add a logs: block, typically inside a conf.d integration config or the Agent’s own logs configuration, and tell it the type of source, the path (or port, for tcp), and two labels that matter a lot downstream: service and source.

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

• type — how the Agent reads the log. file tails a file on disk; other common values are tcp, udp, and docker for container log collection.
• path — where the log lives, for the file type. This is a plain file path, not a query.
• service — the name of the service that produced the log. This is the same service tag used everywhere else in Datadog (metrics, traces), which is exactly what lets a log line, a trace, and a metric all be correlated back to one deployable unit.
• source — a technology identifier, not a free-text label. Setting source: java (or nginx, postgresql, and so on) tells Datadog which built-in, source-specific pipeline to attach automatically, because Datadog ships pre-built Grok parsing rules for dozens of common frameworks and services.

Once the Agent picks up this config on its next reload, it starts tailing /my/app/file.log and shipping each line to Datadog’s intake, tagged with service:payments-api and source:java.

Processing Pipelines: parse and enrich after intake

A Processing Pipeline is an ordered list of processors that runs on every log that arrives at intake and matches the pipeline’s filter query. This stage happens after the log has left the host and reached Datadog, and before it is written into any index — its job is purely to parse unstructured text into structured attributes and enrich the log, not to decide whether the log gets kept.

A pipeline is scoped by a filter, for example service:my-service, so that only logs from the services or sources it’s meant for pass through it. You can have many pipelines active at once, each scoped to a different service or source, including the built-in ones that source: java, source: nginx, etc. attach automatically.

Common processors inside a pipeline, run in order:

• Grok Parser — applies a Grok pattern to the raw log message to pull structured attributes out of unstructured text, e.g. extracting duration, http.status_code, or user.id from a line that started as one long string.
• Date Remapper — tells Datadog which of the now-parsed attributes is the log’s official timestamp (instead of falling back to intake time), which matters for correctly placing the log on the timeline.
• Status Remapper — maps a parsed attribute (like a text field containing ERROR or WARN) onto Datadog’s standard log status (severity) field, so severity-based views and facets work consistently across services.

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

From Agent config to a parsed log

🐶 The one-line takeaway

The Agent’s logs: config block (type, path, service, source) is how a log gets collected at all, and a Processing Pipeline — a filter-scoped, ordered set of processors like Grok Parser, Date Remapper, and Status Remapper — parses and enriches every matching log after intake and before indexing; it does not decide what gets kept.

Proof: a real Agent genuinely tailing a real file

Mounted the config above (as conf.d/mylogs.d/conf.yaml, plus a second type: tcp block) into a real, freshly-started datadog/agent:7 container (DD_API_KEY fake, DD_DD_URL pointed at a closed local port — nothing sent anywhere) and ran 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 accepted with zero errors. To prove the Agent is genuinely tailing, not just parsing config, appended a real line to the mounted file and re-checked agent status’s Logs Agent section before and after:

$ 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 is the exact byte count of the line just appended — real, measured proof the Agent read the new content, not a static config echo. A second config block for type: tcp (the third collection type besides file/docker) also configchecks clean against the same running 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

The fourth collection type, type: docker (container logs), is not repeated here — The Datadog Agent and Integrations already covers container-log Autodiscovery (the ad.datadoghq.com/<container>.logs annotation on a Pod template) in depth; the same logs: collection concept applies, just driven by an annotation instead of a static conf.d file.

Pipeline JSON, validated — and a real gap this course’s own harness has

Built the three-processor pipeline sketched above as real JSON and validated it against the real, installed datadog_api_client==2.61.0’s LogsPipeline model:

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

🐶 A real, surprising harness gap: a broken processor nested in `processors` is NOT caught

Building a deliberately broken pipeline — a category-processor missing its two required fields, categories and target — and validating the whole LogsPipeline, even with _check_type=True, does not raise an 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 is typed as a giant Union of 20+ specific processor models. When a dict doesn’t strictly match any of them, the client silently falls back to the loose, generic LogsProcessor base type instead of raising — it does not tell you which specific processor type you meant, or that it’s missing required fields. Constructing the specific model directly catches it immediately:

# @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)

Lesson: validating a full pipeline JSON through LogsPipeline alone is not sufficient proof that every nested processor is well-formed — this course’s own harness README documents a related caveat (enum/query-string values aren’t eagerly rejected); this is the same shape of gap, one level deeper, inside a Union-typed list field.

Order matters: a remapper before its parser has nothing to remap

The three processors above are listed in the order they must run: Grok Parser first (it’s the only one that turns the raw message string into named attributes like timestamp and level at all), then Date Remapper and Status Remapper (they only have something to remap once those attributes exist). Swap the order — Date Remapper before Grok Parser — and the pipeline still validates as a perfectly well-formed LogsPipeline (both are correctly-shaped processor objects; nothing about the model can know they need to run one after another), but at runtime the Date Remapper looks for a timestamp attribute that does not exist yet, because the Grok Parser hasn’t run to create it:

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

This is a purely semantic bug — every processor is individually valid JSON, LogsPipeline validates the whole file without complaint, and the only way to catch it is knowing (or testing) that processor order is meaningful, not just a list.

🐶 Grok syntax: what this lesson could and could not verify offline

There is no local Grok matching engine installed in this harness — Grok pattern syntax (%{MATCHER:attribute}, named capture groups, the built-in matcher library) cannot be tested against sample text the way terraform validate or a Python model can be exercised. What the LogsGrokParserRules model does check is structural: match_rules is a required, non-empty string — it happily accepts match_rules: "not even valid grok syntax {{{" because the model only knows it should be a string, not that it should parse as Grok. This lesson’s match_rules value above was written by hand against Grok’s documented pattern syntax, not verified by running it.

Reading AI-generated code

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

Show the answer

This JSON is completely well-formed — every processor object has its required fields, and LogsPipeline(**payload, _check_type=True) validates it with zero complaints (proven above, the model checks shape, not order). The bug, proven above by tracing what each processor actually needs: Date Remapper and Status Remapper run before the Grok Parser that would have created the timestamp and level attributes they’re looking for. Both remappers silently do nothing on every log this pipeline processes — no error anywhere, just a log whose official timestamp stays “time of intake” instead of the time embedded in the message, and whose status facet never gets set from the parsed level. The fix is reordering the array: Grok Parser first, then the two remappers.

Mental Model

A Processing Pipeline is not a set of independent rules that all see the same input — it’s an assembly line where each station’s only input is what the previous stations left behind. The raw log’s message string is the only thing guaranteed to exist when the pipeline starts; every other attribute (timestamp, level, anything a Grok Parser extracts) exists only from the moment some earlier processor creates it onward. A LogsPipeline model — proven above — validates that every station is individually well-built; it has no way to know, and doesn’t try to know, whether station 3 depends on something only station 5 produces.

Trade-offs

• Agent-side exclude_at_match vs. a Processing Pipeline — dropping a line on the host (from the logs: block’s log_processing_rules) is free but irreversible; a Processing Pipeline always runs (you cannot skip it), so filtering decisions belong at the index/exclusion-filter stage covered in the next lesson, not here — this lesson’s pipeline only parses and enriches, it never drops.
• Structural JSON validation vs. semantic correctness — proven above twice (the Union-typed processor gap, and the order-dependency bug): a LogsPipeline passing validation tells you the shape is right, not that the pipeline does what you intended. Both require separately reasoning about what each processor actually needs as input.
• source: java (built-in pipeline) vs. a hand-written pipeline — a built-in, source-specific pipeline needs zero authoring and stays maintained by Datadog as the underlying framework’s log format evolves, at the cost of being generic; a hand-written pipeline like the one above is exact to your own log format, at the cost of you owning it forever.

Common Misconceptions

• “A Processing Pipeline decides whether a log gets kept.” No — this lesson opened with this distinction and it holds up under the JSON proof above: pipelines parse and enrich every log matching their filter; whether a log survives long enough to be indexed is a separate decision, covered in the next lesson.
• “If a pipeline’s JSON validates, the pipeline is correct.” No — proven above twice: a LogsPipeline with processors in the wrong order validates cleanly and silently does nothing useful, and a malformed nested processor can silently downgrade to a generic type instead of raising.
• “Grok syntax is checked before a pipeline is created.” Not by this course’s tooling — proven above: LogsGrokParserRules only requires match_rules to be a non-empty string, accepting syntactically invalid Grok without complaint.

In an Agent logs: block with type: file, what does the path field specify?

The Grok pattern to apply The file on disk that the Agent should tail The name of the processing pipeline The index the log should be written to

Setting source: java in a logs config primarily does what?

Deletes non-Java log lines before they leave the host Sets the log retention period Attaches Datadog's built-in, source-specific processing pipeline for Java logs Encrypts the log in transit

What does a Processing Pipeline's filter query (e.g. service:my-service) control?

Which logs are permanently dropped and never billed Which logs from intake are scoped into that pipeline for parsing and enrichment How many days the matching logs are retained in an index Which hosts run the Agent

Where does Processing Pipeline parsing happen relative to indexing?

After a log is written into an index On the host, before the log leaves via the Agent After intake and before the log reaches an index Pipelines and indexing happen at the same stage, in no particular order

A pipeline JSON has Date Remapper and Status Remapper listed BEFORE Grok Parser. What does LogsPipeline(**payload, _check_type=True) do with it, and what actually happens at runtime?

The model raises an error immediately, refusing the out-of-order processors The model validates cleanly (shape is fine); at runtime, both remappers silently do nothing because the attributes they look for do not exist until Grok Parser runs The model automatically reorders the processors to the correct sequence Grok Parser runs first regardless of array order

A category-processor is nested inside a LogsPipeline's processors list, missing its required categories and target fields. What really happens when the whole pipeline is validated?

A TypeError names the missing categories and target fields No error at all — the dict silently resolves to the generic LogsProcessor type instead of LogsCategoryProcessor, and the missing-fields problem is never surfaced The processor is dropped from the list automatically Validation always fails for any processors list with more than 3 items

Submit

Mark as complete