1use crate::execution_guard::ExecutionGuardScope;
2use crate::model::{RunSkipReason, RunStatus};
3use crate::store::StoreOperation;
4use chrono::{DateTime, Utc};
5use log::{debug, info, warn};
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum StateLoadSource {
9 New,
10 Restored,
11 Repaired,
12}
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum SchedulerStopReason {
16 Cancelled,
17 Shutdown,
18 Terminal,
19 ChannelClosed,
20}
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub enum PauseScope {
24 Local,
25 Shared,
26}
27
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub enum SchedulerEvent {
30 StateLoaded {
31 job_id: String,
32 trigger_count: u32,
33 next_run_at: Option<DateTime<Utc>>,
34 source: StateLoadSource,
35 },
36 StateRepaired {
37 job_id: String,
38 trigger_count: u32,
39 previous_next_run_at: Option<DateTime<Utc>>,
40 repaired_next_run_at: Option<DateTime<Utc>>,
41 },
42 TriggerEmitted {
43 job_id: String,
44 scheduled_at: DateTime<Utc>,
45 catch_up: bool,
46 trigger_count: u32,
47 },
48 RunSkipped {
49 job_id: String,
50 scheduled_at: DateTime<Utc>,
51 catch_up: bool,
52 trigger_count: u32,
53 reason: RunSkipReason,
54 },
55 RunCompleted {
56 job_id: String,
57 scheduled_at: DateTime<Utc>,
58 catch_up: bool,
59 trigger_count: u32,
60 status: RunStatus,
61 error: Option<String>,
62 },
63 ExecutionGuardAcquired {
64 job_id: String,
65 resource_id: String,
66 scope: ExecutionGuardScope,
67 lease_key: String,
68 scheduled_at: Option<DateTime<Utc>>,
69 catch_up: bool,
70 trigger_count: u32,
71 },
72 ExecutionGuardContended {
73 job_id: String,
74 resource_id: String,
75 scope: ExecutionGuardScope,
76 scheduled_at: Option<DateTime<Utc>>,
77 catch_up: bool,
78 trigger_count: u32,
79 },
80 ExecutionGuardRenewed {
81 job_id: String,
82 resource_id: String,
83 scope: ExecutionGuardScope,
84 lease_key: String,
85 scheduled_at: Option<DateTime<Utc>>,
86 catch_up: bool,
87 trigger_count: u32,
88 renewal_count: u32,
89 },
90 ExecutionGuardRenewFailed {
91 job_id: String,
92 resource_id: String,
93 scope: ExecutionGuardScope,
94 lease_key: String,
95 scheduled_at: Option<DateTime<Utc>>,
96 catch_up: bool,
97 trigger_count: u32,
98 renewal_count: u32,
99 failed_renewal_count: u32,
100 error: String,
101 },
102 ExecutionGuardLost {
103 job_id: String,
104 resource_id: String,
105 scope: ExecutionGuardScope,
106 lease_key: String,
107 scheduled_at: Option<DateTime<Utc>>,
108 catch_up: bool,
109 trigger_count: u32,
110 renewal_count: u32,
111 failed_renewal_count: u32,
112 },
113 ExecutionGuardReleased {
114 job_id: String,
115 resource_id: String,
116 scope: ExecutionGuardScope,
117 lease_key: String,
118 scheduled_at: Option<DateTime<Utc>>,
119 catch_up: bool,
120 trigger_count: u32,
121 },
122 ExecutionGuardReleaseFailed {
123 job_id: String,
124 resource_id: String,
125 scope: ExecutionGuardScope,
126 lease_key: String,
127 scheduled_at: Option<DateTime<Utc>>,
128 catch_up: bool,
129 trigger_count: u32,
130 error: String,
131 },
132 StoreDegraded {
133 job_id: String,
134 operation: StoreOperation,
135 error: String,
136 },
137 StoreRecovering {
138 job_id: String,
139 operation: StoreOperation,
140 },
141 StoreRecovered {
142 job_id: String,
143 operation: StoreOperation,
144 },
145 StoreRecoveryFailed {
146 job_id: String,
147 operation: StoreOperation,
148 error: String,
149 },
150 StoreRecoveryConflict {
151 job_id: String,
152 operation: StoreOperation,
153 error: String,
154 },
155 ExecutionGuardDegraded {
156 job_id: String,
157 resource_id: String,
158 scope: ExecutionGuardScope,
159 scheduled_at: Option<DateTime<Utc>>,
160 catch_up: bool,
161 trigger_count: u32,
162 error: String,
163 },
164 ExecutionGuardRecovered {
165 job_id: String,
166 resource_id: String,
167 scope: ExecutionGuardScope,
168 scheduled_at: Option<DateTime<Utc>>,
169 catch_up: bool,
170 trigger_count: u32,
171 },
172 TerminalStateDeleted {
173 job_id: String,
174 trigger_count: u32,
175 },
176 SchedulerPaused {
177 job_id: String,
178 trigger_count: u32,
179 scope: PauseScope,
180 },
181 SchedulerResumed {
182 job_id: String,
183 trigger_count: u32,
184 scope: PauseScope,
185 },
186 SchedulerStopped {
187 job_id: String,
188 trigger_count: u32,
189 reason: SchedulerStopReason,
190 },
191}
192
193pub trait SchedulerObserver: Send + Sync + 'static {
194 fn on_event(&self, event: &SchedulerEvent);
195}
196
197#[derive(Debug, Clone, Copy, Default)]
198pub struct NoopObserver;
199
200impl SchedulerObserver for NoopObserver {
201 fn on_event(&self, _event: &SchedulerEvent) {}
202}
203
204#[derive(Debug, Clone, Copy, Default)]
205pub struct LogObserver;
206
207impl SchedulerObserver for LogObserver {
208 fn on_event(&self, event: &SchedulerEvent) {
209 match event {
210 SchedulerEvent::StateLoaded {
211 job_id,
212 trigger_count,
213 next_run_at,
214 source,
215 } => info!(
216 "scheduler state loaded job_id={} source={:?} trigger_count={} next_run_at={:?}",
217 job_id, source, trigger_count, next_run_at
218 ),
219 SchedulerEvent::StateRepaired {
220 job_id,
221 trigger_count,
222 previous_next_run_at,
223 repaired_next_run_at,
224 } => warn!(
225 "scheduler repaired state job_id={} trigger_count={} previous_next_run_at={:?} repaired_next_run_at={:?}",
226 job_id, trigger_count, previous_next_run_at, repaired_next_run_at
227 ),
228 SchedulerEvent::TriggerEmitted {
229 job_id,
230 scheduled_at,
231 catch_up,
232 trigger_count,
233 } => debug!(
234 "scheduler trigger emitted job_id={} scheduled_at={} catch_up={} trigger_count={}",
235 job_id, scheduled_at, catch_up, trigger_count
236 ),
237 SchedulerEvent::RunSkipped {
238 job_id,
239 scheduled_at,
240 catch_up,
241 trigger_count,
242 reason,
243 } => debug!(
244 "scheduler run skipped job_id={} scheduled_at={} catch_up={} trigger_count={} reason={}",
245 job_id, scheduled_at, catch_up, trigger_count, reason
246 ),
247 SchedulerEvent::RunCompleted {
248 job_id,
249 scheduled_at,
250 catch_up,
251 trigger_count,
252 status,
253 error,
254 } => debug!(
255 "scheduler run completed job_id={} scheduled_at={} catch_up={} trigger_count={} status={:?} error={:?}",
256 job_id, scheduled_at, catch_up, trigger_count, status, error
257 ),
258 SchedulerEvent::ExecutionGuardAcquired {
259 job_id,
260 resource_id,
261 scope,
262 lease_key,
263 scheduled_at,
264 catch_up,
265 trigger_count,
266 } => debug!(
267 "scheduler execution guard acquired job_id={} resource_id={} scope={:?} lease_key={} scheduled_at={:?} catch_up={} trigger_count={}",
268 job_id, resource_id, scope, lease_key, scheduled_at, catch_up, trigger_count
269 ),
270 SchedulerEvent::ExecutionGuardContended {
271 job_id,
272 resource_id,
273 scope,
274 scheduled_at,
275 catch_up,
276 trigger_count,
277 } => debug!(
278 "scheduler execution guard contended job_id={} resource_id={} scope={:?} scheduled_at={:?} catch_up={} trigger_count={}",
279 job_id, resource_id, scope, scheduled_at, catch_up, trigger_count
280 ),
281 SchedulerEvent::ExecutionGuardRenewed {
282 job_id,
283 resource_id,
284 scope,
285 lease_key,
286 scheduled_at,
287 catch_up,
288 trigger_count,
289 renewal_count,
290 } => debug!(
291 "scheduler execution guard renewed job_id={} resource_id={} scope={:?} lease_key={} scheduled_at={:?} catch_up={} trigger_count={} renewal_count={}",
292 job_id,
293 resource_id,
294 scope,
295 lease_key,
296 scheduled_at,
297 catch_up,
298 trigger_count,
299 renewal_count
300 ),
301 SchedulerEvent::ExecutionGuardRenewFailed {
302 job_id,
303 resource_id,
304 scope,
305 lease_key,
306 scheduled_at,
307 catch_up,
308 trigger_count,
309 renewal_count,
310 failed_renewal_count,
311 error,
312 } => warn!(
313 "scheduler execution guard renew failed job_id={} resource_id={} scope={:?} lease_key={} scheduled_at={:?} catch_up={} trigger_count={} renewal_count={} failed_renewal_count={} error={}",
314 job_id,
315 resource_id,
316 scope,
317 lease_key,
318 scheduled_at,
319 catch_up,
320 trigger_count,
321 renewal_count,
322 failed_renewal_count,
323 error
324 ),
325 SchedulerEvent::ExecutionGuardLost {
326 job_id,
327 resource_id,
328 scope,
329 lease_key,
330 scheduled_at,
331 catch_up,
332 trigger_count,
333 renewal_count,
334 failed_renewal_count,
335 } => warn!(
336 "scheduler execution guard lost job_id={} resource_id={} scope={:?} lease_key={} scheduled_at={:?} catch_up={} trigger_count={} renewal_count={} failed_renewal_count={}",
337 job_id,
338 resource_id,
339 scope,
340 lease_key,
341 scheduled_at,
342 catch_up,
343 trigger_count,
344 renewal_count,
345 failed_renewal_count
346 ),
347 SchedulerEvent::ExecutionGuardReleased {
348 job_id,
349 resource_id,
350 scope,
351 lease_key,
352 scheduled_at,
353 catch_up,
354 trigger_count,
355 } => debug!(
356 "scheduler execution guard released job_id={} resource_id={} scope={:?} lease_key={} scheduled_at={:?} catch_up={} trigger_count={}",
357 job_id, resource_id, scope, lease_key, scheduled_at, catch_up, trigger_count
358 ),
359 SchedulerEvent::ExecutionGuardReleaseFailed {
360 job_id,
361 resource_id,
362 scope,
363 lease_key,
364 scheduled_at,
365 catch_up,
366 trigger_count,
367 error,
368 } => warn!(
369 "scheduler execution guard release failed job_id={} resource_id={} scope={:?} lease_key={} scheduled_at={:?} catch_up={} trigger_count={} error={}",
370 job_id, resource_id, scope, lease_key, scheduled_at, catch_up, trigger_count, error
371 ),
372 SchedulerEvent::StoreDegraded {
373 job_id,
374 operation,
375 error,
376 } => warn!(
377 "scheduler store degraded job_id={} operation={:?} error={}",
378 job_id, operation, error
379 ),
380 SchedulerEvent::StoreRecovering { job_id, operation } => info!(
381 "scheduler store recovering job_id={} operation={:?}",
382 job_id, operation
383 ),
384 SchedulerEvent::StoreRecovered { job_id, operation } => info!(
385 "scheduler store recovered job_id={} operation={:?}",
386 job_id, operation
387 ),
388 SchedulerEvent::StoreRecoveryFailed {
389 job_id,
390 operation,
391 error,
392 } => warn!(
393 "scheduler store recovery failed job_id={} operation={:?} error={}",
394 job_id, operation, error
395 ),
396 SchedulerEvent::StoreRecoveryConflict {
397 job_id,
398 operation,
399 error,
400 } => warn!(
401 "scheduler store recovery conflict job_id={} operation={:?} error={}",
402 job_id, operation, error
403 ),
404 SchedulerEvent::ExecutionGuardDegraded {
405 job_id,
406 resource_id,
407 scope,
408 scheduled_at,
409 catch_up,
410 trigger_count,
411 error,
412 } => warn!(
413 "scheduler execution guard degraded job_id={} resource_id={} scope={:?} scheduled_at={:?} catch_up={} trigger_count={} error={}",
414 job_id, resource_id, scope, scheduled_at, catch_up, trigger_count, error
415 ),
416 SchedulerEvent::ExecutionGuardRecovered {
417 job_id,
418 resource_id,
419 scope,
420 scheduled_at,
421 catch_up,
422 trigger_count,
423 } => info!(
424 "scheduler execution guard recovered job_id={} resource_id={} scope={:?} scheduled_at={:?} catch_up={} trigger_count={}",
425 job_id, resource_id, scope, scheduled_at, catch_up, trigger_count
426 ),
427 SchedulerEvent::TerminalStateDeleted {
428 job_id,
429 trigger_count,
430 } => info!(
431 "scheduler deleted terminal state job_id={} trigger_count={}",
432 job_id, trigger_count
433 ),
434 SchedulerEvent::SchedulerPaused {
435 job_id,
436 trigger_count,
437 scope,
438 } => info!(
439 "scheduler paused job_id={} trigger_count={} scope={:?}",
440 job_id, trigger_count, scope
441 ),
442 SchedulerEvent::SchedulerResumed {
443 job_id,
444 trigger_count,
445 scope,
446 } => info!(
447 "scheduler resumed job_id={} trigger_count={} scope={:?}",
448 job_id, trigger_count, scope
449 ),
450 SchedulerEvent::SchedulerStopped {
451 job_id,
452 trigger_count,
453 reason,
454 } => info!(
455 "scheduler stopped job_id={} trigger_count={} reason={:?}",
456 job_id, trigger_count, reason
457 ),
458 }
459 }
460}