Structured audit events and operation counters for the API.
An audit event identifies a request, not a person: the API has no user identity, and the shared
operator bearer is not one. Events are ordinary log lines; shipping, retention and tamper evidence
belong to the deployment's log pipeline.
AuditMiddleware
Give every request an ID and record operations and refusals once their outcome is known.
Source code in spark_on_k8s/api/audit.py
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171 | class AuditMiddleware:
"""Give every request an ID and record operations and refusals once their outcome is known."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] not in {"http", "websocket"}:
return await self.app(scope, receive, send)
# Always generated here: a client-supplied ID could collide with or impersonate another.
request_id = scope[REQUEST_ID_SCOPE_KEY] = uuid4().hex
record: dict = {}
response: dict = {}
async def send_with_request_id(message):
if message["type"] == "http.response.start":
response.setdefault("status", message["status"])
headers = [(k, v) for k, v in message.get("headers", []) if k.lower() != b"x-request-id"]
message = {**message, "headers": [*headers, (b"x-request-id", request_id.encode())]}
elif message["type"] == "websocket.accept":
response.setdefault("status", 101)
elif message["type"] == "websocket.close":
# Closing before accepting rejects the handshake; servers answer it with HTTP 403.
response.setdefault("status", 403)
await send(message)
token = _record.set(record)
try:
await self.app(scope, receive, send_with_request_id)
except Exception as error:
count_kubernetes_error(error)
response.setdefault("status", 500)
raise
finally:
_record.reset(token)
_finish(scope, request_id, record, response.get("status"))
|
Write audit lines to stdout unless the deployment already routes this logger.
Source code in spark_on_k8s/api/audit.py
80
81
82
83
84
85
86
87
88
89 | def configure_audit_log() -> None:
"""Write audit lines to stdout unless the deployment already routes this logger."""
if not get_audit_settings().enabled:
return
if logger.level == logging.NOTSET:
logger.setLevel(logging.INFO)
if not logger.hasHandlers():
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
|
count_kubernetes_error(error)
Count a Kubernetes API error response by status class; other exceptions are ignored.
Source code in spark_on_k8s/api/audit.py
| def count_kubernetes_error(error: BaseException) -> None:
"""Count a Kubernetes API error response by status class; other exceptions are ignored."""
if isinstance(error, ApiException | AsyncApiException):
status = error.status if isinstance(error.status, int) else 0
with _lock:
_kubernetes_errors[f"{status // 100}xx" if 400 <= status < 600 else "other"] += 1
|
metric_lines()
Prometheus text lines for this worker process's counters; families appear once counted.
Source code in spark_on_k8s/api/audit.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133 | def metric_lines() -> list[str]:
"""Prometheus text lines for this worker process's counters; families appear once counted."""
with _lock:
events, errors = sorted(_events.items()), sorted(_kubernetes_errors.items())
lines = []
if events:
lines += [
"# HELP spark_on_k8s_api_audit_events_total Audited API operations and access refusals "
"by event, outcome and code, in this worker process.",
"# TYPE spark_on_k8s_api_audit_events_total counter",
]
lines += [
f'spark_on_k8s_api_audit_events_total{{event="{event}",outcome="{outcome}",code="{code}"}} '
f"{count}"
for (event, outcome, code), count in events
]
if errors:
lines += [
"# HELP spark_on_k8s_api_kubernetes_errors_total Kubernetes API error responses while "
"serving API requests, by status class, in this worker process.",
"# TYPE spark_on_k8s_api_kubernetes_errors_total counter",
]
lines += [
f'spark_on_k8s_api_kubernetes_errors_total{{status_class="{status_class}"}} {count}'
for status_class, count in errors
]
return lines
|
note(**fields)
Add what the current request is known to target or return; a no-op outside a request.
Source code in spark_on_k8s/api/audit.py
| def note(**fields) -> None:
"""Add what the current request is known to target or return; a no-op outside a request."""
record = _record.get()
if record is not None:
record.update(fields)
|