Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The broker's operational surface (`Admin`): health, readiness, metrics, the tenant list, size and purge, and the dead-letter admin.
3 :
4 : use super::*;
5 :
6 : /// The broker's own operational surface: health, readiness, metrics, the
7 : /// tenant list and purge, and the dead-letter admin. Health and readiness
8 : /// are always open; the rest sits behind the admin credential when the
9 : /// state carries one (`admin_auth.rs`), `/q/metrics` unless the
10 : /// deployment left it open for a scraper.
11 : pub struct Admin;
12 :
13 : impl Admin {
14 : /// Every path this surface mounts, and the only source of the route
15 : /// count `/q/health` reports. The destructuring in `router` binds one
16 : /// name per entry, so a route added to one and not the other does not
17 : /// compile.
18 : pub(crate) const PATHS: [&'static str; 8] = [
19 : "/health",
20 : "/ready",
21 : "/metrics",
22 : "/tenants",
23 : "/tenants/{tenant}",
24 : "/dead-letters",
25 : "/dead-letters/{id}",
26 : "/dead-letters/{id}/replay",
27 : ];
28 : }
29 :
30 : impl ApiSurface for Admin {
31 88 : fn name(&self) -> &str {
32 88 : "admin"
33 88 : }
34 :
35 19026 : fn prefix(&self) -> &str {
36 19026 : "/q"
37 19026 : }
38 :
39 18848 : fn router(&self, st: AppState) -> Router<AppState> {
40 18848 : let [health_p, ready_p, metrics_p, tenants_p, tenant_p, dl_p, dl_id_p, replay_p] =
41 18848 : Self::PATHS;
42 18848 : let guard = axum::middleware::from_fn_with_state(st.clone(), admin_auth::require);
43 18848 : let metrics = Self::metrics_routes(&st, metrics_p);
44 18848 : let routes = Router::new()
45 18848 : .route(tenants_p, get(tenants_list))
46 18848 : .route(tenant_p, get(tenant_get).delete(tenant_purge))
47 18848 : .route(dl_p, get(dead_letters_list))
48 18848 : .route(dl_id_p, delete(dead_letter_delete))
49 18848 : .route(replay_p, post(dead_letter_replay));
50 18848 : routes
51 18848 : .route_layer(guard)
52 18848 : .route(health_p, get(health))
53 18848 : .route(ready_p, get(ready))
54 18848 : .merge(metrics)
55 18848 : }
56 :
57 86 : fn version_info(&self) -> serde_json::Value {
58 86 : serde_json::json!({"routes": Self::PATHS.len()})
59 86 : }
60 : }
61 :
62 : impl Admin {
63 : /// `/q/metrics` alone, under the credential rule of the rest of the
64 : /// surface: the admin router mounts it, and so does the metrics listener
65 : /// (`metrics_router`). Prometheus text format; 404 until the
66 : /// broker installs the renderer, the api crate never depends on an
67 : /// exporter.
68 18848 : fn metrics_routes(st: &AppState, path: &str) -> Router<AppState> {
69 18848 : let metrics = Router::new().route(path, get(metrics_endpoint));
70 18848 : if st.admin_auth.as_ref().is_some_and(|a| !a.metrics_open) {
71 102 : metrics.route_layer(axum::middleware::from_fn_with_state(
72 102 : st.clone(),
73 : admin_auth::require,
74 : ))
75 : } else {
76 18746 : metrics
77 : }
78 18848 : }
79 :
80 : /// The metrics listener's whole surface (ANTARES_METRICS_PORT):
81 : /// `/q/metrics` and nothing else, so a NetworkPolicy that opens that
82 : /// port to a scraper opens no API or admin path.
83 0 : pub fn metrics_router(state: AppState) -> Router {
84 0 : let [_, _, metrics_p, ..] = Self::PATHS;
85 0 : Router::new()
86 0 : .nest(Admin.prefix(), Self::metrics_routes(&state, metrics_p))
87 0 : .with_state(state)
88 0 : }
89 : }
90 :
91 : /// /q/metrics — Prometheus exposition, rendered by the closure the
92 : /// broker installed; 404 when no recorder exists (tests, embedded builds).
93 18 : pub(super) async fn metrics_endpoint(
94 18 : axum::extract::State(state): axum::extract::State<AppState>,
95 18 : ) -> Response {
96 18 : match &state.metrics_render {
97 0 : Some(render) => Response::builder()
98 0 : .status(axum::http::StatusCode::OK)
99 0 : .header(
100 0 : axum::http::header::CONTENT_TYPE,
101 : "text/plain; version=0.0.4",
102 : )
103 0 : .body(axum::body::Body::from(render()))
104 0 : .unwrap_or_else(|_| axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()),
105 18 : None => axum::http::StatusCode::NOT_FOUND.into_response(),
106 : }
107 18 : }
108 :
109 : /// The build-time git hash (build.rs) — re-exported for --version.
110 : pub const GIT_HASH: &str = env!("ANTARES_GIT_HASH");
111 :
112 : /// Liveness plus the deployment's own description: which drivers are
113 : /// running, what they are, every configured cap with its rejection counters,
114 : /// the mounted surfaces and the notification schemes they can deliver to.
115 : /// 503 while draining, so a load balancer stops routing here before the
116 : /// listener stops accepting.
117 86 : pub(super) async fn health(
118 86 : axum::extract::State(state): axum::extract::State<AppState>,
119 86 : headers: axum::http::HeaderMap,
120 86 : ) -> (StatusCode, axum::Json<serde_json::Value>) {
121 : // Drain: the load balancer decides on the STATUS CODE, so draining has
122 : // to be a 503 — a 200 body saying "DRAINING" would keep traffic arriving.
123 : // This flips BEFORE the listener stops accepting, which is the whole point
124 : // of the ordering: the LB must stop routing while the socket still works.
125 86 : let draining = state.draining.load(std::sync::atomic::Ordering::Relaxed);
126 86 : let code = if draining {
127 2 : StatusCode::SERVICE_UNAVAILABLE
128 : } else {
129 84 : StatusCode::OK
130 : };
131 86 : let mut body = serde_json::Map::new();
132 86 : body.insert(
133 86 : "status".into(),
134 86 : (if draining { "DRAINING" } else { "UP" }).into(),
135 : );
136 86 : body.insert("store".into(), state.store_name.clone().into());
137 86 : body.insert(
138 86 : "temporal".into(),
139 86 : state.temporal_name.as_deref().unwrap_or("none").into(),
140 : );
141 : // Version surface: workspace version + build-time git hash
142 : // (build.rs), both asserted against the checkout by health_is_up.
143 86 : body.insert("version".into(), env!("CARGO_PKG_VERSION").into());
144 86 : body.insert("commit".into(), env!("ANTARES_GIT_HASH").into());
145 : // What each driver runs on — engine, server version, extensions — so an
146 : // operator can tell two deployments of the same backend name apart. Both
147 : // seams answer from state captured at startup; a driver with nothing to
148 : // add answers an empty object and the key stays off the body.
149 172 : for (key, info) in [
150 86 : ("storeInfo", state.store.version_info()),
151 86 : ("temporalInfo", state.temporal.version_info()),
152 86 : ] {
153 172 : if info.as_object().is_some_and(|m| !m.is_empty()) {
154 162 : body.insert(key.into(), info);
155 162 : }
156 : }
157 : // Where commits serialize behind one writer, the queue depth (current,
158 : // peak) is the signal that decides the group-commit lever. A driver
159 : // without such a committer reports nothing, so the branch is the
160 : // driver's, not a backend name read here.
161 86 : if let Some((depth, peak)) = state.store.commit_queue() {
162 2 : body.insert("commitQueueDepth".into(), depth.into());
163 2 : body.insert("commitQueuePeak".into(), peak.into());
164 84 : }
165 : // The policy engine this binary was started with, and how long it has
166 : // to answer before the seam denies for it (ADR-0020). An operator
167 : // reading `allow-all` here knows no engine is attached.
168 86 : let mut policy = serde_json::Map::new();
169 86 : policy.insert("engine".into(), state.policy_name().into());
170 86 : policy.insert(
171 86 : "timeoutMs".into(),
172 86 : u64::try_from(policy::TIMEOUT.as_millis())
173 86 : .unwrap_or(u64::MAX)
174 86 : .into(),
175 : );
176 : // Configured caps + rejection counters, for observability.
177 86 : body.insert("limits".into(), state.limits.snapshot());
178 : // History events a driver failed to record after the write stood.
179 86 : body.insert("temporalDrainErrors".into(), history::drain_errors().into());
180 : // What this binary serves beside the NGSI-LD API: one entry per mounted
181 : // surface, its prefix and whatever the surface reports about itself.
182 86 : body.insert(
183 86 : "surfaces".into(),
184 : serde_json::Value::Object(
185 86 : state
186 86 : .surfaces
187 86 : .iter()
188 94 : .map(|s| {
189 94 : let mut info = match s.version_info() {
190 94 : serde_json::Value::Object(m) => m,
191 0 : other => {
192 0 : let mut m = serde_json::Map::new();
193 0 : m.insert("info".into(), other);
194 0 : m
195 : }
196 : };
197 94 : info.insert("prefix".into(), s.prefix().into());
198 94 : (s.name().to_owned(), serde_json::Value::Object(info))
199 94 : })
200 86 : .collect(),
201 : ),
202 : );
203 : // Notification endpoint schemes this deployment can deliver notifications to: the
204 : // registered bindings (6.3.8, clause 7, and any a deployment added).
205 : // A subscription naming a scheme absent here is refused at creation.
206 86 : body.insert("notificationSchemes".into(), state.sinks.schemes().into());
207 : // Notifications the delivery policy gave up on (this process, since start).
208 86 : body.insert("deadLetters".into(), notify::dead_letters_written().into());
209 : // Each stored view, loaded or invalid with the reason.
210 : // only the view reports below read the credential
211 86 : drop(headers);
212 : // Changes the bounded matcher queue dropped (this process, since start).
213 86 : body.insert("changesDropped".into(), notify::changes_dropped().into());
214 : // Panics absorbed at the notification-task boundary, each one a change
215 : // whose notification was lost. Reported here and not only through the
216 : // metrics facade: the recorder is installed by ANTARES_TELEMETRY, which
217 : // is off by default, so this endpoint is where a default deployment can
218 : // see it at all.
219 86 : body.insert("taskPanics".into(), notify::task_panics().into());
220 : // Jemalloc heap stats (RSS ≈ live×1.2 is the target).
221 86 : if let Some(mem) = &state.mem_stats {
222 16 : body.insert("memory".into(), mem());
223 70 : }
224 : // Bus visibility: present only when the nats wiring installed the
225 : // provider — bus=local carries NO bus member.
226 86 : if let Some(bus) = &state.bus_stats {
227 4 : body.insert("bus".into(), bus());
228 82 : }
229 86 : body.insert("policy".into(), serde_json::Value::Object(policy));
230 86 : (code, axum::Json(serde_json::Value::Object(body)))
231 86 : }
232 :
233 : /// The tenant an admin dead-letter call addresses: `?tenant=` (default
234 : /// tenant when absent), grammar-checked, never a reserved internal one.
235 104 : pub(super) fn admin_tenant(
236 104 : q: &std::collections::HashMap<String, String>,
237 104 : ) -> Result<TenantId, NgsiError> {
238 104 : let raw = q.get("tenant").map_or(TenantId::DEFAULT, String::as_str);
239 104 : let bad = || NgsiError::BadRequestData(format!("invalid tenant: {raw:?}"));
240 104 : if reserved_tenant(raw) {
241 8 : return Err(bad());
242 96 : }
243 96 : TenantId::new(raw).map_err(|_| bad())
244 104 : }
245 :
246 : /// One dead letter as the listing may show it: everything on it that can be
247 : /// a credential is blanked, the keys are not.
248 : ///
249 : /// `receiverInfo` and `notifierInfo` are KeyValuePair arrays (5.2.22), and
250 : /// the example content the datatype names is the HTTP Authentication header;
251 : /// a letter written before the bindings moved behind the registry carries the
252 : /// same values already rendered into `headers`. `notifierInfo` is opaque to
253 : /// every sink but the endpoint's own, so the broker cannot tell which of its
254 : /// keys names a secret and blanks them all. The stored letter keeps the
255 : /// values a replay has to send — only what leaves through this route loses
256 : /// them.
257 28 : pub(super) fn redact_letter(l: &mut serde_json::Value) {
258 : // Assigning through `Value`'s index panics on anything but an object,
259 : // and the row comes from the store rather than from this process.
260 28 : let Some(letter) = l.as_object_mut() else {
261 8 : return;
262 : };
263 20 : if let Some(u) = letter.get("uri").and_then(serde_json::Value::as_str) {
264 20 : let redacted = serde_json::Value::String(notify::redact_userinfo(u));
265 20 : letter.insert("uri".into(), redacted);
266 20 : }
267 60 : for member in ["receiverInfo", "notifierInfo", "headers"] {
268 60 : let Some(pairs) = letter
269 60 : .get_mut(member)
270 60 : .and_then(serde_json::Value::as_array_mut)
271 : else {
272 36 : continue;
273 : };
274 42 : for pair in pairs {
275 42 : if let Some(v) = pair.get_mut(1) {
276 42 : *v = serde_json::Value::String("[redacted]".into());
277 42 : }
278 : }
279 : }
280 28 : }
281 :
282 : /// Dead letters of one tenant, newest first; `subscription=` narrows to one
283 : /// subscription, `limit=` bounds the page (`ANTARES_DEFAULT_LIMIT` unless
284 : /// named, 403 TooManyResults past `ANTARES_MAX_LIMIT`). The endpoint URI is
285 : /// shown with its userinfo redacted and every credential the letter carries
286 : /// is blanked.
287 52 : pub(super) async fn dead_letters_list(
288 52 : axum::extract::State(st): axum::extract::State<AppState>,
289 52 : axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
290 52 : ) -> Response {
291 52 : let tenant = match admin_tenant(&q) {
292 44 : Ok(t) => t,
293 8 : Err(e) => return ApiError::from(e).into_response(),
294 : };
295 44 : let limit = match q.get("limit") {
296 32 : None => st.default_limit,
297 12 : Some(v) => match v.parse::<usize>() {
298 8 : Ok(n) if n > st.max_limit => {
299 2 : return ApiError::from(NgsiError::TooManyResults(format!(
300 2 : "limit {n} exceeds the server maximum {}",
301 2 : st.max_limit
302 2 : )))
303 2 : .into_response()
304 : }
305 6 : Ok(n) if n > 0 => n,
306 : _ => {
307 6 : return ApiError::from(NgsiError::BadRequestData(format!(
308 6 : "limit must be a positive integer, got {v:?}"
309 6 : )))
310 6 : .into_response()
311 : }
312 : },
313 : };
314 36 : let mut letters = match st
315 36 : .store
316 36 : .list(&tenant, antares_store::Kind::DeadLetter)
317 36 : .await
318 : {
319 36 : Ok(l) => l,
320 0 : Err(e) => return ApiError::from(e).into_response(),
321 : };
322 36 : if let Some(sid) = q.get("subscription") {
323 6 : letters.retain(|l| l["subscriptionId"].as_str() == Some(sid.as_str()));
324 34 : }
325 36 : letters.sort_by(|a, b| b["lastAt"].as_str().cmp(&a["lastAt"].as_str()));
326 36 : letters.truncate(limit);
327 36 : for l in &mut letters {
328 28 : redact_letter(l);
329 28 : }
330 36 : (
331 36 : StatusCode::OK,
332 36 : axum::Json(serde_json::Value::Array(letters)),
333 36 : )
334 36 : .into_response()
335 52 : }
336 :
337 : /// Re-deliver one dead letter once through its own binding: 204 and the
338 : /// letter is gone on success; 502 with the failure text and the letter
339 : /// kept (attempt history extended) otherwise.
340 34 : pub(super) async fn dead_letter_replay(
341 34 : axum::extract::State(st): axum::extract::State<AppState>,
342 34 : axum::extract::Path(id): axum::extract::Path<String>,
343 34 : axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
344 34 : ) -> Response {
345 : use antares_store::{CurrentStateDriverExt as _, Kind};
346 34 : let tenant = match admin_tenant(&q) {
347 32 : Ok(t) => t,
348 2 : Err(e) => return ApiError::from(e).into_response(),
349 : };
350 32 : let letter = match st.store.get(&tenant, Kind::DeadLetter, &id).await {
351 16 : Ok(Some(l)) => l,
352 : Ok(None) => {
353 16 : return ApiError::from(NgsiError::ResourceNotFound(format!("dead letter {id}")))
354 16 : .into_response()
355 : }
356 0 : Err(e) => return ApiError::from(e).into_response(),
357 : };
358 16 : match notify::replay_dead_letter(&st, &tenant, &letter).await {
359 4 : Ok(()) => match st.store.delete(&tenant, Kind::DeadLetter, &id).await {
360 4 : Ok(_) => StatusCode::NO_CONTENT.into_response(),
361 0 : Err(e) => ApiError::from(e).into_response(),
362 : },
363 12 : Err(why) => {
364 12 : let ts = state::now_iso();
365 12 : let w = why.clone();
366 12 : let _ = st
367 12 : .store
368 12 : .mutate(&tenant, Kind::DeadLetter, &id, move |d| {
369 12 : let Some(letter) = d.as_object_mut() else {
370 2 : return Ok::<(), NgsiError>(());
371 : };
372 10 : let n = letter
373 10 : .get("attempts")
374 10 : .and_then(serde_json::Value::as_u64)
375 10 : .unwrap_or(0)
376 10 : .saturating_add(1);
377 10 : letter.insert("attempts".into(), n.into());
378 10 : letter.insert("lastError".into(), serde_json::Value::String(w));
379 10 : letter.insert("lastAt".into(), serde_json::Value::String(ts));
380 10 : Ok::<(), NgsiError>(())
381 12 : })
382 12 : .await;
383 12 : (
384 12 : StatusCode::BAD_GATEWAY,
385 12 : axum::Json(serde_json::json!({"detail": why, "id": id})),
386 12 : )
387 12 : .into_response()
388 : }
389 : }
390 34 : }
391 :
392 : /// Drop one dead letter without replaying it: 204 when it was there, 404
393 : /// when it was not. The letter is the only copy of a notification the
394 : /// delivery policy gave up on, so this is the one route that discards it.
395 18 : pub(super) async fn dead_letter_delete(
396 18 : axum::extract::State(st): axum::extract::State<AppState>,
397 18 : axum::extract::Path(id): axum::extract::Path<String>,
398 18 : axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
399 18 : ) -> Response {
400 18 : let tenant = match admin_tenant(&q) {
401 16 : Ok(t) => t,
402 2 : Err(e) => return ApiError::from(e).into_response(),
403 : };
404 16 : match st
405 16 : .store
406 16 : .delete(&tenant, antares_store::Kind::DeadLetter, &id)
407 16 : .await
408 : {
409 2 : Ok(true) => StatusCode::NO_CONTENT.into_response(),
410 : Ok(false) => {
411 14 : ApiError::from(NgsiError::ResourceNotFound(format!("dead letter {id}"))).into_response()
412 : }
413 0 : Err(e) => ApiError::from(e).into_response(),
414 : }
415 18 : }
416 :
417 : /// Tenant inventory: the names the backends know, sorted. Names only —
418 : /// at the 10 000-tenant target (ADR-0001) a list carrying per-kind counts
419 : /// would cost a count per kind per tenant, so a client picks a name here and
420 : /// reads its detail from `GET /q/tenants/{tenant}`. Admin surface, never
421 : /// under the API root.
422 28 : pub(super) async fn tenants_list(
423 28 : axum::extract::State(st): axum::extract::State<AppState>,
424 28 : ) -> Response {
425 28 : match st.store.tenant_ids().await {
426 28 : Ok(ids) => (StatusCode::OK, axum::Json(ids)).into_response(),
427 0 : Err(e) => crate::negotiate::ApiError::from(e).into_response(),
428 : }
429 28 : }
430 :
431 : /// What one tenant holds: 400 for a name no request could carry, 404 when it
432 : /// does not exist (5.5.10 keeps the default Tenant existing even when
433 : /// empty), 200 with its counts otherwise. The path names the tenant; the
434 : /// NGSILD-Tenant header never redirects the read.
435 50 : pub(super) async fn tenant_get(
436 50 : axum::extract::State(st): axum::extract::State<AppState>,
437 50 : axum::extract::Path(raw): axum::extract::Path<String>,
438 50 : ) -> Response {
439 : use crate::negotiate::ApiError;
440 50 : let bad = || {
441 10 : ApiError::from(NgsiError::BadRequestData(format!(
442 10 : "invalid tenant: {raw:?}"
443 10 : )))
444 10 : .into_response()
445 10 : };
446 : // the broker's internal snapshot tenants are not addressable
447 50 : if reserved_tenant(&raw) {
448 8 : return bad();
449 42 : }
450 42 : let Ok(tenant) = TenantId::new(&raw) else {
451 2 : return bad();
452 : };
453 40 : let stats = match st.store.tenant_stats_one(&tenant).await {
454 32 : Ok(Some(s)) => s,
455 : Ok(None) => {
456 8 : return ApiError::from(NgsiError::ResourceNotFound(format!("tenant {raw}")))
457 8 : .into_response()
458 : }
459 0 : Err(e) => return ApiError::from(e).into_response(),
460 : };
461 : // a tenant with a long history counts for seconds: bounded like the
462 : // size scans, and a count that did not finish is left out, never 0
463 32 : let instances = match st
464 32 : .temporal
465 32 : .attr_instance_count_within(&tenant, TENANT_COUNT_BOUND)
466 32 : .await
467 : {
468 30 : Ok(n) => Some(n),
469 2 : Err(e) => {
470 2 : tracing::warn!(tenant = tenant.as_str(), error = %e, "tenant history not counted");
471 2 : None
472 : }
473 : };
474 32 : let bytes = tenant_bytes(&st, &tenant).await;
475 32 : let mut row = serde_json::json!({
476 32 : "tenant": stats.tenant,
477 32 : "counts": {
478 32 : "entities": stats.entities,
479 32 : "subscriptions": stats.subscriptions,
480 32 : "csourceSubscriptions": stats.csource_subscriptions,
481 32 : "registrations": stats.registrations,
482 32 : "snapshots": stats.snapshots,
483 32 : "entityMaps": stats.entity_maps,
484 32 : "distSubs": stats.dist_subs,
485 : },
486 : });
487 30 : if let (Some(n), Some(counts)) = (
488 32 : instances,
489 32 : row.get_mut("counts")
490 32 : .and_then(serde_json::Value::as_object_mut),
491 30 : ) {
492 30 : counts.insert("attrInstances".into(), n.into());
493 30 : }
494 32 : if let Some(row) = row.as_object_mut() {
495 32 : if let Some(c) = stats.created_at {
496 0 : row.insert("createdAt".into(), serde_json::Value::String(c));
497 32 : }
498 32 : if let Some(b) = bytes {
499 32 : row.insert("bytes".into(), b);
500 32 : }
501 0 : }
502 32 : (StatusCode::OK, axum::Json(row)).into_response()
503 50 : }
504 :
505 : /// How long `GET /q/tenants/{tenant}` waits for the history count,
506 : /// the same bound as one size scan.
507 : pub const TENANT_COUNT_BOUND: std::time::Duration = std::time::Duration::from_secs(5);
508 :
509 : /// How long `GET /q/tenants/{tenant}` reuses a measured size.
510 : pub const TENANT_BYTES_TTL: std::time::Duration = std::time::Duration::from_secs(300);
511 : /// How long a size that could not be measured, or is being measured, stays
512 : /// unanswered before the next read tries again.
513 : pub const TENANT_BYTES_RETRY: std::time::Duration = std::time::Duration::from_secs(60);
514 :
515 : /// One tenant's entry in [`crate::state::Shared::tenant_bytes`]: when it was
516 : /// written and the size, `None` while it is measured or after it failed.
517 : pub type BytesEntry = (std::time::Instant, Option<serde_json::Value>);
518 :
519 : /// What a size read takes from the cache: `Some(answer)` while an entry is
520 : /// fresh (a measured size for [`TENANT_BYTES_TTL`], a failure or a
521 : /// measurement in flight for [`TENANT_BYTES_RETRY`]), or `None` after
522 : /// claiming the measurement — the claim is an in-flight entry, so concurrent
523 : /// misses scan once and the others answer without a size.
524 60 : pub(super) fn bytes_cached_or_claim(
525 60 : map: &mut std::collections::HashMap<String, BytesEntry>,
526 60 : tenant: &str,
527 60 : now: std::time::Instant,
528 60 : ) -> Option<Option<serde_json::Value>> {
529 60 : if let Some((at, v)) = map.get(tenant) {
530 22 : let ttl = if v.is_some() {
531 10 : TENANT_BYTES_TTL
532 : } else {
533 12 : TENANT_BYTES_RETRY
534 : };
535 22 : if now.saturating_duration_since(*at) < ttl {
536 14 : return Some(v.clone());
537 8 : }
538 38 : }
539 46 : map.insert(tenant.to_owned(), (now, None));
540 46 : None
541 60 : }
542 :
543 : /// `{entities, attrInstances, byType}` in bytes, measured in the store under
544 : /// the store's own statement bound and kept as [`bytes_cached_or_claim`]
545 : /// says. `None` when the backend cannot measure, the scan failed or another
546 : /// read is measuring: the counts still answer, without a size.
547 32 : pub(super) async fn tenant_bytes(st: &AppState, tenant: &TenantId) -> Option<serde_json::Value> {
548 62 : let cache = || {
549 62 : st.tenant_bytes
550 62 : .lock()
551 62 : .unwrap_or_else(std::sync::PoisonError::into_inner)
552 62 : };
553 2 : if let Some(answer) =
554 32 : bytes_cached_or_claim(&mut cache(), tenant.as_str(), std::time::Instant::now())
555 : {
556 2 : return answer;
557 30 : }
558 30 : let measured = async {
559 30 : let entities = st.store.tenant_bytes(tenant).await?;
560 30 : let instances = st.temporal.attr_instance_bytes(tenant).await?;
561 30 : Ok::<_, NgsiError>(serde_json::json!({
562 30 : "entities": entities.entities,
563 30 : "attrInstances": instances,
564 30 : "byType": entities.by_type,
565 30 : }))
566 30 : }
567 30 : .await;
568 30 : let v = match measured {
569 30 : Ok(v) => Some(v),
570 0 : Err(NgsiError::OperationNotSupported(_)) => None,
571 0 : Err(e) => {
572 0 : tracing::warn!(tenant = tenant.as_str(), error = %e, "tenant size not measured");
573 0 : None
574 : }
575 : };
576 30 : cache().insert(
577 30 : tenant.as_str().to_owned(),
578 30 : (std::time::Instant::now(), v.clone()),
579 30 : );
580 30 : v
581 32 : }
582 :
583 : /// Purge one tenant from both backends: 400 for a name no request could
584 : /// carry, 404 when it does not exist (5.5.10), 409 while it still holds
585 : /// distributed subscriptions (unsubscribe them first), 204 once every
586 : /// document of it is gone. Deleted subscriptions and registrations are
587 : /// announced to the bus mirrors the way an API delete is.
588 58 : pub(super) async fn tenant_purge(
589 58 : axum::extract::State(st): axum::extract::State<AppState>,
590 58 : axum::extract::Path(raw): axum::extract::Path<String>,
591 58 : ) -> Response {
592 : use crate::negotiate::ApiError;
593 : use antares_store::Kind;
594 58 : let bad = || {
595 18 : ApiError::from(NgsiError::BadRequestData(format!(
596 18 : "invalid tenant: {raw:?}"
597 18 : )))
598 18 : .into_response()
599 18 : };
600 58 : if reserved_tenant(&raw) {
601 8 : return bad();
602 50 : }
603 50 : let Ok(tenant) = TenantId::new(&raw) else {
604 10 : return bad();
605 : };
606 40 : match st.store.tenant_exists(&tenant).await {
607 30 : Ok(true) => {}
608 : Ok(false) => {
609 10 : return ApiError::from(NgsiError::ResourceNotFound(format!("tenant {raw}")))
610 10 : .into_response()
611 : }
612 0 : Err(e) => return ApiError::from(e).into_response(),
613 : }
614 : // The paged walk, not `list`: the all-at-once read carries a ceiling
615 : // (the Postgres arm refuses past MAX_UNDECIDED_ROWS), and a purge behind
616 : // that ceiling means the Tenants most worth reclaiming are the ones that
617 : // can never be reclaimed — while their rows stay readable to anyone
618 : // sending the Tenant header.
619 84 : async fn ids(st: &AppState, tenant: &TenantId, kind: Kind) -> Result<Vec<String>, NgsiError> {
620 84 : let mut out = Vec::new();
621 84 : crate::csource::walk_docs(st, tenant, kind, |doc| {
622 20 : if let Some(id) = doc.get("id").and_then(|v| v.as_str()) {
623 20 : out.push(id.to_owned());
624 20 : }
625 20 : Ok(())
626 20 : })
627 84 : .await?;
628 84 : Ok(out)
629 84 : }
630 30 : let run = async {
631 : // 5.8.1.4 stores one mapping document per distributed Subscription;
632 : // its `remotes` names the subscription copies that live AT context
633 : // sources. Those are deleted at their source by deleting the
634 : // Subscription (5.8.5.4), which a purge does not do, so a tenant
635 : // still holding them is refused rather than orphaning them. The
636 : // mapping shell of a Subscription no source has matched yet, and
637 : // the internal Registration Subscription beside it, hold nothing
638 : // remote and block nothing — the id-shaped read this used to do
639 : // saw neither, since a mapping document carries no `id` member.
640 30 : let mut remote_copies = false;
641 38 : crate::csource::walk_docs(&st, &tenant, Kind::DistSub, |doc| {
642 34 : if doc
643 34 : .get("remotes")
644 34 : .and_then(serde_json::Value::as_object)
645 34 : .is_some_and(|m| !m.is_empty())
646 2 : {
647 2 : remote_copies = true;
648 32 : }
649 34 : Ok(())
650 34 : })
651 30 : .await?;
652 30 : if remote_copies {
653 2 : return Err(NgsiError::Conflict(
654 2 : "tenant holds active distributed subscriptions".into(),
655 2 : ));
656 28 : }
657 28 : let subs = ids(&st, &tenant, Kind::Subscription).await?;
658 28 : let csubs = ids(&st, &tenant, Kind::CSourceSubscription).await?;
659 28 : let regs = ids(&st, &tenant, Kind::Registration).await?;
660 : // 5.2.41: a Snapshot's isolated copy does not live under the Tenant.
661 : // It lives under the synthetic tenant its internal `__tenant` member
662 : // names, and the Snapshot document is the only pointer to it — so the
663 : // two purges below would delete that pointer and leave the copy
664 : // behind, reachable by nothing and freeable by nothing. Each snapshot
665 : // goes through the same removal a DELETE of it does, first.
666 28 : let mut snaps = Vec::new();
667 28 : crate::csource::walk_docs(&st, &tenant, Kind::Snapshot, |meta| {
668 2 : snaps.push(meta);
669 2 : Ok(())
670 2 : })
671 28 : .await?;
672 28 : for meta in &snaps {
673 2 : if let Some(id) = meta.get("id").and_then(|v| v.as_str()) {
674 2 : crate::snapshots::snap_remove(&st, &tenant, id, meta).await;
675 0 : }
676 : }
677 : // 5.13.1: a Hosted @context belongs to the Tenant that stored it.
678 : // Its rows go first, while the store still lists them: this pass
679 : // also drops the loader's warm copy of each, which the store purge
680 : // below cannot reach (run after it, the pass listed
681 : // nothing and a Tenant recreated under the name resolved the old
682 : // term mappings).
683 28 : crate::contexts::purge_tenant(&st, &tenant).await?;
684 28 : st.temporal.purge_tenant(&tenant).await?;
685 28 : st.store.purge_tenant(&tenant).await?;
686 28 : st.known_tenants.forget(&tenant);
687 28 : if let Some(sync) = &st.sub_sync {
688 : // Kept apart: the mirror entry is keyed per kind, so one id
689 : // naming both a Subscription and a Context Source Registration
690 : // Subscription needs a tombstone for each.
691 18 : for id in &subs {
692 14 : sync(&tenant, Kind::Subscription, id, None);
693 14 : }
694 18 : for id in &csubs {
695 0 : sync(&tenant, Kind::CSourceSubscription, id, None);
696 0 : }
697 10 : }
698 28 : if let Some(sync) = &st.reg_sync {
699 18 : for id in ®s {
700 0 : sync(&tenant, id, None);
701 0 : }
702 10 : }
703 28 : Ok::<(), NgsiError>(())
704 30 : };
705 30 : if let Err(e) = run.await {
706 2 : return ApiError::from(e).into_response();
707 28 : }
708 28 : st.tenant_bytes
709 28 : .lock()
710 28 : .unwrap_or_else(std::sync::PoisonError::into_inner)
711 28 : .remove(tenant.as_str());
712 28 : StatusCode::NO_CONTENT.into_response()
713 58 : }
714 :
715 : /// Readiness, distinct from `/q/health` liveness: ready = not draining AND
716 : /// the store answers a trivial request AND (when bus=nats) the bus is
717 : /// connected. On a Postgres failover or a NATS partition the pod is still
718 : /// alive (liveness stays 200 — a restart fixes nothing) but must stop
719 : /// receiving traffic, so the readinessProbe points HERE.
720 16 : pub(super) async fn ready(
721 16 : axum::extract::State(state): axum::extract::State<AppState>,
722 16 : ) -> (StatusCode, axum::Json<serde_json::Value>) {
723 16 : let draining = state.draining.load(std::sync::atomic::Ordering::Relaxed);
724 16 : let store_ok = state.store.ping().await.is_ok();
725 16 : let bus = state.bus_stats.as_ref().map(|b| b());
726 16 : let bus_ok = bus
727 16 : .as_ref()
728 16 : .is_none_or(|b| b.get("connected").and_then(serde_json::Value::as_bool) == Some(true));
729 16 : let ready = !draining && store_ok && bus_ok;
730 16 : let mut body = serde_json::json!({
731 16 : "status": if ready { "READY" } else { "NOT_READY" },
732 16 : "store": store_ok,
733 : });
734 16 : if let (Some(b), Some(body)) = (bus, body.as_object_mut()) {
735 4 : body.insert("bus".into(), b);
736 12 : }
737 16 : let code = if ready {
738 12 : StatusCode::OK
739 : } else {
740 4 : StatusCode::SERVICE_UNAVAILABLE
741 : };
742 16 : (code, axum::Json(body))
743 16 : }
744 :
745 : /// CIM 009 5.15.1 / 6.33 — Context Source identity.
746 : /// 5.15.1 Retrieve Context Source Identity Information: the 5.2.40
747 : /// ContextSourceIdentity object for this source (per tenant in the
748 : /// multi-tenancy case). The NotImplemented arm of 5.15.1.4 is vacuous here —
749 : /// this broker can always supply its identity.
750 56 : pub(super) async fn source_identity(
751 56 : axum::extract::State(state): axum::extract::State<AppState>,
752 56 : headers: HeaderMap,
753 56 : ) -> Response {
754 56 : let go = async {
755 56 : let tenant = tenant_from(&headers)?;
756 56 : gate!(state, &tenant, &headers, "5.15.1").await?;
757 56 : let accept = negotiate::parse_accept(&headers)?;
758 56 : let ctx = state.loader.core();
759 56 : let uptime = state.started.elapsed().as_secs();
760 : // Table 5.2.40-1: contextSourceAlias, contextSourceUptime and
761 : // contextSourceTimeAt are all cardinality 1. `hostAlias`/`uptime` were
762 : // not spec members at all — neither expands to an NGSI-LD IRI, so the
763 : // payload was not valid JSON-LD against the core context either.
764 : // Table 5.2.40-1: in the multi-tenancy case the alias "shall be
765 : // identifying a specific Tenant within a registered Context Source",
766 : // so what this resource serves depends on NGSILD-Tenant — a peer
767 : // retrieves it per tenant and registers it as `contextSourceAlias`.
768 56 : let alias = crate::federation::alias_for(&state.host_alias, &tenant);
769 56 : let body = serde_json::json!({
770 56 : "id": format!("urn:ngsi-ld:ContextSourceIdentity:{alias}"),
771 56 : "type": "ContextSourceIdentity",
772 56 : "contextSourceAlias": alias,
773 56 : "contextSourceUptime": format!("PT{uptime}S"),
774 56 : "contextSourceTimeAt": crate::state::now_iso(),
775 : });
776 56 : Ok::<_, ApiError>(respond(
777 56 : axum::http::StatusCode::OK,
778 56 : body,
779 56 : &ctx,
780 56 : accept,
781 56 : &tenant,
782 56 : ))
783 56 : };
784 56 : go.await.unwrap_or_else(|e| e.into_response())
785 56 : }
|
