feat(health): implement dependency health monitoring for RocketMQ and S3

Modified app/src/main.rs +32 −0 View app/src/main.rs

@@ -95,6 +95,8 @@

9595 ],

9696 );

9797

98 spawn_dependency_health_sampler(state.clone());

99

98100 tokio::spawn(chat::worker::run_background_tasks(deps.chat.clone()));

99101 tokio::spawn(deletion::run_worker(state.clone(), deps.accounts.clone()));

100102

@@ -401,6 +403,36 @@

401403

402404const QUEUE_DEPTH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);

403405

406const HEALTH_SAMPLE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);

407const HEALTH_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);

408

409/// 周期性采样 RocketMQ/S3 健康：`/readyz` 只读缓存，避免每次探活都新建连接。

410fn spawn_dependency_health_sampler(state: bootstrap::AppState) {

411 tokio::spawn(async move {

412 let mut ticker = tokio::time::interval(HEALTH_SAMPLE_INTERVAL);

413 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

414 loop {

415 ticker.tick().await;

416

417 let rocketmq_up = queue::probe_nameservers_with(

418 &state.config.rocketmq,

419 std::time::Duration::from_secs(1),

420 )

421 .await

422 .is_ok();

423 state.health.mark_rocketmq(rocketmq_up);

424 oltp::metrics::health_dependency("rocketmq", rocketmq_up);

425

426 let s3_up = matches!(

427 tokio::time::timeout(HEALTH_PROBE_TIMEOUT, state.storage.probe()).await,

428 Ok(Ok(()))

429 );

430 state.health.mark_s3(s3_up);

431 oltp::metrics::health_dependency("s3", s3_up);

432 }

433 });

434}

435

404436/// 周期性上报队列与死信队列的积压深度；进程退出时随运行时一起结束。

