1use axum::{Extension, Json, extract::Query};
4use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
5use chrono::{DateTime, Utc};
6use lenso_service::system_plane::{
7 CapabilityAdvertisement, RUNTIME_OBSERVABILITY_FEATURE_QUEUE_SUMMARY,
8 RUNTIME_OBSERVABILITY_FEATURE_RECOVERY_FEED, RUNTIME_OBSERVABILITY_PATH,
9 RUNTIME_OBSERVABILITY_PROTOCOL, RuntimeObservabilitySnapshot, RuntimeObservabilityStatus,
10 RuntimeObservationChange, RuntimeObservationChangeKind, RuntimeObservationContinuity,
11 RuntimeObservationEvidenceGap, RuntimeObservationFeed, RuntimeObservationGapReason,
12 RuntimeQueueKind, RuntimeQueueSummary, runtime_observability_schema_digest,
13};
14use platform_core::Migration;
15use platform_system_plane::{
16 AuthorizedSystemPlaneCaller, SystemPlaneErrorBody, SystemPlaneRejection,
17};
18use serde::{Deserialize, Serialize};
19use sha2::{Digest as _, Sha256};
20use sqlx::PgPool;
21use std::{collections::BTreeSet, sync::Arc};
22use utoipa_axum::{router::OpenApiRouter, routes};
23
24type QueueRow = (
25 i64,
26 i64,
27 i64,
28 i64,
29 i64,
30 Option<i64>,
31 Option<i64>,
32 Option<DateTime<Utc>>,
33);
34
35type ChangeRow = (i64, String, String, String, DateTime<Utc>);
36
37pub const RUNTIME_OBSERVABILITY_MIGRATIONS: &[Migration] = &[Migration {
38 name: "runtime-observability/0001_record_runtime_observation_changes",
39 sql: include_str!("../migrations/0001_record_runtime_observation_changes.sql"),
40}];
41
42const DEFAULT_FEED_LIMIT: u16 = 100;
43const MAX_FEED_LIMIT: u16 = 500;
44
45#[derive(Debug, Deserialize)]
46struct FeedQuery {
47 cursor: String,
48 #[serde(default = "default_feed_limit")]
49 limit: u16,
50}
51
52#[derive(Debug, Serialize, Deserialize)]
53#[serde(rename_all = "camelCase", deny_unknown_fields)]
54struct ObservationCursor {
55 version: u8,
56 service_id: String,
57 service_revision: String,
58 schema_digest: String,
59 sequence: u64,
60 checksum: String,
61}
62
63#[derive(Debug, Clone)]
64pub struct RuntimeObservabilityProvider {
65 pool: PgPool,
66 service_id: String,
67 service_revision: String,
68}
69
70impl RuntimeObservabilityProvider {
71 #[must_use]
72 pub fn new(
73 pool: PgPool,
74 service_id: impl Into<String>,
75 service_revision: impl Into<String>,
76 ) -> Self {
77 Self {
78 pool,
79 service_id: service_id.into(),
80 service_revision: service_revision.into(),
81 }
82 }
83
84 #[must_use]
85 pub fn advertisement() -> CapabilityAdvertisement {
86 CapabilityAdvertisement {
87 contract_id: RUNTIME_OBSERVABILITY_PROTOCOL.to_owned(),
88 major_version: 1,
89 feature_ids: BTreeSet::from([
90 RUNTIME_OBSERVABILITY_FEATURE_QUEUE_SUMMARY.to_owned(),
91 RUNTIME_OBSERVABILITY_FEATURE_RECOVERY_FEED.to_owned(),
92 ]),
93 schema_digest: runtime_observability_schema_digest(),
94 endpoint: RUNTIME_OBSERVABILITY_PATH.to_owned(),
95 }
96 }
97
98 #[must_use]
99 pub fn service_id(&self) -> &str {
100 &self.service_id
101 }
102
103 #[must_use]
104 pub fn service_revision(&self) -> &str {
105 &self.service_revision
106 }
107
108 #[must_use]
109 pub fn validation_error(&self) -> Option<&'static str> {
110 if self.service_id.trim().is_empty() {
111 return Some("Runtime Observability provider Service identity must not be empty");
112 }
113 if self.service_revision.trim().is_empty() {
114 return Some("Runtime Observability provider Service revision must not be empty");
115 }
116 None
117 }
118
119 pub async fn snapshot(&self) -> Result<RuntimeObservabilitySnapshot, sqlx::Error> {
120 let mut transaction = self.pool.begin().await?;
121 sqlx::query("set transaction isolation level repeatable read read only")
122 .execute(&mut *transaction)
123 .await?;
124 let outbox = sqlx::query_as::<_, QueueRow>(
125 r#"
126 select
127 count(*) filter (where status = 'pending')::bigint,
128 count(*) filter (where status = 'processing')::bigint,
129 count(*) filter (where status = 'published')::bigint,
130 count(*) filter (where status = 'failed')::bigint,
131 count(*) filter (where status = 'dead')::bigint,
132 extract(epoch from now() - min(created_at) filter (where status = 'pending'))::bigint,
133 extract(epoch from now() - min(created_at) filter (where status in ('failed', 'dead')))::bigint,
134 max(greatest(
135 created_at,
136 available_at,
137 coalesce(locked_at, created_at),
138 coalesce(published_at, created_at)
139 ))
140 from platform.outbox
141 "#,
142 )
143 .fetch_one(&mut *transaction)
144 .await?;
145 let functions = sqlx::query_as::<_, QueueRow>(
146 r#"
147 select
148 count(*) filter (where status = 'pending')::bigint,
149 count(*) filter (where status in ('processing', 'running'))::bigint,
150 count(*) filter (where status = 'completed')::bigint,
151 count(*) filter (where status = 'failed')::bigint,
152 count(*) filter (where status = 'dead')::bigint,
153 extract(epoch from now() - min(created_at) filter (where status = 'pending'))::bigint,
154 extract(epoch from now() - min(created_at) filter (where status in ('failed', 'dead')))::bigint,
155 max(updated_at)
156 from runtime.function_runs
157 "#,
158 )
159 .fetch_one(&mut *transaction)
160 .await?;
161 let sequence = sqlx::query_scalar::<_, i64>(
162 "select coalesce(max(sequence), 0)::bigint from platform.runtime_observation_changes",
163 )
164 .fetch_one(&mut *transaction)
165 .await?;
166 transaction.commit().await?;
167 let status = runtime_status(&outbox, &functions);
168 let snapshot_revision = snapshot_revision(&outbox, &functions);
169 let schema_digest = runtime_observability_schema_digest();
170 let next_cursor = self.cursor(count(sequence), &schema_digest);
171 Ok(RuntimeObservabilitySnapshot {
172 protocol: RUNTIME_OBSERVABILITY_PROTOCOL.to_owned(),
173 service_id: self.service_id.clone(),
174 service_revision: self.service_revision.clone(),
175 snapshot_revision,
176 schema_digest,
177 next_cursor,
178 observed_at: Utc::now(),
179 status,
180 queues: vec![
181 queue_summary(RuntimeQueueKind::Outbox, &outbox),
182 queue_summary(RuntimeQueueKind::Functions, &functions),
183 ],
184 })
185 }
186
187 pub async fn feed(
188 &self,
189 cursor: &str,
190 limit: u16,
191 ) -> Result<RuntimeObservationFeed, sqlx::Error> {
192 let schema_digest = runtime_observability_schema_digest();
193 let decoded = self.decode_cursor(cursor, &schema_digest);
194 let sequence = match decoded {
195 Ok(sequence) => sequence,
196 Err(reason) => return Ok(self.gap_feed(reason, &schema_digest)),
197 };
198 let mut transaction = self.pool.begin().await?;
199 sqlx::query("set transaction isolation level repeatable read read only")
200 .execute(&mut *transaction)
201 .await?;
202 let minimum = sqlx::query_scalar::<_, Option<i64>>(
203 "select min(sequence) from platform.runtime_observation_changes",
204 )
205 .fetch_one(&mut *transaction)
206 .await?;
207 if retention_lost(sequence, minimum) {
208 transaction.commit().await?;
209 return Ok(self.gap_feed(RuntimeObservationGapReason::RetentionLost, &schema_digest));
210 }
211 let limit = limit.clamp(1, MAX_FEED_LIMIT);
212 let rows = sqlx::query_as::<_, ChangeRow>(
213 r#"
214 select sequence, queue_kind, resource_id, change_kind, recorded_at
215 from platform.runtime_observation_changes
216 where sequence > $1
217 order by sequence
218 limit $2
219 "#,
220 )
221 .bind(i64::try_from(sequence).unwrap_or(i64::MAX))
222 .bind(i64::from(limit) + 1)
223 .fetch_all(&mut *transaction)
224 .await?;
225 transaction.commit().await?;
226 let has_more = rows.len() > usize::from(limit);
227 let changes = rows
228 .into_iter()
229 .take(usize::from(limit))
230 .map(change)
231 .collect::<Vec<_>>();
232 let next_sequence = changes.last().map_or(sequence, |item| item.sequence);
233 Ok(RuntimeObservationFeed {
234 protocol: RUNTIME_OBSERVABILITY_PROTOCOL.to_owned(),
235 service_id: self.service_id.clone(),
236 service_revision: self.service_revision.clone(),
237 schema_digest: schema_digest.clone(),
238 collected_at: Utc::now(),
239 continuity: RuntimeObservationContinuity::Continuous,
240 evidence_gap: None,
241 changes,
242 next_cursor: self.cursor(next_sequence, &schema_digest),
243 has_more,
244 })
245 }
246
247 fn cursor(&self, sequence: u64, schema_digest: &str) -> String {
248 let mut cursor = ObservationCursor {
249 version: 1,
250 service_id: self.service_id.clone(),
251 service_revision: self.service_revision.clone(),
252 schema_digest: schema_digest.to_owned(),
253 sequence,
254 checksum: String::new(),
255 };
256 cursor.checksum = cursor_checksum(&cursor);
257 URL_SAFE_NO_PAD.encode(serde_json::to_vec(&cursor).expect("observation cursor serializes"))
258 }
259
260 fn decode_cursor(
261 &self,
262 encoded: &str,
263 schema_digest: &str,
264 ) -> Result<u64, RuntimeObservationGapReason> {
265 let bytes = URL_SAFE_NO_PAD
266 .decode(encoded)
267 .map_err(|_| RuntimeObservationGapReason::InvalidCursor)?;
268 let cursor: ObservationCursor = serde_json::from_slice(&bytes)
269 .map_err(|_| RuntimeObservationGapReason::InvalidCursor)?;
270 if cursor.version != 1
271 || cursor.service_id != self.service_id
272 || cursor.checksum != cursor_checksum(&cursor)
273 {
274 return Err(RuntimeObservationGapReason::InvalidCursor);
275 }
276 if cursor.service_revision != self.service_revision {
277 return Err(RuntimeObservationGapReason::ServiceRevisionChanged);
278 }
279 if cursor.schema_digest != schema_digest {
280 return Err(RuntimeObservationGapReason::SchemaChanged);
281 }
282 Ok(cursor.sequence)
283 }
284
285 fn gap_feed(
286 &self,
287 reason: RuntimeObservationGapReason,
288 schema_digest: &str,
289 ) -> RuntimeObservationFeed {
290 let message = match reason {
291 RuntimeObservationGapReason::InvalidCursor => {
292 "The observation cursor is invalid or belongs to another Service."
293 }
294 RuntimeObservationGapReason::ServiceRevisionChanged => {
295 "The Service revision changed after the observation cursor was issued."
296 }
297 RuntimeObservationGapReason::SchemaChanged => {
298 "The Runtime Observability schema changed after the cursor was issued."
299 }
300 RuntimeObservationGapReason::RetentionLost => {
301 "Required observation changes are no longer retained by the Service."
302 }
303 };
304 RuntimeObservationFeed {
305 protocol: RUNTIME_OBSERVABILITY_PROTOCOL.to_owned(),
306 service_id: self.service_id.clone(),
307 service_revision: self.service_revision.clone(),
308 schema_digest: schema_digest.to_owned(),
309 collected_at: Utc::now(),
310 continuity: RuntimeObservationContinuity::ResetRequired,
311 evidence_gap: Some(RuntimeObservationEvidenceGap {
312 reason,
313 message: message.to_owned(),
314 required_action: "fetch_fresh_runtime_observability_snapshot".to_owned(),
315 }),
316 changes: Vec::new(),
317 next_cursor: String::new(),
318 has_more: false,
319 }
320 }
321}
322
323#[must_use]
324pub fn router<S>(provider: Option<Arc<RuntimeObservabilityProvider>>) -> OpenApiRouter<S>
325where
326 S: Clone + Send + Sync + 'static,
327{
328 OpenApiRouter::new()
329 .routes(routes!(runtime_observability_snapshot))
330 .routes(routes!(runtime_observability_feed))
331 .layer(Extension(provider))
332}
333
334#[utoipa::path(
335 get,
336 path = "/system-plane/v1/runtime-observability/changes",
337 params(("cursor" = String, Query, description = "Opaque cursor returned by a snapshot or prior feed page"), ("limit" = Option<u16>, Query)),
338 responses(
339 (status = 200, description = "Runtime observation changes or an explicit Evidence Gap", body = RuntimeObservationFeed),
340 (status = 401, description = "Workload Identity or transport binding was not accepted", body = SystemPlaneErrorBody, content_type = "application/problem+json"),
341 (status = 403, description = "Caller has no active enrollment grant", body = SystemPlaneErrorBody, content_type = "application/problem+json"),
342 (status = 503, description = "Runtime observation is unavailable", body = SystemPlaneErrorBody, content_type = "application/problem+json")
343 ),
344 security(("bearer_auth" = [])),
345 tag = "system-plane-runtime-observability"
346)]
347async fn runtime_observability_feed(
348 caller: AuthorizedSystemPlaneCaller,
349 Extension(provider): Extension<Option<Arc<RuntimeObservabilityProvider>>>,
350 Query(query): Query<FeedQuery>,
351) -> Result<Json<RuntimeObservationFeed>, SystemPlaneRejection> {
352 let provider = require_provider(provider)?;
353 caller.require_capability(
354 RUNTIME_OBSERVABILITY_PROTOCOL,
355 &runtime_observability_schema_digest(),
356 [RUNTIME_OBSERVABILITY_FEATURE_RECOVERY_FEED],
357 )?;
358 provider
359 .feed(&query.cursor, query.limit)
360 .await
361 .map(Json)
362 .map_err(|_| observation_failed())
363}
364
365#[utoipa::path(
366 get,
367 path = "/system-plane/v1/runtime-observability",
368 responses(
369 (status = 200, description = "Revisioned runtime queue snapshot", body = RuntimeObservabilitySnapshot),
370 (status = 401, description = "Workload Identity or transport binding was not accepted", body = SystemPlaneErrorBody, content_type = "application/problem+json"),
371 (status = 403, description = "Caller has no active enrollment grant", body = SystemPlaneErrorBody, content_type = "application/problem+json"),
372 (status = 503, description = "Runtime observation is unavailable", body = SystemPlaneErrorBody, content_type = "application/problem+json")
373 ),
374 security(("bearer_auth" = [])),
375 tag = "system-plane-runtime-observability"
376)]
377async fn runtime_observability_snapshot(
378 caller: AuthorizedSystemPlaneCaller,
379 Extension(provider): Extension<Option<Arc<RuntimeObservabilityProvider>>>,
380) -> Result<Json<RuntimeObservabilitySnapshot>, SystemPlaneRejection> {
381 let provider = require_provider(provider)?;
382 caller.require_capability(
383 RUNTIME_OBSERVABILITY_PROTOCOL,
384 &runtime_observability_schema_digest(),
385 [RUNTIME_OBSERVABILITY_FEATURE_QUEUE_SUMMARY],
386 )?;
387 provider
388 .snapshot()
389 .await
390 .map(Json)
391 .map_err(|_| observation_failed())
392}
393
394fn require_provider(
395 provider: Option<Arc<RuntimeObservabilityProvider>>,
396) -> Result<Arc<RuntimeObservabilityProvider>, SystemPlaneRejection> {
397 provider.ok_or_else(|| {
398 SystemPlaneRejection::unavailable(
399 "runtime_observability_unavailable",
400 "Runtime Observability capability is not configured for this Service",
401 "configure_runtime_observability",
402 )
403 })
404}
405
406fn observation_failed() -> SystemPlaneRejection {
407 SystemPlaneRejection::unavailable(
408 "runtime_observation_failed",
409 "Runtime observation query failed",
410 "restore_service_store_observation",
411 )
412}
413
414fn queue_summary(kind: RuntimeQueueKind, row: &QueueRow) -> RuntimeQueueSummary {
415 RuntimeQueueSummary {
416 queue: kind,
417 pending: count(row.0),
418 active: count(row.1),
419 completed: count(row.2),
420 failed: count(row.3),
421 dead: count(row.4),
422 oldest_pending_age_seconds: row.5.map(count),
423 oldest_failed_age_seconds: row.6.map(count),
424 }
425}
426
427fn runtime_status(outbox: &QueueRow, functions: &QueueRow) -> RuntimeObservabilityStatus {
428 if outbox.4 > 0 || functions.4 > 0 {
429 RuntimeObservabilityStatus::Failing
430 } else if outbox.3 > 0 || functions.3 > 0 {
431 RuntimeObservabilityStatus::Degraded
432 } else {
433 RuntimeObservabilityStatus::Healthy
434 }
435}
436
437fn snapshot_revision(outbox: &QueueRow, functions: &QueueRow) -> String {
438 let material = format!(
439 "{:?}:{:?}",
440 (outbox.0, outbox.1, outbox.2, outbox.3, outbox.4, outbox.7),
441 (
442 functions.0,
443 functions.1,
444 functions.2,
445 functions.3,
446 functions.4,
447 functions.7
448 )
449 );
450 format!("sha256:{}", hex(&Sha256::digest(material)))
451}
452
453fn change(row: ChangeRow) -> RuntimeObservationChange {
454 RuntimeObservationChange {
455 sequence: count(row.0),
456 queue: match row.1.as_str() {
457 "outbox" => RuntimeQueueKind::Outbox,
458 "functions" => RuntimeQueueKind::Functions,
459 _ => unreachable!("migration constrains runtime observation queue kinds"),
460 },
461 resource_id: row.2,
462 change_kind: match row.3.as_str() {
463 "upserted" => RuntimeObservationChangeKind::Upserted,
464 "deleted" => RuntimeObservationChangeKind::Deleted,
465 _ => unreachable!("migration constrains runtime observation change kinds"),
466 },
467 recorded_at: row.4,
468 }
469}
470
471fn cursor_checksum(cursor: &ObservationCursor) -> String {
472 let material = format!(
473 "{}:{}:{}:{}:{}",
474 cursor.version,
475 cursor.service_id,
476 cursor.service_revision,
477 cursor.schema_digest,
478 cursor.sequence
479 );
480 format!("sha256:{}", hex(&Sha256::digest(material)))
481}
482
483const fn default_feed_limit() -> u16 {
484 DEFAULT_FEED_LIMIT
485}
486
487fn count(value: i64) -> u64 {
488 u64::try_from(value).unwrap_or(0)
489}
490
491fn hex(bytes: &[u8]) -> String {
492 bytes.iter().map(|byte| format!("{byte:02x}")).collect()
493}
494
495fn retention_lost(sequence: u64, minimum: Option<i64>) -> bool {
496 minimum.is_some_and(|minimum| sequence.saturating_add(1) < count(minimum))
497}
498
499#[cfg(test)]
500mod tests {
501 use super::*;
502 use sqlx::postgres::PgPoolOptions;
503
504 fn provider(revision: &str) -> RuntimeObservabilityProvider {
505 RuntimeObservabilityProvider::new(
506 PgPoolOptions::new()
507 .connect_lazy("postgres://lenso:lenso@127.0.0.1/lenso")
508 .unwrap(),
509 "support",
510 revision,
511 )
512 }
513
514 #[tokio::test]
515 async fn cursor_is_scoped_to_service_revision_and_schema() {
516 let first = provider("release:one");
517 let schema = runtime_observability_schema_digest();
518 let cursor = first.cursor(42, &schema);
519
520 assert_eq!(first.decode_cursor(&cursor, &schema), Ok(42));
521 assert_eq!(
522 provider("release:two").decode_cursor(&cursor, &schema),
523 Err(RuntimeObservationGapReason::ServiceRevisionChanged)
524 );
525 assert_eq!(
526 first.decode_cursor(&cursor, "sha256:changed"),
527 Err(RuntimeObservationGapReason::SchemaChanged)
528 );
529 assert_eq!(
530 first.decode_cursor("invalid", &schema),
531 Err(RuntimeObservationGapReason::InvalidCursor)
532 );
533 }
534
535 #[test]
536 fn retention_gap_requires_a_missing_sequence() {
537 assert!(!retention_lost(4, Some(5)));
538 assert!(retention_lost(4, Some(6)));
539 assert!(!retention_lost(4, None));
540 }
541}