1use pharos_corefile::{Cursor, Error};
4use pharos_server::{Context, Next, Outcome, Plugin};
5use regex::Regex;
6use std::{
7 sync::{
8 Arc,
9 atomic::{AtomicU64, Ordering},
10 },
11 time::Duration,
12};
13
14#[derive(Debug)]
15struct Pattern {
16 period: Duration,
17 period_text: String,
18 regex: Regex,
19 level: &'static str,
20 show_first: bool,
21 count: AtomicU64,
22 stop: tokio::sync::watch::Sender<bool>,
23}
24
25#[derive(Debug)]
26pub struct Errors {
27 patterns: Vec<Arc<Pattern>>,
28 pub stacktrace: bool,
29}
30
31pub(crate) fn parse(cursor: &mut Cursor<'_>) -> Result<Errors, Error> {
32 let mut patterns = Vec::new();
33 let mut stacktrace = false;
34 while cursor.next() {
35 let args = cursor.remaining_args();
36 if args.len() > 1 || args.first().is_some_and(|arg| arg != "stdout") {
37 return Err(cursor.arg_err());
38 }
39 while cursor.next_block() {
40 match cursor.val() {
41 "stacktrace" => stacktrace = true,
42 "consolidate" => {
43 let args = cursor.remaining_args();
44 if !(2..=4).contains(&args.len()) || args[1].len() > 10_000 {
45 return Err(cursor.arg_err());
46 }
47 let period = crate::health::parse_duration(&args[0])
48 .ok_or_else(|| cursor.error("invalid duration"))?;
49 let period_text = if let Some(value) = args[0].strip_prefix('-') {
50 let magnitude = crate::health::parse_duration(value)
51 .unwrap_or(Duration::from_nanos(1_u64 << 63));
52 if magnitude.is_zero() {
53 "0s".to_owned()
54 } else {
55 format!("-{}", display_duration(magnitude))
56 }
57 } else {
58 display_duration(period)
59 };
60 let regex =
61 Regex::new(&args[1]).map_err(|err| cursor.error(err.to_string()))?;
62 let mut level = None;
63 let mut show_first = false;
64 for arg in &args[2..] {
65 if arg == "show_first" {
66 show_first = true;
67 } else {
68 let value = match arg.as_str() {
69 "warning" => "WARNING",
70 "error" => "ERROR",
71 "info" => "INFO",
72 "debug" => "DEBUG",
73 _ => return Err(cursor.error("invalid consolidate option")),
74 };
75 if level.is_some() || show_first {
76 return Err(cursor.error("invalid consolidate option order"));
77 }
78 level = Some(value);
79 }
80 }
81 patterns.push(Arc::new(Pattern {
82 period,
83 period_text,
84 regex,
85 level: level.unwrap_or("ERROR"),
86 show_first,
87 count: AtomicU64::new(0),
88 stop: tokio::sync::watch::channel(false).0,
89 }));
90 }
91 _ => return Err(cursor.error("unknown errors option")),
92 }
93 }
94 }
95 Ok(Errors {
96 patterns,
97 stacktrace,
98 })
99}
100
101impl Plugin for Errors {
102 async fn serve<N: Next + Send>(&self, cx: &mut Context<'_>, next: N) -> Outcome {
103 let outcome = next.run(cx).await;
104 if let Outcome::Error(error) = outcome {
105 cx.metadata
106 .insert(pharos_server::logging::ERROR_LOGGED_METADATA, "true");
107 let message = error.to_string();
108 let mut level = "ERROR";
109 for pattern in &self.patterns {
110 if pattern.regex.is_match(&message) {
111 let Ok(previous) =
112 pattern
113 .count
114 .try_update(Ordering::Relaxed, Ordering::Relaxed, |count| {
115 (count != STOPPED).then(|| count.saturating_add(1).min(STOPPED - 1))
116 })
117 else {
118 level = pattern.level;
119 break;
120 };
121 if previous == 0 {
122 let deadline = tokio::time::Instant::now() + pattern.period;
123 let timer_pattern = Arc::clone(pattern);
124 let mut stop = pattern.stop.subscribe();
125 tokio::spawn(async move {
126 if !*stop.borrow() {
127 tokio::select! {
128 () = tokio::time::sleep_until(deadline) => timer_pattern.flush(false),
129 _ = stop.changed() => {},
130 }
131 }
132 });
133 if pattern.show_first {
134 level = pattern.level;
135 break;
136 }
137 }
138 return outcome;
139 }
140 }
141 if let Some(question) = cx.query.questions().next() {
142 let name = question.name.to_string().to_lowercase();
143 pharos_server::logging::log(
144 level,
145 "errors",
146 &format!("2 {name} {}: {message}", question.qtype),
147 );
148 } else {
149 pharos_server::logging::log(level, "errors", &format!("2 . : {message}"));
150 }
151 }
152 outcome
153 }
154}
155
156const STOPPED: u64 = u64::MAX;
157
158impl Pattern {
159 fn flush(&self, stop: bool) {
160 let count = if stop {
161 self.count.swap(STOPPED, Ordering::Relaxed)
162 } else {
163 self.count
164 .try_update(Ordering::Relaxed, Ordering::Relaxed, |count| {
165 (count != STOPPED).then_some(0)
166 })
167 .unwrap_or(STOPPED)
168 };
169 if count != STOPPED && count != 0 && (count > 1 || !self.show_first) {
170 pharos_server::logging::log(
171 self.level,
172 "errors",
173 &format!(
174 "{count} errors like '{}' occurred in last {}",
175 self.regex.as_str(),
176 self.period_text
177 ),
178 );
179 }
180 }
181}
182
183impl Errors {
184 pub(crate) fn register_shutdown(&self, lifecycle: &mut pharos_server::Lifecycle) {
185 let patterns = self.patterns.clone();
186 lifecycle.register_sync(pharos_server::Phase::Shutdown, move || {
187 for pattern in &patterns {
188 pattern.flush(true);
189 pattern.stop.send_replace(true);
190 }
191 Ok(())
192 });
193 }
194}
195
196fn display_duration(duration: Duration) -> String {
197 let nanos = duration.as_nanos();
198 if nanos == 0 {
199 return "0s".to_owned();
200 }
201 if nanos < 1_000_000_000 {
202 let (unit, scale) = if nanos < 1_000 {
203 ("ns", 1)
204 } else if nanos < 1_000_000 {
205 ("µs", 1_000)
206 } else {
207 ("ms", 1_000_000)
208 };
209 return decimal(nanos, scale, unit);
210 }
211 let seconds = nanos / 1_000_000_000;
212 let mut out = String::new();
213 if seconds >= 3600 {
214 out.push_str(&format!("{}h", seconds / 3600));
215 }
216 if seconds >= 60 {
217 out.push_str(&format!("{}m", seconds % 3600 / 60));
218 }
219 out.push_str(&decimal(nanos % 60_000_000_000, 1_000_000_000, "s"));
220 out
221}
222
223fn decimal(value: u128, scale: u128, unit: &str) -> String {
224 let whole = value / scale;
225 let fraction = value % scale;
226 if fraction == 0 {
227 return format!("{whole}{unit}");
228 }
229 let width = scale.ilog10() as usize;
230 let fraction = format!("{fraction:0width$}");
231 format!("{whole}.{}{unit}", fraction.trim_end_matches('0'))
232}