405437fn spawn_queue_depth_reporter(redis: cache::AppRedis, queues: &[(&'static str, String)]) {

406438 if queues.is_empty() {

Modified lib/infra/bootstrap/src/health.rs +62 −12 View lib/infra/bootstrap/src/health.rs

@@ -1,8 +1,7 @@

11use std::sync::Arc;

2use std::time::Duration;

2use std::sync::atomic::{AtomicU8, Ordering};

33

44use actix_web::HttpResponse;

5use config::rocketmq::RocketMqConfig;

65use sea_orm::{ConnectionTrait, DbBackend, Statement};

76

87use api::http::{Router, State, get};

@@ -11,11 +10,51 @@

1110

1211use crate::AppState;

1312

13const UP: u8 = 1;

14const DOWN: u8 = 2;

15

16/// 后台采样得到的依赖健康状态；探针只读取这里的值，不发起探测。

17/// 原子量初值为 0，表示尚未采样（同样按未就绪处理）。

18#[derive(Clone, Default)]

19pub struct HealthCache {

20 inner: Arc<HealthCacheInner>,

21}

22

23#[derive(Default)]

24struct HealthCacheInner {

25 rocketmq: AtomicU8,

26 s3: AtomicU8,

27}

28

29impl HealthCache {

30 pub fn mark_rocketmq(&self, up: bool) {

31 self.inner

32 .rocketmq

33 .store(if up { UP } else { DOWN }, Ordering::Relaxed);

34 }

35

36 pub fn mark_s3(&self, up: bool) {

37 self.inner

38 .s3

39 .store(if up { UP } else { DOWN }, Ordering::Relaxed);

40 }

41

42 fn dependencies(&self) -> [(&'static str, bool); 2] {

43 [

44 (

45 "rocketmq",

46 self.inner.rocketmq.load(Ordering::Relaxed) == UP,

47 ),

48 ("s3", self.inner.s3.load(Ordering::Relaxed) == UP),

49 ]

50 }

51}

52

1453#[derive(Clone)]

1554struct HealthState {

1655 database: Option<Arc<AppDatabase>>,

1756 redis: AppRedis,

18 rocketmq: RocketMqConfig,

57 health: HealthCache,

1958}

2059

2160async fn healthz() -> HttpResponse {

@@ -26,7 +65,7 @@

2665 let db_ok = match state.database.as_ref() {

2766 None => true,

2867 Some(db) => db

29 .read()

68 .write()

3069 .query_one_raw(Statement::from_sql_and_values(

3170 DbBackend::Postgres,

3271 "SELECT 1",

@@ -36,20 +75,31 @@

3675 .is_ok(),

3776 };

3877 let redis_ok = state.redis.ping().await.is_ok();

39 let rocketmq_ok = queue::probe_nameservers_with(&state.rocketmq, Duration::from_secs(1))

40 .await

41 .is_ok();

42 if db_ok && redis_ok && rocketmq_ok {

78

79 let mut failed = Vec::new();

80 if !db_ok {

81 failed.push("database");

82 }

83 if !redis_ok {

84 failed.push("redis");

85 }

86 for (dependency, up) in state.health.dependencies() {

87 if !up {

88 failed.push(dependency);

89 }

90 }

91

92 if failed.is_empty() {

4393 HttpResponse::Ok().body("ready")

4494 } else {

45 HttpResponse::ServiceUnavailable().body("not ready")

95 HttpResponse::ServiceUnavailable().body(format!("not ready: {}", failed.join(",")))

4696 }

4797}

4898

4999pub fn router_with(

50100 database: Option<Arc<AppDatabase>>,

51101 redis: AppRedis,

52 rocketmq: RocketMqConfig,

102 health: HealthCache,

53103) -> Router {

54104 Router::<()>::new()

55105 .route("/healthz", get(healthz))

@@ -57,7 +107,7 @@

57107 .with_state(HealthState {

58108 database,

59109 redis,

60 rocketmq,

110 health,

61111 })

62112}

63113

@@ -65,6 +115,6 @@

65115 router_with(

66116 Some(state.database.clone()),

67117 state.redis.clone(),

68 state.config.rocketmq.clone(),

118 state.health.clone(),

69119 )

70120}

Modified lib/infra/bootstrap/src/lib.rs +2 −0 View lib/infra/bootstrap/src/lib.rs

@@ -14,6 +14,7 @@

1414 pub cache: cache::Cache,

1515 pub storage: Arc<storage::S3Client>,

1616 pub config: AppConfig,

17 pub health: health::HealthCache,

1718 pub mail: Option<email::MailPublisher>,

1819 pub events: Option<events::EventHandler>,

1920}

@@ -63,6 +64,7 @@

6364 cache,

6465 storage,

6566 config,

67 health: health::HealthCache::default(),

6668 mail,

6769 events,

6870 }

Modified lib/infra/bootstrap/Cargo.toml +0 −1 View lib/infra/bootstrap/Cargo.toml

@@ -10,7 +10,6 @@

1010db = { workspace = true }

1111email = { workspace = true }

1212events = { workspace = true }

13queue = { workspace = true }

1413storage = { workspace = true }

1514actix-web = { workspace = true }

1615redis = { workspace = true }

Modified lib/infra/oltp/src/metrics.rs +14 −0 View lib/infra/oltp/src/metrics.rs

@@ -185,6 +185,13 @@

185185 .build()

186186});

187187

188static DEPENDENCY_UP: LazyLock<Gauge<i64>> = LazyLock::new(|| {

189 app_meter()

190 .i64_gauge("health.dependency.up")

191 .with_description("Sampled dependency health (1 = up, 0 = down)")

192 .build()

193});

194

188195pub fn auth_login(outcome: &'static str) {

189196 AUTH_LOGINS.add(1, &[KeyValue::new("outcome", outcome)]);

190197}

@@ -228,3 +235,10 @@

228235pub fn queue_depth(topic: &'static str, depth: i64) {

229236 QUEUE_DEPTH.record(depth, &[KeyValue::new("topic", topic)]);

230237}

238

239pub fn health_dependency(dependency: &'static str, up: bool) {

240 DEPENDENCY_UP.record(

241 if up { 1 } else { 0 },

242 &[KeyValue::new("dependency", dependency)],

243 );

244}

Modified lib/infra/storage/src/s3.rs +4 −0 View lib/infra/storage/src/s3.rs

@@ -274,6 +274,10 @@

274274 )))

275275 }

276276

277 pub async fn probe(&self) -> Result<(), StorageError> {

278 self.list_page("", None, Some(1)).await.map(|_| ())

279 }

280

277281 pub async fn list_page(

278282 &self,

279283 prefix: &str,

Modified lib/service/webapi/src/openapi.rs +2 −1 View lib/service/webapi/src/openapi.rs

@@ -65,7 +65,8 @@

6565

6666### Operations

6767- `GET /healthz` — liveness probe (no auth, outside `/api/v1`).

68- `GET /readyz` — readiness probe: database + redis check, `503` when down.

68- `GET /readyz` — readiness probe: database (primary) + redis live checks,

69 rocketmq + s3 sampled every 10s, `503` when any is down.

6970- Runtime metrics are pushed to the configured OTLP collector (30s interval);

7071 there is no scrape endpoint.

7172- Cross-origin browser access is disabled by default; the API expects a

Modified Cargo.lock +0 −1 View Cargo.lock

@@ -1153,7 +1153,6 @@

11531153 "db",

11541154 "email",

11551155 "events",

1156 "queue",

11571156 "redis",

11581157 "sea-orm",

11591158 "storage",