filebeat-8.11.3-nginx-access-pipeline-custom.json

PUT _ingest/pipeline/filebeat-8.11.3-nginx-access-pipeline-custom
{
  "description": "Pipeline for parsing Nginx access logs. Requires the geoip and user_agent plugins.",
  "processors": [
    {
      "set": {
        "field": "event.ingested",
        "value": "{{_ingest.timestamp}}"
      }
    },
    {
      "rename": {
        "field": "message",
        "target_field": "event.original"
      }
    },
    {
      "grok": {
        "field": "event.original",
        "pattern_definitions": {
          "NGINX_HOST": "(?:%{IP:destination.ip}|%{NGINX_NOTSEPARATOR:destination.domain})(:%{NUMBER:destination.port})?",
          "NGINX_NOTSEPARATOR": "[^\t ,:]+",
          "NGINX_ADDRESS_LIST": "(?:%{IP}|%{WORD})(\"?,?\\s*(?:%{IP}|%{WORD}))*"
        },
        "ignore_missing": true,
        "patterns": [
          "(%{NGINX_HOST} )?\"?(?:%{NGINX_ADDRESS_LIST:nginx.access.remote_ip_list}|%{NOTSPACE:source.address}) - (-|%{DATA:user.name}) \\[%{HTTPDATE:nginx.access.time}\\] \"%{DATA:nginx.access.info}\" %{NUMBER:http.response.status_code:long} %{NUMBER:http.response.body.bytes:long} \"(-|%{DATA:http.request.referrer})\" \"(-|%{DATA:user_agent.original})\" \"-\" \"(-|%{IPORHOST:nginx.access.host.name})\" sn=\"(-|%{DATA:nginx.access.host.domain})\" rt=(-|%{NUMBER:nginx.access.request_time:float}) ua=\"(-|%{DATA:nginx.access.upstream_addr})\" us=\"(-|%{DATA:nginx.access.upstream_status})\" ut=\"(-|%{NUMBER:nginx.access.upstream_response_time:float})\" ul=\"(-|%{NUMBER:nginx.access.upstream_response_length:long})\" cs=-",
          "(%{NGINX_HOST} )?\"?(?:%{NGINX_ADDRESS_LIST:nginx.access.remote_ip_list}|%{NOTSPACE:source.address}) - (-|%{DATA:user.name}) \\[%{HTTPDATE:nginx.access.time}\\] \"%{DATA:nginx.access.info}\" %{NUMBER:http.response.status_code:long} %{NUMBER:http.response.body.bytes:long} \"(-|%{DATA:http.request.referrer})\" \"(-|%{DATA:user_agent.original})\""
        ]
      }
    },
    {
      "grok": {
        "field": "nginx.access.info",
        "patterns": [
          "%{WORD:http.request.method} %{DATA:_tmp.url_orig} HTTP/%{NUMBER:http.version}",
          ""
        ],
        "ignore_missing": true
      }
    },
    {
      "uri_parts": {
        "field": "_tmp.url_orig",
        "ignore_failure": true
      }
    },
    {
      "set": {
        "value": "{{destination.domain}}",
        "if": "ctx.url?.domain == null && ctx.destination?.domain != null",
        "field": "url.domain"
      }
    },
    {
      "remove": {
        "field": [
          "nginx.access.info",
          "_tmp.url_orig"
        ],
        "ignore_missing": true
      }
    },
    {
      "split": {
        "field": "nginx.access.remote_ip_list",
        "separator": "\"?,?\\s+",
        "ignore_missing": true
      }
    },
    {
      "split": {
        "field": "nginx.access.origin",
        "separator": "\"?,?\\s+",
        "ignore_missing": true
      }
    },
    {
      "set": {
        "field": "source.address",
        "if": "ctx.source?.address == null",
        "value": ""
      }
    },
    {
      "script": {
        "if": "ctx.nginx?.access?.remote_ip_list != null && ctx.nginx.access.remote_ip_list.length > 0",
        "lang": "painless",
        "source": "boolean isPrivate(def dot, def ip) {\n  try {\n    StringTokenizer tok = new StringTokenizer(ip, dot);\n    int firstByte = Integer.parseInt(tok.nextToken());\n    int secondByte = Integer.parseInt(tok.nextToken());\n    if (firstByte == 10) {\n      return true;\n    }\n    if (firstByte == 192 && secondByte == 168) {\n      return true;\n    }\n    if (firstByte == 172 && secondByte >= 16 && secondByte <= 31) {\n      return true;\n    }\n    if (firstByte == 127) {\n      return true;\n    }\n    return false;\n  }\n  catch (Exception e) {\n    return false;\n  }\n} try {\n  ctx.source.address = null;\n  if (ctx.nginx.access.remote_ip_list == null) {\n    return;\n  }\n  def found = false;\n  for (def item : ctx.nginx.access.remote_ip_list) {\n    if (!isPrivate(params.dot, item)) {\n      ctx.source.address = item;\n      found = true;\n      break;\n    }\n  }\n  if (!found) {\n    ctx.source.address = ctx.nginx.access.remote_ip_list[0];\n  }\n} catch (Exception e) {\n  ctx.source.address = null;\n}",
        "params": {
          "dot": "."
        }
      }
    },
    {
      "remove": {
        "field": "source.address",
        "if": "ctx.source.address == null"
      }
    },
    {
      "grok": {
        "field": "source.address",
        "patterns": [
          "^%{IP:source.ip}$"
        ],
        "ignore_failure": true
      }
    },
    {
      "set": {
        "field": "event.created",
        "copy_from": "@timestamp"
      }
    },
    {
      "date": {
        "field": "nginx.access.time",
        "target_field": "@timestamp",
        "formats": [
          "dd/MMM/yyyy:H:m:s Z"
        ],
        "on_failure": [
          {
            "append": {
              "field": "error.message",
              "value": "{{ _ingest.on_failure_message }}"
            }
          }
        ]
      }
    },
    {
      "remove": {
        "field": "nginx.access.time"
      }
    },
    {
      "user_agent": {
        "field": "user_agent.original",
        "ignore_missing": true
      }
    },
    {
      "geoip": {
        "field": "source.ip",
        "target_field": "source.geo",
        "ignore_missing": true
      }
    },
    {
      "geoip": {
        "field": "source.ip",
        "target_field": "source.as",
        "properties": [
          "asn",
          "organization_name"
        ],
        "ignore_missing": true,
        "database_file": "GeoLite2-ASN.mmdb"
      }
    },
    {
      "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"
      }
    },
    {
      "append": {
        "value": "access",
        "field": "event.type"
      }
    },
    {
      "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": {
        "value": "{{user.name}}",
        "if": "ctx?.user?.name != null",
        "field": "related.user"
      }
    },
    {
      "script": {
        "description": "This script processor iterates over the whole document to remove fields with null values.",
        "source": "void handleMap(Map map) {\n  for (def x : map.values()) {\n    if (x instanceof Map) {\n        handleMap(x);\n    } else if (x instanceof List) {\n        handleList(x);\n    }\n  }\n  map.values().removeIf(v -> v == null);\n}\nvoid handleList(List list) {\n  for (def x : list) {\n      if (x instanceof Map) {\n          handleMap(x);\n      } else if (x instanceof List) {\n          handleList(x);\n      }\n  }\n}\nhandleMap(ctx);\n",
        "lang": "painless"
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "field": "error.message",
        "value": "{{ _ingest.on_failure_message }}"
      }
    }
  ]
}