runnerd_core/api/
events.rs

1//! What `GET /v1/events` streams: one enum of everything that happens, wrapped with its ID and time. Durable events
2//! are stored and get an ID, so a client resumes after them with `Last-Event-ID`; live events (samples and log
3//! lines) are only sent to connected clients, and a client that missed them reads the job's samples or log instead.
4
5use serde::{Deserialize, Serialize};
6
7use super::jobs::{HostSample, JobSummary, Sample, Step};
8use super::{Budget, RepositoryStatus, SlotStatus};
9
10/// One event as `GET /v1/events` sends it: the SSE `data` field. The SSE `event` field is `event.kind()` and the
11/// SSE `id` field is `id`.
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
13pub struct EventEnvelope {
14    /// Increases with every durable event; `None` for live events.
15    pub id: Option<u64>,
16    /// When it happened.
17    pub at: u64,
18    /// What happened.
19    #[serde(flatten)]
20    pub event: Event,
21}
22
23/// Everything that happens, tagged by `type`.
24#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(tag = "type")]
26pub enum Event {
27    /// A job that runners of this daemon may run was queued.
28    #[serde(rename = "job.queued")]
29    JobQueued {
30        /// The job.
31        job: JobSummary,
32    },
33    /// A runner of this daemon took the job.
34    #[serde(rename = "job.started")]
35    JobStarted {
36        /// The job.
37        job: JobSummary,
38    },
39    /// A step of a running job started or completed.
40    #[serde(rename = "job.step")]
41    JobStep {
42        /// The job's ID.
43        job: u64,
44        /// The step as it is now.
45        step: Step,
46    },
47    /// The job's record is final: it completed and its runner was removed, it was lost, or it ran elsewhere.
48    #[serde(rename = "job.finished")]
49    JobFinished {
50        /// The job.
51        job: JobSummary,
52    },
53    /// A slot's runner started, took a job, or was removed.
54    #[serde(rename = "slot.changed")]
55    SlotChanged {
56        /// The profile.
57        profile: String,
58        /// The slot as it is now.
59        slot: SlotStatus,
60    },
61    /// The memory caps in use changed.
62    #[serde(rename = "budget.changed")]
63    BudgetChanged {
64        /// The budget as it is now.
65        budget: Budget,
66    },
67    /// A repository was paused or resumed, its queue changed, a reload changed its profile or labels, or a reload
68    /// added it.
69    #[serde(rename = "repository.changed")]
70    RepositoryChanged {
71        /// The repository as it is now.
72        repository: RepositoryStatus,
73    },
74    /// A reload removed the repository from the config.
75    #[serde(rename = "repository.removed")]
76    RepositoryRemoved {
77        /// Its `owner/name`.
78        repository: String,
79    },
80    /// A profile's snapshot check finished.
81    #[serde(rename = "cache.reseeded")]
82    CacheReseeded {
83        /// The profile.
84        profile: String,
85        /// What the seed container printed, such as why it did not reseed.
86        output: String,
87        /// Set when reseeding failed.
88        error: Option<String>,
89    },
90    /// Live: a running job's container was sampled.
91    #[serde(rename = "job.sample")]
92    JobSample {
93        /// The job's ID.
94        job: u64,
95        /// The sample.
96        sample: Sample,
97    },
98    /// Live: the slice runners run in was sampled.
99    #[serde(rename = "host.sample")]
100    HostSample {
101        /// The sample.
102        sample: HostSample,
103    },
104    /// Live: a line of a running job's log was stored. Only sent with `logs=true`.
105    #[serde(rename = "log.line")]
106    LogLine {
107        /// The job's ID.
108        job: u64,
109        /// The line's number in the job's log, 0-based.
110        line: u64,
111        /// The line, without its newline.
112        text: String,
113    },
114}
115
116impl Event {
117    /// The `type` tag, also the SSE event name.
118    pub fn kind(&self) -> &'static str {
119        match self {
120            Event::JobQueued { .. } => "job.queued",
121            Event::JobStarted { .. } => "job.started",
122            Event::JobStep { .. } => "job.step",
123            Event::JobFinished { .. } => "job.finished",
124            Event::SlotChanged { .. } => "slot.changed",
125            Event::BudgetChanged { .. } => "budget.changed",
126            Event::RepositoryChanged { .. } => "repository.changed",
127            Event::RepositoryRemoved { .. } => "repository.removed",
128            Event::CacheReseeded { .. } => "cache.reseeded",
129            Event::JobSample { .. } => "job.sample",
130            Event::HostSample { .. } => "host.sample",
131            Event::LogLine { .. } => "log.line",
132        }
133    }
134
135    /// Whether it is stored and can be resumed after; live events are not.
136    pub fn is_durable(&self) -> bool {
137        !matches!(
138            self,
139            Event::JobSample { .. } | Event::HostSample { .. } | Event::LogLine { .. }
140        )
141    }
142
143    /// The job it is about, if any.
144    pub fn job(&self) -> Option<u64> {
145        match self {
146            Event::JobQueued { job } | Event::JobStarted { job } | Event::JobFinished { job } => {
147                Some(job.id)
148            }
149            Event::JobStep { job, .. }
150            | Event::JobSample { job, .. }
151            | Event::LogLine { job, .. } => Some(*job),
152            Event::SlotChanged { .. }
153            | Event::BudgetChanged { .. }
154            | Event::RepositoryChanged { .. }
155            | Event::RepositoryRemoved { .. }
156            | Event::CacheReseeded { .. }
157            | Event::HostSample { .. } => None,
158        }
159    }
160}
161
162/// `GET /v1/events` query parameters. Resume with the `Last-Event-ID` header.
163#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
164pub struct EventQuery {
165    /// Only events about this job.
166    pub job: Option<u64>,
167    /// Also send `log.line` events.
168    #[serde(default)]
169    pub logs: bool,
170}
171
172impl EventQuery {
173    /// Whether an event passes the filter.
174    pub fn matches(&self, event: &Event) -> bool {
175        (self.logs || !matches!(event, Event::LogLine { .. }))
176            && self.job.is_none_or(|job| event.job() == Some(job))
177    }
178}
runnerd_core/api/events.rs

runnerd_core/api/
events.rs
