Skip to content

Commit ce989ae

Browse files
lance0lance0
authored
Merge pull request #102 from bswinnerton/fix/wire-up-prometheus-metrics
fix: wire up missing Prometheus metrics
2 parents 69e11b4 + 2476761 commit ce989ae

3 files changed

Lines changed: 175 additions & 10 deletions

File tree

‎src/api/handlers.rs‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -533,20 +533,36 @@ async fn handle_unban(
533533
let action = FlowSpecAction::from((mitigation.action_type, &mitigation.action_params));
534534
let rule = FlowSpecRule::new(nlri, action);
535535

536+
let start = std::time::Instant::now();
536537
if let Err(e) = state.announcer.withdraw(&rule).await {
537538
tracing::error!(error = %e, "BGP withdrawal failed");
538539
// Continue anyway - mark as withdrawn in DB
540+
} else {
541+
crate::observability::metrics::ANNOUNCEMENTS_TOTAL
542+
.with_label_values(&["withdrawn"])
543+
.inc();
544+
crate::observability::metrics::ANNOUNCEMENTS_LATENCY
545+
.observe(start.elapsed().as_secs_f64());
539546
}
540547
}
541548

542549
// Update mitigation status
550+
let action_type_str = mitigation.action_type.to_string();
543551
mitigation.withdraw(Some(format!("Detector unban: {}", source)));
544552
state
545553
.repo
546554
.update_mitigation(&mitigation)
547555
.await
548556
.map_err(AppError)?;
549557

