| 237 | } |
| 238 | |
| 239 | pub(super) fn prometheus_backpressure(&self, out: &mut String) { |
| 240 | use std::fmt::Write as _; |
| 241 | let critical = self |
| 242 | .backpressure_critical_by_engine |
| 243 | .read() |
| 244 | .unwrap_or_else(|p| p.into_inner()); |
| 245 | let emergency = self |
| 246 | .backpressure_emergency_by_engine |
| 247 | .read() |
| 248 | .unwrap_or_else(|p| p.into_inner()); |
| 249 | if !critical.is_empty() { |
| 250 | let _ = out.write_str( |
| 251 | "# HELP nodedb_backpressure_critical_total Write handlers that entered Critical-pressure flush path\n\ |
| 252 | # TYPE nodedb_backpressure_critical_total counter\n", |
| 253 | ); |
| 254 | let mut pairs: Vec<_> = critical.iter().collect(); |
| 255 | pairs.sort_by(|a, b| a.0.cmp(b.0)); |
| 256 | for (engine, count) in pairs { |
| 257 | let _ = writeln!( |
| 258 | out, |
| 259 | r#"nodedb_backpressure_critical_total{{engine="{engine}"}} {count}"# |
| 260 | ); |
| 261 | } |
| 262 | } |
| 263 | if !emergency.is_empty() { |
| 264 | let _ = out.write_str( |
| 265 | "# HELP nodedb_backpressure_emergency_total Write handlers rejected by Emergency-pressure\n\ |
| 266 | # TYPE nodedb_backpressure_emergency_total counter\n", |
| 267 | ); |
| 268 | let mut pairs: Vec<_> = emergency.iter().collect(); |
| 269 | pairs.sort_by(|a, b| a.0.cmp(b.0)); |
| 270 | for (engine, count) in pairs { |
| 271 | let _ = writeln!( |
| 272 | out, |
| 273 | r#"nodedb_backpressure_emergency_total{{engine="{engine}"}} {count}"# |
| 274 | ); |
| 275 | } |
| 276 | } |
| 277 | } |
| 278 | |
| 279 | /// Emit `nodedb_cdc_events_dropped_total{tenant,stream}` labelled counters. |
| 280 | pub(super) fn prometheus_cdc_stream_drops(&self, out: &mut String) { |