feat: added sidecar to turn logs into metrics and forward remaining logs

README.md

@@ -30,7 +30,9 @@ Since we are using [Open Policy Agent](https://www.openpolicyagent.org/) (aka `O
 
 ### Log Format
 
-The docker image provides default log format (`/etc/nginx/log_format`). It's not possible to extend the log format, so if you'd want to add/remove certain fields you have to override it.
+The docker image provides a default JSON log format (`/etc/nginx/log_format.conf`). It can't be extended in place, so adding or removing fields means overriding the whole file.
+
+When the Fluent Bit sidecar is enabled, the chart renders a second, human-readable format alongside it: `kubectl logs` gets the readable one while the JSON format goes to the sidecar. Controlled by `fluentbit.accessLog.stdoutReadable`.
 
 ## Helm Chart
 
@@ -46,6 +48,8 @@ There's an option to dynamically add annotations to the pod. You might find it u
 
 There's support for instrumenting NGINX with OpenTelemetry (currently only for tracing). Simply the relevant parameters in the `values.yaml` file.
 
+There's also an optional Fluent Bit log-processing sidecar, off by default. When enabled it forwards only the selected access-log statuses and error-log severities to a central OTLP/HTTP endpoint instead of shipping every line, and derives Prometheus metrics from the access log — merged with the NGINX exporter's series onto one `/metrics` endpoint. See the `fluentbit` parameters in [values.md](./helm/values.md).
+
 #### Overriding NGINX configuration files
 
 If you wish to override the default configuration files, you can do it by providing an external ConfigMap and supplying Volumes & VolumeMounts that'll be added to the Deployment.

helm/config/fluent-bit.yaml

@@ -0,0 +1,180 @@
+# Fluent Bit configuration, rendered through Helm `tpl` from the chart ConfigMap.
+#
+# YAML, not classic `.conf`: only YAML config supports `processors`, the one way to reach an
+# outgoing record's OTLP resource attributes. Classic config can set log attributes but never
+# resource ones, so every forwarded record would arrive at Loki as `unknown_service`.
+#
+# Fluent Bit's own `fluentbit_metrics` input is deliberately not collected — too noisy.
+{{- $debug := .Values.fluentbit.debug }}
+{{- $stages := ternary $debug.stages (dict) $debug.enabled }}
+
+service:
+  flush: 1
+  daemon: off
+  log_level: {{ $debug.enabled | ternary $debug.logLevel "info" }}
+  # Relative to this file, i.e. the image's own /fluent-bit/etc/parsers.conf (json, syslog, …).
+  parsers_file: parsers.conf
+{{- if .Values.fluentbit.errorLog.enabled }}
+
+# Parses the standard open-source nginx error-log line into structured fields. Format:
+#   2024/01/15 10:23:45 [error] 1234#5678: *90 <message>, client: …, server: …, request: "…"
+# The leading timestamp is optional (some syslog paths strip it), the connection id (*cid) is
+# optional (worker/startup messages omit it), and the client/server/request tail is optional
+# and matched as a unit so a comma inside <message> does not truncate it.
+parsers:
+  - name: nginx_error
+    format: regex
+    regex: '^(?:(?<time>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) )?\[(?<level>\w+)\] (?<pid>\d+)#\d+:(?: \*\d+)? (?<message>.*?)(?:, client: (?<client>[^,]+), server: (?<server>[^,]*), request: "(?<request>[^"]*)".*)?$'
+    time_key: time
+    time_format: '%Y/%m/%d %H:%M:%S'
+    time_keep: off
+{{- end }}
+
+pipeline:
+  inputs:
+    # nginx `access_log syslog:server=127.0.0.1:<port>` delivers the JSON `main` format here.
+    # Parser is mandatory: nginx frames its datagrams as RFC3164 (`<PRI>Mmm dd hh:mm:ss host
+    # tag: msg`), while Fluent Bit's syslog input defaults to syslog-rfc5424 — that mismatch
+    # fails to parse and silently drops every record before any filter runs.
+    - name: syslog
+      tag: nginx.access
+      mode: udp
+      listen: 127.0.0.1
+      port: {{ .Values.fluentbit.accessLog.syslogPort }}
+      parser: syslog-rfc3164
+      buffer_chunk_size: 32k
+      buffer_max_size: 64k
+{{- if .Values.fluentbit.errorLog.enabled }}
+    # Same RFC3164 framing as the access log, so the same explicit parser is required.
+    - name: syslog
+      tag: nginx.error
+      mode: udp
+      listen: 127.0.0.1
+      port: {{ .Values.fluentbit.errorLog.syslogPort }}
+      parser: syslog-rfc3164
+      buffer_chunk_size: 32k
+      buffer_max_size: 64k
+{{- end }}
+{{- if and .Values.fluentbit.accessLog.metrics.enabled .Values.fluentbit.accessLog.metrics.scrapeExporter }}
+    # Scrape the existing nginx-prometheus-exporter so its stub_status gauges merge into the
+    # single /metrics endpoint below (port = the exporter's own mclabels.prometheus.port).
+    - name: prometheus_scrape
+      tag: nginx.metrics.exporter
+      host: 127.0.0.1
+      port: {{ .Values.mclabels.prometheus.port }}
+      metrics_path: /metrics
+{{- end }}
+
+  filters:
+{{- if $stages.received }}
+    # Debug stage `received`: both pipelines straight off the syslog input, before any decoding.
+    # `stdout` is a filter, so it prints the records as they are at this point and passes them on.
+    - name: stdout
+      match_regex: ^nginx\.(access|error)$
+{{- end }}
+    # The syslog message body is the nginx JSON line; decode it so the OTel-shaped fields
+    # (including the nested http.response.status_code) become queryable by the grep filter.
+    - name: parser
+      match: nginx.access
+      key_name: message
+      parser: json
+      reserve_data: off
+{{- if $stages.parsed }}
+    # Debug stage `parsed`: access records after JSON decode. Deliberately ahead of the guard
+    # below, so lines the decode failed on are still visible here (they keep a raw `message`).
+    - name: stdout
+      match: nginx.access
+{{- end }}
+    # Drop records the JSON parser could not decode. log_to_metrics segfaults on a record whose
+    # value_field is missing, so an unparseable line would otherwise crash-loop the sidecar.
+    - name: grep
+      match: nginx.access
+      regex: "$Attributes['http.response.status_code'] ."
+{{- if .Values.fluentbit.accessLog.exclude.enabled }}
+    # Drop health-check / probe traffic before anything downstream sees it, so it neither
+    # inflates the derived metrics nor reaches Loki. Placed after the JSON parser (the matched
+    # field only exists once decoded) and before log_to_metrics.
+    - name: grep
+      match: nginx.access
+      exclude: {{ printf "%s %s" .Values.fluentbit.accessLog.exclude.key .Values.fluentbit.accessLog.exclude.regex | quote }}
+{{- end }}
+{{- if .Values.fluentbit.accessLog.metrics.enabled }}
+    # Derive metrics from the full stream, before grep drops the non-forwarded records, so the
+    # counters cover every request. Overridable via fluentbit.accessLog.metrics.filters (tpl).
+{{- tpl (toYaml .Values.fluentbit.accessLog.metrics.filters) . | nindent 4 }}
+{{- end }}
+    # Keep only the status codes selected by fluentbit.accessLog.forward; drop everything else.
+    # When no forwarding rule is active the regex matches nothing, so all records are dropped.
+    - name: grep
+      match: nginx.access
+      regex: {{ printf "$Attributes['http.response.status_code'] %s" (include "nginx.fluentbit.accessLogRegex" .) | quote }}
+{{- if .Values.fluentbit.maskQueryParams }}
+    # Mask fluentbit.maskQueryParams in the fields that carry the query string: url.query,
+    # url.full and the request line in `Body`. Only the forwarded records need it, and it has to
+    # precede both filters below: the hook would see the raw token, flatten puts Attributes out of
+    # reach. The error pipeline is masked in envelope_error, once its request line is parsed.
+    - name: lua
+      match: nginx.access
+      script: /fluent-bit/etc/metadata.lua
+      call: mask_query_params
+{{- end }}
+{{- if .Values.fluentbit.lua.enabled }}
+    # Operator hook (fluentbit.lua): after the forwarding grep, so the script only sees records
+    # already selected for shipping, and before the flatten filter below, which is what keeps
+    # `record["Attributes"]` reachable from the script.
+    - name: lua
+      match: nginx.access
+      script: /fluent-bit/scripts/custom.lua
+      call: {{ .Values.fluentbit.lua.call }}
+{{- end }}
+{{- if .Values.fluentbit.errorLog.enabled }}
+    # Parse the plaintext nginx error line into structured fields, using the nginx_error parser
+    # defined above. Severity is already filtered at nginx, so no grep.
+    - name: parser
+      match: nginx.error
+      key_name: message
+      parser: nginx_error
+      reserve_data: off
+{{- if $stages.parsed }}
+    # Debug stage `parsed`, error pipeline.
+    - name: stdout
+      match: nginx.error
+{{- end }}
+{{- end }}
+    # Last filter, deliberately: it moves the decoded fields out of the record body into the
+    # record's metadata, after which `$Attributes['…']` accessors (the greps and log_to_metrics
+    # above, and anything supplied through fluentbit.accessLog.metrics.filters) no longer resolve.
+    - name: lua
+      match: nginx.access
+      script: /fluent-bit/etc/metadata.lua
+      call: flatten_attributes
+{{- if .Values.fluentbit.errorLog.enabled }}
+    # Same treatment for the error pipeline, whose records reach here with the nginx_error
+    # parser's field names: rewritten to semconv attributes in metadata + an OTel envelope, so
+    # both outputs below are identical apart from the tag they match.
+    - name: lua
+      match: nginx.error
+      script: /fluent-bit/etc/metadata.lua
+      call: envelope_error
+{{- end }}
+
+  outputs:
+    {{- include "nginx.fluentbit.logsOutput" (dict "ctx" . "match" "nginx.access") | nindent 4 }}
+{{- if .Values.fluentbit.errorLog.enabled }}
+    {{- include "nginx.fluentbit.logsOutput" (dict "ctx" . "match" "nginx.error") | nindent 4 }}
+{{- end }}
+{{- if $debug.enabled }}
+    # Debug: the records handed to the OTLP outputs above, i.e. what actually reaches Alloy once
+    # every filter has run. Always printed while debug is on — out_stdout flushes the process's
+    # stdout buffer, which is what makes the `stdout` filter dumps above appear promptly too.
+    - name: stdout
+      match_regex: ^nginx\.(access|error)$
+{{- end }}
+{{- if .Values.fluentbit.accessLog.metrics.enabled }}
+    # Serve the merged /metrics (log-derived + scraped exporter) on metrics.port.
+    # `match: '*'` is safe — prometheus_exporter only handles metric events.
+    - name: prometheus_exporter
+      match: '*'
+      host: 0.0.0.0
+      port: {{ .Values.fluentbit.accessLog.metrics.port }}
+{{- end }}

helm/config/log_format.conf

@@ -28,8 +28,11 @@ log_format main escape=json
         '"mapcolonies.http.upstream_connect_time":"$upstream_connect_time",'
         '"mapcolonies.http.upstream_response_time":"$upstream_response_time",'
         '"mapcolonies.http.upstream_response_length":"$upstream_response_length",'
-        '"mapcolonies.http.upstream_bytes_sent": $upstream_bytes_sent,'
-        '"mapcolonies.http.upstream_bytes_received": $upstream_bytes_received,'
+        # Quoted, unlike the other size fields: nginx writes a bare `-` for these when the
+        # request had no upstream, which is invalid JSON as a number. The record then fails
+        # Fluent Bit's JSON parser and is dropped by the guard grep in config/fluent-bit.yaml.
+        '"mapcolonies.http.upstream_bytes_sent":"$upstream_bytes_sent",'
+        '"mapcolonies.http.upstream_bytes_received":"$upstream_bytes_received",'
         '"mapcolonies.http.upstream_cache_status":"$upstream_cache_status",'
         '"server.address":"$host",'
         '"server.port":"$server_port",'
@@ -53,3 +56,15 @@ log_format main escape=json
     '"InstrumentationScope":"access.log",'
     '"Body":"$request"'
 '}';
+{{- if .Values.fluentbit.enabled }}
+
+# Human-readable (combined-style) access log for `kubectl logs`, augmented with the
+# authenticated client name, request time and upstream status. Written to stdout when
+# fluentbit.accessLog.stdoutReadable is true; the JSON `main` format always goes to
+# Fluent Bit over syslog.
+log_format readable
+'$remote_addr {{ if .Values.authorization.enabled }}$jwt_payload_sub {{ end }}[$time_local] '
+'"$request" $status $body_bytes_sent '
+'"$http_referer" "$http_user_agent" '
+'rt=$request_time upstream_status=$upstream_status';
+{{- end }}

helm/config/metadata.lua

@@ -0,0 +1,148 @@
+-- Rendered through Helm `tpl` for the parameter list below, so no Go-template braces in Lua code.
+-- Parameters carrying credentials (the OPA check reads the client's JWT from a token arg), from
+-- fluentbit.maskQueryParams; an empty list makes mask() a no-op.
+local MASKED_PARAMS = { {{ range $i, $param := .Values.fluentbit.maskQueryParams }}{{ if $i }}, {{ end }}{{ $param | quote }}{{ end }} }
+local MASK = "[MASKED]"
+local MASKED_FIELDS = { "url.query", "url.full" }
+
+-- Mask in a bare query ("a=1&token=x"), a target ("/p?token=x") or a request line alike. Lua
+-- patterns have no alternation or word boundaries, so each name is matched with its `?`/`&`
+-- delimiter (else `token` also matches `access_token=`) and the leading "&" gives the first
+-- parameter one too. Stopping the value at `&` or whitespace keeps the trailing " HTTP/1.1".
+local function mask(value)
+    if type(value) ~= "string" then
+        return value
+    end
+
+    local masked, total = "&" .. value, 0
+    for _, param in ipairs(MASKED_PARAMS) do
+        -- %W escapes the name's non-alphanumerics into pattern literals.
+        local pattern = "([?&]" .. param:gsub("(%W)", "%%%1") .. "=)[^&%s]*"
+        local count
+        masked, count = masked:gsub(pattern, "%1" .. MASK)
+        total = total + count
+    end
+
+    if total == 0 then
+        return value
+    end
+    return masked:sub(2)
+end
+
+-- Its own filter rather than part of flatten_attributes below, so it can run ahead of the
+-- operator's Lua hook while `record["Attributes"]` is still reachable.
+function mask_query_params(tag, timestamp, group, metadata, record)
+    local modified = false
+
+    local attributes = record["Attributes"]
+    if type(attributes) == "table" then
+        for _, field in ipairs(MASKED_FIELDS) do
+            local masked = mask(attributes[field])
+            if masked ~= attributes[field] then
+                attributes[field] = masked
+                modified = true
+            end
+        end
+    end
+
+    -- `Body` is nginx's $request: the whole request line, query included.
+    local body = mask(record["Body"])
+    if body ~= record["Body"] then
+        record["Body"] = body
+        modified = true
+    end
+
+    -- Code 0 keeps the record as it arrived, discarding the edits above.
+    if modified then
+        return 1, timestamp, metadata, record
+    else
+        return 0, timestamp, metadata, record
+    end
+end
+
+-- Lift the decoded access-log `Attributes` into the record's OTLP metadata, which the
+-- opentelemetry output maps to the log record's attributes. No filter can write metadata, so
+-- this has to be Lua: returning 4 values marks the 3rd as metadata. `Resource` is dropped —
+-- resource attributes are set by the output's processors instead.
+function flatten_attributes(tag, timestamp, group, metadata, record)
+    local metadata = {}
+    local modified = false
+
+    if type(record["Attributes"]) == "table" then
+        for k, v in pairs(record["Attributes"]) do
+            -- nginx writes "" for variables with no value on this request; skip those.
+            if v ~= "" then
+                metadata[k] = v
+            end
+        end
+        record["Attributes"] = nil
+        modified = true
+    end
+
+    if record["Resource"] ~= nil then
+        record["Resource"] = nil
+        modified = true
+    end
+
+    if modified then
+        return 1, timestamp, metadata, record
+    else
+        return 0, timestamp, metadata, record
+    end
+end
+
+-- nginx severity -> OTel SeverityNumber (INFO 9-12, WARN 13-16, ERROR 17-20, FATAL 21-24).
+local SEVERITY_NUMBERS = {
+    debug = 5, info = 9, notice = 10, warn = 13,
+    error = 17, crit = 18, alert = 21, emerg = 22,
+}
+
+-- Rewrite the error-log record the nginx_error parser produced into the same envelope the access
+-- log arrives in: semantic-convention attributes in metadata, the message as `Body`, the nginx
+-- level as OTel severity. Without this its fields keep the parser's own names (client, pid,
+-- request, …) and land in Loki as attributes no dashboard shares with the access log.
+function envelope_error(tag, timestamp, group, metadata, record)
+    local attributes = {}
+
+    if record["client"] ~= nil and record["client"] ~= "" then
+        attributes["client.address"] = record["client"]
+    end
+    if record["server"] ~= nil and record["server"] ~= "" then
+        attributes["server.address"] = record["server"]
+    end
+    if record["pid"] ~= nil then
+        attributes["process.pid"] = tonumber(record["pid"]) or record["pid"]
+    end
+
+    -- The request line ("GET /path?q=1 HTTP/1.1") only appears on messages tied to a request.
+    local request = record["request"]
+    if request ~= nil and request ~= "" then
+        local method, target, version = string.match(request, "^(%S+) (%S+) HTTP/(%S+)$")
+        if method ~= nil then
+            attributes["http.request.method"] = method
+            attributes["network.protocol.name"] = "http"
+            attributes["network.protocol.version"] = version
+            local path, query = string.match(target, "^([^?]*)%??(.*)$")
+            attributes["url.path"] = path
+            if query ~= "" then
+                -- The error line quotes the same request line as the access log, token and all.
+                attributes["url.query"] = mask(query)
+            end
+        else
+            -- Not a well-formed request line (a malformed request is itself a common error).
+            attributes["mapcolonies.request"] = request
+        end
+    end
+
+    local envelope = {
+        Body = record["message"],
+        InstrumentationScope = "error.log",
+    }
+    local level = record["level"]
+    if level ~= nil then
+        envelope["SeverityText"] = level
+        envelope["SeverityNumber"] = SEVERITY_NUMBERS[string.lower(level)] or 9
+    end
+
+    return 1, timestamp, attributes, envelope
+end