558+
crate::observability::metrics::MITIGATIONS_WITHDRAWN
559+
.with_label_values(&[
560+
action_type_str.as_str(),
561+
mitigation.pop.as_str(),
562+
"detector_unban",
563+
])
564+
.inc();
565+
550566
// Broadcast withdrawal via WebSocket
551567
let _ = state
552568
.ws_broadcast
@@ -589,6 +605,9 @@ async fn handle_ban(
589605
.find_ban_event_by_external_id(&input.source, ext_id)
590606
.await
591607
{
608+
crate::observability::metrics::EVENTS_REJECTED
609+
.with_label_values(&[input.source.as_str(), "duplicate"])
610+
.inc();
592611
return Err(AppError(PrefixdError::DuplicateEvent {
593612
detector_source: input.source.clone(),
594613
external_id: ext_id.clone(),
@@ -602,6 +621,10 @@ async fn handle_ban(
602621
// Store event
603622
state.repo.insert_event(&event).await.map_err(AppError)?;
604623

624+
crate::observability::metrics::EVENTS_INGESTED
625+
.with_label_values(&[&event.source, &event.attack_vector().to_string()])
626+
.inc();
627+
605628
// Check if shutting down
606629
if state.is_shutting_down() {
607630
return Err(AppError(PrefixdError::ShuttingDown));
@@ -873,6 +896,20 @@ async fn handle_ban(
873896
.validate(&intent, state.repo.as_ref(), is_safelisted)
874897
.await
875898
{
899+
crate::observability::metrics::EVENTS_REJECTED
900+
.with_label_values(&[event.source.as_str(), "guardrail"])
901+
.inc();
902+
let reason = match &e {
903+
PrefixdError::GuardrailViolation(g) => format!("{:?}", g)
904+
.split_whitespace()
905+
.next()
906+
.unwrap_or("unknown")
907+
.to_string(),
908+
_ => "unknown".to_string(),
909+
};
910+
crate::observability::metrics::GUARDRAIL_REJECTIONS
911+
.with_label_values(&[&reason])
912+
.inc();
876913
tracing::warn!(error = %e, "guardrail rejected mitigation");
877914
return Err(AppError(e));
878915
}
@@ -888,6 +925,7 @@ async fn handle_ban(
888925
let action = FlowSpecAction::from((mitigation.action_type, &mitigation.action_params));
889926
let rule = FlowSpecRule::new(nlri, action);
890927

928+
let start = std::time::Instant::now();
891929
if let Err(e) = state.announcer.announce(&rule).await {
892930
tracing::error!(error = %e, "BGP announcement failed");
893931
mitigation.reject(e.to_string());
@@ -898,6 +936,10 @@ async fn handle_ban(
898936
.map_err(AppError)?;
899937
return Err(AppError(e));
900938
}
939+
crate::observability::metrics::ANNOUNCEMENTS_TOTAL
940+
.with_label_values(&["announced"])
941+
.inc();
942+
crate::observability::metrics::ANNOUNCEMENTS_LATENCY.observe(start.elapsed().as_secs_f64());
901943
}
902944

903945
mitigation.activate();
@@ -907,6 +949,10 @@ async fn handle_ban(
907949
.await
908950
.map_err(AppError)?;
909951

952+
crate::observability::metrics::MITIGATIONS_CREATED
953+
.with_label_values(&[&mitigation.action_type.to_string(), &state.settings.pop])
954+
.inc();
955+
910956
// Resolve signal group to 'resolved' now that mitigation is confirmed
911957
if let Some(group_id) = signal_group_id {
912958
if let Ok(Some(mut group)) = state.repo.get_signal_group(group_id).await {
@@ -1370,6 +1416,17 @@ pub async fn create_mitigation(
13701416
.validate(&intent, state.repo.as_ref(), is_safelisted)
13711417
.await
13721418
{
1419+
let reason = match &e {
1420+
PrefixdError::GuardrailViolation(g) => format!("{:?}", g)
1421+
.split_whitespace()
1422+
.next()
1423+
.unwrap_or("unknown")
1424+
.to_string(),
1425+
_ => "unknown".to_string(),
1426+
};
1427+
crate::observability::metrics::GUARDRAIL_REJECTIONS
1428+
.with_label_values(&[&reason])
1429+
.inc();
13731430
return Ok(AppError(e).into_response());
13741431
}
13751432

@@ -1381,16 +1438,25 @@ pub async fn create_mitigation(
13811438
let nlri = FlowSpecNlri::from(&mitigation.match_criteria);
13821439
let action = FlowSpecAction::from((mitigation.action_type, &mitigation.action_params));
13831440
let rule = FlowSpecRule::new(nlri, action);
1441+
let start = std::time::Instant::now();
13841442
if let Err(e) = state.announcer.announce(&rule).await {
13851443
return Ok(AppError(e).into_response());
13861444
}
1445+
crate::observability::metrics::ANNOUNCEMENTS_TOTAL
1446+
.with_label_values(&["announced"])
1447+
.inc();
1448+
crate::observability::metrics::ANNOUNCEMENTS_LATENCY.observe(start.elapsed().as_secs_f64());
13871449
}
13881450

13891451
mitigation.activate();
13901452
if let Err(e) = state.repo.insert_mitigation(&mitigation).await {
13911453
return Ok(AppError(e).into_response());
13921454
}
13931455

1456+
crate::observability::metrics::MITIGATIONS_CREATED
1457+
.with_label_values(&[&mitigation.action_type.to_string(), &state.settings.pop])
1458+
.inc();
1459+
13941460
Ok((
13951461
StatusCode::CREATED,
13961462
Json(MitigationResponse::from(&mitigation)),
@@ -1434,20 +1500,34 @@ pub async fn withdraw_mitigation(
14341500
let nlri = FlowSpecNlri::from(&mitigation.match_criteria);
14351501
let action = FlowSpecAction::from((mitigation.action_type, &mitigation.action_params));
14361502
let rule = FlowSpecRule::new(nlri, action);
1503+
let start = std::time::Instant::now();
14371504
state
14381505
.announcer
14391506
.withdraw(&rule)
14401507
.await
14411508
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
1509+
crate::observability::metrics::ANNOUNCEMENTS_TOTAL
1510+
.with_label_values(&["withdrawn"])
1511+
.inc();
1512+
crate::observability::metrics::ANNOUNCEMENTS_LATENCY.observe(start.elapsed().as_secs_f64());
14421513
}
14431514

1515+
let action_type_str = mitigation.action_type.to_string();
14441516
mitigation.withdraw(Some(format!("{}: {}", req.operator_id, req.reason)));
14451517
state
14461518
.repo
14471519
.update_mitigation(&mitigation)
14481520
.await
14491521
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
14501522

1523+
crate::observability::metrics::MITIGATIONS_WITHDRAWN
1524+
.with_label_values(&[
1525+
action_type_str.as_str(),
1526+
mitigation.pop.as_str(),
1527+
"operator",
1528+
])
1529+
.inc();
1530+
14511531
// Broadcast withdrawal via WebSocket
14521532
let _ = state
14531533
.ws_broadcast
@@ -1565,8 +1645,15 @@ pub async fn bulk_withdraw_mitigations(
15651645
let nlri = FlowSpecNlri::from(&mitigation.match_criteria);
15661646
let action = FlowSpecAction::from((mitigation.action_type, &mitigation.action_params));
15671647
let rule = FlowSpecRule::new(nlri, action);
1648+
let start = std::time::Instant::now();
15681649
if let Err(e) = state.announcer.withdraw(&rule).await {
15691650
tracing::error!(error = %e, mitigation_id = %id, "BGP withdrawal failed in bulk withdraw");
1651+
} else {
1652+
crate::observability::metrics::ANNOUNCEMENTS_TOTAL
1653+
.with_label_values(&["withdrawn"])
1654+
.inc();
1655+
crate::observability::metrics::ANNOUNCEMENTS_LATENCY
1656+
.observe(start.elapsed().as_secs_f64());
15701657
}
15711658
}
15721659

‎src/observability/metrics.rs‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use once_cell::sync::Lazy;
22
use prometheus::{
3-
CounterVec, Encoder, GaugeVec, HistogramVec, TextEncoder, register_counter_vec,
4-
register_gauge_vec, register_histogram_vec,
3+
CounterVec, Encoder, GaugeVec, Histogram, HistogramVec, TextEncoder, register_counter_vec,
4+
register_gauge_vec, register_histogram, register_histogram_vec,
55
};
66

77
// Event metrics
@@ -65,16 +65,15 @@ pub static ANNOUNCEMENTS_TOTAL: Lazy<CounterVec> = Lazy::new(|| {
6565
register_counter_vec!(
6666
"prefixd_announcements_total",
6767
"Total number of BGP announcements",
68-
&["peer", "status"]
68+
&["status"]
6969
)
7070
.unwrap()
7171
});
7272

73-
pub static ANNOUNCEMENTS_LATENCY: Lazy<HistogramVec> = Lazy::new(|| {
74-
register_histogram_vec!(
73+
pub static ANNOUNCEMENTS_LATENCY: Lazy<Histogram> = Lazy::new(|| {
74+
register_histogram!(
7575
"prefixd_announcements_latency_seconds",
7676
"BGP announcement latency in seconds",
77-
&["peer"],
7877
vec![0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
7978
)
8079
.unwrap()

‎src/scheduler/reconcile.rs‎

Lines changed: 83 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -55,8 +55,18 @@ impl ReconciliationLoop {
5555
);
5656

5757
// Initial reconciliation
58-
if let Err(e) = self.reconcile().await {
59-
tracing::error!(error = %e, "initial reconciliation failed");
58+
match self.reconcile().await {
59+
Ok(()) => {
60+
crate::observability::metrics::RECONCILIATION_RUNS
61+
.with_label_values(&["success"])
62+
.inc();
63+
}
64+
Err(e) => {
65+
crate::observability::metrics::RECONCILIATION_RUNS
66+
.with_label_values(&["error"])
67+
.inc();
68+
tracing::error!(error = %e, "initial reconciliation failed");
69+
}
6070
}
6171

6272
let mut interval = tokio::time::interval(self.interval);
@@ -65,8 +75,18 @@ impl ReconciliationLoop {
6575
loop {
6676
tokio::select! {
6777
_ = interval.tick() => {
68-
if let Err(e) = self.reconcile().await {
69-
tracing::error!(error = %e, "reconciliation failed");
78+
match self.reconcile().await {
79+
Ok(()) => {
80+
crate::observability::metrics::RECONCILIATION_RUNS
81+
.with_label_values(&["success"])
82+
.inc();
83+
}
84+
Err(e) => {
85+
crate::observability::metrics::RECONCILIATION_RUNS
86+
.with_label_values(&["error"])
87+
.inc();
88+
tracing::error!(error = %e, "reconciliation failed");
89+
}
7090
}
7191
}
7292
_ = shutdown.recv() => {
@@ -88,6 +108,9 @@ impl ReconciliationLoop {
88108
// 3. Sync desired vs actual state
89109
self.sync_announcements().await?;
90110

111+
// 4. Update BGP session metrics
112+
self.update_bgp_session_metrics().await;
113+
91114
Ok(())
92115
}
93116

@@ -104,19 +127,32 @@ impl ReconciliationLoop {
104127
// Withdraw BGP announcement
105128
if !self.dry_run {
106129
let rule = self.build_flowspec_rule(&mitigation);
130+
let start = std::time::Instant::now();
107131
if let Err(e) = self.announcer.withdraw(&rule).await {
108132
tracing::warn!(
109133
mitigation_id = %mitigation.mitigation_id,
110134
error = %e,
111135
"failed to withdraw expired mitigation"
112136
);
137+
} else {
138+
crate::observability::metrics::ANNOUNCEMENTS_TOTAL
139+
.with_label_values(&["withdrawn"])
140+
.inc();
141+
crate::observability::metrics::ANNOUNCEMENTS_LATENCY
142+
.observe(start.elapsed().as_secs_f64());
113143
}
114144
}
115145

116146
// Update status
147+
let action_type_str = mitigation.action_type.to_string();
148+
let pop = mitigation.pop.clone();
117149
mitigation.expire();
118150
self.repo.update_mitigation(&mitigation).await?;
119151

152+
crate::observability::metrics::MITIGATIONS_EXPIRED
153+
.with_label_values(&[&action_type_str, &pop])
154+
.inc();
155+
120156
// Broadcast expiry via WebSocket
121157
if let Some(ref tx) = self.ws_broadcast {
122158
let _ = tx.send(WsMessage::MitigationExpired {
@@ -197,6 +233,22 @@ impl ReconciliationLoop {
197233
.with_label_values(&["local"])
198234
.set(active.len() as f64);
199235

236+
{
237+
use std::collections::HashMap;
238+
let mut counts: HashMap<(String, String), f64> = HashMap::new();
239+
for m in &active {
240+
*counts
241+
.entry((m.action_type.to_string(), m.pop.clone()))
242+
.or_default() += 1.0;
243+
}
244+
crate::observability::metrics::MITIGATIONS_ACTIVE.reset();
245+
for ((action_type, pop), count) in &counts {
246+
crate::observability::metrics::MITIGATIONS_ACTIVE
247+
.with_label_values(&[action_type, pop])
248+
.set(*count);
249+
}
250+
}
251+
200252
// Get actual state from BGP
201253
let announced = self.announcer.list_active().await?;
202254
let announced_hashes: std::collections::HashSet<_> =
@@ -215,12 +267,19 @@ impl ReconciliationLoop {
215267
);
216268

217269
if !self.dry_run {
270+
let start = std::time::Instant::now();
218271
if let Err(e) = self.announcer.announce(&rule).await {
219272
tracing::error!(
220273
mitigation_id = %mitigation.mitigation_id,
221274
error = %e,
222275
"failed to re-announce"
223276
);
277+
} else {
278+
crate::observability::metrics::ANNOUNCEMENTS_TOTAL
279+
.with_label_values(&["announced"])
280+
.inc();
281+
crate::observability::metrics::ANNOUNCEMENTS_LATENCY
282+
.observe(start.elapsed().as_secs_f64());
224283
}
225284
}
226285
}
@@ -245,6 +304,26 @@ impl ReconciliationLoop {
245304
Ok(())
246305
}
247306

307+
async fn update_bgp_session_metrics(&self) {
308+
match self.announcer.session_status().await {
309+
Ok(peers) => {
310+
for peer in &peers {
311+
let value = if peer.state.is_established() {
312+
1.0
313+
} else {
314+
0.0
315+
};
316+
crate::observability::metrics::BGP_SESSION_UP
317+
.with_label_values(&[&peer.name])
318+
.set(value);
319+
}
320+
}
321+
Err(e) => {
322+
tracing::warn!(error = %e, "failed to fetch BGP session status for metrics");
323+
}
324+
}
325+
}
326+
248327
fn build_flowspec_rule(&self, m: &crate::domain::Mitigation) -> FlowSpecRule {
249328
let nlri = FlowSpecNlri::from(&m.match_criteria);
250329
let action = FlowSpecAction::from((m.action_type, &m.action_params));

0 commit comments

Comments
 (0)