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, ProfileStatus, Record, RecordRemoval, 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 daemon started or a reload added the profile, a reload changed its image, CPUs, memory or slots, or it
62    /// started or stopped reseeding. A slot's runner changing is `slot.changed` only; this event carries every slot as it is now, so a
63    /// client can replace the whole profile.
64    #[serde(rename = "profile.changed")]
65    ProfileChanged {
66        /// The profile as it is now.
67        profile: ProfileStatus,
68    },
69    /// A reload removed the profile and its last runner and reseed are gone.
70    #[serde(rename = "profile.removed")]
71    ProfileRemoved {
72        /// Its name.
73        profile: String,
74    },
75    /// The memory caps in use changed.
76    #[serde(rename = "budget.changed")]
77    BudgetChanged {
78        /// The budget as it is now.
79        budget: Budget,
80    },
81    /// A repository was paused or resumed, its queue changed, a reload changed its profile or labels, or a reload
82    /// added it.
83    #[serde(rename = "repository.changed")]
84    RepositoryChanged {
85        /// The repository as it is now.
86        repository: RepositoryStatus,
87    },
88    /// A reload removed the repository from the config.
89    #[serde(rename = "repository.removed")]
90    RepositoryRemoved {
91        /// Its `owner/name`.
92        repository: String,
93    },
94    /// A profile's snapshot check finished.
95    #[serde(rename = "cache.reseeded")]
96    CacheReseeded {
97        /// The profile.
98        profile: String,
99        /// What the seed container printed, such as why it did not reseed.
100        output: String,
101        /// Set when reseeding failed.
102        error: Option<String>,
103    },
104    /// A failed job's runner was kept as a record.
105    #[serde(rename = "record.created")]
106    RecordCreated {
107        /// The record.
108        record: Record,
109    },
110    /// A record was removed.
111    #[serde(rename = "record.deleted")]
112    RecordDeleted {
113        /// The job whose record it was.
114        job: u64,
115        /// Why.
116        removed_by: RecordRemoval,
117    },
118    /// Live: a running job's container was sampled.
119    #[serde(rename = "job.sample")]
120    JobSample {
121        /// The job's ID.
122        job: u64,
123        /// The sample.
124        sample: Sample,
125    },
126    /// Live: the slice runners run in was sampled.
127    #[serde(rename = "host.sample")]
128    HostSample {
129        /// The sample.
130        sample: HostSample,
131    },
132    /// Live: a line of a running job's log was stored. Only sent with `logs=true`.
133    #[serde(rename = "log.line")]
134    LogLine {
135        /// The job's ID.
136        job: u64,
137        /// The line's number in the job's log, 0-based.
138        line: u64,
139        /// The line, without its newline.
140        text: String,
141    },
142}
143
144impl Event {
145    /// The `type` tag, also the SSE event name.
146    pub fn kind(&self) -> &'static str {
147        match self {
148            Event::JobQueued { .. } => "job.queued",
149            Event::JobStarted { .. } => "job.started",
150            Event::JobStep { .. } => "job.step",
151            Event::JobFinished { .. } => "job.finished",
152            Event::SlotChanged { .. } => "slot.changed",
153            Event::ProfileChanged { .. } => "profile.changed",
154            Event::ProfileRemoved { .. } => "profile.removed",
155            Event::BudgetChanged { .. } => "budget.changed",
156            Event::RepositoryChanged { .. } => "repository.changed",
157            Event::RepositoryRemoved { .. } => "repository.removed",
158            Event::CacheReseeded { .. } => "cache.reseeded",
159            Event::RecordCreated { .. } => "record.created",
160            Event::RecordDeleted { .. } => "record.deleted",
161            Event::JobSample { .. } => "job.sample",
162            Event::HostSample { .. } => "host.sample",
163            Event::LogLine { .. } => "log.line",
164        }
165    }
166
167    /// Whether it is stored and can be resumed after; live events are not.
168    pub fn is_durable(&self) -> bool {
169        !matches!(
170            self,
171            Event::JobSample { .. } | Event::HostSample { .. } | Event::LogLine { .. }
172        )
173    }
174
175    /// The job it is about, if any.
176    pub fn job(&self) -> Option<u64> {
177        match self {
178            Event::JobQueued { job } | Event::JobStarted { job } | Event::JobFinished { job } => {
179                Some(job.id)
180            }
181            Event::JobStep { job, .. }
182            | Event::JobSample { job, .. }
183            | Event::LogLine { job, .. }
184            | Event::RecordDeleted { job, .. } => Some(*job),
185            Event::RecordCreated { record } => Some(record.job),
186            Event::SlotChanged { .. }
187            | Event::ProfileChanged { .. }
188            | Event::ProfileRemoved { .. }
189            | Event::BudgetChanged { .. }
190            | Event::RepositoryChanged { .. }
191            | Event::RepositoryRemoved { .. }
192            | Event::CacheReseeded { .. }
193            | Event::HostSample { .. } => None,
194        }
195    }
196}
197
198/// `GET /v1/events` query parameters. Resume with the `Last-Event-ID` header.
199#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
200pub struct EventQuery {
201    /// Only events about this job.
202    pub job: Option<u64>,
203    /// Also send `log.line` events.
204    #[serde(default)]
205    pub logs: bool,
206}
207
208impl EventQuery {
209    /// Whether an event passes the filter.
210    pub fn matches(&self, event: &Event) -> bool {
211        (self.logs || !matches!(event, Event::LogLine { .. }))
212            && self.job.is_none_or(|job| event.job() == Some(job))
213    }
214}
runnerd_core/api/events.rs

runnerd_core/api/
events.rs
