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::{
9 Budget, ProfileStatus, Record, RecordRemoval, RepositoryStatus, SizingChange, SlotStatus,
10};
11
12/// One event as `GET /v1/events` sends it: the SSE `data` field. The SSE `event` field is `event.kind()` and the
13/// SSE `id` field is `id`.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct EventEnvelope {
16 /// Increases with every durable event; `None` for live events.
17 pub id: Option<u64>,
18 /// When it happened.
19 pub at: u64,
20 /// What happened.
21 #[serde(flatten)]
22 pub event: Event,
23}
24
25/// Everything that happens, tagged by `type`.
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(tag = "type")]
28pub enum Event {
29 /// A job that runners of this daemon may run was queued.
30 #[serde(rename = "job.queued")]
31 JobQueued {
32 /// The job.
33 job: JobSummary,
34 },
35 /// A runner of this daemon took the job.
36 #[serde(rename = "job.started")]
37 JobStarted {
38 /// The job.
39 job: JobSummary,
40 },
41 /// A step of a running job started or completed.
42 #[serde(rename = "job.step")]
43 JobStep {
44 /// The job's ID.
45 job: u64,
46 /// The step as it is now.
47 step: Step,
48 },
49 /// The job's record is final: it completed and its runner was removed, it was lost, or it ran elsewhere.
50 #[serde(rename = "job.finished")]
51 JobFinished {
52 /// The job.
53 job: JobSummary,
54 },
55 /// A slot's runner started, took a job, or was removed.
56 #[serde(rename = "slot.changed")]
57 SlotChanged {
58 /// The profile.
59 profile: String,
60 /// The slot as it is now.
61 slot: SlotStatus,
62 },
63 /// The daemon started or a reload added the profile, a reload changed its image, CPUs, memory or slots, or it
64 /// started or stopped reseeding. A slot's runner changing is `slot.changed` only; this event carries every slot as it is now, so a
65 /// client can replace the whole profile.
66 #[serde(rename = "profile.changed")]
67 ProfileChanged {
68 /// The profile as it is now.
69 profile: ProfileStatus,
70 },
71 /// A reload removed the profile and its last runner and reseed are gone.
72 #[serde(rename = "profile.removed")]
73 ProfileRemoved {
74 /// Its name.
75 profile: String,
76 },
77 /// The memory caps in use changed.
78 #[serde(rename = "budget.changed")]
79 BudgetChanged {
80 /// The budget as it is now.
81 budget: Budget,
82 },
83 /// A repository was paused or resumed, its queue changed, a reload changed its profile or labels, or a reload
84 /// added it.
85 #[serde(rename = "repository.changed")]
86 RepositoryChanged {
87 /// The repository as it is now.
88 repository: RepositoryStatus,
89 },
90 /// A reload removed the repository from the config.
91 #[serde(rename = "repository.removed")]
92 RepositoryRemoved {
93 /// Its `owner/name`.
94 repository: String,
95 },
96 /// A profile's snapshot check finished.
97 #[serde(rename = "cache.reseeded")]
98 CacheReseeded {
99 /// The profile.
100 profile: String,
101 /// What the seed container printed, such as why it did not reseed.
102 output: String,
103 /// Set when reseeding failed.
104 error: Option<String>,
105 },
106 /// A failed job's runner was kept as a record.
107 #[serde(rename = "record.created")]
108 RecordCreated {
109 /// The record.
110 record: Record,
111 },
112 /// A record was removed.
113 #[serde(rename = "record.deleted")]
114 RecordDeleted {
115 /// The job whose record it was.
116 job: u64,
117 /// Why.
118 removed_by: RecordRemoval,
119 },
120 /// Auto sizing wrote a profile's new `memory` and `slots` into the config, and the daemon runs with them.
121 #[serde(rename = "sizing.changed")]
122 SizingChanged {
123 /// The change.
124 change: SizingChange,
125 },
126 /// A runner could not be registered or started, and its repository gets no runner for a while: once, until one of
127 /// its runners takes a job again.
128 #[serde(rename = "runner.failed")]
129 RunnerFailed {
130 /// What failed.
131 failure: RunnerFailure,
132 },
133 /// Live: a running job's container was sampled.
134 #[serde(rename = "job.sample")]
135 JobSample {
136 /// The job's ID.
137 job: u64,
138 /// The sample.
139 sample: Sample,
140 },
141 /// Live: the slice runners run in was sampled.
142 #[serde(rename = "host.sample")]
143 HostSample {
144 /// The sample.
145 sample: HostSample,
146 },
147 /// Live: a line of a running job's log was stored. Only sent with `logs=true`.
148 #[serde(rename = "log.line")]
149 LogLine {
150 /// The job's ID.
151 job: u64,
152 /// The line's number in the job's log, 0-based.
153 line: u64,
154 /// The line, without its newline.
155 text: String,
156 },
157}
158
159impl Event {
160 /// The `type` tag, also the SSE event name.
161 pub fn kind(&self) -> &'static str {
162 match self {
163 Event::JobQueued { .. } => "job.queued",
164 Event::JobStarted { .. } => "job.started",
165 Event::JobStep { .. } => "job.step",
166 Event::JobFinished { .. } => "job.finished",
167 Event::SlotChanged { .. } => "slot.changed",
168 Event::ProfileChanged { .. } => "profile.changed",
169 Event::ProfileRemoved { .. } => "profile.removed",
170 Event::BudgetChanged { .. } => "budget.changed",
171 Event::RepositoryChanged { .. } => "repository.changed",
172 Event::RepositoryRemoved { .. } => "repository.removed",
173 Event::CacheReseeded { .. } => "cache.reseeded",
174 Event::RecordCreated { .. } => "record.created",
175 Event::RecordDeleted { .. } => "record.deleted",
176 Event::SizingChanged { .. } => "sizing.changed",
177 Event::RunnerFailed { .. } => "runner.failed",
178 Event::JobSample { .. } => "job.sample",
179 Event::HostSample { .. } => "host.sample",
180 Event::LogLine { .. } => "log.line",
181 }
182 }
183
184 /// Whether it is stored and can be resumed after; live events are not.
185 pub fn is_durable(&self) -> bool {
186 !matches!(
187 self,
188 Event::JobSample { .. } | Event::HostSample { .. } | Event::LogLine { .. }
189 )
190 }
191
192 /// The job it is about, if any.
193 pub fn job(&self) -> Option<u64> {
194 match self {
195 Event::JobQueued { job } | Event::JobStarted { job } | Event::JobFinished { job } => {
196 Some(job.id)
197 }
198 Event::JobStep { job, .. }
199 | Event::JobSample { job, .. }
200 | Event::LogLine { job, .. }
201 | Event::RecordDeleted { job, .. } => Some(*job),
202 Event::RecordCreated { record } => Some(record.job),
203 Event::SlotChanged { .. }
204 | Event::ProfileChanged { .. }
205 | Event::ProfileRemoved { .. }
206 | Event::BudgetChanged { .. }
207 | Event::RepositoryChanged { .. }
208 | Event::RepositoryRemoved { .. }
209 | Event::CacheReseeded { .. }
210 | Event::SizingChanged { .. }
211 | Event::RunnerFailed { .. }
212 | Event::HostSample { .. } => None,
213 }
214 }
215}
216
217/// A runner that could not be registered or started (`runner.failed`).
218#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
219pub struct RunnerFailure {
220 /// `owner/name`.
221 pub repository: String,
222 /// The profile it was for.
223 pub profile: String,
224 /// Why, as GitHub or Docker said.
225 pub error: String,
226 /// Seconds before the repository gets another runner.
227 pub held_for: u64,
228}
229
230/// `GET /v1/events` query parameters. Resume with the `Last-Event-ID` header.
231#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
232pub struct EventQuery {
233 /// Only events about this job.
234 pub job: Option<u64>,
235 /// Also send `log.line` events.
236 #[serde(default)]
237 pub logs: bool,
238}
239
240impl EventQuery {
241 /// Whether an event passes the filter.
242 pub fn matches(&self, event: &Event) -> bool {
243 (self.logs || !matches!(event, Event::LogLine { .. }))
244 && self.job.is_none_or(|job| event.job() == Some(job))
245 }
246}