feat: add fluent-bit support for Loki log ingestion

CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.17.0] - 2026-04-10
+
+### Added
+
+- fluent-bit support: Loki log lines wrapped in a `{"log": "..."}` JSON envelope are now automatically detected and unwrapped, so flows ingested via fluent-bit work out of the box alongside promtail.
+- Default `--loki-query` changed from `{container="cilium-agent"}` (promtail) to `{app_kubernetes_io_name="cilium-agent"}` (fluent-bit). Promtail users can override with `--loki-query '{container="cilium-agent"}'`.
+
+### Fixed
+
+- Loki server-side line filters now use format-agnostic regex patterns (`verdict.{0,5}:.{0,5}AUDIT`) instead of literal JSON patterns (`"verdict":"AUDIT"`), so they match both promtail (plain quotes) and fluent-bit (escaped quotes) log formats.
+
 ## [0.13.0] - 2026-04-02
 
 ### Added

README.md

@@ -88,10 +88,10 @@ hubble-audit2policy flows.json --report-only
 
 Query a Grafana Loki instance directly -- ideal when Hubble flows are already being shipped to Loki via fluentd, promtail, or another collector.
 
-The default LogQL selector is `{container="cilium-agent"}`, which matches the standard Cilium agent container label. Override it with `--loki-query` if your setup uses different labels.
+The default LogQL selector is `{app_kubernetes_io_name="cilium-agent"}`, which matches the standard fluent-bit Kubernetes label for the Cilium agent. For promtail setups use `--loki-query '{container="cilium-agent"}'`. Override with `--loki-query` if your setup uses different labels.
 
 ```bash
-# All flows from the last hour (uses default query {container="cilium-agent"}):
+# All flows from the last hour (uses default query {app_kubernetes_io_name="cilium-agent"}):
 hubble-audit2policy --from loki --loki-url http://loki:3100 --dry-run
 
 # Scoped to a namespace with a custom time window:
@@ -239,7 +239,7 @@ hubble-audit2policy [-h] [-o OUTPUT_DIR] [-n NAMESPACE]
 | `--no-enrich` | Skip live cluster enrichment via Cilium endpoints |
 | `--from {file,loki}` | Flow source backend (default: `file`) |
 | `--loki-url URL` | Loki base URL, e.g. `http://loki:3100` |
-| `--loki-query LOGQL` | LogQL stream selector (default: `{container="cilium-agent"}`) |
+| `--loki-query LOGQL` | LogQL stream selector (default: `{app_kubernetes_io_name="cilium-agent"}`; for promtail: `{container="cilium-agent"}`) |
 | `--since DURATION` | How far back to query, e.g. `30m`, `2h`, `1d` (default: `1h`) |
 | `--until DURATION` | End of query window as duration before now (default: `0s` = now) |
 | `--loki-limit N` | Max entries per Loki request batch (default: `5000`) |

hubble_audit2policy.py

@@ -8,7 +8,7 @@
 
 from __future__ import annotations
 
-__version__ = "0.16.0"
+__version__ = "0.17.0"
 __author__ = "noexecstack"
 __license__ = "Apache-2.0"
 
