[Filebeat] Improve ECS categorization field mappings in traefik module

CHANGELOG.next.asciidoc

@@ -413,6 +413,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Explicitly set ECS version in all Filebeat modules. {pull}19198[19198]
 - Add new mode to multiline reader to aggregate constant number of lines {pull}18352[18352]
 - Add automatic retries and exponential backoff to httpjson input. {pull}18956[18956]
+- Improve ECS categorization field mappings in traefik module. {issue}16183[16183] {pull}19379[19379]
 
 *Heartbeat*
 

filebeat/module/traefik/access/ingest/pipeline.yml

@@ -0,0 +1,106 @@
+description: Pipeline for parsing Traefik access logs. Requires the geoip and user_agent
+  plugins.
+processors:
+- dissect:
+    field: message
+    pattern: '%{source.address} %{traefik.access.user_identifier} %{user.name} [%{traefik.access.time}]
+      "%{http.request.method} %{url.original} HTTP/%{http.version}" %{http.response.status_code}
+      %{traefik.access.message}'
+- grok:
+    field: traefik.access.message
+    patterns:
+    - (?:%{NUMBER:http.response.body.bytes:long}|-)( (?:"%{DATA:http.request.referrer}"|-)?(
+      (?:"%{DATA:user_agent.original}"|-)?)?( (?:%{NUMBER:traefik.access.request_count:long}|-)?)?(
+      (?:"%{DATA:traefik.access.frontend_name}"|-)?)?( "%{DATA:traefik.access.backend_url}")?(
+      %{NUMBER:temp.duration:long}ms)?)?
+    ignore_missing: true
+- remove:
+    field: message
+    ignore_missing: true
+- remove:
+    field: traefik.access.message
+    ignore_missing: true
+- rename:
+    field: '@timestamp'
+    target_field: event.created
+- date:
+    field: traefik.access.time
+    target_field: '@timestamp'
+    formats:
+    - dd/MMM/yyyy:H:m:s Z
+- remove:
+    field: traefik.access.time
+- convert:
+    field: http.response.status_code
+    type: long
+- grok:
+    field: source.address
+    patterns:
+    - ^(%{IP:source.ip}|%{HOSTNAME:source.domain})$
+- script:
+    lang: painless
+    source: ctx.event.duration = Math.round(ctx.temp.duration * params.scale)
+    params:
+      scale: 1000000
+    if: ctx.temp?.duration != null
+- remove:
+    field: temp.duration
+    ignore_missing: true
+- user_agent:
+    field: user_agent.original
+    ignore_failure: true
+- geoip:
+    field: source.ip
+    target_field: source.geo
+    ignore_missing: true
+- geoip:
+    database_file: GeoLite2-ASN.mmdb
+    field: source.ip
+    target_field: source.as
+    properties:
+    - asn
+    - organization_name
+    ignore_missing: true
+- rename:
+    field: source.as.asn
+    target_field: source.as.number
+    ignore_missing: true
+- rename:
+    field: source.as.organization_name
+    target_field: source.as.organization.name
+    ignore_missing: true
+- set:
+    field: event.kind
+    value: event
+- append:
+    field: event.category
+    value: web
+    if: "ctx?.http?.request?.method != null && ctx.http.request.method != '-'"
+- append:
+    field: event.type
+    value: access
+    if: "ctx?.http?.request?.method != null && ctx.http.request.method != '-'"
+- set:
+    field: event.outcome
+    value: success
+    if: "ctx?.http?.response?.status_code != null && ctx.http.response.status_code < 400"
+- set:
+    field: event.outcome
+    value: failure
+    if: "ctx?.http?.response?.status_code != null && ctx.http.response.status_code >= 400"
+- append:
+    field: related.ip
+    value: "{{source.ip}}"
+    if: "ctx?.source?.ip != null"
+- append:
+    field: related.ip
+    value: "{{destination.ip}}"
+    if: "ctx?.destination?.ip != null"
+- append:
+    field: related.user
+    value: "{{user.name}}"
+    if: "ctx?.user?.name != null && ctx.user.name != '-'"
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'

filebeat/module/traefik/access/manifest.yml

@@ -9,7 +9,7 @@ var:
     os.windows:
       - c:/programdata/traefik/logs/*access.log*
 
-ingest_pipeline: ingest/pipeline.json
+ingest_pipeline: ingest/pipeline.yml
 input: config/traefik-access.yml
 
 requires.processors: