forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
14.1 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 with no JavaScript: while the job is not terminal it refreshes
4//! itself with a meta tag. `log.txt` streams the whole log and keeps following it until the job is
5//! terminal and the file is read to its end. Artifacts are blobs under `ci/<artifact id>` and are
6//! always served as attachments.
7
8use std::path::{Path, PathBuf};
9use std::time::Duration;
10
11use bytes::Bytes;
12use forge_core::Store;
13use forge_core::ci::{CiArtifact, CiJob, CiRun};
14use futures_util::{Stream, TryStreamExt};
15use http::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE};
16use http_body::Frame;
17use http_body_util::StreamBody;
18use tokio::io::{AsyncReadExt, AsyncSeekExt, SeekFrom};
19use topcoat::Result;
20use topcoat::context::{Cx, app_context};
21use topcoat::router::error::not_found;
22use topcoat::router::response::Response;
23use topcoat::router::{Body, HeaderName, HeaderValue, StatusCode, page, path_param, route};
24use topcoat::view::{View, component, view};
25
26use crate::ci::{artifact_key, log_path};
27use crate::i18n::{
28 self, Lang,
29 releases_ci::{self as g, K},
30};
31use crate::pages::ci_runs::{job_href, run_href, state_text};
32use crate::pages::repo::repo_head;
33use crate::pages::repo_ctx::{RepoCtx, RepoHead, load_repo};
34use crate::state::AppState;
35use crate::views::{enc, human_size, shell};
36
37path_param!(owner);
38path_param!(repo);
39path_param!(job_id);
40path_param!(artifact_name);
41
42/// The most of the log the job page shows, from its end.
43pub const TAIL_BYTES: u64 = 512 * 1024;
44/// Bytes read per chunk while following a log.
45const CHUNK_BYTES: usize = 64 * 1024;
46/// How long `log.txt` waits before it looks at the file again.
47const FOLLOW_POLL: Duration = Duration::from_secs(1);
48
49/// The job and its run, both checked to belong to this repo. Anything else is 404.
50async fn job_of(state: &AppState, ctx: &RepoCtx, id: &str) -> Result<(CiJob, CiRun)> {
51 let Some(job) = state.store.job_by_id(id).await? else {
52 return Err(not_found().into());
53 };
54 let Some(run) = state.store.run_by_id(&job.run_id).await? else {
55 return Err(not_found().into());
56 };
57 if run.repo_id != ctx.repo.id {
58 return Err(not_found().into());
59 }
60 Ok((job, run))
61}
62
63/// The last `TAIL_BYTES` of a log, from a line boundary, and whether anything was cut. A log that
64/// does not exist yet is empty.
65async fn log_tail(path: &Path) -> Result<(String, bool)> {
66 let mut file = match tokio::fs::File::open(path).await {
67 Ok(file) => file,
68 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((String::new(), false)),
69 Err(e) => return Err(forge_core::Error::Io(e).into()),
70 };
71 let len = file.metadata().await.map_err(forge_core::Error::Io)?.len();
72 let truncated = len > TAIL_BYTES;
73 if truncated {
74 file.seek(SeekFrom::Start(len - TAIL_BYTES))
75 .await
76 .map_err(forge_core::Error::Io)?;
77 }
78 let mut buf = Vec::new();
79 (&mut file)
80 .take(TAIL_BYTES)
81 .read_to_end(&mut buf)
82 .await
83 .map_err(forge_core::Error::Io)?;
84 if truncated && let Some(end) = buf.iter().position(|byte| *byte == b'\n') {
85 buf.drain(..=end);
86 }
87 Ok((String::from_utf8_lossy(&buf).into_owned(), truncated))
88}
89
90/// Follows a log: yields what the file holds, then polls for new bytes once a second. The stream
91/// ends when the job is terminal and a read after seeing that returns nothing, so every byte the
92/// runner wrote is sent. A job that is gone, or a read error, ends the stream.
93fn follow_log(
94 store: Store,
95 job_id: String,
96 path: PathBuf,
97) -> impl Stream<Item = std::io::Result<Bytes>> + Send + 'static {
98 struct Follow {
99 store: Store,
100 job_id: String,
101 path: PathBuf,
102 file: Option<tokio::fs::File>,
103 }
104 futures_util::stream::unfold(
105 Follow {
106 store,
107 job_id,
108 path,
109 file: None,
110 },
111 |mut follow| async move {
112 loop {
113 // Checked before the read, so a read that finds nothing after this is the end.
114 let terminal = match follow.store.job_by_id(&follow.job_id).await {
115 Ok(Some(job)) => job.state.is_terminal(),
116 Ok(None) => true,
117 Err(e) => {
118 tracing::error!(error = %e, "ci: log follow could not read the job");
119 return None;
120 }
121 };
122 if follow.file.is_none() {
123 follow.file = tokio::fs::File::open(&follow.path).await.ok();
124 }
125 if let Some(file) = follow.file.as_mut() {
126 let mut buf = vec![0u8; CHUNK_BYTES];
127 match file.read(&mut buf).await {
128 Ok(0) => {}
129 Ok(n) => {
130 buf.truncate(n);
131 return Some((Ok(Bytes::from(buf)), follow));
132 }
133 Err(e) => {
134 tracing::error!(error = %e, "ci: log follow could not read the file");
135 return None;
136 }
137 }
138 }
139 if terminal {
140 return None;
141 }
142 tokio::time::sleep(FOLLOW_POLL).await;
143 }
144 },
145 )
146}
147
148/// A header value for a download's file name: printable ASCII, with quotes and backslashes cut.
149fn attachment_name(name: &str) -> String {
150 name.chars()
151 .map(|c| {
152 if c.is_ascii_graphic() && c != '"' && c != '\\' {
153 c
154 } else {
155 '_'
156 }
157 })
158 .collect()
159}
160
161/// The job page's values, already shown as text.
162struct JobView {
163 name: String,
164 state: String,
165 state_class: String,
166 runner: String,
167 exit_code: String,
168 message: String,
169 started: String,
170 finished: String,
171 run_href: String,
172 run_number: i64,
173 log_href: String,
174}
175
176/// One artifact row: its link and its size.
177struct ArtifactRow {
178 href: String,
179 name: String,
180 size: String,
181}
182
183/// The job page body. The meta refresh is in the body, since the shell has no head hook.
184#[component]
185async fn job_body(
186 head: RepoHead,
187 lang: Lang,
188 job: JobView,
189 log: String,
190 truncated: bool,
191 artifacts: Vec<ArtifactRow>,
192 refresh: bool,
193) -> Result<impl View> {
194 Ok(view! {
195 repo_head(head: head)
196 if refresh {
197 <meta http-equiv="refresh" content="3">
198 }
199 <p class="muted small"><a href=(job.run_href.as_str())>(g::t(lang, K::RunNumber)) (job.run_number)</a></p>
200 <section class="card">
201 <h2>(job.name.as_str())</h2>
202 <p class="muted small">
203 <span class=(job.state_class.as_str())>(job.state.as_str())</span>
204 if !job.runner.is_empty() {
205 " \u{b7} " (g::t(lang, K::Runner)) " " (job.runner.as_str())
206 }
207 if !job.exit_code.is_empty() {
208 " \u{b7} " (g::t(lang, K::Exit)) " " (job.exit_code.as_str())
209 }
210 </p>
211 if !job.started.is_empty() {
212 <p class="muted small">(g::t(lang, K::Started)) " " (job.started.as_str())</p>
213 }
214 if !job.finished.is_empty() {
215 <p class="muted small">(g::t(lang, K::Finished)) " " (job.finished.as_str())</p>
216 }
217 if !job.message.is_empty() {
218 <p>(job.message.as_str())</p>
219 }
220 </section>
221 <section class="card">
222 <h2>(g::t(lang, K::Log))</h2>
223 <p class="muted small"><a href=(job.log_href.as_str())>(g::t(lang, K::FullLog))</a></p>
224 if truncated {
225 <p class="muted small">(g::t(lang, K::LogTruncated))</p>
226 }
227 <pre class="ci-log">(log.as_str())</pre>
228 </section>
229 if !artifacts.is_empty() {
230 <section class="card">
231 <h2>(g::t(lang, K::Artifacts))</h2>
232 <ul class="ci-artifacts">
233 for artifact in artifacts.iter() {
234 <li><a href=(artifact.href.as_str())>(artifact.name.as_str())</a> " (" (artifact.size.as_str()) ")"</li>
235 }
236 </ul>
237 </section>
238 }
239 })
240}
241
242/// One job: its state, the last 512 KiB of its log and its artifacts.
243#[page("/{owner}/{repo}/ci/jobs/{job_id}")]
244async fn job_page(cx: &Cx) -> Result<impl View> {
245 let owner = path_param::<Owner>(cx);
246 let name = path_param::<Repo>(cx);
247 let id = path_param::<JobId>(cx).to_owned();
248 let ctx = load_repo(cx, owner, name).await?;
249 let lang = i18n::lang(cx, ctx.viewer.as_ref().map(|(user, _)| user));
250 let state = app_context::<AppState>(cx);
251 let (job, run) = job_of(state, &ctx, &id).await?;
252 let (log, truncated) = log_tail(&log_path(&state.config.data_dir, &job.id)).await?;
253 let artifacts = state
254 .store
255 .list_artifacts(&job.id)
256 .await?
257 .into_iter()
258 .map(|artifact| ArtifactRow {
259 href: format!(
260 "{}/artifacts/{}",
261 job_href(&title_of(&ctx), &job.id),
262 enc(&artifact.name)
263 ),
264 size: human_size(lang, u64::try_from(artifact.size).unwrap_or(0)),
265 name: artifact.name,
266 })
267 .collect();
268 let runner = match job.runner_id.as_deref() {
269 Some(runner_id) => state
270 .store
271 .list_runners()
272 .await?
273 .into_iter()
274 .find(|runner| runner.id == runner_id)
275 .map(|runner| runner.name)
276 .unwrap_or_default(),
277 None => String::new(),
278 };
279 let title = title_of(&ctx);
280 let refresh = !job.state.is_terminal();
281 let view = JobView {
282 name: job.name.clone(),
283 state: state_text(lang, job.state.as_str()),
284 state_class: format!("ci-state ci-{}", job.state.as_str()),
285 runner,
286 exit_code: job
287 .exit_code
288 .map(|code| code.to_string())
289 .unwrap_or_default(),
290 message: job.message.clone().unwrap_or_default(),
291 started: job.started_at.clone().unwrap_or_default(),
292 finished: job.finished_at.clone().unwrap_or_default(),
293 run_href: run_href(&title, run.number),
294 run_number: run.number,
295 log_href: format!("{}/log.txt", job_href(&title, &job.id)),
296 };
297 let head = ctx.head(state, "ci").await?;
298 let page_title = format!("{title} \u{b7} CI \u{b7} {}", view.name);
299 Ok(view! {
300 shell(title: page_title.as_str(), viewer: ctx.viewer.as_ref(),
301 job_body(head: head, lang: lang, job: view, log: log, truncated: truncated,
302 artifacts: artifacts, refresh: refresh))
303 })
304}
305
306fn title_of(ctx: &RepoCtx) -> String {
307 format!("{}/{}", ctx.repo.owner_name, ctx.repo.name)
308}
309
310/// Streams the job's log as plain text. It sends the file so far and then follows it until the job
311/// is terminal.
312#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/log.txt")]
313async fn job_log(cx: &Cx) -> Result<Response> {
314 let owner = path_param::<Owner>(cx);
315 let name = path_param::<Repo>(cx);
316 let id = path_param::<JobId>(cx).to_owned();
317 let ctx = load_repo(cx, owner, name).await?;
318 let state = app_context::<AppState>(cx);
319 let (job, _) = job_of(state, &ctx, &id).await?;
320 let path = log_path(&state.config.data_dir, &job.id);
321 let chunks = follow_log(state.store.clone(), job.id.clone(), path).map_ok(Frame::data);
322 let mut res = Response::new(Body::new(StreamBody::new(chunks)));
323 *res.status_mut() = StatusCode::OK;
324 let headers = res.headers_mut();
325 headers.insert(
326 CONTENT_TYPE,
327 HeaderValue::from_static("text/plain; charset=utf-8"),
328 );
329 headers.insert(
330 HeaderName::from_static("x-content-type-options"),
331 HeaderValue::from_static("nosniff"),
332 );
333 headers.insert(
334 HeaderName::from_static("cache-control"),
335 HeaderValue::from_static("no-store"),
336 );
337 Ok(res)
338}
339
340/// Downloads an artifact of a job as an attachment.
341#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/artifacts/{artifact_name}")]
342async fn artifact_download(cx: &Cx) -> Result<Response> {
343 let owner = path_param::<Owner>(cx);
344 let name = path_param::<Repo>(cx);
345 let id = path_param::<JobId>(cx).to_owned();
346 let file_name = path_param::<ArtifactName>(cx).to_owned();
347 let ctx = load_repo(cx, owner, name).await?;
348 let state = app_context::<AppState>(cx);
349 let (job, _) = job_of(state, &ctx, &id).await?;
350 let Some(artifact) = state.store.artifact(&job.id, &file_name).await? else {
351 return Err(not_found().into());
352 };
353 let Some(read) = state.blobs.read(&artifact_key(&artifact.id)).await? else {
354 tracing::error!(artifact_id = %artifact.id, "ci: artifact row without a blob");
355 return Err(not_found().into());
356 };
357 artifact_response(read.size, read.stream, &artifact)
358}
359
360/// The attachment response for an artifact's bytes.
361fn artifact_response(
362 size: u64,
363 stream: futures_util::stream::BoxStream<'static, std::io::Result<Bytes>>,
364 artifact: &CiArtifact,
365) -> Result<Response> {
366 let chunks = stream.map_ok(Frame::data);
367 let mut res = Response::new(Body::new(StreamBody::new(chunks)));
368 *res.status_mut() = StatusCode::OK;
369 let headers = res.headers_mut();
370 headers.insert(
371 CONTENT_TYPE,
372 HeaderValue::from_static("application/octet-stream"),
373 );
374 headers.insert(CONTENT_LENGTH, HeaderValue::from(size));
375 if let Ok(value) = HeaderValue::from_str(&format!(
376 "attachment; filename=\"{}\"",
377 attachment_name(&artifact.name)
378 )) {
379 headers.insert(CONTENT_DISPOSITION, value);
380 }
381 headers.insert(
382 HeaderName::from_static("x-content-type-options"),
383 HeaderValue::from_static("nosniff"),
384 );
385 headers.insert(
386 HeaderName::from_static("cache-control"),
387 HeaderValue::from_static("private, no-cache"),
388 );
389 Ok(res)
390}