forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
21.4 KiB
1//! Pages: one CI job (`/{owner}/{repo}/ci/jobs/{job_id}`), its live log (`.../log.txt`) and its
2//! artifacts (`.../artifacts/{name}`). Reading follows repo Read, like the run pages. The job page
3//! shows the last 512 KiB of the log. While the job is not terminal the page is live: it carries a
4//! `data-live` marker for its `.../events` stream, and nav.js re-renders it when the job's status
5//! changes. `log.txt` streams the whole log and keeps following it until the job is
6//! terminal and the file is read to its end. Artifacts are blobs under `ci/<artifact id>` and are
7//! always served as attachments.
8
9use std::path::{Path, PathBuf};
10use std::sync::Arc;
11use std::time::Duration;
12
13use bytes::Bytes;
14use forge_core::Store;
15use forge_core::ci::{CiArtifact, CiJob, CiRun};
16use futures_util::{Stream, TryStreamExt};
17use http::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE};
18use http_body::Frame;
19use http_body_util::StreamBody;
20use tokio::io::{AsyncReadExt, AsyncSeekExt, SeekFrom};
21use tokio::sync::watch;
22use topcoat::Result;
23use topcoat::context::{Cx, app_context};
24use topcoat::router::content::sse::{Event, last_event_id};
25use topcoat::router::error::not_found;
26use topcoat::router::request::uri;
27use topcoat::router::response::Response;
28use topcoat::router::{Body, HeaderName, HeaderValue, StatusCode, page, path_param, route};
29use topcoat::view::{View, component, view};
30
31use crate::ci::{CiLive, artifact_key, log_path};
32use crate::i18n::{
33 self, Lang,
34 releases_ci::{self as g, K},
35};
36use crate::pages::ci_runs::{
37 changed_event, events_response, job_href, job_status, run_href, state_text,
38};
39use crate::pages::repo::repo_head;
40use crate::pages::repo_ctx::{RepoCtx, RepoHead, load_repo};
41use crate::state::AppState;
42use crate::views::{enc, human_size, shell};
43
44path_param!(owner);
45path_param!(repo);
46path_param!(job_id);
47path_param!(artifact_name);
48
49/// The most of the log the job page shows, from its end.
50pub const TAIL_BYTES: u64 = 512 * 1024;
51/// Bytes read per chunk while following a log.
52const CHUNK_BYTES: usize = 64 * 1024;
53/// The longest a follower waits for a change to its job before it reads again anyway. A change
54/// wakes it at once; this only guards against a missed wake.
55const WAKE_BACKSTOP: Duration = Duration::from_secs(60);
56
57/// The job and its run, both checked to belong to this repo. Anything else is 404.
58async fn job_of(state: &AppState, ctx: &RepoCtx, id: &str) -> Result<(CiJob, CiRun)> {
59 let Some(job) = state.store.job_by_id(id).await? else {
60 return Err(not_found().into());
61 };
62 let Some(run) = state.store.run_by_id(&job.run_id).await? else {
63 return Err(not_found().into());
64 };
65 if run.repo_id != ctx.repo.id {
66 return Err(not_found().into());
67 }
68 Ok((job, run))
69}
70
71/// The length of the longest whole-UTF-8 prefix of `buf`. A character that a read cut short at the
72/// end is left out, so the next read sends it whole. Other invalid bytes are kept.
73fn whole_utf8_len(buf: &[u8]) -> usize {
74 match std::str::from_utf8(buf) {
75 Ok(_) => buf.len(),
76 Err(e) if e.error_len().is_none() => e.valid_up_to(),
77 Err(_) => buf.len(),
78 }
79}
80
81/// The last `TAIL_BYTES` of a log, from a line boundary, whether anything was cut, and the byte
82/// offset just past the text shown. A log that does not exist yet is empty.
83async fn log_tail(path: &Path) -> Result<(String, bool, u64)> {
84 let mut file = match tokio::fs::File::open(path).await {
85 Ok(file) => file,
86 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((String::new(), false, 0)),
87 Err(e) => return Err(forge_core::Error::Io(e).into()),
88 };
89 let len = file.metadata().await.map_err(forge_core::Error::Io)?.len();
90 let truncated = len > TAIL_BYTES;
91 let mut start = 0;
92 if truncated {
93 start = len - TAIL_BYTES;
94 file.seek(SeekFrom::Start(start))
95 .await
96 .map_err(forge_core::Error::Io)?;
97 }
98 let mut buf = Vec::new();
99 (&mut file)
100 .take(TAIL_BYTES)
101 .read_to_end(&mut buf)
102 .await
103 .map_err(forge_core::Error::Io)?;
104 if truncated && let Some(end) = buf.iter().position(|byte| *byte == b'\n') {
105 buf.drain(..=end);
106 start += end as u64 + 1;
107 }
108 buf.truncate(whole_utf8_len(&buf));
109 let offset = start + buf.len() as u64;
110 Ok((
111 String::from_utf8_lossy(&buf).into_owned(),
112 truncated,
113 offset,
114 ))
115}
116
117/// Follows a log: yields what the file holds, then sends new bytes each time the job changes (its
118/// log grew or its state did). The stream ends when the job is terminal and a read after seeing
119/// that returns nothing, so every byte the runner wrote is sent. A job that is gone, or a read
120/// error, ends the stream.
121pub(crate) fn follow_log(
122 store: Store,
123 live: Arc<CiLive>,
124 job_id: String,
125 path: PathBuf,
126) -> impl Stream<Item = std::io::Result<Bytes>> + Send + 'static {
127 struct Follow {
128 store: Store,
129 live: Arc<CiLive>,
130 job_id: String,
131 path: PathBuf,
132 file: Option<tokio::fs::File>,
133 /// Subscribed before the first read, so a change between the read and the wait is seen.
134 wake: watch::Receiver<u64>,
135 }
136 let wake = live.subscribe_job(&job_id);
137 futures_util::stream::unfold(
138 Follow {
139 store,
140 live,
141 job_id,
142 path,
143 file: None,
144 wake,
145 },
146 |mut follow| async move {
147 loop {
148 // Checked before the read, so a read that finds nothing after this is the end.
149 let terminal = match follow.store.job_by_id(&follow.job_id).await {
150 Ok(Some(job)) => job.state.is_terminal(),
151 Ok(None) => true,
152 Err(e) => {
153 tracing::error!(error = %e, "ci: log follow could not read the job");
154 return None;
155 }
156 };
157 if follow.file.is_none() {
158 follow.file = tokio::fs::File::open(&follow.path).await.ok();
159 }
160 if let Some(file) = follow.file.as_mut() {
161 let mut buf = vec![0u8; CHUNK_BYTES];
162 match file.read(&mut buf).await {
163 Ok(0) => {}
164 Ok(n) => {
165 buf.truncate(n);
166 return Some((Ok(Bytes::from(buf)), follow));
167 }
168 Err(e) => {
169 tracing::error!(error = %e, "ci: log follow could not read the file");
170 return None;
171 }
172 }
173 }
174 if terminal {
175 return None;
176 }
177 // A closed signal means the job finished, which the next check sees as terminal.
178 // It is subscribed again so this wait does not return at once.
179 if let Ok(Err(_)) = tokio::time::timeout(WAKE_BACKSTOP, follow.wake.changed()).await
180 {
181 follow.wake = follow.live.subscribe_job(&follow.job_id);
182 }
183 }
184 },
185 )
186}
187
188/// A header value for a download's file name: printable ASCII, with quotes and backslashes cut.
189fn attachment_name(name: &str) -> String {
190 name.chars()
191 .map(|c| {
192 if c.is_ascii_graphic() && c != '"' && c != '\\' {
193 c
194 } else {
195 '_'
196 }
197 })
198 .collect()
199}
200
201/// The job page's values, already shown as text.
202struct JobView {
203 name: String,
204 /// The status word as the events carry it, for the live check.
205 status: String,
206 state: String,
207 state_class: String,
208 runner: String,
209 exit_code: String,
210 message: String,
211 started: String,
212 finished: String,
213 run_href: String,
214 run_number: i64,
215 log_href: String,
216}
217
218/// One artifact row: its link and its size.
219struct ArtifactRow {
220 href: String,
221 name: String,
222 size: String,
223}
224
225/// The job page body. A job that is not terminal carries `live`, the URL of its event stream, and
226/// the status the page shows. The log tail carries the offset its text ends at.
227#[component]
228async fn job_body(
229 head: RepoHead,
230 lang: Lang,
231 job: JobView,
232 log: String,
233 log_offset: String,
234 truncated: bool,
235 artifacts: Vec<ArtifactRow>,
236 live: Option<String>,
237) -> Result<impl View> {
238 Ok(view! {
239 repo_head(head: head)
240 <div data-live=(live.as_deref()) data-live-state=(job.status.as_str())>
241 <p class="muted small"><a href=(job.run_href.as_str())>(g::t(lang, K::RunNumber)) (job.run_number)</a></p>
242 <section class="card">
243 <h2>(job.name.as_str())</h2>
244 <p class="muted small">
245 <span class=(job.state_class.as_str())>(job.state.as_str())</span>
246 if !job.runner.is_empty() {
247 " \u{b7} " (g::t(lang, K::Runner)) " " (job.runner.as_str())
248 }
249 if !job.exit_code.is_empty() {
250 " \u{b7} " (g::t(lang, K::Exit)) " " (job.exit_code.as_str())
251 }
252 </p>
253 if !job.started.is_empty() {
254 <p class="muted small">(g::t(lang, K::Started)) " " (job.started.as_str())</p>
255 }
256 if !job.finished.is_empty() {
257 <p class="muted small">(g::t(lang, K::Finished)) " " (job.finished.as_str())</p>
258 }
259 if !job.message.is_empty() {
260 <p>(job.message.as_str())</p>
261 }
262 </section>
263 <section class="card">
264 <h2>(g::t(lang, K::Log))</h2>
265 <p class="muted small"><a href=(job.log_href.as_str())>(g::t(lang, K::FullLog))</a></p>
266 if truncated {
267 <p class="muted small">(g::t(lang, K::LogTruncated))</p>
268 }
269 <pre class="ci-log" data-live-log="true" data-offset=(log_offset.as_str())>(log.as_str())</pre>
270 </section>
271 if !artifacts.is_empty() {
272 <section class="card">
273 <h2>(g::t(lang, K::Artifacts))</h2>
274 <ul class="ci-artifacts">
275 for artifact in artifacts.iter() {
276 <li><a href=(artifact.href.as_str())>(artifact.name.as_str())</a> " (" (artifact.size.as_str()) ")"</li>
277 }
278 </ul>
279 </section>
280 }
281 </div>
282 })
283}
284
285/// One job: its state, the last 512 KiB of its log and its artifacts.
286#[page("/{owner}/{repo}/ci/jobs/{job_id}")]
287async fn job_page(cx: &Cx) -> Result<impl View> {
288 let owner = path_param::<Owner>(cx);
289 let name = path_param::<Repo>(cx);
290 let id = path_param::<JobId>(cx).to_owned();
291 let ctx = load_repo(cx, owner, name).await?;
292 let lang = i18n::lang(cx, ctx.viewer.as_ref().map(|(user, _)| user));
293 let state = app_context::<AppState>(cx);
294 let (job, run) = job_of(state, &ctx, &id).await?;
295 let (log, truncated, log_offset) = log_tail(&log_path(&state.config.data_dir, &job.id)).await?;
296 let artifacts = state
297 .store
298 .list_artifacts(&job.id)
299 .await?
300 .into_iter()
301 .map(|artifact| ArtifactRow {
302 href: format!(
303 "{}/artifacts/{}",
304 job_href(&title_of(&ctx), &job.id),
305 enc(&artifact.name)
306 ),
307 size: human_size(lang, u64::try_from(artifact.size).unwrap_or(0)),
308 name: artifact.name,
309 })
310 .collect();
311 let runner = match job.runner_id.as_deref() {
312 Some(runner_id) => state
313 .store
314 .list_runners()
315 .await?
316 .into_iter()
317 .find(|runner| runner.id == runner_id)
318 .map(|runner| runner.name)
319 .unwrap_or_default(),
320 None => String::new(),
321 };
322 let title = title_of(&ctx);
323 let live = (!job.state.is_terminal()).then(|| format!("{}/events", job_href(&title, &job.id)));
324 let view = JobView {
325 name: job.name.clone(),
326 status: job.state.as_str().to_owned(),
327 state: state_text(lang, job.state.as_str()),
328 state_class: format!("ci-state ci-{}", job.state.as_str()),
329 runner,
330 exit_code: job
331 .exit_code
332 .map(|code| code.to_string())
333 .unwrap_or_default(),
334 message: job.message.clone().unwrap_or_default(),
335 started: job.started_at.clone().unwrap_or_default(),
336 finished: job.finished_at.clone().unwrap_or_default(),
337 run_href: run_href(&title, run.number),
338 run_number: run.number,
339 log_href: format!("{}/log.txt", job_href(&title, &job.id)),
340 };
341 let head = ctx.head(state, "ci").await?;
342 let page_title = format!("{title} \u{b7} CI \u{b7} {}", view.name);
343 Ok(view! {
344 shell(title: page_title.as_str(), viewer: ctx.viewer.as_ref(),
345 job_body(head: head, lang: lang, job: view, log: log, log_offset: log_offset.to_string(),
346 truncated: truncated, artifacts: artifacts, live: live))
347 })
348}
349
350fn title_of(ctx: &RepoCtx) -> String {
351 format!("{}/{}", ctx.repo.owner_name, ctx.repo.name)
352}
353
354/// Bytes of log per `log` event. A bigger append is sent as several events.
355const LOG_EVENT_BYTES: u64 = 64 * 1024;
356
357/// The job's live stream. It opens with the status, sends a `changed` event when the status changes
358/// and a `log` event with each new run of log bytes. A `log` event's id is the offset past its text,
359/// so a reconnect (`Last-Event-ID`) resumes there. A new stream starts at `?from=`, the offset the
360/// page already shows. The stream ends once the job is terminal and its log is read to the end.
361#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/events")]
362async fn job_events(cx: &Cx) -> Result<Response> {
363 let owner = path_param::<Owner>(cx);
364 let name = path_param::<Repo>(cx);
365 let id = path_param::<JobId>(cx).to_owned();
366 let ctx = load_repo(cx, owner, name).await?;
367 let state = app_context::<AppState>(cx);
368 let (job, _) = job_of(state, &ctx, &id).await?;
369 let offset = last_event_id(cx)
370 .and_then(|id| id.parse::<u64>().ok())
371 .or_else(|| from_query(cx))
372 .unwrap_or(0);
373 // Subscribed before the first read, so a change in between is sent.
374 let wake = state.ci_live.subscribe_job(&job.id);
375 let follow = JobFollow {
376 store: state.store.clone(),
377 job_id: job.id.clone(),
378 path: log_path(&state.config.data_dir, &job.id),
379 wake,
380 offset,
381 status: None,
382 terminal: false,
383 more: false,
384 primed: false,
385 pending_status: None,
386 pending_log: None,
387 };
388 events_response(cx, job_stream(follow))
389}
390
391/// The `from` offset of the events URL, if it is a number.
392fn from_query(cx: &Cx) -> Option<u64> {
393 uri(cx).query().and_then(|query| {
394 form_urlencoded::parse(query.as_bytes())
395 .find(|(key, _)| key == "from")
396 .and_then(|(_, value)| value.parse::<u64>().ok())
397 })
398}
399
400/// One job's stream state. Each read of the status and the log happens after a wake, and a read that
401/// sent log bytes is followed by another read at once, so a long append drains in order.
402struct JobFollow {
403 store: Store,
404 job_id: String,
405 path: PathBuf,
406 wake: watch::Receiver<u64>,
407 /// The log offset read so far. The next read starts here.
408 offset: u64,
409 /// The status last read.
410 status: Option<String>,
411 /// The job was terminal at the last read.
412 terminal: bool,
413 /// The last read sent log bytes, so more may be waiting.
414 more: bool,
415 /// False until the first read. Later reads wait for a wake first, unless `more` is set.
416 primed: bool,
417 pending_status: Option<String>,
418 /// The text and the offset past it.
419 pending_log: Option<(String, u64)>,
420}
421
422fn job_stream(follow: JobFollow) -> impl Stream<Item = Result<Event>> + Send + 'static {
423 futures_util::stream::unfold(follow, |mut f| async move {
424 loop {
425 if let Some(status) = f.pending_status.take() {
426 return Some((changed_event(status), f));
427 }
428 if let Some((text, end)) = f.pending_log.take() {
429 let event = Event::new().event("log").data(text).id(end.to_string());
430 return Some((Ok(event), f));
431 }
432 if f.terminal && !f.more {
433 return None;
434 }
435 if f.primed && !f.more && f.wake.changed().await.is_err() {
436 return None;
437 }
438 f.primed = true;
439 f.more = false;
440 let (status, terminal) = match job_status(&f.store, &f.job_id).await {
441 Ok(Some(found)) => found,
442 Ok(None) => return None,
443 Err(e) => {
444 tracing::error!(error = %e, "ci: job stream could not read the job");
445 return None;
446 }
447 };
448 f.terminal = terminal;
449 if f.status.as_deref() != Some(status.as_str()) {
450 f.status = Some(status.clone());
451 f.pending_status = Some(status);
452 }
453 match read_log(&f.path, f.offset).await {
454 Ok(Some((text, end))) => {
455 f.offset = end;
456 f.more = true;
457 f.pending_log = Some((text, end));
458 }
459 Ok(None) => {}
460 Err(e) => {
461 tracing::error!(error = %e, "ci: job stream could not read the log");
462 return None;
463 }
464 }
465 }
466 })
467}
468
469/// The whole text of the log from `offset`, at most `LOG_EVENT_BYTES`, and the offset past it. None
470/// when there is nothing new (no file, no bytes past `offset`, or only part of a character).
471async fn read_log(path: &Path, offset: u64) -> std::io::Result<Option<(String, u64)>> {
472 let mut file = match tokio::fs::File::open(path).await {
473 Ok(file) => file,
474 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
475 Err(e) => return Err(e),
476 };
477 file.seek(SeekFrom::Start(offset)).await?;
478 let mut buf = Vec::new();
479 (&mut file)
480 .take(LOG_EVENT_BYTES)
481 .read_to_end(&mut buf)
482 .await?;
483 let whole = whole_utf8_len(&buf);
484 if whole == 0 {
485 return Ok(None);
486 }
487 buf.truncate(whole);
488 Ok(Some((
489 String::from_utf8_lossy(&buf).into_owned(),
490 offset + whole as u64,
491 )))
492}
493
494/// Streams the job's log as plain text. It sends the file so far and then follows it until the job
495/// is terminal.
496#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/log.txt")]
497async fn job_log(cx: &Cx) -> Result<Response> {
498 let owner = path_param::<Owner>(cx);
499 let name = path_param::<Repo>(cx);
500 let id = path_param::<JobId>(cx).to_owned();
501 let ctx = load_repo(cx, owner, name).await?;
502 let state = app_context::<AppState>(cx);
503 let (job, _) = job_of(state, &ctx, &id).await?;
504 let path = log_path(&state.config.data_dir, &job.id);
505 let chunks = follow_log(
506 state.store.clone(),
507 Arc::clone(&state.ci_live),
508 job.id.clone(),
509 path,
510 )
511 .map_ok(Frame::data);
512 let mut res = Response::new(Body::new(StreamBody::new(chunks)));
513 *res.status_mut() = StatusCode::OK;
514 let headers = res.headers_mut();
515 headers.insert(
516 CONTENT_TYPE,
517 HeaderValue::from_static("text/plain; charset=utf-8"),
518 );
519 headers.insert(
520 HeaderName::from_static("x-content-type-options"),
521 HeaderValue::from_static("nosniff"),
522 );
523 headers.insert(
524 HeaderName::from_static("cache-control"),
525 HeaderValue::from_static("no-store"),
526 );
527 Ok(res)
528}
529
530/// Downloads an artifact of a job as an attachment.
531#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/artifacts/{artifact_name}")]
532async fn artifact_download(cx: &Cx) -> Result<Response> {
533 let owner = path_param::<Owner>(cx);
534 let name = path_param::<Repo>(cx);
535 let id = path_param::<JobId>(cx).to_owned();
536 let file_name = path_param::<ArtifactName>(cx).to_owned();
537 let ctx = load_repo(cx, owner, name).await?;
538 let state = app_context::<AppState>(cx);
539 let (job, _) = job_of(state, &ctx, &id).await?;
540 let Some(artifact) = state.store.artifact(&job.id, &file_name).await? else {
541 return Err(not_found().into());
542 };
543 let Some(read) = state.blobs.read(&artifact_key(&artifact.id)).await? else {
544 tracing::error!(artifact_id = %artifact.id, "ci: artifact row without a blob");
545 return Err(not_found().into());
546 };
547 artifact_response(read.size, read.stream, &artifact)
548}
549
550/// The attachment response for an artifact's bytes.
551fn artifact_response(
552 size: u64,
553 stream: futures_util::stream::BoxStream<'static, std::io::Result<Bytes>>,
554 artifact: &CiArtifact,
555) -> Result<Response> {
556 let chunks = stream.map_ok(Frame::data);
557 let mut res = Response::new(Body::new(StreamBody::new(chunks)));
558 *res.status_mut() = StatusCode::OK;
559 let headers = res.headers_mut();
560 headers.insert(
561 CONTENT_TYPE,
562 HeaderValue::from_static("application/octet-stream"),
563 );
564 headers.insert(CONTENT_LENGTH, HeaderValue::from(size));
565 if let Ok(value) = HeaderValue::from_str(&format!(
566 "attachment; filename=\"{}\"",
567 attachment_name(&artifact.name)
568 )) {
569 headers.insert(CONTENT_DISPOSITION, value);
570 }
571 headers.insert(
572 HeaderName::from_static("x-content-type-options"),
573 HeaderValue::from_static("nosniff"),
574 );
575 headers.insert(
576 HeaderName::from_static("cache-control"),
577 HeaderValue::from_static("private, no-cache"),
578 );
579 Ok(res)
580}