@@ -20,7 +20,8 @@ use config::regex::Regex;
2020use prober:: manager:: STORE as PROBER_STORE ;
2121use prober:: mode:: Mode ;
2222use super :: states:: { ServiceStates , ServiceStatesProbe , ServiceStatesProbeNode ,
23- ServiceStatesProbeNodeReplica } ;
23+ ServiceStatesProbeNodeReplica , ServiceStatesProbeNodeReplicaMetrics ,
24+ ServiceStatesProbeNodeReplicaMetricsRabbitMQ } ;
2425use super :: replica:: ReplicaURL ;
2526use super :: status:: Status ;
2627use APP_CONF ;
@@ -105,9 +106,8 @@ fn map_poll_replicas() -> Vec<(String, String, String, ReplicaURL, Option<Regex>
105106fn proceed_replica_probe_with_retry (
106107 replica_url : & ReplicaURL ,
107108 body_match : & Option < Regex > ,
108- ) -> Status {
109- let mut status = Status :: Dead ;
110- let mut retry_count = 0 ;
109+ ) -> ( Status , Option < Duration > ) {
110+ let ( mut status, mut latency, mut retry_count) = ( Status :: Dead , None , 0 ) ;
111111
112112 while retry_count < APP_CONF . metrics . poll_retry && status == Status :: Dead {
113113 retry_count += 1 ;
@@ -120,13 +120,19 @@ fn proceed_replica_probe_with_retry(
120120
121121 thread:: sleep ( Duration :: from_millis ( PROBE_HOLD_MILLISECONDS ) ) ;
122122
123- status = proceed_replica_probe ( replica_url, body_match) ;
123+ let probe_results = proceed_replica_probe ( replica_url, body_match) ;
124+
125+ status = probe_results. 0 ;
126+ latency = Some ( probe_results. 1 ) ;
124127 }
125128
126- status
129+ ( status, latency )
127130}
128131
129- fn proceed_replica_probe ( replica_url : & ReplicaURL , body_match : & Option < Regex > ) -> Status {
132+ fn proceed_replica_probe (
133+ replica_url : & ReplicaURL ,
134+ body_match : & Option < Regex > ,
135+ ) -> ( Status , Duration ) {
130136 let start_time = SystemTime :: now ( ) ;
131137
132138 let is_up = match replica_url {
@@ -135,17 +141,19 @@ fn proceed_replica_probe(replica_url: &ReplicaURL, body_match: &Option<Regex>) -
135141 & ReplicaURL :: HTTPS ( ref url) => proceed_replica_probe_http ( url, body_match) ,
136142 } ;
137143
144+ let duration_latency = SystemTime :: now ( ) . duration_since ( start_time) . unwrap_or (
145+ Duration :: from_secs ( 0 ) ,
146+ ) ;
147+
138148 if is_up == true {
139149 // Probe reports as sick?
140- if let Ok ( duration_since) = SystemTime :: now ( ) . duration_since ( start_time) {
141- if duration_since >= Duration :: from_secs ( APP_CONF . metrics . poll_delay_sick ) {
142- return Status :: Sick ;
143- }
150+ if duration_latency >= Duration :: from_secs ( APP_CONF . metrics . poll_delay_sick ) {
151+ return ( Status :: Sick , duration_latency) ;
144152 }
145153
146- Status :: Healthy
154+ ( Status :: Healthy , duration_latency )
147155 } else {
148- Status :: Dead
156+ ( Status :: Dead , duration_latency )
149157 }
150158}
151159
@@ -233,7 +241,10 @@ fn proceed_replica_probe_http(url: &str, body_match: &Option<Regex>) -> bool {
233241 false
234242}
235243
236- fn proceed_rabbitmq_queue_probe ( rabbitmq : & ConfigPluginsRabbitMQ , rabbitmq_queue : & str ) -> bool {
244+ fn proceed_rabbitmq_queue_probe (
245+ rabbitmq : & ConfigPluginsRabbitMQ ,
246+ rabbitmq_queue : & str ,
247+ ) -> ( bool , Option < ( u32 , u32 ) > ) {
237248 let url_queue = rabbitmq. api_url . join ( & format ! (
238249 "/api/queues/{}/{}" ,
239250 rabbitmq. virtualhost,
@@ -268,6 +279,11 @@ fn proceed_rabbitmq_queue_probe(rabbitmq: &ConfigPluginsRabbitMQ, rabbitmq_queue
268279 // Check JSON result?
269280 if status == StatusCode :: OK {
270281 if let Ok ( response_json) = response_inner. json :: < RabbitMQAPIQueueResponse > ( ) {
282+ let queue_counts = Some ( (
283+ response_json. messages_ready ,
284+ response_json. messages_unacknowledged ,
285+ ) ) ;
286+
271287 // Queue full?
272288 if response_json. messages_ready >= rabbitmq. queue_ready_healthy_below ||
273289 response_json. messages_unacknowledged >= rabbitmq. queue_nack_healthy_below
@@ -279,8 +295,10 @@ fn proceed_rabbitmq_queue_probe(rabbitmq: &ConfigPluginsRabbitMQ, rabbitmq_queue
279295 response_json. messages_unacknowledged
280296 ) ;
281297
282- return true ;
298+ return ( true , queue_counts ) ;
283299 }
300+
301+ return ( false , queue_counts) ;
284302 }
285303 } else {
286304 warn ! (
@@ -293,13 +311,14 @@ fn proceed_rabbitmq_queue_probe(rabbitmq: &ConfigPluginsRabbitMQ, rabbitmq_queue
293311 }
294312 }
295313
296- false
314+ ( false , None )
297315}
298316
299317fn dispatch_polls ( ) {
300318 // Probe hosts
301319 for probe_replica in map_poll_replicas ( ) {
302- let replica_status = proceed_replica_probe_with_retry ( & probe_replica. 3 , & probe_replica. 4 ) ;
320+ let ( replica_status, replica_latency) =
321+ proceed_replica_probe_with_retry ( & probe_replica. 3 , & probe_replica. 4 ) ;
303322
304323 debug ! (
305324 "replica probe result: {}:{}:{} => {:?}" ,
@@ -317,6 +336,9 @@ fn dispatch_polls() {
317336 if let Some ( ref mut node) = probe. nodes . get_mut ( & probe_replica. 1 ) {
318337 if let Some ( ref mut replica) = node. replicas . get_mut ( & probe_replica. 2 ) {
319338 replica. status = replica_status;
339+
340+ replica. metrics . latency =
341+ replica_latency. map ( |duration| duration. as_millis ( ) as u64 ) ;
320342 }
321343 }
322344 }
@@ -331,12 +353,12 @@ fn dispatch_plugins_rabbitmq(probe_id: String, node_id: String, queue: Option<St
331353 // Any queue for node?
332354 if let Some ( ref queue_value) = queue {
333355 // Check if RabbitMQ queue is loaded
334- let mut rabbitmq_queue_loaded = proceed_rabbitmq_queue_probe ( rabbitmq, queue_value) ;
356+ let mut rabbitmq_queue_load = proceed_rabbitmq_queue_probe ( rabbitmq, queue_value) ;
335357
336358 // Check once again? (the queue can be seen as loaded from check #1, but if we \
337359 // check again a few milliseconds later, it will actually be empty; so not loaded)
338360 // Notice: this prevents false-positive 'sick' statuses.
339- if rabbitmq_queue_loaded == true {
361+ if rabbitmq_queue_load . 0 == true {
340362 if let Some ( retry_delay) = rabbitmq. queue_loaded_retry_delay {
341363 debug ! (
342364 "rabbitmq queue is loaded, checking once again in {}ms: {}:{} [{}]" ,
@@ -349,7 +371,7 @@ fn dispatch_plugins_rabbitmq(probe_id: String, node_id: String, queue: Option<St
349371 thread:: sleep ( Duration :: from_millis ( retry_delay) ) ;
350372
351373 // Check again if RabbitMQ queue is loaded
352- rabbitmq_queue_loaded = proceed_rabbitmq_queue_probe ( rabbitmq, queue_value) ;
374+ rabbitmq_queue_load = proceed_rabbitmq_queue_probe ( rabbitmq, queue_value) ;
353375 }
354376 }
355377
@@ -358,7 +380,7 @@ fn dispatch_plugins_rabbitmq(probe_id: String, node_id: String, queue: Option<St
358380 & probe_id,
359381 & node_id,
360382 queue_value,
361- rabbitmq_queue_loaded
383+ rabbitmq_queue_load . 0
362384 ) ;
363385
364386 // Update replica status (write-lock the store)
@@ -368,9 +390,19 @@ fn dispatch_plugins_rabbitmq(probe_id: String, node_id: String, queue: Option<St
368390 if let Some ( ref mut probe) = store. states . probes . get_mut ( & probe_id) {
369391 if let Some ( ref mut node) = probe. nodes . get_mut ( & node_id) {
370392 for ( _, replica) in node. replicas . iter_mut ( ) {
371- // Only alter healthy replicas
372393 if let Some ( ref mut replica_load) = replica. load {
373- replica_load. queue = rabbitmq_queue_loaded;
394+ replica_load. queue = rabbitmq_queue_load. 0 ;
395+ }
396+
397+ // Store RabbitMQ metrics
398+ if let Some ( ( queue_ready, queue_nack) ) = rabbitmq_queue_load. 1 {
399+ replica. metrics . rabbitmq =
400+ Some ( ServiceStatesProbeNodeReplicaMetricsRabbitMQ {
401+ queue_ready : queue_ready,
402+ queue_nack : queue_nack,
403+ } ) ;
404+ } else {
405+ replica. metrics . rabbitmq = None
374406 }
375407 }
376408 }
@@ -440,6 +472,7 @@ pub fn initialize_store() {
440472 ServiceStatesProbeNodeReplica {
441473 status : Status :: Healthy ,
442474 url : Some ( replica_url) ,
475+ metrics : ServiceStatesProbeNodeReplicaMetrics :: default ( ) ,
443476 load : None ,
444477 report : None ,
445478 } ,
0 commit comments