From 36050b100f0063cd1cf7fff5bc32b78e801d3bc8 Mon Sep 17 00:00:00 2001 From: codex Date: Wed, 21 Oct 2026 08:15:18 +0200 Subject: [PATCH] CUST-WP-0071-T05: prepare weekly allocation assessment CronJob (suspended, unregistered) New platform addon custodian-allocation-review: namespace custodian-assessment, read-only ClusterRole (get/list nodes, pods, PVCs, PVs, storageclasses, metrics.k8s.io nodes/pods), retained local-path PVC (Prune=false,Delete=false), CronJob 0 8 * * 1 Europe/Berlin, Forbid, suspend: true. Code/data vendored from the-custodian, railiance-cluster and resource-control with a sha256 lock and an offline guard test. Draft Application only in drafts/; nothing is synced. Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 8340@bnt-lap001 Assistant-Session: 2e192b3d-b3e8-4403-94d4-d55a2c8b476c --- .../custodian-allocation-review/README.md | 80 +++++ .../custodian-allocation-review/cronjob.yaml | 80 +++++ .../kustomization.yaml | 27 ++ .../namespace.yaml | 9 + .../custodian-allocation-review/pvc.yaml | 21 ++ .../custodian-allocation-review/rbac.yaml | 49 +++ .../src/namespace-ownership.json | 205 +++++++++++ .../src/observe_cluster_resources.py | 334 ++++++++++++++++++ .../src/reconcile_railiance_allocation.py | 153 ++++++++ .../src/reef-railiance-k3s.json | 63 ++++ .../src/weekly-allocation-queries.json | 16 + .../src/weekly_allocation_incluster.py | 272 ++++++++++++++ .../src/weekly_allocation_report.py | 321 +++++++++++++++++ .../vendor.lock.json | 55 +++ ...stodian-allocation-review.application.yaml | 29 ++ scripts/vendor_custodian_allocation_review.py | 92 +++++ tests/test_custodian_allocation_review.py | 84 +++++ 17 files changed, 1890 insertions(+) create mode 100644 argocd/platform-addons/custodian-allocation-review/README.md create mode 100644 argocd/platform-addons/custodian-allocation-review/cronjob.yaml create mode 100644 argocd/platform-addons/custodian-allocation-review/kustomization.yaml create mode 100644 argocd/platform-addons/custodian-allocation-review/namespace.yaml create mode 100644 argocd/platform-addons/custodian-allocation-review/pvc.yaml create mode 100644 argocd/platform-addons/custodian-allocation-review/rbac.yaml create mode 100644 argocd/platform-addons/custodian-allocation-review/src/namespace-ownership.json create mode 100644 argocd/platform-addons/custodian-allocation-review/src/observe_cluster_resources.py create mode 100644 argocd/platform-addons/custodian-allocation-review/src/reconcile_railiance_allocation.py create mode 100644 argocd/platform-addons/custodian-allocation-review/src/reef-railiance-k3s.json create mode 100644 argocd/platform-addons/custodian-allocation-review/src/weekly-allocation-queries.json create mode 100644 argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_incluster.py create mode 100644 argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_report.py create mode 100644 argocd/platform-addons/custodian-allocation-review/vendor.lock.json create mode 100644 argocd/railiance01/drafts/custodian-allocation-review.application.yaml create mode 100644 scripts/vendor_custodian_allocation_review.py create mode 100644 tests/test_custodian_allocation_review.py diff --git a/argocd/platform-addons/custodian-allocation-review/README.md b/argocd/platform-addons/custodian-allocation-review/README.md new file mode 100644 index 0000000..daa118b --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/README.md @@ -0,0 +1,80 @@ +# custodian-allocation-review (CUST-WP-0071-T05) — prepared, not deployed + +Weekly workload-demand and allocation assessment for railiance01, selected by +the founder on 2026-10-21 (the-custodian +`docs/decisions/2026-10-21-cust-wp-0071-delivery-interview.md`). Not +referenced by any Application. The draft Application is in +`argocd/railiance01/drafts/`. The CronJob is `suspend: true`. + +| Item | Value | +|---|---| +| Executor owner | railiance-platform (platform-addons, ArgoCD `railiance-platform-addons` project) | +| Assessment/report owner | the-custodian (CUST-WP-0071-T05); code in `the-custodian/scripts/weekly_allocation_incluster.py` | +| Schedule | `0 8 * * 1`, `timeZone: Europe/Berlin` (Monday 08:00), `concurrencyPolicy: Forbid`, `startingDeadlineSeconds: 21600` | +| Telemetry window | 7d ending Monday 06:00 UTC of the run's ISO week (DST-independent, so consecutive weeks are comparable) | +| Image | `python:3.12-slim` pinned by digest (same as eso-token-renewer); code and data come from a ConfigMap | +| Identity | SA `custodian-assessment/custodian-allocation-review`, ClusterRole `custodian-allocation-review-reader` | +| RBAC | get/list on core `nodes`, `pods`, `persistentvolumeclaims`, `persistentvolumes`; `storage.k8s.io/storageclasses`; `metrics.k8s.io/nodes,pods`. No watch, no writes, no secrets, configmaps, logs, exec or proxy | +| Prometheus | `http://telemetry-prometheus.telemetry.svc:9090`, plain HTTP, no credential | +| Storage | PVC `custodian-allocation-review-data` (local-path, 1Gi, ArgoCD `Prune=false,Delete=false`): `reports/-W.json`, `runs///` inputs, append-only `receipts.jsonl` | +| Resources | requests 20m/96Mi, limits 500m/256Mi, `activeDeadlineSeconds: 900`, `backoffLimit: 2` | + +## Behaviour + +- Writes only to the PVC. It never resizes anything, creates tasks or sends + messages (`delivery: not emitted`). +- Idempotency: an existing report for the week means the run does nothing and + succeeds. A failed attempt writes a `failed` receipt and no report, so the Job + retry or a manual Job collects again. +- Missed weeks: each retained receipt lists `missed_weeks` since the previous + report. Comparison across a gap stays unknown. +- Workload rows remain `investigate`. The queries are namespace-level because + kube-state-metrics does not export `kube_pod_labels`. + +## Vendored sources + +`src/` is generated by `scripts/vendor_custodian_allocation_review.py` from +clean owner commits, and `vendor.lock.json` records each commit and sha256. +`--check` verifies them, and `tests/test_custodian_allocation_review.py` +enforces the check offline. To change the job code, commit in the owner repo +and re-vendor. Never edit `src/` by hand. + +## Activation (founder go-ahead required; after CUST-WP-0071-T04 completes) + +1. Review this directory and the vendored sources. Re-vendor if the owner + repositories have moved on. +2. rapp-telemetry: apply its prepared + `manifests/custodian-assessment-prometheus-ingress.pending.yaml`. Without + it, `telemetry-isolation` blocks port 9090 from other namespaces, and every + query is recorded as a transport error. +3. Add `{group: "", kind: PersistentVolumeClaim}` to the + `namespaceResourceWhitelist` in + `argocd/railiance01/bootstrap/02-railiance-platform-addons-project.yaml` + and apply the AppProject. +4. Commit, then set `targetRevision` in the draft Application to that commit. + Move the draft to `argocd/railiance01/applications/`, run + `make argocd-pin-drift`, and hand-sync `railiance-apps-root` and then + `custodian-allocation-review`. +5. Acceptance run, still suspended: + `kubectl -n custodian-assessment create job --from=cronjob/custodian-allocation-review accept-1`. + Check that `receipts.jsonl` shows `retained` and `failed_queries: []`. Run + it on a Monday: it claims that ISO week's report slot, and coverage falls + later in the week as the 7-day retention expires. +6. Unsuspend through a reviewed commit (`suspend: false`) and a new + `targetRevision`, not a live patch. + +## Rollback + +- Stop: commit `suspend: true` and sync. In an emergency, run + `kubectl -n custodian-assessment patch cronjob custodian-allocation-review -p '{"spec":{"suspend":true}}'`, + then commit the same change. +- Remove: delete the Application (it has no finalizer, so the resources stay), + then delete the CronJob, RBAC and ConfigMap. Keep the PVC unless the history + is explicitly abandoned, because local-path deletes the data with the claim. + +## Still open before T05 can be accepted + +These items are not provided here: delivery to and acknowledgment by the +weekly review owner, upserting recommendations into owner work records, +alerting on a `failed` receipt or missing weekly receipt, and how owner reviews +reach `/data/reviews.json`. diff --git a/argocd/platform-addons/custodian-allocation-review/cronjob.yaml b/argocd/platform-addons/custodian-allocation-review/cronjob.yaml new file mode 100644 index 0000000..29b8081 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/cronjob.yaml @@ -0,0 +1,80 @@ +# CUST-WP-0071-T05: weekly workload-demand and allocation assessment. +# Founder decision 2026-10-21 (the-custodian +# docs/decisions/2026-10-21-cust-wp-0071-delivery-interview.md): in-cluster +# CronJob, Monday 08:00 Europe/Berlin, scoped read-only ServiceAccount, +# persisted weekly aggregates. suspend: true until review and T04 completion; +# unsuspend only by a reviewed commit (see README.md). +apiVersion: batch/v1 +kind: CronJob +metadata: + name: custodian-allocation-review + namespace: custodian-assessment + labels: + app.kubernetes.io/name: custodian-allocation-review + app.kubernetes.io/part-of: railiance-platform +spec: + schedule: '0 8 * * 1' + timeZone: Europe/Berlin + suspend: true + concurrencyPolicy: Forbid + # Late starts lose coverage at the 7-day retention edge; after 6h the week + # is left to the next run, whose receipt lists it under missed_weeks. + startingDeadlineSeconds: 21600 + successfulJobsHistoryLimit: 4 + failedJobsHistoryLimit: 4 + jobTemplate: + spec: + activeDeadlineSeconds: 900 + backoffLimit: 2 + ttlSecondsAfterFinished: 1209600 + template: + metadata: + labels: + app.kubernetes.io/name: custodian-allocation-review + spec: + serviceAccountName: custodian-allocation-review + # Token needed for the read-only API calls; the SA itself does not + # automount, so only this pod template receives it. + automountServiceAccountToken: true + restartPolicy: Never + securityContext: + runAsNonRoot: true + runAsUser: 65534 + runAsGroup: 65534 + fsGroup: 65534 + seccompProfile: + type: RuntimeDefault + containers: + - name: assess + image: python:3.12-slim@sha256:d764629ce0ddd8c71fd371e9901efb324a95789d2315a47db7e4d27e78f1b0e9 + command: [python3, /app/weekly_allocation_incluster.py] + env: + - {name: PYTHONDONTWRITEBYTECODE, value: '1'} + - {name: ASSESSMENT_DATA_DIR, value: /data} + - {name: ASSESSMENT_SOURCE_DIR, value: /app} + - {name: ASSESSMENT_OBSERVER, value: /app/observe_cluster_resources.py} + - {name: ASSESSMENT_QUERIES, value: /app/weekly-allocation-queries.json} + - {name: ASSESSMENT_OWNERSHIP, value: /app/namespace-ownership.json} + - {name: ASSESSMENT_INVENTORY, value: /app/reef-railiance-k3s.json} + - {name: ASSESSMENT_REVIEWS, value: /data/reviews.json} + - {name: ASSESSMENT_EXECUTOR, value: railiance-platform/custodian-assessment/cronjob/custodian-allocation-review} + - {name: PROMETHEUS_URL, value: 'http://telemetry-prometheus.telemetry.svc:9090'} + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: [ALL] + resources: + requests: {cpu: 20m, memory: 96Mi} + limits: {cpu: 500m, memory: 256Mi} + volumeMounts: + - {name: source, mountPath: /app, readOnly: true} + - {name: data, mountPath: /data} + - {name: tmp, mountPath: /tmp} + volumes: + - name: source + configMap: {name: custodian-allocation-review-source, defaultMode: 0444} + - name: data + persistentVolumeClaim: {claimName: custodian-allocation-review-data} + - name: tmp + emptyDir: {sizeLimit: 16Mi} diff --git a/argocd/platform-addons/custodian-allocation-review/kustomization.yaml b/argocd/platform-addons/custodian-allocation-review/kustomization.yaml new file mode 100644 index 0000000..3f44842 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/kustomization.yaml @@ -0,0 +1,27 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +# CUST-WP-0071-T05. NOT referenced by any Application yet: the draft lives in +# argocd/railiance01/drafts/custodian-allocation-review.application.yaml. +# src/ is vendored from owner repos (the-custodian, railiance-cluster, +# resource-control); vendor.lock.json pins the source commits and sha256. +# Refresh with tools/vendor_custodian_allocation_review.py, never by hand. + +namespace: custodian-assessment + +resources: + - namespace.yaml + - rbac.yaml + - pvc.yaml + - cronjob.yaml + +configMapGenerator: + - name: custodian-allocation-review-source + files: + - src/weekly_allocation_incluster.py + - src/weekly_allocation_report.py + - src/reconcile_railiance_allocation.py + - src/observe_cluster_resources.py + - src/weekly-allocation-queries.json + - src/namespace-ownership.json + - src/reef-railiance-k3s.json diff --git a/argocd/platform-addons/custodian-allocation-review/namespace.yaml b/argocd/platform-addons/custodian-allocation-review/namespace.yaml new file mode 100644 index 0000000..18ed7cd --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/namespace.yaml @@ -0,0 +1,9 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: custodian-assessment + labels: + app.kubernetes.io/part-of: railiance-gitops + railiance-platform/component: custodian-allocation-review + pod-security.kubernetes.io/enforce: restricted + pod-security.kubernetes.io/enforce-version: latest diff --git a/argocd/platform-addons/custodian-allocation-review/pvc.yaml b/argocd/platform-addons/custodian-allocation-review/pvc.yaml new file mode 100644 index 0000000..da2e117 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/pvc.yaml @@ -0,0 +1,21 @@ +# Weekly reports, per-attempt inputs and receipts.jsonl outlive the 7-day +# Prometheus retention here. local-path reclaimPolicy is Delete, so ArgoCD must +# never prune or delete this claim: removing it destroys the history. +# Requires PersistentVolumeClaim in the railiance-platform-addons AppProject +# namespaceResourceWhitelist before the Application can sync (activation step). +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: custodian-allocation-review-data + namespace: custodian-assessment + labels: + app.kubernetes.io/name: custodian-allocation-review + app.kubernetes.io/part-of: railiance-platform + annotations: + argocd.argoproj.io/sync-options: Prune=false,Delete=false +spec: + accessModes: [ReadWriteOnce] + storageClassName: local-path + resources: + requests: + storage: 1Gi diff --git a/argocd/platform-addons/custodian-allocation-review/rbac.yaml b/argocd/platform-addons/custodian-allocation-review/rbac.yaml new file mode 100644 index 0000000..161062e --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/rbac.yaml @@ -0,0 +1,49 @@ +# CUST-WP-0071-T05 (founder decision 2026-10-21): read-only, scoped to exactly +# the reads made by weekly_allocation_incluster.py (see its API_PATHS and +# METRICS_PATHS). Verbs get/list only; no watch, no write verb, no wildcard, +# no secrets, configmaps, logs, exec, proxy or token subresources. Prometheus +# access is network-level (rapp-telemetry NetworkPolicy), not RBAC. +# Enforced offline by tests/test_custodian_allocation_review.py. +apiVersion: v1 +kind: ServiceAccount +metadata: + name: custodian-allocation-review + namespace: custodian-assessment + labels: + app.kubernetes.io/name: custodian-allocation-review + app.kubernetes.io/part-of: railiance-platform +automountServiceAccountToken: false +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: custodian-allocation-review-reader + labels: + app.kubernetes.io/name: custodian-allocation-review + app.kubernetes.io/part-of: railiance-platform +rules: + - apiGroups: [""] + resources: [nodes, pods, persistentvolumeclaims, persistentvolumes] + verbs: [get, list] + - apiGroups: [storage.k8s.io] + resources: [storageclasses] + verbs: [get, list] + - apiGroups: [metrics.k8s.io] + resources: [nodes, pods] + verbs: [get, list] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: custodian-allocation-review-reader + labels: + app.kubernetes.io/name: custodian-allocation-review + app.kubernetes.io/part-of: railiance-platform +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: custodian-allocation-review-reader +subjects: + - kind: ServiceAccount + name: custodian-allocation-review + namespace: custodian-assessment diff --git a/argocd/platform-addons/custodian-allocation-review/src/namespace-ownership.json b/argocd/platform-addons/custodian-allocation-review/src/namespace-ownership.json new file mode 100644 index 0000000..a1b4967 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/src/namespace-ownership.json @@ -0,0 +1,205 @@ +{ + "host": { + "cluster_resource_id": "resource:railiance:reef-railiance:k3s", + "owner": "railiance-cluster", + "resource_id": "resource:hosteurope:railiance01", + "same_cpus_as_cluster": true + }, + "namespaces": { + "activity-core": { + "owner": "activity-core", + "service": "activity-core", + "tenant": "unknown" + }, + "approval-engine": { + "owner": "approval-engine", + "service": "approval-engine", + "tenant": "unknown" + }, + "argocd": { + "owner": "railiance-enablement", + "service": "argocd", + "tenant": "unknown" + }, + "audit-core": { + "owner": "audit-core", + "service": "audit-core", + "tenant": "unknown" + }, + "bao-notice": { + "owner": "railiance-platform", + "service": "bao-notice", + "tenant": "unknown" + }, + "canned-prompts": { + "owner": "canned-prompts", + "service": "canned-prompts", + "tenant": "unknown" + }, + "cert-manager": { + "owner": "railiance-cluster", + "service": "cert-manager", + "tenant": "unknown" + }, + "cnpg-system": { + "owner": "rapp-postgres", + "service": "cloudnative-pg", + "tenant": "unknown" + }, + "core-hub": { + "owner": "core-hub", + "service": "core-hub", + "tenant": "unknown" + }, + "coulomb": { + "owner": "unknown", + "service": "ihp-railiance-probe", + "tenant": "unknown" + }, + "coulomb-social": { + "owner": "coulomb-social", + "service": "coulomb-social", + "tenant": "unknown" + }, + "databases": { + "owner": "rapp-postgres", + "service": "apps-pg", + "tenant": "unknown" + }, + "email-connect": { + "owner": "email-connect", + "service": "email-connect", + "tenant": "unknown" + }, + "external-secrets": { + "owner": "railiance-platform", + "service": "external-secrets", + "tenant": "unknown" + }, + "flex-auth": { + "owner": "flex-auth", + "service": "flex-auth", + "tenant": "unknown" + }, + "forgejo": { + "owner": "railiance-forge", + "service": "forgejo", + "tenant": "unknown" + }, + "informed-decision": { + "owner": "informed-decision", + "service": "informed-decision", + "tenant": "unknown" + }, + "issue-core": { + "owner": "issue-core", + "service": "issue-core", + "tenant": "unknown" + }, + "knative-serving": { + "owner": "rail-knative", + "service": "knative-serving", + "tenant": "unknown" + }, + "kourier-system": { + "owner": "rail-knative", + "service": "kourier", + "tenant": "unknown" + }, + "kube-system": { + "owner": "railiance-cluster", + "service": "k3s-control-plane", + "tenant": "unknown" + }, + "mas": { + "owner": "mail-address-semantics", + "service": "mas-live-test", + "tenant": "unknown" + }, + "mfa": { + "owner": "net-kingdom", + "service": "lldap-mfa", + "tenant": "unknown" + }, + "openbao": { + "owner": "railiance-platform", + "service": "openbao", + "tenant": "unknown" + }, + "platform-pg-drill": { + "owner": "rapp-postgres", + "service": "platform-pg-drill", + "tenant": "unknown" + }, + "playwright": { + "owner": "railiance-platform", + "service": "rapp-playwright", + "tenant": "unknown" + }, + "policy-nexus": { + "owner": "policy-nexus", + "service": "policy-nexus", + "tenant": "unknown" + }, + "rapp-qonto": { + "owner": "rapp-qonto", + "service": "qonto", + "tenant": "unknown" + }, + "rapp-qonto-egress": { + "owner": "rapp-qonto", + "service": "qonto-egress", + "tenant": "tenant:friendly:binky" + }, + "rein-aharness": { + "owner": "rein-aharness", + "service": "rein-aharness", + "tenant": "unknown" + }, + "reuse": { + "owner": "reuse-surface", + "service": "reuse-surface", + "tenant": "unknown" + }, + "sbom-nexus": { + "owner": "sbom-nexus", + "service": "sbom-nexus", + "tenant": "unknown" + }, + "sso": { + "owner": "net-kingdom", + "service": "keycape-authelia", + "tenant": "unknown" + }, + "state-hub": { + "owner": "state-hub", + "service": "state-hub", + "tenant": "unknown" + }, + "target-revenue": { + "owner": "target-revenue", + "service": "target-revenue", + "tenant": "unknown" + }, + "telemetry": { + "owner": "rapp-telemetry", + "service": "prometheus-grafana", + "tenant": "unknown" + }, + "tenant-engine": { + "owner": "tenant-engine", + "service": "tenant-engine", + "tenant": "unknown" + }, + "user-engine": { + "owner": "user-engine", + "service": "user-engine", + "tenant": "unknown" + }, + "vergabe-demo-company": { + "owner": "railiance-apps", + "service": "vergabe-teilnahme", + "tenant": "unknown" + } + } +} diff --git a/argocd/platform-addons/custodian-allocation-review/src/observe_cluster_resources.py b/argocd/platform-addons/custodian-allocation-review/src/observe_cluster_resources.py new file mode 100644 index 0000000..08e5a33 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/src/observe_cluster_resources.py @@ -0,0 +1,334 @@ +#!/usr/bin/env python3 +"""Capture non-secret cluster capacity, utilization, and allocation evidence.""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + + +CPU_FACTORS = {"n": 0.000001, "u": 0.001, "m": 1.0, "": 1000.0} +BYTE_FACTORS = { + "": 1, + "Ki": 1024, + "Mi": 1024**2, + "Gi": 1024**3, + "Ti": 1024**4, + "K": 1000, + "M": 1000**2, + "G": 1000**3, + "T": 1000**4, +} + + +def cpu_millicores(value: str | None) -> int: + match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)(n|u|m)?", str(value or "0")) + if not match: + raise ValueError(f"unsupported CPU quantity: {value}") + return round(float(match.group(1)) * CPU_FACTORS[match.group(2) or ""]) + + +def bytes_value(value: str | None) -> int: + match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)(Ki|Mi|Gi|Ti|K|M|G|T)?", str(value or "0")) + if not match: + raise ValueError(f"unsupported byte quantity: {value}") + return round(float(match.group(1)) * BYTE_FACTORS[match.group(2) or ""]) + + +def workload_name(pod: dict[str, Any]) -> str: + metadata = pod.get("metadata") or {} + labels = metadata.get("labels") or {} + for key in ("app.kubernetes.io/part-of", "app.kubernetes.io/name", "app"): + if labels.get(key): + return str(labels[key]) + owners = metadata.get("ownerReferences") or [] + if owners and owners[0].get("name"): + return str(owners[0]["name"]) + return re.sub(r"-[a-f0-9]{6,}(?:-[a-z0-9]{5})?$", "", str(metadata.get("name", "unknown"))) + + +def pod_effective_requests(pod: dict[str, Any]) -> tuple[int, int]: + """Pod request is max(sum(containers), max(init containers)).""" + spec = pod.get("spec") or {} + containers = spec.get("containers") or [] + inits = spec.get("initContainers") or [] + + def req(container: dict[str, Any], key: str) -> str | None: + return ((container.get("resources") or {}).get("requests") or {}).get(key) + + cpu = sum(cpu_millicores(req(c, "cpu")) for c in containers) + mem = sum(bytes_value(req(c, "memory")) for c in containers) + init_cpu = max((cpu_millicores(req(c, "cpu")) for c in inits), default=0) + init_mem = max((bytes_value(req(c, "memory")) for c in inits), default=0) + return max(cpu, init_cpu), max(mem, init_mem) + + +def parse_top(lines: str | None, *, pods: bool) -> dict[tuple[str, str] | str, dict[str, int]]: + result: dict[tuple[str, str] | str, dict[str, int]] = {} + for line in (lines or "").splitlines(): + fields = line.split() + if pods and len(fields) >= 4: + result[(fields[0], fields[1])] = { + "cpu_millicores": cpu_millicores(fields[2]), + "memory_bytes": bytes_value(fields[3]), + } + elif not pods and len(fields) >= 4: + result[fields[0]] = { + "cpu_millicores": cpu_millicores(fields[1]), + # kubectl top nodes: NAME CPU CPU% MEMORY MEMORY% + "memory_bytes": bytes_value(fields[3]), + } + return result + + +def build_observation(snapshots: dict[str, Any], captured_at: str) -> dict[str, Any]: + gaps: list[str] = [] + node_top = parse_top(snapshots.get("node_top", ""), pods=False) + pod_top = parse_top(snapshots.get("pod_top", ""), pods=True) + if snapshots.get("node_top") is None: + gaps.append("node metrics unavailable") + if snapshots.get("pod_top") is None: + gaps.append("pod metrics unavailable") + + nodes = [] + alloc_cpu = alloc_memory = alloc_pods = 0 + for node in snapshots["nodes"].get("items") or []: + meta = node.get("metadata") or {} + status = node.get("status") or {} + alloc = status.get("allocatable") or {} + labels = meta.get("labels") or {} + name = str(meta.get("name")) + topology = { + "region": labels.get("topology.kubernetes.io/region"), + "zone": labels.get("topology.kubernetes.io/zone"), + "hostname": labels.get("kubernetes.io/hostname"), + "provider_id": (node.get("spec") or {}).get("providerID"), + } + if not topology["region"] or not topology["zone"]: + gaps.append(f"node {name} lacks independent region/zone labels") + row = { + "name": name, + "ready": any(c.get("type") == "Ready" and c.get("status") == "True" for c in status.get("conditions") or []), + "allocatable": { + "cpu_millicores": cpu_millicores(alloc.get("cpu")), + "memory_bytes": bytes_value(alloc.get("memory")), + "ephemeral_storage_bytes": bytes_value(alloc.get("ephemeral-storage")), + "pods": int(alloc.get("pods") or 0), + }, + "topology": topology, + "observed": node_top.get(name), + } + nodes.append(row) + alloc_cpu += row["allocatable"]["cpu_millicores"] + alloc_memory += row["allocatable"]["memory_bytes"] + alloc_pods += row["allocatable"]["pods"] + + workloads: dict[tuple[str, str], dict[str, Any]] = {} + pending_unscheduled: list[dict[str, Any]] = [] + node_allocated: dict[str, dict[str, int]] = { + node["name"]: {"cpu_millicores": 0, "memory_bytes": 0} for node in nodes + } + scheduled_cpu = scheduled_memory = 0 + for pod in snapshots["pods"].get("items") or []: + meta = pod.get("metadata") or {} + spec = pod.get("spec") or {} + status = pod.get("status") or {} + phase = str(status.get("phase") or "") + if phase in {"Succeeded", "Failed"}: + continue + namespace = str(meta.get("namespace", "default")) + pod_name = str(meta.get("name")) + cpu, mem = pod_effective_requests(pod) + node_name = spec.get("nodeName") + if phase == "Pending" and not node_name: + pending_unscheduled.append({ + "namespace": namespace, + "name": pod_name, + "cpu_millicores": cpu, + "memory_bytes": mem, + }) + continue + if node_name in node_allocated: + node_allocated[node_name]["cpu_millicores"] += cpu + node_allocated[node_name]["memory_bytes"] += mem + scheduled_cpu += cpu + scheduled_memory += mem + name = workload_name(pod) + key = (namespace, name) + row = workloads.setdefault(key, { + "namespace": namespace, + "workload": name, + "pods": 0, + "requests": {"cpu_millicores": 0, "memory_bytes": 0}, + "limits": {"cpu_millicores": 0, "memory_bytes": 0}, + "observed": {"cpu_millicores": 0, "memory_bytes": 0} if snapshots.get("pod_top") is not None else None, + }) + row["pods"] += 1 + row["requests"]["cpu_millicores"] += cpu + row["requests"]["memory_bytes"] += mem + for container in spec.get("containers") or []: + limits = (container.get("resources") or {}).get("limits") or {} + row["limits"]["cpu_millicores"] += cpu_millicores(limits.get("cpu")) + row["limits"]["memory_bytes"] += bytes_value(limits.get("memory")) + metric = pod_top.get((namespace, pod_name)) + if metric and row["observed"] is not None: + row["observed"]["cpu_millicores"] += metric["cpu_millicores"] + row["observed"]["memory_bytes"] += metric["memory_bytes"] + + storage_classes = [] + for item in snapshots["storageclasses"].get("items") or []: + storage_classes.append({ + "name": (item.get("metadata") or {}).get("name"), + "provisioner": item.get("provisioner"), + "reclaim_policy": item.get("reclaimPolicy"), + "volume_binding_mode": item.get("volumeBindingMode"), + "default": (item.get("metadata") or {}).get("annotations", {}).get("storageclass.kubernetes.io/is-default-class") == "true", + }) + pv_capacity = sum(bytes_value(((item.get("spec") or {}).get("capacity") or {}).get("storage")) for item in snapshots["pvs"].get("items") or []) + pvc_rows = [] + pvc_requested = 0 + for item in snapshots["pvcs"].get("items") or []: + meta = item.get("metadata") or {} + spec = item.get("spec") or {} + requested = bytes_value(((spec.get("resources") or {}).get("requests") or {}).get("storage")) + pvc_requested += requested + pvc_rows.append({ + "namespace": meta.get("namespace"), + "name": meta.get("name"), + "storage_class": spec.get("storageClassName"), + "requested_bytes": requested, + "phase": (item.get("status") or {}).get("phase"), + }) + + workload_rows = sorted(workloads.values(), key=lambda row: (row["namespace"], row["workload"])) + requested_cpu = scheduled_cpu + requested_memory = scheduled_memory + pending_cpu = sum(item["cpu_millicores"] for item in pending_unscheduled) + pending_memory = sum(item["memory_bytes"] for item in pending_unscheduled) + preflight_nodes = [] + for node in nodes: + allocated = node_allocated.get(node["name"], {"cpu_millicores": 0, "memory_bytes": 0}) + preflight_nodes.append({ + "name": node["name"], + "ready": node["ready"], + "unschedulable": False, + "allocatable_cpu_m": node["allocatable"]["cpu_millicores"], + "allocatable_memory_bytes": node["allocatable"]["memory_bytes"], + "allocated_cpu_m": allocated["cpu_millicores"], + "allocated_memory_bytes": allocated["memory_bytes"], + }) + return { + "schema_version": "railiance.cluster-resource-observation.v1", + "record_type": "usage_observation", + "resource_id": "resource:railiance:reef-railiance:k3s", + "source": "railiance-cluster", + "workplan_id": "RCLUSTER-WP-0014", + "reef": "reef-railiance", + "captured_at": captured_at, + "capacity": { + "nodes": nodes, + "totals": {"cpu_millicores": alloc_cpu, "memory_bytes": alloc_memory, "pods": alloc_pods, "pv_bytes": pv_capacity}, + }, + "utilization": {"workloads": workload_rows}, + "storage": {"classes": storage_classes, "claims": pvc_rows}, + "failure_domain": { + "current_contract": "single Ready node on railiance01; local storage is correlated with node loss", + "independent_domains_claimed": 1 if nodes else 0, + "threephoenix_target_is_live": False, + }, + "allocation": { + "method_version": "cluster-raw-drivers-v1", + "drivers": ["cpu_requests", "memory_requests", "pvc_requested_bytes", "observed_usage"], + "denominators": {"cpu_requested_millicores": requested_cpu, "memory_requested_bytes": requested_memory, "pvc_requested_bytes": pvc_requested}, + "pending_unscheduled": { + "cpu_millicores": pending_cpu, + "memory_bytes": pending_memory, + "pods": pending_unscheduled, + }, + "residual_capacity": {"cpu_millicores": max(alloc_cpu - requested_cpu, 0), "memory_bytes": max(alloc_memory - requested_memory, 0), "pv_bytes": max(pv_capacity - pvc_requested, 0)}, + "not_a_scheduling_guarantee": True, + "selection_owner": "resource-control", + }, + "state_hub_preflight_observation": { + "schema": "state-hub.release-headroom-preflight.v1-input", + "observed_at": captured_at, + "freshness_seconds": 0, + "nodes": preflight_nodes, + "pending_unrelated": [ + { + "name": f"{item['namespace']}/{item['name']}", + "cpu_m": item["cpu_millicores"], + "memory_bytes": item["memory_bytes"], + } + for item in pending_unscheduled + ], + }, + "measurement_gaps": sorted(set(gaps)), + "provenance": { + "commands": [ + "kubectl get nodes -o json", "kubectl get pods -A -o json", + "kubectl get pvc -A -o json", "kubectl get pv -o json", + "kubectl get storageclass -o json", "kubectl top nodes --no-headers", + "kubectl top pods -A --no-headers", + ], + "secret_surfaces_read": False, + }, + } + + +def run_json(command: list[str]) -> dict[str, Any]: + return json.loads(subprocess.check_output(command, text=True)) + + +def run_optional(command: list[str]) -> str | None: + completed = subprocess.run(command, text=True, capture_output=True, check=False) + return completed.stdout if completed.returncode == 0 else None + + +def collect(kubectl: str, remote: str | None = None) -> dict[str, Any]: + base = ["ssh", "-o", "BatchMode=yes", remote, kubectl] if remote else [kubectl] + snapshots = { + "nodes": run_json(base + ["get", "nodes", "-o", "json"]), + "pods": run_json(base + ["get", "pods", "-A", "-o", "json"]), + "pvcs": run_json(base + ["get", "pvc", "-A", "-o", "json"]), + "pvs": run_json(base + ["get", "pv", "-o", "json"]), + "storageclasses": run_json(base + ["get", "storageclass", "-o", "json"]), + "node_top": run_optional(base + ["top", "nodes", "--no-headers"]), + "pod_top": run_optional(base + ["top", "pods", "-A", "--no-headers"]), + } + captured_at = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + return build_observation(snapshots, captured_at) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--kubectl", default="kubectl") + parser.add_argument("--remote", help="Optional SSH host on which kubectl runs") + parser.add_argument("--output-dir", type=Path) + args = parser.parse_args() + try: + observation = collect(args.kubectl, args.remote) + except (OSError, subprocess.CalledProcessError, json.JSONDecodeError, ValueError) as exc: + print(f"cluster observation failed: {exc}", file=sys.stderr) + return 1 + rendered = json.dumps(observation, indent=2) + "\n" + if args.output_dir: + args.output_dir.mkdir(parents=True, exist_ok=True) + stamp = observation["captured_at"].replace(":", "").replace("-", "") + destination = args.output_dir / f"{stamp}.json" + destination.write_text(rendered, encoding="utf-8") + (args.output_dir / "latest.json").write_text(rendered, encoding="utf-8") + print(destination) + else: + sys.stdout.write(rendered) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/argocd/platform-addons/custodian-allocation-review/src/reconcile_railiance_allocation.py b/argocd/platform-addons/custodian-allocation-review/src/reconcile_railiance_allocation.py new file mode 100644 index 0000000..a7ca729 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/src/reconcile_railiance_allocation.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""CUST-WP-0071-T01: join live cluster allocation with owner/service/tenant.""" +from __future__ import annotations + +import argparse +import importlib.util +import json +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +CLUSTER_OBS = Path.home() / "railiance-cluster" / "tools" / "observe_cluster_resources.py" +HEADROOM = Path.home() / "state-hub" / "scripts" / "release_headroom_preflight.py" +OWNERS = ROOT / "docs" / "evidence" / "namespace-ownership.yaml" + + +def load_module(path: Path, name: str): + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise RuntimeError(f"cannot load {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def join(observation: dict[str, Any], owners: dict[str, Any]) -> dict[str, Any]: + mapping = owners.get("namespaces") or {} + rows = [] + unknown = [] + zero_request = [] + for workload in observation.get("utilization", {}).get("workloads") or []: + ns = workload["namespace"] + meta = mapping.get(ns, {"owner": "unknown", "service": "unknown", "tenant": "unknown"}) + row = { + "namespace": ns, + "workload": workload["workload"], + "pods": workload["pods"], + "cpu_request_m": workload["requests"]["cpu_millicores"], + "memory_request_bytes": workload["requests"]["memory_bytes"], + "owner": meta.get("owner", "unknown"), + "service": meta.get("service", "unknown"), + "tenant": meta.get("tenant", "unknown"), + } + rows.append(row) + if ns not in mapping: + unknown.append(ns) + if workload["requests"]["cpu_millicores"] == 0: + zero_request.append(f"{ns}/{workload['workload']}") + pending = observation.get("allocation", {}).get("pending_unscheduled") or {} + alloc = observation.get("allocation") or {} + capacity = observation.get("capacity", {}).get("totals") or {} + residual = (alloc.get("residual_capacity") or {}).get("cpu_millicores", 0) + requested = (alloc.get("denominators") or {}).get("cpu_requested_millicores", 0) + return { + "schema": "custodian.allocation-reconcile.v1", + "workplan_id": "CUST-WP-0071-T01", + "captured_at": observation.get("captured_at"), + "cluster_observation_schema": observation.get("schema_version"), + "count_host_and_cluster_cpus_once": True, + "host": owners.get("host"), + "capacity_cpu_m": capacity.get("cpu_millicores"), + "scheduled_request_cpu_m": requested, + "pending_unscheduled_cpu_m": pending.get("cpu_millicores", 0), + "residual_cpu_m": residual, + "not_a_scheduling_guarantee": True, + "workloads": sorted(rows, key=lambda r: (-r["cpu_request_m"], r["namespace"], r["workload"])), + "unknown_namespaces": sorted(set(unknown)), + "zero_request_workloads": sorted(zero_request), + "pending_unscheduled": pending.get("pods") or [], + "measurement_gaps": observation.get("measurement_gaps") or [], + "state_hub_preflight_observation": observation.get("state_hub_preflight_observation"), + "signal_notes": [ + "Host resource:hosteurope:railiance01 and cluster resource:railiance:reef-railiance:k3s are the same 4 vCPU; do not add them.", + "Prometheus namespace_cpu:kube_pod_container_resource_requests:sum omitted Forgejo 100m on 2026-09-11; Kubernetes API is the scheduler-effective source.", + "node-exporter was disabled; host CPU usage is not a reservation and is not treated as spare capacity.", + "Zero-request pods are demand, not zero. Missing metrics are listed as gaps, not zeros.", + "STATE-WP-0091 preflight consumes state_hub_preflight_observation; residual millicores are not admission.", + ], + } + + +def render_markdown(report: dict[str, Any], preflight: dict[str, Any] | None) -> str: + lines = [ + f"# Railiance allocation reconcile — {report['captured_at']}", + "", + "CUST-WP-0071-T01. Scheduler-effective requests from the Kubernetes API", + "via `railiance-cluster` observe (RCLUSTER-WP-0014 / RAIL-BS-WP-0014).", + "Host and cluster are the same 4 vCPU and are counted once.", + "", + f"- Capacity: {report['capacity_cpu_m']}m", + f"- Scheduled requests: {report['scheduled_request_cpu_m']}m", + f"- Pending unscheduled: {report['pending_unscheduled_cpu_m']}m", + f"- Residual (not a guarantee): {report['residual_cpu_m']}m", + f"- Unknown namespaces: {', '.join(report['unknown_namespaces']) or 'none'}", + f"- Zero-request workloads: {len(report['zero_request_workloads'])}", + "", + "| Namespace | Workload | Owner | Tenant | CPU request (m) | Pods |", + "|---|---|---|---|---:|---:|", + ] + for row in report["workloads"]: + lines.append( + f"| {row['namespace']} | {row['workload']} | {row['owner']} | {row['tenant']} | {row['cpu_request_m']} | {row['pods']} |" + ) + lines += ["", "## STATE-WP-0091 preflight", ""] + if preflight: + lines.append(f"- ok: `{preflight.get('ok')}`") + lines.append(f"- remaining_cpu_m: {preflight.get('remaining_cpu_m')}") + for reason in preflight.get("reasons") or []: + lines.append(f"- refuse: {reason}") + for note in preflight.get("notes") or []: + lines.append(f"- note: {note}") + lines += ["", "## Signal notes", ""] + for note in report["signal_notes"]: + lines.append(f"- {note}") + lines.append("") + return "\n".join(lines) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--observation", type=Path, help="Existing cluster observation JSON") + parser.add_argument("--kubectl", default="kubectl") + parser.add_argument("--output-dir", type=Path, default=ROOT / "docs" / "evidence") + args = parser.parse_args() + cluster = load_module(CLUSTER_OBS, "cluster_resources") + if args.observation: + observation = json.loads(args.observation.read_text()) + else: + observation = cluster.collect(args.kubectl) + import yaml # main only: join() stays stdlib-only for the in-cluster runner + + owners = yaml.safe_load(OWNERS.read_text()) + report = join(observation, owners) + preflight = None + if observation.get("state_hub_preflight_observation"): + headroom = load_module(HEADROOM, "release_headroom") + preflight = headroom.evaluate(observation["state_hub_preflight_observation"]) + report["state_hub_preflight"] = preflight + stamp = datetime.now(UTC).strftime("%Y-%m-%d") + out_json = args.output_dir / f"{stamp}-allocation-reconcile.json" + out_md = args.output_dir / f"{stamp}-allocation-reconcile.md" + args.output_dir.mkdir(parents=True, exist_ok=True) + out_json.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + out_md.write_text(render_markdown(report, preflight), encoding="utf-8") + print(out_json) + print(out_md) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/argocd/platform-addons/custodian-allocation-review/src/reef-railiance-k3s.json b/argocd/platform-addons/custodian-allocation-review/src/reef-railiance-k3s.json new file mode 100644 index 0000000..86a601c --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/src/reef-railiance-k3s.json @@ -0,0 +1,63 @@ +{ + "schema_version": "0.3", "record_scope": "inventory", + "id": "resource:railiance:reef-railiance:k3s", + "financial_entity_id": null, + "procuring_entity_id": "entity:railiance", + "entity_gap": "no allocation driver; cost remains a Railiance residual", + "description": "Single-node k3s runtime on railiance01. Cost and consumption view of reef-railiance; not a copy of IPs or kubeconfig.", + "decision": { + "status": "gap", + "chosen": "Live discovered cluster; no formal procurement decision in this repo", + "rejected": [], + "approved_by": null, + "approved_on": null, + "ref": "/home/worsch/railiance-cluster/docs/rail-kubernetes-substrate-profile.md" + }, + "operational_refs": [ + "reef:railiance/declarations/reef.yaml" + ], + "credential_handles": [], + "consumers": { + "potential": [ + "rapp-postgres", + "rapp-qonto", + "rapp-openbao", + "coulomb-social", + "helix-forge", + "railiance-platform" + ], + "actual": [ + "rapp-postgres", + "rapp-qonto", + "coulomb-social", + "helix-forge", + "railiance-platform", + "state-hub" + ] + }, + "resource_class": "kubernetes_capacity", + "status": "active", "management_model": "self_managed", + "provider": {"name": "Railiance", "account_ref": null, "product_ref": "k3s", "provider_resource_id": "k3s://239.62.205.92.host.secureserver.net"}, + "service": {"name": "reef-railiance k3s", "service_id": "reef-railiance", "class": "single-node Kubernetes runtime", "capacity_model": "shared"}, + "location": {"region": null, "country": null, "failure_domains": ["provider:host-europe", "host:railiance01", "cluster:reef-railiance"], "residency": null}, + "capacity": [ + {"metric": "cpu", "value": 4, "unit": "vCPU", "kind": "usable", "observed_at": "2026-08-11"}, + {"metric": "memory", "value": 15.62, "unit": "GiB", "kind": "usable", "observed_at": "2026-08-11"}, + {"metric": "ephemeral_storage", "value": 183.06, "unit": "GiB", "kind": "usable", "observed_at": "2026-08-11"}, + {"metric": "pods", "value": 110, "unit": "count", "kind": "limit", "observed_at": "2026-08-11"}, + {"metric": "nodes", "value": 1, "unit": "count", "kind": "observed", "observed_at": "2026-08-11"}, + {"metric": "cpu_usage", "value": 0.564, "unit": "vCPU", "kind": "observed", "observed_at": "2026-08-11"}, + {"metric": "memory_usage", "value": 5.83, "unit": "GiB", "kind": "observed", "observed_at": "2026-08-11"} + ], + "ownership": {"owner": "railiance-cluster", "environment": "production", "workload_ids": ["helix-forge", "coulomb-social", "rapp-qonto", "railiance-platform", "state-hub"], "tenant_id": null, "allocation": {"mode": "unattributed", "cost_attribution_key": null, "driver": null, "method_version": null}}, + "cost": {"currency": "EUR", "tax_status": "unknown", "billing_model": "allocated share of railiance01 plus cluster operations labor; method not established", "commitment_ref": null, "price_evidence": null}, + "lifecycle": {"proposed_on": null, "ordered_on": null, "commissioned_on": "2026-03-10", "renews_on": null, "cancel_by": null, "retired_on": null, "exit_path": "Deploy a replacement cluster, restore stateful services, promote workloads and ingress, verify health, then remove the old node and credentials."}, + "relationships": [{"type": "hosted_on", "resource_id": "resource:hosteurope:railiance01"}], + "requirements": [{"kind": "service_objective", "ref": "/home/worsch/railiance-cluster/INTENT.md"}, {"kind": "recovery", "ref": "/home/worsch/railiance-cluster/workplans/RAIL-BS-WP-0007-threephoenix-ha-cluster.md"}], + "evidence": [ + {"kind": "telemetry", "ref": "kubernetes:reef-railiance/node/239.62.205.92.host.secureserver.net@2026-08-11T09:05:25Z", "observed_at": "2026-08-11", "authority": "reef-railiance Kubernetes API"}, + {"kind": "workload", "ref": "/home/worsch/railiance-cluster/docs/rail-kubernetes-substrate-profile.md", "observed_at": "2026-08-11", "authority": "railiance-cluster"}, + {"kind": "workload", "ref": "STATE-WP-0091: state-hub MCP CPU request 50m→10m after partial telemetry (p99 2.35m, peak 30.45m, 1078 samples). 105m node remaining is not reef admission and does not authorize procurement or scaling.", "observed_at": "2026-09-14", "authority": "state-hub"}, + {"kind": "workload", "ref": "docs/evidence/2026-10-19-custodian-allocation-review.md", "observed_at": "2026-10-19", "authority": "resource-control review of Custodian8e67cfb and Vergabe11f7c8a; reservation snapshot and provisional synthetic evidence, not admission"} + ] +} diff --git a/argocd/platform-addons/custodian-allocation-review/src/weekly-allocation-queries.json b/argocd/platform-addons/custodian-allocation-review/src/weekly-allocation-queries.json new file mode 100644 index 0000000..8170e56 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/src/weekly-allocation-queries.json @@ -0,0 +1,16 @@ +{ + "schema": "custodian.weekly-allocation-queries.v1", + "note": "CUST-WP-0071-T05. Namespace-level aggregates proven on 2026-09-28 against telemetry/telemetry-prometheus. They populate namespace_context only; workload rows stay 'investigate' until workload-labelled queries exist (kube_pod_labels is not exported by the current kube-state-metrics). Changing any query text changes the report query_signature and breaks week-on-week comparability for one week.", + "window": "7d", + "step": "5m", + "queries": { + "cpu_p95_m": "quantile_over_time(0.95, (sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])", + "cpu_peak_m": "max_over_time((sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])", + "cpu_sample_count": "count_over_time((sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])", + "memory_p95_bytes": "quantile_over_time(0.95, (sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])", + "memory_peak_bytes": "max_over_time((sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])", + "memory_sample_count": "count_over_time((sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])", + "cpu_throttle_ratio": "sum by(namespace) (increase(container_cpu_cfs_throttled_periods_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[7d])) / sum by(namespace) (increase(container_cpu_cfs_periods_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[7d]))", + "restarts": "sum by(namespace) (increase(kube_pod_container_status_restarts_total[7d]))" + } +} diff --git a/argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_incluster.py b/argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_incluster.py new file mode 100644 index 0000000..ddfea2c --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_incluster.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python3 +"""CUST-WP-0071-T05 in-cluster weekly runner: read-only collection, join, retain. + +Standard library only, so it runs on a pinned python:3.12-slim image with the +scripts mounted from a ConfigMap. It reads the Kubernetes API with the pod's +ServiceAccount (get/list only), queries Prometheus over plain HTTP, reuses the +existing owner code unchanged (railiance-cluster build_observation, custodian +reconcile join, weekly_allocation_report build_report/retain) and writes only +under the data directory. It never resizes, creates tasks or delivers messages. + +Week identity is the ISO week of (run time - 6h); the telemetry window always +ends at Monday 06:00 UTC of that week, so consecutive reports are exactly +604800 s apart regardless of DST or scheduler jitter. 06:00 UTC is at or before +the Monday 08:00 Europe/Berlin schedule in both CET and CEST and keeps the +window start as close as possible to the seven-day Prometheus retention edge. A week whose report is already +retained is a no-op success (idempotent retry); a failed attempt leaves no +report, so the Job's retry or the next manual run collects again. +""" +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import re +import ssl +import sys +import urllib.error +import urllib.parse +import urllib.request + +HERE = Path(__file__).resolve().parent +SA_DIR = Path("/var/run/secrets/kubernetes.io/serviceaccount") +WEEK = re.compile(r"^(\d{4})-W(\d{2})\.json$") +# Exactly the reads the collector performs; mirrored by the ClusterRole. +API_PATHS = { + "nodes": "/api/v1/nodes", + "pods": "/api/v1/pods", + "pvcs": "/api/v1/persistentvolumeclaims", + "pvs": "/api/v1/persistentvolumes", + "storageclasses": "/apis/storage.k8s.io/v1/storageclasses", +} +METRICS_PATHS = { + "node_top": "/apis/metrics.k8s.io/v1beta1/nodes", + "pod_top": "/apis/metrics.k8s.io/v1beta1/pods", +} + + +def load(path: Path, name: str): + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise RuntimeError(f"cannot load {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def iso(value: datetime) -> str: + return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +WINDOW_END_HOUR = 6 + + +def week_of(now: datetime) -> tuple[str, datetime]: + """ISO week label and its Monday 06:00 UTC (the telemetry window end).""" + shifted = now.astimezone(timezone.utc) - timedelta(hours=WINDOW_END_HOUR) + year, week, weekday = shifted.isocalendar() + monday = (shifted - timedelta(days=weekday - 1)).replace(hour=0, minute=0, second=0, microsecond=0) + return f"{year}-W{week:02d}", monday + timedelta(hours=WINDOW_END_HOUR) + + +def missed_weeks(previous: str | None, current: str) -> list[str]: + if previous is None: + return [] + def monday(label): + year, week = int(label[:4]), int(label[6:8]) + return datetime.fromisocalendar(year, week, 1).replace(tzinfo=timezone.utc, hour=WINDOW_END_HOUR) + out, cursor = [], monday(previous) + timedelta(days=7) + while cursor < monday(current): + out.append(week_of(cursor)[0]) + cursor += timedelta(days=7) + return out + + +def previous_report(reports: Path, label: str) -> Path | None: + earlier = sorted(p for p in reports.glob("*.json") if WEEK.match(p.name) and p.stem < label) + return earlier[-1] if earlier else None + + +class Kube: + """Minimal in-cluster API reader; the token is never logged or persisted.""" + + def __init__(self, base: str, token_path: Path, ca_path: Path): + self.base = base.rstrip("/") + self.token_path = token_path + self.context = ssl.create_default_context(cafile=str(ca_path)) + + def get(self, path: str): + request = urllib.request.Request(self.base + path, headers={ + "Authorization": "Bearer " + self.token_path.read_text().strip(), + "Accept": "application/json"}) + with urllib.request.urlopen(request, context=self.context, timeout=60) as response: + return json.load(response) + + +def top_lines(metrics: dict, *, pods: bool, observer) -> str: + """Render metrics.k8s.io lists in the `kubectl top` shape parse_top reads.""" + lines = [] + for item in metrics.get("items") or []: + meta = item.get("metadata") or {} + if pods: + cpu = sum(observer.cpu_millicores((c.get("usage") or {}).get("cpu")) for c in item.get("containers") or []) + mem = sum(observer.bytes_value((c.get("usage") or {}).get("memory")) for c in item.get("containers") or []) + lines.append(f"{meta.get('namespace')} {meta.get('name')} {cpu}m {mem}") + else: + usage = item.get("usage") or {} + cpu = observer.cpu_millicores(usage.get("cpu")) + mem = observer.bytes_value(usage.get("memory")) + lines.append(f"{meta.get('name')} {cpu}m 0% {mem} 0%") + return "\n".join(lines) + ("\n" if lines else "") + + +def collect_snapshots(get, observer) -> dict: + snapshots = {name: get(path) for name, path in API_PATHS.items()} + for name, path in METRICS_PATHS.items(): + try: + snapshots[name] = top_lines(get(path), pods=name == "pod_top", observer=observer) + except (OSError, ValueError, urllib.error.URLError): + snapshots[name] = None # recorded by build_observation as a measurement gap + return snapshots + + +def prometheus_query(base: str, promql: str, at: datetime, opener=urllib.request.urlopen) -> dict: + """Return Prometheus' JSON response, including error bodies, never raise on 4xx/5xx.""" + url = base.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode( + {"query": promql, "time": f"{at.timestamp():.3f}"}) + try: + with opener(url, timeout=120) as response: + return json.load(response) + except urllib.error.HTTPError as exc: + try: + return json.loads(exc.read()) + except ValueError: + return {"status": "error", "errorType": "http", "error": f"HTTP {exc.code}"} + except (OSError, ValueError) as exc: + return {"status": "error", "errorType": "transport", "error": type(exc).__name__} + + +def write_json(path: Path, value) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(path.name + ".tmp") + temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n") + os.replace(temporary, path) + + +def append_receipt(data: Path, receipt: dict) -> None: + data.mkdir(parents=True, exist_ok=True) + with (data / "receipts.jsonl").open("a") as stream: + stream.write(json.dumps(receipt, sort_keys=True) + "\n") + stream.flush() + os.fsync(stream.fileno()) + print(json.dumps(receipt, sort_keys=True)) + + +def run(args, *, now: datetime, get, query) -> dict: + sources = { + "weekly_allocation_report": args.source_dir / "weekly_allocation_report.py", + "reconcile_railiance_allocation": args.source_dir / "reconcile_railiance_allocation.py", + "observe_cluster_resources": args.observer, + "weekly_allocation_incluster": Path(__file__).resolve(), + "queries": args.queries, "ownership": args.ownership, "inventory": args.inventory, + } + weekly = load(sources["weekly_allocation_report"], "weekly_allocation_report") + reconcile = load(sources["reconcile_railiance_allocation"], "reconcile_railiance_allocation") + observer = load(sources["observe_cluster_resources"], "observe_cluster_resources") + label, window_end = week_of(now) + reports = args.data_dir / "reports" + target = reports / f"{label}.json" + receipt = {"schema": "custodian.weekly-allocation-receipt.v1", "week": label, + "run_at": iso(now), "report": str(target), "delivery": "not emitted", + "acknowledgment": None} + if target.exists(): + return {**receipt, "status": "already-retained", "report_sha256": sha256(target)} + previous_path = previous_report(reports, label) + previous_label = previous_path.stem if previous_path else None + + observation = observer.build_observation(collect_snapshots(get, observer), iso(now)) + owners = json.loads(args.ownership.read_text()) + allocation = reconcile.join(observation, owners) + spec = json.loads(args.queries.read_text()) + if spec.get("schema") != "custodian.weekly-allocation-queries.v1": + raise ValueError("unsupported query specification") + telemetry = {"captured_at": window_end.isoformat(), "window": spec["window"], "step": spec["step"], + "source": args.prometheus_url, + "queries": {name: {"promql": promql, "response": query(promql, window_end)} + for name, promql in sorted(spec["queries"].items())}} + provenance = { + "schema": "custodian.allocation-capture-provenance.v1", "workplan": "CUST-WP-0071-T05", + "source_observation": observation["captured_at"], "read_only": True, + "secret_surfaces_read": False, "executor": args.executor, + "kubernetes_reads": sorted(API_PATHS.values()) + sorted(METRICS_PATHS.values()), + "prometheus": args.prometheus_url, "telemetry_window_end": window_end.isoformat(), + "inputs": [{"name": n, "path": str(p), "sha256": sha256(p)} for n, p in sorted(sources.items())], + } + inventory = json.loads(args.inventory.read_text()) + inputs = {"cluster": observation, "allocation": allocation, "telemetry": telemetry, + "inventory": inventory, "provenance": provenance} + attempt = args.data_dir / "runs" / label / iso(now).replace(":", "") + for name, value in inputs.items(): + write_json(attempt / f"{name}.json", value) + report = weekly.build_report( + **inputs, as_of=iso(now), max_age_hours=args.max_age_hours, + previous=json.loads(previous_path.read_text()) if previous_path else None, + reviews=json.loads(args.reviews.read_text()) if args.reviews and args.reviews.exists() else None) + weekly.retain(target, report) + failed = sorted(n for n, q in telemetry["queries"].items() if q["response"].get("status") != "success") + return {**receipt, "status": "retained", "report_sha256": sha256(target), + "inputs_dir": str(attempt), "previous_week": previous_label, + "missed_weeks": missed_weeks(previous_label, label), + "comparison_compatible": report["comparison"]["compatible"], + "workloads": len(report["workloads"]), "failed_queries": failed, + "measurement_gaps": len(report["measurement_gaps"])} + + +def parser() -> argparse.ArgumentParser: + env = os.environ.get + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--data-dir", type=Path, default=Path(env("ASSESSMENT_DATA_DIR", "/data"))) + p.add_argument("--source-dir", type=Path, default=Path(env("ASSESSMENT_SOURCE_DIR", str(HERE)))) + p.add_argument("--observer", type=Path, default=Path(env("ASSESSMENT_OBSERVER", + str(Path(env("ASSESSMENT_SOURCE_DIR", str(HERE))) / "observe_cluster_resources.py")))) + p.add_argument("--queries", type=Path, default=Path(env("ASSESSMENT_QUERIES", str(HERE.parent / "config/weekly-allocation-queries.json")))) + p.add_argument("--ownership", type=Path, default=env("ASSESSMENT_OWNERSHIP") and Path(env("ASSESSMENT_OWNERSHIP"))) + p.add_argument("--inventory", type=Path, default=env("ASSESSMENT_INVENTORY") and Path(env("ASSESSMENT_INVENTORY"))) + p.add_argument("--reviews", type=Path, default=env("ASSESSMENT_REVIEWS") and Path(env("ASSESSMENT_REVIEWS"))) + p.add_argument("--prometheus-url", default=env("PROMETHEUS_URL", "http://telemetry-prometheus.telemetry.svc:9090")) + p.add_argument("--executor", default=env("ASSESSMENT_EXECUTOR", "unspecified")) + p.add_argument("--max-age-hours", type=float, default=float(env("ASSESSMENT_MAX_AGE_HOURS", "48"))) + return p + + +def main() -> int: + args = parser().parse_args() + now = datetime.now(timezone.utc) + label = week_of(now)[0] + try: + if not args.ownership or not args.inventory: + raise ValueError("ownership and inventory inputs must be bound") + host = os.environ["KUBERNETES_SERVICE_HOST"] + port = os.environ.get("KUBERNETES_SERVICE_PORT", "443") + kube = Kube(f"https://{host}:{port}", SA_DIR / "token", SA_DIR / "ca.crt") + receipt = run(args, now=now, get=kube.get, + query=lambda promql, at: prometheus_query(args.prometheus_url, promql, at)) + except Exception as exc: # noqa: BLE001 - every failure must leave a receipt + append_receipt(args.data_dir, {"schema": "custodian.weekly-allocation-receipt.v1", + "week": label, "run_at": iso(now), "status": "failed", + "error": f"{type(exc).__name__}: {exc}"[:500]}) + return 1 + append_receipt(args.data_dir, receipt) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_report.py b/argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_report.py new file mode 100644 index 0000000..4917a23 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/src/weekly_allocation_report.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +"""Retain CUST-WP-0071 sizing evidence; no network, resizing or task emission.""" +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import hashlib +import json +import math +import os +from pathlib import Path +import re +import tempfile + +SCHEMA = "custodian.weekly-allocation.v1" +METRICS = ( + "cpu_p95_m", "cpu_peak_m", "memory_p95_bytes", "memory_peak_bytes", + "cpu_throttle_ratio", "latency_p95_seconds", "errors", "request_count", + "restarts", "oom_events", "cpu_sample_count", "memory_sample_count", +) +REQUIRED = tuple(m for m in METRICS if not m.endswith("sample_count")) + + +def timestamp(value): + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + if result.tzinfo is None: + raise ValueError("timestamps must include timezone") + return result.astimezone(timezone.utc) + + +def duration(value): + match = re.fullmatch(r"([1-9][0-9]*)([smhdw])", value) + if not match: + raise ValueError(f"unsupported duration: {value}") + return int(match[1]) * {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}[match[2]] + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, allow_nan=False).encode()).hexdigest() + + +def finite(value): + number = float(value) + if not math.isfinite(number) or number < 0: + raise ValueError("metric must be finite and nonnegative") + return number + + +def metric_series(telemetry): + """Do not distribute namespace aggregates over workloads, or sum percentiles.""" + series, queries = {}, {} + seconds = duration(telemetry["window"]) + for name, query in sorted(telemetry.get("queries", {}).items()): + response = query.get("response", {}) + data = response.get("data", {}) + queries[name] = { + "promql": query.get("promql"), "response_status": response.get("status"), + "result_type": data.get("resultType"), "warnings": response.get("warnings", []), + "error": response.get("error"), "response_sha256": digest(response), + "evaluation_times": [], + } + if name not in METRICS or response.get("status") != "success" or data.get("resultType") != "vector": + continue + for sample in data.get("result", []): + labels = sample.get("metric", {}) + namespace, workload = labels.get("namespace"), labels.get("workload") + # Additional dimensions are ambiguous: do not silently collapse pod/job series. + if not namespace or set(labels) - {"namespace", "workload", "__name__"}: + queries[name].setdefault("rejected", []).append("unaggregated or absent identity labels") + continue + key = (namespace, workload) + values = series.setdefault(key, {}) + if name in values: + values[name] = {"value": None, "reason": "ambiguous duplicate series"} + continue + try: + epoch, raw = sample["value"] + end = datetime.fromtimestamp(finite(epoch), timezone.utc) + number = finite(raw) + except (KeyError, ValueError, TypeError, OverflowError): + values[name] = {"value": None, "reason": "invalid sample"} + continue + queries[name]["evaluation_times"].append(end.isoformat()) + values[name] = { + "value": number, "window_start": (end - timedelta(seconds=seconds)).isoformat(), + "window_end": end.isoformat(), "query": name, + "capture_alignment_seconds": (end - timestamp(telemetry["captured_at"])).total_seconds(), + } + return series, queries + + +def recommendation_id(resource, namespace, workload): + return "allocation-" + digest([resource, namespace, workload, "sizing-review"])[:24] + + +def build_report(cluster, allocation, telemetry, inventory, provenance, *, as_of, + previous=None, reviews=None, max_age_hours=48): + now = timestamp(as_of) + if cluster.get("schema_version") != "railiance.cluster-resource-observation.v1": + raise ValueError("unsupported cluster schema") + if allocation.get("schema") != "custodian.allocation-reconcile.v1": + raise ValueError("unsupported allocation schema") + resource = cluster["resource_id"] + if inventory.get("id") != resource or allocation.get("host", {}).get("cluster_resource_id") != resource: + raise ValueError("resource identity mismatch") + if allocation.get("count_host_and_cluster_cpus_once") is not True: + raise ValueError("allocation must deduplicate host and cluster CPUs") + if cluster["captured_at"] != allocation["captured_at"]: + raise ValueError("cluster and reconcile snapshots differ") + if provenance.get("source_observation") != cluster["captured_at"]: + raise ValueError("provenance does not identify cluster snapshot") + seconds = duration(telemetry["window"]) + if seconds != 604800: + raise ValueError("weekly report requires a seven-day telemetry window") + step = duration(telemetry["step"]) + series, queries = metric_series(telemetry) + captured = {"cluster": cluster["captured_at"], "telemetry": telemetry["captured_at"]} + freshness = {} + for name, value in captured.items(): + age = (now - timestamp(value)).total_seconds() + freshness[name] = {"age_seconds": age, "status": "future" if age < 0 else "stale" if age > max_age_hours * 3600 else "fresh"} + query_signature = digest({name: q["promql"] for name, q in queries.items()}) + end = timestamp(telemetry["captured_at"]) + comparison_reason = "no previous retained report" + compatible = False + if previous: + prev_end = timestamp(previous["period"]["capture_end"]) + compatible = ( + previous.get("schema") == SCHEMA and previous.get("resource_id") == resource + and previous["period"].get("seconds") == seconds + and previous["period"].get("step_seconds") == step + and previous.get("query_signature") == query_signature + and abs((end - prev_end).total_seconds() - seconds) <= step + ) + comparison_reason = "comparable consecutive capture windows" if compatible else "nonconsecutive windows or changed query/resource/step contract" + previous_rows = {(r["namespace"], r["workload"]): r for r in (previous or {}).get("workloads", [])} + owners = {(r["namespace"], r["workload"]): r for r in allocation.get("workloads", [])} + if len(owners) != len(allocation.get("workloads", [])): + raise ValueError("duplicate allocation workload identity") + reviews = reviews or {} + rows, recommendations = [], [] + seen = set() + for workload in sorted(cluster.get("utilization", {}).get("workloads", []), key=lambda x: (x["namespace"], x["workload"])): + key = workload["namespace"], workload["workload"] + if key in seen: + raise ValueError("duplicate cluster workload identity") + seen.add(key) + owner = owners.get(key, {}) + measurements = {m: series.get(key, {}).get(m, {"value": None, "reason": "no workload series"}) for m in METRICS} + gaps = [m for m in REQUIRED if measurements[m]["value"] is None] + metric_stale = [m for m, v in measurements.items() if v["value"] is not None and not 0 <= (now - timestamp(v["window_end"])).total_seconds() <= max_age_hours * 3600] + metric_unaligned = [m for m, v in measurements.items() if v["value"] is not None and abs(v["capture_alignment_seconds"]) > step] + rid = recommendation_id(resource, *key) + reasons = [] + if gaps: + reasons.append("missing workload metrics: " + ", ".join(gaps)) + if metric_stale or any(v["status"] != "fresh" for v in freshness.values()): + reasons.append("stale or future observation") + if metric_unaligned: + reasons.append("metric evaluation windows differ from capture by more than one step") + if owner.get("owner", "unknown") == "unknown": + reasons.append("unknown owner") + if workload.get("requests", {}).get("cpu_millicores") == 0: + reasons.append("zero CPU request requires owner review") + review = reviews.get(rid) + decision, confidence, evidence = "investigate", "low", [] + if review: + if review.get("decision") not in {"keep", "increase", "decrease", "investigate"}: + raise ValueError("invalid owner review decision") + required = ("owner", "evidence_refs", "receiving_task", "configuration", "rollback_ref", "reviewed_at", "period_end", "coverage_ref", "service_acceptance_ref", "margin_ref") + if any(not review.get(k) for k in required): + raise ValueError("owner review missing evidence, receiving task or configuration/rollback") + if review["owner"] != owner.get("owner") or review["period_end"] != telemetry["captured_at"]: + reasons.append("owner review does not match this owner/window") + elif timestamp(review["reviewed_at"]) > now: + reasons.append("owner review is future-dated") + elif not reasons: + decision, confidence, evidence = review["decision"], "owner-reviewed", review["evidence_refs"] + else: + reasons.append("owner proposal retained; incomplete evidence prevents accepted sizing recommendation") + else: + reasons.append("no owner-reviewed sizing decision") + row = { + "namespace": key[0], "workload": key[1], "owner": owner.get("owner", "unknown"), + "service": owner.get("service", "unknown"), "tenant": owner.get("tenant", "unknown"), + "pods": workload.get("pods"), "requests": workload.get("requests"), + "limits": workload.get("limits"), "instantaneous_observed": workload.get("observed"), + "metrics": measurements, "missing_metrics": gaps, + "namespace_context_ref": key[0] if (key[0], None) in series else None, + "raw_scrape_coverage": "unknown; evaluation counts do not prove source coverage", + "recommendation_id": rid, "week_on_week": {}, "comparison_reasons": {}, + } + prior = previous_rows.get(key) + for metric in METRICS: + current = measurements[metric]["value"] + old_measurement = (prior or {}).get("metrics", {}).get(metric, {}) + old = old_measurement.get("value") + metric_compatible = False + reason = comparison_reason + if compatible and current is not None and old is not None: + try: + old_end, old_start = timestamp(old_measurement["window_end"]), timestamp(old_measurement["window_start"]) + new_end, new_start = timestamp(measurements[metric]["window_end"]), timestamp(measurements[metric]["window_start"]) + metric_compatible = ( + (old_end - old_start).total_seconds() == seconds + and (new_end - new_start).total_seconds() == seconds + and abs((new_end - old_end).total_seconds() - seconds) <= step + and abs((new_start - old_end).total_seconds()) <= step + and abs(measurements[metric]["capture_alignment_seconds"]) <= step + and abs((old_end - timestamp(previous["period"]["capture_end"])).total_seconds()) <= step + ) + reason = "comparable consecutive metric windows" if metric_compatible else "metric evaluation windows are not consecutive or capture-aligned" + except (KeyError, ValueError, TypeError): + reason = "missing or invalid previous metric window" + elif compatible: + reason = "missing current or previous metric" + row["week_on_week"][metric] = current - old if metric_compatible else None + row["comparison_reasons"][metric] = reason + rows.append(row) + prior_recommendation = next((r for r in (previous or {}).get("recommendations", []) if r["id"] == rid), {}) + recommendations.append({ + "id": rid, "decision": decision, "confidence": confidence, "owner": row["owner"], + "reasons": reasons, "evidence_refs": evidence, "owner_proposal": review, + "receiving_task": review.get("receiving_task") if review else prior_recommendation.get("receiving_task"), + "previous_decision": prior_recommendation.get("decision"), + "owner_outcome": review.get("outcome") if review else None, + "delivery": "not emitted", "acknowledgment": None, + }) + if seen != set(owners): + raise ValueError("cluster and allocation workload inventories differ") + return { + "schema": SCHEMA, "workplan": "CUST-WP-0071-T05", "as_of": as_of, + "adapter_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + "resource_id": resource, "period": {"capture_end": telemetry["captured_at"], "seconds": seconds, "step_seconds": step, + "note": "Actual per-query windows are retained on samples; capture time is not a common query evaluation time."}, + "freshness": freshness, "max_age_hours": max_age_hours, "query_signature": query_signature, + "queries": queries, "collector_provenance": provenance, + "source_sha256": {"cluster": digest(cluster), "allocation": digest(allocation), "telemetry": digest(telemetry), "inventory": digest(inventory)}, + "node": {"capacity": cluster.get("capacity"), "allocation": cluster.get("allocation"), + "scheduled_request_cpu_m": allocation.get("scheduled_request_cpu_m"), + "pending_unscheduled_cpu_m": allocation.get("pending_unscheduled_cpu_m"), + "reservation_residual_cpu_m": allocation.get("residual_cpu_m"), + "release_preflight": allocation.get("state_hub_preflight_observation"), + "operating_margin": None, "release_margin_acceptance": "not established by reservation residual", + "host_cluster_counted_once": True}, + "portfolio": {k: inventory.get(k) for k in ("id", "schema_version", "ownership", "capacity", "relationships", "evidence")}, + "workloads": rows, + "namespace_context": {ns: values for (ns, workload), values in sorted(series.items(), key=lambda x: str(x[0])) if workload is None}, + "comparison": {"compatible": compatible, "reason": comparison_reason, + "previous_report_sha256": digest(previous) if previous else None, + "disappeared_workloads": [list(key) for key in sorted(set(previous_rows) - seen)]}, + "measurement_gaps": cluster.get("measurement_gaps", []) + allocation.get("measurement_gaps", []), + "recommendations": recommendations, + "limitations": ["No representative pilot acceptance inferred from telemetry.", + "Namespace percentiles and peaks are not workload metrics or additive node peaks.", + "Portfolio capacity is dated evidence, not additional host capacity.", + "No procurement, sizing changes, owner-task creation, delivery or acknowledgment performed.", + "Installed executor, scoped reads, independent missed-run handling and first unattended receipt remain T05 gates."], + } + + +def retain(path, report): + """Never overwrite a retained week; identical retries succeed.""" + encoded = json.dumps(report, indent=2, sort_keys=True, allow_nan=False) + "\n" + path.parent.mkdir(parents=True, exist_ok=True) + temporary = None + try: + with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as stream: + temporary = Path(stream.name) + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + try: + os.link(temporary, path) + except FileExistsError: + if path.is_symlink() or path.read_text() != encoded: + raise ValueError("retained report differs; choose a new explicit revision path") + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--request", type=Path, help="Executor-owned JSON request; paths relative to its directory") + for name in ("cluster", "allocation", "telemetry", "inventory", "provenance"): + parser.add_argument("--" + name, type=Path) + parser.add_argument("--previous", type=Path) + parser.add_argument("--reviews", type=Path) + parser.add_argument("--as-of") + parser.add_argument("--max-age-hours", type=float, default=48) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + try: + if args.request: + request = json.loads(args.request.read_text()) + if request.get("schema") != "custodian.weekly-allocation-request.v1": + raise ValueError("unsupported executor request schema") + if any(getattr(args, key) for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "as_of", "output")): + raise ValueError("request cannot be mixed with input/output overrides") + for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "output"): + value = request.get(key) + setattr(args, key, args.request.parent / value if value else None) + args.as_of = request.get("as_of") + if any(not getattr(args, key) for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "as_of", "output")): + raise ValueError("executor input paths, as_of and retention output must be explicitly bound") + if not math.isfinite(args.max_age_hours) or args.max_age_hours <= 0: + raise ValueError("max-age-hours must be positive and finite") + inputs = {name: json.loads(getattr(args, name).read_text()) for name in ("cluster", "allocation", "telemetry", "inventory", "provenance")} + report = build_report(**inputs, as_of=args.as_of, max_age_hours=args.max_age_hours, + previous=json.loads(args.previous.read_text()) if args.previous else None, + reviews=json.loads(args.reviews.read_text()) if args.reviews else None) + retain(args.output, report) + except (OSError, ValueError, KeyError, TypeError) as exc: + parser.exit(2, f"weekly allocation report refused: {exc}\n") + print(json.dumps({"status": "retained", "path": str(args.output), "sha256": digest(report), + "workloads": len(report["workloads"]), "delivery": "not emitted"})) + + +if __name__ == "__main__": + main() diff --git a/argocd/platform-addons/custodian-allocation-review/vendor.lock.json b/argocd/platform-addons/custodian-allocation-review/vendor.lock.json new file mode 100644 index 0000000..6496a14 --- /dev/null +++ b/argocd/platform-addons/custodian-allocation-review/vendor.lock.json @@ -0,0 +1,55 @@ +{ + "files": { + "namespace-ownership.json": { + "commit": "7dc1133f4cb94f6e4f7bd032abb6aa5357197c74", + "path": "docs/evidence/namespace-ownership.yaml", + "repo": "the-custodian", + "source_sha256": "a3663c03f685db35d5e0eabca59e1d74810b1b56c39db0892a58903302bb771f", + "vendored_sha256": "dd73079defcc26db460ee1ed0ef17a5eab39b04d9dbc209cb6d02064df8f3f05" + }, + "observe_cluster_resources.py": { + "commit": "a5948408a565d383a99f43d414145277a43ce233", + "path": "tools/observe_cluster_resources.py", + "repo": "railiance-cluster", + "source_sha256": "7a613f7ba692169ec9ddf498f7250ee2a56f8fb1088585498af6df2302e15002", + "vendored_sha256": "7a613f7ba692169ec9ddf498f7250ee2a56f8fb1088585498af6df2302e15002" + }, + "reconcile_railiance_allocation.py": { + "commit": "7dc1133f4cb94f6e4f7bd032abb6aa5357197c74", + "path": "scripts/reconcile_railiance_allocation.py", + "repo": "the-custodian", + "source_sha256": "3e118023397bbb6621116f6f3009cb6ec7f894809671efeb5c165dd2c3e32a3a", + "vendored_sha256": "3e118023397bbb6621116f6f3009cb6ec7f894809671efeb5c165dd2c3e32a3a" + }, + "reef-railiance-k3s.json": { + "commit": "e3e996fc95b3cfa8bcfd0ba62b08b332fe5cae71", + "path": "data/resources/reef-railiance-k3s.json", + "repo": "resource-control", + "source_sha256": "c97cb2c505cc52dcf1e1f394d35a8d87dfec09a9033f6cd4c0c05717111f913f", + "vendored_sha256": "c97cb2c505cc52dcf1e1f394d35a8d87dfec09a9033f6cd4c0c05717111f913f" + }, + "weekly-allocation-queries.json": { + "commit": "7dc1133f4cb94f6e4f7bd032abb6aa5357197c74", + "path": "config/weekly-allocation-queries.json", + "repo": "the-custodian", + "source_sha256": "83bde8a6e53c4b86b9624195b9a00297310eefb1e271ba4506622c1efdd7e7a9", + "vendored_sha256": "83bde8a6e53c4b86b9624195b9a00297310eefb1e271ba4506622c1efdd7e7a9" + }, + "weekly_allocation_incluster.py": { + "commit": "7dc1133f4cb94f6e4f7bd032abb6aa5357197c74", + "path": "scripts/weekly_allocation_incluster.py", + "repo": "the-custodian", + "source_sha256": "92068a349a6faaf87495c0811c0ad94dac1ab195c29fd000a9e9fa85bb28364c", + "vendored_sha256": "92068a349a6faaf87495c0811c0ad94dac1ab195c29fd000a9e9fa85bb28364c" + }, + "weekly_allocation_report.py": { + "commit": "7dc1133f4cb94f6e4f7bd032abb6aa5357197c74", + "path": "scripts/weekly_allocation_report.py", + "repo": "the-custodian", + "source_sha256": "93f546243331308d7e60f8db36e988d9847537a2fe46cce49ee2fffba1f7f247", + "vendored_sha256": "93f546243331308d7e60f8db36e988d9847537a2fe46cce49ee2fffba1f7f247" + } + }, + "schema": "railiance-platform.vendor-lock.v1", + "workplan": "CUST-WP-0071-T05" +} diff --git a/argocd/railiance01/drafts/custodian-allocation-review.application.yaml b/argocd/railiance01/drafts/custodian-allocation-review.application.yaml new file mode 100644 index 0000000..ca38df6 --- /dev/null +++ b/argocd/railiance01/drafts/custodian-allocation-review.application.yaml @@ -0,0 +1,29 @@ +# DRAFT — not synced by any root (drafts/ is never referenced). CUST-WP-0071-T05. +# Activation (founder go-ahead, after CUST-WP-0071-T04 and review): add +# PersistentVolumeClaim to the railiance-platform-addons AppProject, set +# targetRevision to the reviewed commit, move this file to ../applications/, +# run `make argocd-pin-drift`, then hand-sync the root and this Application. +# See argocd/platform-addons/custodian-allocation-review/README.md. +apiVersion: argoproj.io/v1alpha1 +kind: Application +metadata: + name: custodian-allocation-review + namespace: argocd + labels: + app.kubernetes.io/part-of: railiance-gitops + railiance-platform/component: custodian-allocation-review + annotations: + argocd.argoproj.io/sync-wave: "10" +spec: + project: railiance-platform-addons + source: + repoURL: https://forgejo.coulomb.social/coulomb/railiance-platform.git + targetRevision: REPLACE_WITH_REVIEWED_COMMIT_SHA + path: argocd/platform-addons/custodian-allocation-review + destination: + server: https://kubernetes.default.svc + namespace: custodian-assessment + syncPolicy: + syncOptions: + - ApplyOutOfSyncOnly=true + - PruneLast=true diff --git a/scripts/vendor_custodian_allocation_review.py b/scripts/vendor_custodian_allocation_review.py new file mode 100644 index 0000000..cdf677e --- /dev/null +++ b/scripts/vendor_custodian_allocation_review.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Vendor or verify the CUST-WP-0071-T05 weekly allocation assessment sources. + +The CronJob mounts its code and data from a ConfigMap generated from +argocd/platform-addons/custodian-allocation-review/src/. Owner sources stay in +their repositories; this script copies them at an exact clean commit and writes +vendor.lock.json (repo, commit, path, sha256). The YAML ownership map is +rendered to JSON so the job needs no PyYAML. + + scripts/vendor_custodian_allocation_review.py # refresh from ~/ + scripts/vendor_custodian_allocation_review.py --check # lock matches src/ +""" +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +import subprocess +import sys + +ROOT = Path(__file__).resolve().parents[1] +ADDON = ROOT / "argocd/platform-addons/custodian-allocation-review" +SOURCES = { + "weekly_allocation_incluster.py": ("the-custodian", "scripts/weekly_allocation_incluster.py"), + "weekly_allocation_report.py": ("the-custodian", "scripts/weekly_allocation_report.py"), + "reconcile_railiance_allocation.py": ("the-custodian", "scripts/reconcile_railiance_allocation.py"), + "weekly-allocation-queries.json": ("the-custodian", "config/weekly-allocation-queries.json"), + "namespace-ownership.json": ("the-custodian", "docs/evidence/namespace-ownership.yaml"), + "observe_cluster_resources.py": ("railiance-cluster", "tools/observe_cluster_resources.py"), + "reef-railiance-k3s.json": ("resource-control", "data/resources/reef-railiance-k3s.json"), +} + + +def sha(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def git(repo: Path, *args: str) -> str: + return subprocess.check_output(["git", "-C", str(repo), *args], text=True).strip() + + +def refresh(base: Path) -> dict: + lock = {"schema": "railiance-platform.vendor-lock.v1", "workplan": "CUST-WP-0071-T05", "files": {}} + (ADDON / "src").mkdir(parents=True, exist_ok=True) + for name, (repo, path) in sorted(SOURCES.items()): + checkout = base / repo + if git(checkout, "status", "--porcelain", "--", path): + raise SystemExit(f"{repo}/{path} has uncommitted changes; commit it first") + commit = git(checkout, "rev-parse", "HEAD") + source = subprocess.check_output(["git", "-C", str(checkout), "show", f"{commit}:{path}"]) + data = source + if path.endswith(".yaml"): + import yaml + data = (json.dumps(yaml.safe_load(source), indent=2, sort_keys=True) + "\n").encode() + (ADDON / "src" / name).write_bytes(data) + lock["files"][name] = {"repo": repo, "commit": commit, "path": path, + "source_sha256": sha(source), "vendored_sha256": sha(data)} + (ADDON / "vendor.lock.json").write_text(json.dumps(lock, indent=2, sort_keys=True) + "\n") + return lock + + +def check() -> list[str]: + lock = json.loads((ADDON / "vendor.lock.json").read_text()) + problems = [] + if set(lock["files"]) != set(SOURCES): + problems.append("lock file set differs from SOURCES") + for name, entry in lock["files"].items(): + target = ADDON / "src" / name + if not target.exists() or sha(target.read_bytes()) != entry["vendored_sha256"]: + problems.append(f"{name}: vendored content does not match vendor.lock.json") + return problems + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--check", action="store_true") + parser.add_argument("--base", type=Path, default=Path.home(), help="directory holding the owner checkouts") + args = parser.parse_args() + if args.check: + problems = check() + for problem in problems: + print(problem, file=sys.stderr) + return 1 if problems else 0 + lock = refresh(args.base) + for name, entry in sorted(lock["files"].items()): + print(f"{name}: {entry['repo']}@{entry['commit'][:12]} {entry['vendored_sha256'][:12]}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_custodian_allocation_review.py b/tests/test_custodian_allocation_review.py new file mode 100644 index 0000000..769a0e4 --- /dev/null +++ b/tests/test_custodian_allocation_review.py @@ -0,0 +1,84 @@ +"""Offline guard for the CUST-WP-0071-T05 weekly allocation assessment addon. + +Asserts the founder's 2026-10-21 boundary: read-only get/list RBAC limited to +the runner's reads, Monday 08:00 Europe/Berlin, Forbid, suspended, retained +PVC that ArgoCD never deletes, not referenced by any live Application, and +vendored sources matching vendor.lock.json. +""" +from pathlib import Path +import subprocess +import sys + +import yaml + +ROOT = Path(__file__).resolve().parents[1] +ADDON = ROOT / "argocd/platform-addons/custodian-allocation-review" +APPS = ROOT / "argocd/railiance01/applications" +DRAFT = ROOT / "argocd/railiance01/drafts/custodian-allocation-review.application.yaml" +EXPECTED_RULES = { + ("", "nodes"), ("", "pods"), ("", "persistentvolumeclaims"), ("", "persistentvolumes"), + ("storage.k8s.io", "storageclasses"), ("metrics.k8s.io", "nodes"), ("metrics.k8s.io", "pods"), +} + + +def docs(): + kust = yaml.safe_load((ADDON / "kustomization.yaml").read_text()) + out = [] + for resource in kust["resources"]: + out.extend(d for d in yaml.safe_load_all((ADDON / resource).read_text()) if d) + return out + + +def one(kind): + found = [d for d in docs() if d["kind"] == kind] + assert len(found) == 1, kind + return found[0] + + +def test_rbac_is_read_only_and_exact(): + role = one("ClusterRole") + seen = set() + for rule in role["rules"]: + assert set(rule["verbs"]) == {"get", "list"} + assert "*" not in rule["apiGroups"] and "*" not in rule["resources"] + assert "resourceNames" not in rule and "nonResourceURLs" not in rule + seen |= {(g, r) for g in rule["apiGroups"] for r in rule["resources"]} + assert seen == EXPECTED_RULES + binding = one("ClusterRoleBinding") + assert binding["roleRef"]["name"] == role["metadata"]["name"] + assert binding["subjects"] == [{"kind": "ServiceAccount", "name": "custodian-allocation-review", + "namespace": "custodian-assessment"}] + assert not [d for d in docs() if d["kind"] in {"Role", "RoleBinding", "Secret"}] + + +def test_cronjob_schedule_suspended_and_hardened(): + spec = one("CronJob")["spec"] + assert spec["schedule"] == "0 8 * * 1" and spec["timeZone"] == "Europe/Berlin" + assert spec["suspend"] is True and spec["concurrencyPolicy"] == "Forbid" + pod = spec["jobTemplate"]["spec"]["template"]["spec"] + assert pod["serviceAccountName"] == "custodian-allocation-review" + assert pod["securityContext"]["runAsNonRoot"] is True + (container,) = pod["containers"] + assert "@sha256:" in container["image"] + assert container["securityContext"]["readOnlyRootFilesystem"] is True + assert container["securityContext"]["allowPrivilegeEscalation"] is False + assert container["resources"]["requests"]["cpu"] == "20m" + assert not any("secret" in v for v in pod["volumes"]) + claim = next(v for v in pod["volumes"] if "persistentVolumeClaim" in v)["persistentVolumeClaim"]["claimName"] + pvc = one("PersistentVolumeClaim") + assert pvc["metadata"]["name"] == claim + assert pvc["metadata"]["annotations"]["argocd.argoproj.io/sync-options"] == "Prune=false,Delete=false" + + +def test_not_registered_with_any_live_application(): + for path in APPS.glob("*.y*ml"): + assert "custodian-allocation-review" not in path.read_text(), path + draft = yaml.safe_load(DRAFT.read_text()) + assert "automated" not in draft["spec"].get("syncPolicy", {}) + assert "finalizers" not in draft["metadata"] + + +def test_vendored_sources_match_lock(): + result = subprocess.run([sys.executable, str(ROOT / "scripts/vendor_custodian_allocation_review.py"), "--check"], + capture_output=True, text=True) + assert result.returncode == 0, result.stderr