Publish aggregate Hubble traffic

kubernetes/apps/showcase/snapshot-publisher/README.md

@@ -6,6 +6,8 @@ Push, not pull: CGNAT means nothing outside can reach in, and home infrastructur
 
 Beside the snapshot, every run maintains `heartbeat.json`: a rolling window of the last 336 run timestamps (14 days at the hourly cadence). It exists because the portfolio page needs to read the ledger, and walking the commit log through the GitHub REST API rate-limits anonymous browsers at 60 requests an hour per address; a raw file rides the CDN with no such ceiling. The file carries nothing read from the cluster, only the job's own clock, so it sits outside the allowlist and the schema gate by construction. The commit history stays the audit trail: every beat in the file has a matching commit, and anyone skeptical can diff the two.
 
+The optional `traffic` block carries six hours of aggregate Hubble flow and drop rates. The publisher reads the same Prometheus series used by the Devata overview dashboard and emits 15-minute samples. Individual flows, workload identities, addresses, ports, and metric labels remain inside the cluster. If Prometheus is unavailable, the publisher omits this block and still publishes the core snapshot.
+
 ## The safety model
 
 1. **Read-only on the cluster side.** The ClusterRole in `rbac.yaml` is the complete list of what the publisher may see: get and list on nodes, namespaces, services, persistentvolumeclaims, pods, the apps workload kinds, and Argo Applications. No secrets, no writes, no watch.

kubernetes/apps/showcase/snapshot-publisher/configmap-schema.yaml

@@ -166,6 +166,32 @@ data:
             "pods": { "type": "integer", "minimum": 0 }
           }
         },
+        "traffic": {
+          "type": "object",
+          "additionalProperties": false,
+          "description": "Added in 1.4.0 (optional, additive). Six hours of aggregate Hubble rates read from Prometheus. No individual flows, identities, addresses, ports, or labels are published.",
+          "required": ["source", "windowHours", "sampleIntervalMinutes", "samples"],
+          "properties": {
+            "source": { "const": "hubble" },
+            "windowHours": { "type": "integer", "const": 6 },
+            "sampleIntervalMinutes": { "type": "integer", "const": 15 },
+            "samples": {
+              "type": "array",
+              "minItems": 2,
+              "maxItems": 25,
+              "items": {
+                "type": "object",
+                "additionalProperties": false,
+                "required": ["timestamp", "flowsPerSecond", "dropsPerSecond"],
+                "properties": {
+                  "timestamp": { "type": "string", "format": "date-time" },
+                  "flowsPerSecond": { "type": "number", "minimum": 0 },
+                  "dropsPerSecond": { "type": "number", "minimum": 0 }
+                }
+              }
+            }
+          }
+        },
         "logos": {
           "type": "object",
           "description": "Added in 1.2.0 (optional, additive). Marks for the tools this document names, keyed by the exact string emitted elsewhere (e.g. 'MetalLB (L2)'), so the consumer renders logos[key] as a plain lookup. Resolved by the publisher from Artifact Hub at render time, best-effort: a tool without a confident match is simply absent and the consumer falls back to its own rendering. The url pattern pins every entry to Artifact Hub's image endpoint; the document cannot send a browser anywhere else.",

kubernetes/apps/showcase/snapshot-publisher/configmap-scripts.yaml

@@ -23,6 +23,40 @@ data:
 
     NOW="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
 
+    # Hubble aggregate traffic, read through the Prometheus series already
+    # scraped for the Devata overview dashboard. Only rates and timestamps are
+    # exported. Individual flows, identities, addresses, ports, and labels
+    # never enter the public document. Telemetry is optional so a Prometheus
+    # outage cannot block the core snapshot.
+    PROMETHEUS_URL="http://kps-kube-prometheus-stack-prometheus.monitoring.svc:9090/api/v1/query_range"
+    END="$(date -u +%s)"
+    START="$((END - 21600))"
+    FLOW_QUERY="$(jq -rn --arg q 'sum(rate(hubble_flows_processed_total[5m]))' '$q | @uri')"
+    DROP_QUERY="$(jq -rn --arg q 'sum(rate(hubble_drop_total[5m]))' '$q | @uri')"
+    if curl -fsS -m 15 "${PROMETHEUS_URL}?query=${FLOW_QUERY}&start=${START}&end=${END}&step=900" > /tmp/hubble-flow.json \
+      && curl -fsS -m 15 "${PROMETHEUS_URL}?query=${DROP_QUERY}&start=${START}&end=${END}&step=900" > /tmp/hubble-drop.json; then
+      jq -n --slurpfile flow /tmp/hubble-flow.json --slurpfile drops /tmp/hubble-drop.json '
+        ($drops[0].data.result[0].values // []
+          | map({key: (.[0] | tostring), value: (.[1] | tonumber)})
+          | from_entries) as $dropByTime
+        | {
+            source: "hubble",
+            windowHours: 6,
+            sampleIntervalMinutes: 15,
+            samples: [
+              $flow[0].data.result[0].values[]?
+              | {
+                  timestamp: (.[0] | todateiso8601),
+                  flowsPerSecond: (((.[1] | tonumber) * 1000) | round / 1000),
+                  dropsPerSecond: (((($dropByTime[(.[0] | tostring)] // 0) * 1000) | round) / 1000)
+                }
+            ]
+          }
+      ' > /tmp/traffic.json || printf '%s\n' '{"samples":[]}' > /tmp/traffic.json
+    else
+      printf '%s\n' '{"samples":[]}' > /tmp/traffic.json
+    fi
+
     # Build the public document by selecting safe fields into a new object.
     # This is an allowlist: a field that is not named here does not exist in the
     # output, so a new field Kubernetes adds next year cannot leak by default.
@@ -38,7 +72,8 @@ data:
       --slurpfile deployments /tmp/deployments.json \
       --slurpfile daemonsets /tmp/daemonsets.json \
       --slurpfile statefulsets /tmp/statefulsets.json \
-      --slurpfile pods /tmp/pods.json '
+      --slurpfile pods /tmp/pods.json \
+      --slurpfile traffic /tmp/traffic.json '
       def iscp($n): $n.metadata.labels | has("node-role.kubernetes.io/control-plane");
       def ready($n): ([$n.status.conditions[] | select(.type == "Ready") | .status] | first) == "True";
       def gib($q): ($q | rtrimstr("Ki") | tonumber / 1048576);
@@ -53,11 +88,11 @@ data:
       ($svc[0].items) as $S |
       ($pvc[0].items) as $P |
       {
-        schemaVersion: "1.3.0",
+        schemaVersion: "1.4.0",
         generatedAt: $now,
         generator: {
           name: "devata-snapshot-publisher",
-          version: "0.3.0",
+          version: "0.4.0",
           method: "read-only Kubernetes API via a scoped ServiceAccount"
         },
         freshness: { clusterPowered: true, maxAgeHours: 24 },
@@ -133,7 +168,8 @@ data:
           statefulsets: ($statefulsets[0].items | length),
           pods: ($pods[0].items | length)
         }
-      }' > /tmp/snapshot.json
+      }
+      + (if ($traffic[0].samples | length) > 1 then {traffic: $traffic[0]} else {} end)' > /tmp/snapshot.json
 
     # Clone before the logo pass: the previous published snapshot doubles as the
     # fallback cache when Artifact Hub is unreachable. The push credential is a