Adds advanced Flow examples page and examples from kube-logging/logging-operator#1234

Examples provided by genofire

Author: fekete-robert

content/docs/examples/_index.md

@@ -5,7 +5,7 @@ weight: 330
 
 ## Flow examples
 
-The following examples show some simple flows. For more examples that use filters, see {{% xref "/docs/examples/filters-in-flows.md" %}}.
+The following examples show some simple flows. For more advanced examples, see {{% xref "/docs/examples/filters-in-flows/_index.md" %}} and {{% xref "/docs/examples/flows/_index.md" %}}.
 
 ### Flow with a single output
 

content/docs/examples/flows/_index.md

@@ -0,0 +1,25 @@
+---
+title: Advanced Flow examples
+weight: 100
+---
+
+## CoreDNS
+
+This Flow:
+
+- selects [CoreDNS](https://coredns.io/) messages (every message with the `k8s-app: coredns` label),
+- parses them, and
+- sets a number of related [Elastic Common Schema (ECS)](https://www.elastic.co/guide/en/ecs/current/ecs-getting-started.html) fields based on the content of the message using the {{% xref "/docs/configuration/plugins/filters/record_modifier.md" %}} plugin.
+
+{{< include-code "logging_flow_coredns.yaml" "yaml" >}}
+
+## NGINX Ingress Controller
+
+This Flow:
+
+- selects [NGINX Ingress Controller](https://docs.nginx.com/nginx-ingress-controller/) messages (every message with the `app-kubernetes-io/name: ingress-nginx` label),
+- parses them, and
+- sets a number of related [Elastic Common Schema (ECS)](https://www.elastic.co/guide/en/ecs/current/ecs-getting-started.html) fields based on the content of the message using the {{% xref "/docs/configuration/plugins/filters/record_modifier.md" %}} plugin.
+- It also adds GeoIP-related fields based on the source of the traffic using the [Fluentd GeoIP filter]({{< relref "/docs/configuration/plugins/filters/geoip.md" >}}).
+
+{{< include-code "logging_flow_nginx_ingress.yaml" "yaml" >}}

content/docs/examples/flows/logging_flow_coredns.yaml

@@ -0,0 +1,36 @@
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Flow
+metadata:
+  name: coredns
+  namespace: kube-system
+spec:
+  filters:
+    - tag_normaliser: {}
+    - parser:
+        reserve_data: true
+        remove_key_name_field: true
+        parse:
+          type: "regexp"
+          expression: '^\[(?<log.level>.*)\] \[?(?<source.address>.*)\]?:(?<source.port>.*) - (?<dns.id>.*) "(?<dns.question.type>.*) (?<dns.question.class>.*) (?<dns.question.name>.*)\.? (?<network.transport>.*) (?<coredns.query.size>.*) (?<coredns.dnssec_ok>.*) (?<bufsize>.*)" (?<dns.response_code>.*) (?<dns.header_flags>.*) (?<coredns.response.size>.*) (?<coredns.duration>.*)s'
+          types: "source.port:integer,dns.id:integer,coredns.query.size:integer,coredns.dnssec_ok:bool,bufsize:integer,dns.header_flags:array,coredns.response.size:integer,coredns.duration:float"
+    - record_modifier:
+        records:
+          - source.ip: '${ record["source.address"] }'
+          - dns.header_flags: '${ record["dns.header_flags"].map(&:upcase) }'
+          - event.duration: '${ record["coredns.duration"] * 1000000000 }'
+          - event.kind: "event"
+          - event.category: "network"
+          - event.type: "protocol"
+          - event.outcome: '${ record["dns.response_code"] == "NOERROR" ? "success" : "failure" }'
+          - event.protocol: "dns"
+          - event.module: "coredns"
+          - related.ip: '${ record["source.address"] }'
+          # for dashboard
+          - fileset.name: "kubernetes"
+          # alias in elastic
+          - coredns.query.name: '${ record["dns.question.name"] }'
+        remove_keys: "coredns.duration,coredns.dnssec_ok"
+  match:
+    - select:
+        labels:
+          k8s-app: "coredns"

content/docs/examples/flows/logging_flow_nginx_ingress.yaml

@@ -0,0 +1,85 @@
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Flow
+metadata:
+  name: "ingress-nginx"
+spec:
+  filters:
+    - tag_normaliser: {}
+    - parser:
+        reserve_data: true
+        remove_key_name_field: true
+        inject_key_prefix: "nginx."
+        parse:
+          type: "regexp"
+          # from https://raw.githubusercontent.com/fluent/fluentd/master/lib/fluent/plugin/parser_nginx.rb
+          # enharance for ingress-controller by e.g.
+          # https://github.com/elastic/beats/blob/v8.6.1/filebeat/module/nginx/ingress_controller/ingest/pipeline.yml
+          expression: '^(?<remote>[^ ]*) -?(?<host>[^ ]*) -?(?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +(?<httpversion>HTTP\/[0-9\.]+)))" (?<code>[^ ]*) (?<size>[^ ]*)(?: "-?(?<referer>[^\"]*)" "(?<agent>[^\"]*)"\s+(?<reqlength>[0-9]+)\s(?<reqtime>[0-9\.]+)\s\[(?<upstream_name>[^ ]*)\]\s\[(?<alternative_name>[^\]]*)\] -?(?<upstream_address_list>[^ -]*) -?(?<reslength_list>[0-9\,]*) -?(?<restime_list>[0-9\.\,]*) -?(?<rescode_list>[0-9\,]*) (?<reqid>[^ ]+))?$'
+          types: 'code:integer,size:integer,reqlength:integer,reqtime:float,upstream_address_list:array,reslenght_list:array,restime_list:array,rescode_list:array'
+          time_key: "time"
+          time_format: "%d/%b/%Y:%H:%M:%S %z"
+    - record_modifier:
+        records:
+          - destination.domain: '${ !(record["nginx.referer"].nil?) ? URI(record["nginx.referer"]).host : record["nginx.host"] }'
+            url.original: '${ record["nginx.referer"] }${ record["nginx.path"] }'
+            url.domain: '${ !(record["nginx.referer"].nil?) ? URI(record["nginx.referer"]).host : record["nginx.host"] }'
+            url.path: '${ record["nginx.path"] }'
+            http.version: '${ record["nginx.httpversion"] }'
+            nginx.access.remote_ip_list: '${ record["nginx.remote"] }'
+            source.address: '${ record["nginx.remote"] }'
+            source.ip: '${ record["nginx.remote"] }'
+            related.ip: '${ record["nginx.remote"] }'
+            http.request.method: '${ record["nginx.method"] }'
+            http.request.referrer: '${ record["nginx.referer"] }'
+            user.name: '${ record["nginx.user"] }'
+            related.user: '${ record["nginx.user"] }'
+            user_agent.original: '${ record["nginx.agent"] }'
+            http.response.status_code: '${ record["nginx.code"] }'
+            nginx.ingress_controller.http.request.length: '${ record["nginx.reqlength"] }'
+            nginx.ingress_controller.http.request.time: '${ record["nginx.reqtime"] }'
+            nginx.ingress_controller.upstream.name: '${ record["nginx.upstream_name"] }'
+            nginx.ingress_controller.upstream.alternative_name: '${ record["nginx.alternative_name"] }'
+            nginx.ingress_controller.upstream_address_list: '${ record["nginx.upstream_address_list"] }'
+            # TODO split ip and port
+            nginx.ingress_controller.upstream.address.merged: '${ [record["upstream_address_list"]].flatten&.last }'
+            nginx.ingress_controller.upstream.response.length_list: '${ [record["nginx.reslength_list"]].flatten&.map(&:to_i) }'
+            nginx.ingress_controller.upstream.response.length: '${ [record["nginx.reslength_list"]].flatten&.last&.to_i }'
+            nginx.ingress_controller.upstream.response.time_list: '${ [record["nginx.restime_list"]].flatten&.map(&:to_f) }'
+            nginx.ingress_controller.upstream.response.time: '${ [record["nginx.restime_list"]].flatten&.last&.to_f }'
+            nginx.ingress_controller.upstream.response.status_code_list: '${ [record["nginx.rescode_list"]].flatten&.map(&:to_i) }'
+            nginx.ingress_controller.upstream.response.status_code: '${ [record["nginx.rescode_list"]].flatten&.last&.to_i }'
+            nginx.ingress_controller.http.request.id: '${ record["nginx.reqid"] }'
+            http.request.id: '${ record["nginx.reqid"] }'
+            http.response.body.bytes: '${ record["nginx.size"] }'
+            event.created: '${ time * 1000 }'
+            event.kind: "event"
+            event.category: "web"
+            event.type: "access"
+            event.module: "nginx"
+            event.outcome: '${ record["nginx.code"].to_i < 400 ? "success" : "failure" }'
+            # for dashboard
+            fileset.name: '${[ "ingress_controller", "access" ]}'
+        remove_keys: "nginx.remote,nginx.host,nginx.user,nginx.method,nginx.path,nginx.httpversion,nginx.code,nginx.size,nginx.referer,nginx.agent,nginx.reqlength,nginx.reqtime,nginx.upstream_name,nginx.alternative_name,nginx.upstream_address_list,nginx.reslength_list,nginx.restime_list,nginx.rescode_list,nginx.reqid"
+    - geoip:
+       geoip_lookup_keys: "source.ip"
+       backend_library: geoip2_c
+       skip_adding_null_record: false
+       records:
+         - source.geo.country_name: '${country.names.en["source.ip"]}'
+           source.geo.country_iso_code: '${country.iso_code["source.ip"]}'
+           source.geo.city_name: '${city.names.en["source.ip"]}'
+           source.geo.region_iso_code: '${subdivisions.0.iso_code["source.ip"]}'
+           source.geo.region_name: '${subdivisions.0.names.en["source.ip"]}'
+           source.geo.location: '''{ "lat": ${location.latitude["source.ip"]}, "lon": ${location.longitude["source.ip"]} }'''
+    # - geoip:
+    #     geoip_lookup_keys: "source.ip"
+    #     backend_library: geoip
+    #     skip_adding_null_record: false
+    #     records:
+    #       - source.geo.continent_name: '${continent.names.en["source.ip"]}'
+    #       # source.as.number: '${asn["source.ip"]}'
+    #       # source.as.organization.name: '${organization["source.ip"]}'
+  match:
+    - select:
+        labels:
+          app-kubernetes-io/name: "ingress-nginx"