@@ -453,26 +453,32 @@ def _loki_enrich_query(
     filters lines server-side.
 
     Multiple namespaces/verdicts are OR-joined into a single regex filter.
+
+    The regex patterns use ``.{0,5}`` gaps between JSON keys, colons, and
+    values so they match both promtail-style lines (where the log text
+    contains ``"verdict":"AUDIT"``) and fluent-bit-style lines (where the
+    flow JSON is escaped inside a wrapper object, producing
+    ``\"verdict\":\"AUDIT\"``).
     """
     # Verdict filter -- push as specific a match as possible so Loki
     # discards non-matching lines before sending them to us.
     verdict_list = sorted(set(verdicts))
     if len(verdict_list) == 1:
-        query += r' |= "\"verdict\":\"' + verdict_list[0] + r'\""'
+        query += ' |~ "verdict.{0,5}:.{0,5}' + re.escape(verdict_list[0]) + '"'
     elif len(verdict_list) > 1:
         vpat = "|".join(re.escape(v) for v in verdict_list)
-        query += r' |~ "\"verdict\":\"(' + vpat + r')\""'
+        query += ' |~ "verdict.{0,5}:.{0,5}(' + vpat + ')"'
     else:
         # No specific verdict -- still filter to flow records only.
-        query += r' |= "\"verdict\":"'
+        query += ' |~ "verdict.{0,5}:"'
 
     ns_list = sorted(set(namespaces))
     if not ns_list:
         return query
     if len(ns_list) == 1:
-        return query + r' |= "\"namespace\":\"' + ns_list[0] + r'\""'
+        return query + ' |~ "namespace.{0,5}:.{0,5}' + re.escape(ns_list[0]) + '"'
     pattern = "|".join(re.escape(ns) for ns in ns_list)
-    return query + r' |~ "\"namespace\":\"(' + pattern + r')\""'
+    return query + ' |~ "namespace.{0,5}:.{0,5}(' + pattern + ')"'
 
 
 class _LokiProgress:
@@ -558,6 +564,40 @@ class _ChunkStats:
     failed: bool = False
 
 
+_FLUENT_BIT_LOG_PREFIX = re.compile(
+    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z\s+\S+\s+\S+\s+"
+)
+
+
+def _extract_flow_from_loki_line(parsed: Any) -> dict[str, Any] | None:
+    """Return a Hubble flow dict from a parsed Loki log line.
+
+    Handles two formats:
+
+    * **Raw / promtail**: the log line is the Hubble JSON directly, e.g.
+      ``{"flow": {...}, ...}`` or ``{"time": ..., "verdict": ...}``.
+    * **fluent-bit envelope**: the log line is a JSON object with a ``"log"``
+      key whose value is the original container stdout, e.g.
+      ``{"log": "2026-... stdout F {\"flow\":{...}}", "kubernetes": {...}}``.
+      The inner text is stripped of the timestamp/stream prefix and then
+      parsed as JSON.
+
+    Returns ``None`` if no flow can be extracted.
+    """
+    if not isinstance(parsed, dict):
+        return None
+    # fluent-bit format: outer envelope with a "log" string field.
+    if "log" in parsed and isinstance(parsed["log"], str):
+        inner = _FLUENT_BIT_LOG_PREFIX.sub("", parsed["log"], count=1).strip()
+        try:
+            parsed = json.loads(inner)
+        except (json.JSONDecodeError, ValueError):
+            return None
+        if not isinstance(parsed, dict):
+            return None
+    return cast("dict[str, Any]", parsed)
+
+
 def _loki_fetch_chunk(
     base_url: str,
     query: str,
@@ -657,7 +697,10 @@ def _loki_fetch_chunk(
                     continue
                 ts_line_hashes.setdefault(ts, set()).add(line_h)
                 try:
-                    entries.append((ts, json.loads(line)))
+                    parsed = json.loads(line)
+                    flow = _extract_flow_from_loki_line(parsed)
+                    if flow is not None:
+                        entries.append((ts, flow))
                 except json.JSONDecodeError:
                     pass  # Non-JSON lines (e.g. cilium agent logs) are expected.
 
@@ -2400,9 +2443,10 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     loki_group.add_argument(
         "--loki-query",
-        default='{container="cilium-agent"}',
+        default='{app_kubernetes_io_name="cilium-agent"}',
         metavar="LOGQL",
-        help='LogQL stream selector (default: {container="cilium-agent"})',
+        help='LogQL stream selector (default: {app_kubernetes_io_name="cilium-agent"})'
+        '; for promtail use {container="cilium-agent"}',
     )
     loki_group.add_argument(
         "--since",

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hubble-audit2policy"
-version = "0.16.0"
+version = "0.17.0"
 description = "Generate least-privilege CiliumNetworkPolicy YAML from Hubble flow logs."
 readme = "README.md"
 license = "Apache-2.0"

tests/test_flow_parsing.py

@@ -690,6 +690,119 @@ def test_loki_flows_produce_policies(self) -> None:
             assert "db" in apps
 
 
+class TestExtractFlowFromLokiLine:
+    """Tests for _extract_flow_from_loki_line (promtail vs fluent-bit)."""
+
+    def test_raw_flow_dict(self) -> None:
+        """promtail-style: the parsed line IS the flow dict."""
+        flow = _make_flow(port=80)
+        assert h._extract_flow_from_loki_line(flow) is flow
+
+    def test_flow_envelope(self) -> None:
+        """promtail-style with a {flow: ...} wrapper."""
+        inner = _make_flow(port=443)
+        # parse_flows handles the {"flow": ...} unwrap; _extract_flow_from_loki_line
+        # just returns the outer dict.
+        envelope = {"flow": inner, "time": "2026-01-01T00:00:00Z"}
+        result = h._extract_flow_from_loki_line(envelope)
+        assert result is envelope
+
+    def test_fluentbit_envelope(self) -> None:
+        """fluent-bit-style: flow JSON is escaped inside a 'log' string."""
+        inner = _make_flow(port=8080)
+        log_value = "2026-04-10T09:08:39.418Z stdout F " + json.dumps(inner)
+        line = {"log": log_value, "kubernetes": {"namespace_name": "kube-system"}}
+        result = h._extract_flow_from_loki_line(line)
+        assert result is not None
+        assert result["l4"]["TCP"]["destination_port"] == 8080
+        assert result["verdict"] == "AUDIT"
+
+    def test_fluentbit_envelope_with_flow_wrapper(self) -> None:
+        """fluent-bit log value contains {flow: {<actual flow>}}."""
+        inner = _make_flow(port=9090)
+        wrapper = {"flow": inner, "time": "2026-04-10T09:00:00Z"}
+        log_value = "2026-04-10T09:00:00.000Z stdout F " + json.dumps(wrapper)
+        line = {"log": log_value, "kubernetes": {}}
+        result = h._extract_flow_from_loki_line(line)
+        assert result is not None
+        assert result["flow"]["l4"]["TCP"]["destination_port"] == 9090
+
+    def test_fluentbit_non_json_log(self) -> None:
+        """fluent-bit line whose 'log' is not parseable JSON returns None."""
+        line = {"log": "2026-04-10T09:00:00Z stdout F level=info msg=starting"}
+        assert h._extract_flow_from_loki_line(line) is None
+
+    def test_fluentbit_log_not_dict(self) -> None:
+        """fluent-bit line whose inner JSON is a list returns None."""
+        line = {"log": "2026-04-10T09:00:00Z stdout F [1, 2, 3]"}
+        assert h._extract_flow_from_loki_line(line) is None
+
+    def test_non_dict_input(self) -> None:
+        assert h._extract_flow_from_loki_line("not a dict") is None
+        assert h._extract_flow_from_loki_line(42) is None
+        assert h._extract_flow_from_loki_line(None) is None
+
+
+class TestFluentBitLokiEndToEnd:
+    """End-to-end: fluent-bit Loki response -> parse_flows -> policies."""
+
+    _EMPTY_RESPONSE = json.dumps({"status": "success", "data": {"result": []}}).encode()
+
+    @staticmethod
+    def _fluentbit_loki_response(flows: list[dict[str, Any]]) -> bytes:
+        """Build a Loki response where each line is a fluent-bit envelope."""
+        values = []
+        for i, flow in enumerate(flows, 1):
+            ts = str(i * 1_000_000_000)
+            log_text = f"2026-04-10T09:08:{i:02d}.000Z stdout F " + json.dumps(flow)
+            envelope = {"log": log_text, "kubernetes": {"namespace_name": "kube-system"}}
+            values.append([ts, json.dumps(envelope)])
+        body = {
+            "status": "success",
+            "data": {"result": [{"stream": {}, "values": values}]},
+        }
+        return json.dumps(body).encode()
+
+    def test_fluentbit_flows_produce_policies(self) -> None:
+        flows = [
+            _make_flow(src_app="web", dst_app="api", port=8080),
+            _make_flow(src_app="api", dst_app="db", port=5432),
+        ]
+        resp_bytes = self._fluentbit_loki_response(flows)
+
+        empty_resp = mock.MagicMock()
+        empty_resp.read.return_value = self._EMPTY_RESPONSE
+        empty_resp.__enter__ = mock.Mock(return_value=empty_resp)
+        empty_resp.__exit__ = mock.Mock(return_value=False)
+
+        data_resp = mock.MagicMock()
+        data_resp.read.return_value = resp_bytes
+        data_resp.__enter__ = mock.Mock(return_value=data_resp)
+        data_resp.__exit__ = mock.Mock(return_value=False)
+
+        with mock.patch("hubble_audit2policy.urllib.request.urlopen") as mock_open:
+            mock_open.side_effect = [data_resp, empty_resp]
+
+            loki_result = h._read_flows_loki(
+                "http://loki:3100",
+                '{app="hubble"}',
+                3600,
+                0,
+                threads=1,
+                chunk_seconds=7200,
+            )
+            assert len(loki_result.flows) == 2
+            policies, _, total, matched, _ = h.parse_flows(
+                "", LABEL_KEYS, set(), set(), flow_iter=iter(loki_result.flows)
+            )
+            assert total == 2
+            assert matched == 2
+            apps = {app for _, app in policies}
+            assert "web" in apps
+            assert "api" in apps
+            assert "db" in apps
+
+
 class TestLokiAuth:
     """Test Loki authentication modes in _read_flows_loki."""
 
@@ -809,56 +922,84 @@ def test_tls_ca_with_bearer(self) -> None:
 
     def test_enrich_query_no_namespaces(self) -> None:
         q = '{container="cilium-agent"}'
-        verdict = r' |= "\"verdict\":"'
+        verdict = ' |~ "verdict.{0,5}:"'
         assert h._loki_enrich_query(q, []) == q + verdict
         assert h._loki_enrich_query(q, set()) == q + verdict
 
     def test_enrich_query_single_namespace(self) -> None:
         q = '{container="cilium-agent"}'
         result = h._loki_enrich_query(q, ["argocd"])
         assert result == (
-            r'{container="cilium-agent"} |= "\"verdict\":"'
-            r' |= "\"namespace\":\"argocd\""'
+            '{container="cilium-agent"} |~ "verdict.{0,5}:" |~ "namespace.{0,5}:.{0,5}argocd"'
         )
 
     def test_enrich_query_multiple_namespaces(self) -> None:
         q = '{container="cilium-agent"}'
         result = h._loki_enrich_query(q, ["monitoring", "argocd"])
         assert result == (
-            r'{container="cilium-agent"} |= "\"verdict\":"'
-            r' |~ "\"namespace\":\"(argocd|monitoring)\""'
+            '{container="cilium-agent"} |~ "verdict.{0,5}:"'
+            ' |~ "namespace.{0,5}:.{0,5}(argocd|monitoring)"'
         )
 
     def test_enrich_query_deduplicates(self) -> None:
         q = '{container="cilium-agent"}'
         result = h._loki_enrich_query(q, ["argocd", "argocd"])
         assert result == (
-            r'{container="cilium-agent"} |= "\"verdict\":"'
-            r' |= "\"namespace\":\"argocd\""'
+            '{container="cilium-agent"} |~ "verdict.{0,5}:" |~ "namespace.{0,5}:.{0,5}argocd"'
         )
 
     def test_enrich_query_single_verdict(self) -> None:
         q = '{container="cilium-agent"}'
         result = h._loki_enrich_query(q, [], verdicts=["AUDIT"])
-        assert result == (r'{container="cilium-agent"} |= "\"verdict\":\"AUDIT\""')
+        assert result == '{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}AUDIT"'
 
     def test_enrich_query_multiple_verdicts(self) -> None:
         q = '{container="cilium-agent"}'
         result = h._loki_enrich_query(q, [], verdicts=["DROPPED", "AUDIT"])
-        assert result == (r'{container="cilium-agent"} |~ "\"verdict\":\"(AUDIT|DROPPED)\""')
+        assert result == ('{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}(AUDIT|DROPPED)"')
 
     def test_enrich_query_verdict_deduplicates(self) -> None:
         q = '{container="cilium-agent"}'
         result = h._loki_enrich_query(q, [], verdicts=["AUDIT", "AUDIT"])
-        assert result == (r'{container="cilium-agent"} |= "\"verdict\":\"AUDIT\""')
+        assert result == '{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}AUDIT"'
 
     def test_enrich_query_verdict_and_namespace(self) -> None:
         q = '{container="cilium-agent"}'
         result = h._loki_enrich_query(q, ["argocd"], verdicts=["AUDIT"])
         assert result == (
-            r'{container="cilium-agent"} |= "\"verdict\":\"AUDIT\""'
-            r' |= "\"namespace\":\"argocd\""'
+            '{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}AUDIT"'
+            ' |~ "namespace.{0,5}:.{0,5}argocd"'
+        )
+
+    def test_enrich_query_filters_match_promtail_format(self) -> None:
+        """Verify generated regex matches promtail-style log text."""
+        import re
+
+        q = '{app="cilium-agent"}'
+        result = h._loki_enrich_query(q, ["loki"], verdicts=["AUDIT"])
+        # Extract the two regex patterns from the LogQL filters.
+        patterns = re.findall(r'\|~ "([^"]+)"', result)
+        assert len(patterns) == 2
+        # promtail text has plain JSON: "verdict":"AUDIT"  "namespace":"loki"
+        promtail_line = '{"verdict":"AUDIT","source":{"namespace":"loki"}}'
+        for pat in patterns:
+            assert re.search(pat, promtail_line), f"pattern {pat!r} should match promtail"
+
+    def test_enrich_query_filters_match_fluentbit_format(self) -> None:
+        """Verify generated regex matches fluent-bit-style log text."""
+        import re
+
+        q = '{app="cilium-agent"}'
+        result = h._loki_enrich_query(q, ["loki"], verdicts=["AUDIT"])
+        patterns = re.findall(r'\|~ "([^"]+)"', result)
+        assert len(patterns) == 2
+        # fluent-bit text has escaped JSON inside a "log" wrapper:
+        fluentbit_line = (
+            r'{"log":"2026-04-10T09:08:39Z stdout F '
+            r'{\"verdict\":\"AUDIT\",\"source\":{\"namespace\":\"loki\"}}"}'
         )
+        for pat in patterns:
+            assert re.search(pat, fluentbit_line), f"pattern {pat!r} should match fluent-bit"
 
     def test_build_loki_ssl_context_none(self) -> None:
         assert h._build_loki_ssl_context(None) is None