azums-core 1.0.0

Zero-dependency core traits, models, and QueueError for azums
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

use crate::model::Job;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobObservationEvent {
    pub at: DateTime<Utc>,
    pub job_id: Uuid,
    pub attempt: Option<i32>,
    pub worker_id: Option<String>,
    pub queue: String,
    pub duration_ms: Option<i32>,
    pub status: String,
    pub retry_count: i32,
    pub error: Option<String>,
    pub trace_id: Option<String>,
}

impl JobObservationEvent {
    /// Returns OpenTelemetry-compatible span attributes for this job lifecycle observation.
    pub fn span_attributes(&self) -> BTreeMap<&'static str, String> {
        let mut attrs = BTreeMap::new();
        attrs.insert("azums.job_id", self.job_id.to_string());
        attrs.insert("azums.queue", self.queue.clone());
        attrs.insert("azums.status", self.status.clone());
        attrs.insert("azums.retry_count", self.retry_count.to_string());

        if let Some(attempt) = self.attempt {
            attrs.insert("azums.attempt", attempt.to_string());
        }
        if let Some(worker_id) = &self.worker_id {
            attrs.insert("azums.worker_id", worker_id.clone());
        }
        if let Some(duration_ms) = self.duration_ms {
            attrs.insert("azums.duration_ms", duration_ms.to_string());
        }
        if let Some(error) = &self.error {
            attrs.insert("azums.error", error.clone());
        }
        if let Some(trace_id) = &self.trace_id {
            attrs.insert("trace_id", trace_id.clone());
        }

        attrs
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobExplanation {
    pub job_id: Uuid,
    pub job_type: String,
    pub queue: String,
    pub status: String,
    pub retry_count: i32,
    pub last_worker_id: Option<String>,
    pub last_error: Option<String>,
    pub trace_id: Option<String>,
    pub events: Vec<JobObservationEvent>,
    pub summary: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueMetrics {
    pub at: DateTime<Utc>,
    pub queue: String,
    pub jobs_total: u64,
    pub jobs_completed: u64,
    pub jobs_failed: u64,
    pub jobs_retried: u64,
    pub jobs_dlq: u64,
    pub queue_depth: u64,
    pub execution_latency_ms_avg: f64,
    pub claim_latency_ms_avg: f64,
    pub retry_latency_ms_avg: f64,
    pub worker_count: u64,
}

#[async_trait]
pub trait ObservabilityBackend: Send + Sync {
    async fn explain_job(&self, job_id: Uuid) -> anyhow::Result<Option<JobExplanation>>;

    async fn queue_metrics(&self, queue: Option<&str>) -> anyhow::Result<Vec<QueueMetrics>>;
}

pub fn trace_id_from_job(job: &Job) -> Option<String> {
    job.payload
        .get("trace_id")
        .and_then(|value| value.as_str())
        .map(ToString::to_string)
        .or_else(|| {
            job.payload
                .get("metadata")
                .and_then(|value| value.get("trace_id"))
                .and_then(|value| value.as_str())
                .map(ToString::to_string)
        })
}