Skip to content

Commit a6f8c0e

Browse files
Molter73Molter73
committed
feat(grpc): enforce a maximum number of reconnection attempts
If fact fails to connect to the gRPC server in the specified number of attemtps it will crash, which can be used as a clear signal in k8s that something is not working as expected. If the number of retries is set to 0, there will be no limit to the amount of times fact attempts to reconnect. In order to allow the reconnection failure to trigger an application wide crash, the main output task monitors the result of the grpc task's handle propagating the error further up. This method should allow for other output components to be added in the future and follow this same pattern without having to change anything outside the output module. The stdout component is not included in this logic because it has no condition that could merit an application wide crash.
1 parent 5dd2a85 commit a6f8c0e

5 files changed

Lines changed: 333 additions & 83 deletions

File tree

‎fact/src/config/mod.rs‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -332,6 +332,7 @@ pub struct BackoffConfig {
332332
max: Option<Duration>,
333333
jitter: Option<bool>,
334334
multiplier: Option<f64>,
335+
retries_max: Option<u64>,
335336
}
336337

337338
impl BackoffConfig {
@@ -348,6 +349,9 @@ impl BackoffConfig {
348349
if let Some(multiplier) = from.multiplier {
349350
self.multiplier = Some(multiplier);
350351
}
352+
if let Some(retries) = from.retries_max {
353+
self.retries_max = Some(retries);
354+
}
351355
}
352356

353357
pub fn initial(&self) -> Duration {
@@ -365,6 +369,10 @@ impl BackoffConfig {
365369
pub fn multiplier(&self) -> f64 {
366370
self.multiplier.unwrap_or(1.5)
367371
}
372+
373+
pub fn retries(&self) -> u64 {
374+
self.retries_max.unwrap_or(10)
375+
}
368376
}
369377

370378
impl TryFrom<&yaml::Hash> for BackoffConfig {
@@ -405,6 +413,12 @@ impl TryFrom<&yaml::Hash> for BackoffConfig {
405413
};
406414
backoff.multiplier = Some(multiplier);
407415
}
416+
"retries" => {
417+
let Some(retries) = v.as_i64() else {
418+
bail!("invalid grpc.backoff.retries: {v:?}");
419+
};
420+
backoff.retries_max = Some(retries as u64);
421+
}
408422
name => bail!("Invalid field 'grpc.backoff.{name}' with value: {v:?}"),
409423
}
410424
}
@@ -606,6 +620,13 @@ pub struct FactCli {
606620
#[arg(long, env = "FACT_GRPC_BACKOFF_JITTER")]
607621
backoff_jitter: Option<bool>,
608622

623+
/// Maximum number of times a gRPC connection will be attempted
624+
/// before giving up
625+
///
626+
/// 0 means infinite retries
627+
#[arg(long, env = "FACT_GRPC_BACKOFF_RETRIES_MAX")]
628+
backoff_retries_max: Option<u64>,
629+
609630
/// The port to bind for all exposed endpoints
610631
#[arg(long, short, env = "FACT_ENDPOINT_ADDRESS")]
611632
address: Option<SocketAddr>,
@@ -701,6 +722,7 @@ impl FactCli {
701722
max: self.backoff_max,
702723
jitter: self.backoff_jitter,
703724
multiplier: self.backoff_multiplier,
725+
retries_max: self.backoff_retries_max,
704726
},
705727
},
706728
endpoint: EndpointConfig {

‎fact/src/config/tests.rs‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -340,6 +340,23 @@ fn parsing() {
340340
..Default::default()
341341
},
342342
),
343+
(
344+
r#"
345+
grpc:
346+
backoff:
347+
retries: 5
348+
"#,
349+
FactConfig {
350+
grpc: GrpcConfig {
351+
backoff: BackoffConfig {
352+
retries_max: Some(5),
353+
..Default::default()
354+
},
355+
..Default::default()
356+
},
357+
..Default::default()
358+
},
359+
),
343360
(
344361
r#"
345362
grpc:
@@ -348,6 +365,7 @@ fn parsing() {
348365
max: 120
349366
jitter: false
350367
multiplier: 2
368+
retries: 5
351369
"#,
352370
FactConfig {
353371
grpc: GrpcConfig {
@@ -356,6 +374,7 @@ fn parsing() {
356374
max: Some(Duration::from_secs(120)),
357375
jitter: Some(false),
358376
multiplier: Some(2.0),
377+
retries_max: Some(5),
359378
},
360379
..Default::default()
361380
},
@@ -374,6 +393,7 @@ fn parsing() {
374393
max: 120
375394
jitter: false
376395
multiplier: 2
396+
retries: 5
377397
endpoint:
378398
address: 0.0.0.0:8080
379399
expose_metrics: true
@@ -396,6 +416,7 @@ fn parsing() {
396416
max: Some(Duration::from_secs(120)),
397417
jitter: Some(false),
398418
multiplier: Some(2.0),
419+
retries_max: Some(5),
399420
},
400421
},
401422
endpoint: EndpointConfig {
@@ -543,6 +564,22 @@ paths:
543564
"#,
544565
"invalid grpc.backoff.multiplier: Real(\"0.5\")",
545566
),
567+
(
568+
r#"
569+
grpc:
570+
backoff:
571+
retries: 0.5
572+
"#,
573+
"invalid grpc.backoff.retries: Real(\"0.5\")",
574+
),
575+
(
576+
r#"
577+
grpc:
578+
backoff:
579+
retries: true
580+
"#,
581+
"invalid grpc.backoff.retries: Boolean(true)",
582+
),
546583
(
547584
r#"
548585
grpc:
@@ -1058,6 +1095,51 @@ fn update() {
10581095
..Default::default()
10591096
},
10601097
),
1098+
(
1099+
r#"
1100+
grpc:
1101+
backoff:
1102+
retries: 5
1103+
"#,
1104+
FactConfig::default(),
1105+
FactConfig {
1106+
grpc: GrpcConfig {
1107+
backoff: BackoffConfig {
1108+
retries_max: Some(5),
1109+
..Default::default()
1110+
},
1111+
..Default::default()
1112+
},
1113+
..Default::default()
1114+
},
1115+
),
1116+
(
1117+
r#"
1118+
grpc:
1119+
backoff:
1120+
retries: 5
1121+
"#,
1122+
FactConfig {
1123+
grpc: GrpcConfig {
1124+
backoff: BackoffConfig {
1125+
retries_max: Some(10),
1126+
..Default::default()
1127+
},
1128+
..Default::default()
1129+
},
1130+
..Default::default()
1131+
},
1132+
FactConfig {
1133+
grpc: GrpcConfig {
1134+
backoff: BackoffConfig {
1135+
retries_max: Some(5),
1136+
..Default::default()
1137+
},
1138+
..Default::default()
1139+
},
1140+
..Default::default()
1141+
},
1142+
),
10611143
(
10621144
r#"
10631145
endpoint:
@@ -1425,6 +1507,7 @@ fn update() {
14251507
max: 120
14261508
jitter: false
14271509
multiplier: 3.0
1510+
retries: 5
14281511
endpoint:
14291512
address: 127.0.0.1:8080
14301513
expose_metrics: true
@@ -1447,6 +1530,7 @@ fn update() {
14471530
max: Some(Duration::from_secs(30)),
14481531
jitter: Some(true),
14491532
multiplier: Some(2.0),
1533+
retries_max: Some(20),
14501534
},
14511535
},
14521536
endpoint: EndpointConfig {
@@ -1474,6 +1558,7 @@ fn update() {
14741558
max: Some(Duration::from_secs(120)),
14751559
jitter: Some(false),
14761560
multiplier: Some(3.0),
1561+
retries_max: Some(5),
14771562
},
14781563
},
14791564
endpoint: EndpointConfig {
@@ -1525,6 +1610,7 @@ fn defaults() {
15251610
assert_eq!(config.grpc.backoff.max(), Duration::from_secs(60));
15261611
assert!(config.grpc.backoff.jitter());
15271612
assert_eq!(config.grpc.backoff.multiplier(), 1.5);
1613+
assert_eq!(config.grpc.backoff.retries(), 10);
15281614
}
15291615

15301616
static ENV_MUTEX: Mutex<()> = Mutex::new(());
@@ -1756,6 +1842,22 @@ fn env_vars() {
17561842
..Default::default()
17571843
},
17581844
),
1845+
(
1846+
EnvVar {
1847+
name: "FACT_GRPC_BACKOFF_RETRIES_MAX",
1848+
value: "5",
1849+
},
1850+
FactConfig {
1851+
grpc: GrpcConfig {
1852+
backoff: BackoffConfig {
1853+
retries_max: Some(5),
1854+
..Default::default()
1855+
},
1856+
..Default::default()
1857+
},
1858+
..Default::default()
1859+
},
1860+
),
17591861
(
17601862
EnvVar {
17611863
name: "FACT_ENDPOINT_ADDRESS",

‎fact/src/lib.rs‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ use rate_limiter::RateLimiter;
1010
use tokio::{
1111
signal::unix::{SignalKind, signal},
1212
sync::watch,
13+
task::JoinError,
1314
};
1415

1516
mod bpf;
@@ -64,6 +65,16 @@ pub fn log_system_information() {
6465
info!("Hostname: {}", get_hostname());
6566
}
6667

68+
fn log_task_result(component: &str, res: Result<anyhow::Result<()>, JoinError>) {
69+
match res {
70+
Ok(Ok(_)) => {}
71+
Ok(Err(e)) => {
72+
warn!("{component} worker errored out: {e:?}");
73+
}
74+
Err(e) => warn!("{component} task errored out: {e:?}"),
75+
}
76+
}
77+
6778
pub async fn run(config: FactConfig) -> anyhow::Result<()> {
6879
// Log system information as early as possible so we have it
6980
// available in case of a crash
@@ -99,13 +110,13 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> {
99110
exporter.metrics.rate_limiter.clone(),
100111
)?;
101112

102-
output::start(
113+
let mut output_handle = output::start(
103114
rx,
104115
running.subscribe(),
105116
exporter.metrics.output.clone(),
106117
reloader.grpc(),
107118
reloader.config().json(),
108-
)?;
119+
);
109120
let mut host_scanner_handle = host_scanner.start();
110121
let mut rate_limiter_handle = rate_limiter.start();
111122
endpoints::Server::new(exporter.clone(), reloader.endpoint(), running.subscribe()).start();
@@ -120,30 +131,19 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> {
120131
_ = sigterm.recv() => break,
121132
_ = sighup.recv() => config_trigger.notify_one(),
122133
res = bpf_handle.borrow_mut() => {
123-
match res {
124-
Ok(res) => if let Err(e) = res {
125-
warn!("BPF worker errored out: {e:?}");
126-
}
127-
Err(e) => warn!("BPF task errored out: {e:?}"),
128-
}
134+
log_task_result("BPF", res);
129135
break;
130136
}
131137
res = host_scanner_handle.borrow_mut() => {
132-
match res {
133-
Ok(res) => if let Err(e) = res {
134-
warn!("HostScanner worker errored out: {e:?}");
135-
}
136-
Err(e) => warn!("HostScanner task errored out: {e:?}"),
137-
}
138+
log_task_result("HostScanner", res);
138139
break;
139140
}
140141
res = rate_limiter_handle.borrow_mut() => {
141-
match res {
142-
Ok(res) => if let Err(e) = res {
143-
warn!("Rate limiter worker errored out: {e:?}");
144-
}
145-
Err(e) => warn!("Rate limiter task errored out: {e:?}"),
146-
}
142+
log_task_result("Rate limiter", res);
143+
break;
144+
}
145+
res = output_handle.borrow_mut() => {
146+
log_task_result("Output", res);
147147
break;
148148
}
149149
}

0 commit comments

Comments
 (0)