1use std::fs::{
2 self,
3 File,
4};
5use std::io::{
6 self,
7 BufWriter,
8 Write,
9};
10use std::path::{
11 Path,
12 PathBuf,
13};
14use std::sync::atomic::{
15 AtomicBool,
16 AtomicU64,
17 Ordering,
18};
19use std::sync::Arc;
20
21use bytes::Bytes;
22use chrono::Utc;
23use memchr::{
24 memchr,
25 memchr_iter,
26};
27use sha2::{
28 Digest,
29 Sha256,
30};
31use tokio::sync::mpsc;
32
33use super::ansi::{
34 self,
35 Span,
36 StyleTable,
37};
38use super::block::{
39 BlockRef,
40 BlockWriter,
41};
42use super::checkpoint::{
43 self,
44 copy_prefix,
45 steps_hash,
46 Base,
47 Checkpoint,
48 Resume,
49 TAIL_FILE,
50};
51use super::failure;
52use super::outline::{
53 Outline,
54 StepBound,
55};
56use super::store::{
57 short_hex,
58 DocKey,
59 StoredDocMeta,
60 STORED_FORMAT,
61};
62use super::{
63 BLOCK_MAX_LINES,
64 BLOCK_TARGET_BYTES,
65 MAX_DOC_TEXT_BYTES,
66 MAX_LINE_BYTES,
67};
68use crate::domain::{
69 LineKind,
70 LineMarker,
71 LogCursor,
72 LogDialect,
73 LogRead,
74 LogReadMode,
75};
76
77pub(super) const BLOCKS_FILE: &str = "blocks.bin";
78pub(super) const INDEX_FILE: &str = "index.bin";
79pub(super) const META_FILE: &str = "meta.json";
80const BLOCKS_BUFFER_BYTES: usize = 1024 * 1024;
81
82#[derive(Debug, Clone)]
83pub struct BlockLimits {
84 pub lines: usize,
85 pub target_bytes: usize,
86 pub max_doc_text_bytes: u64,
87}
88
89impl Default for BlockLimits {
90 fn default() -> Self {
91 Self {
92 lines: BLOCK_MAX_LINES,
93 target_bytes: BLOCK_TARGET_BYTES,
94 max_doc_text_bytes: MAX_DOC_TEXT_BYTES,
95 }
96 }
97}
98
99pub struct Ingester {
100 pub key: DocKey,
101 pub revision: i64,
102 pub complete: bool,
103 pub dir: PathBuf,
104 pub dialect: &'static dyn LogDialect,
105 pub steps: Vec<StepBound>,
106 pub received: Arc<AtomicU64>,
107 pub limits: BlockLimits,
108 pub base: Option<Base>,
109 pub restart: Arc<AtomicBool>,
110}
111
112#[derive(Debug)]
113pub struct IngestOutcome {
114 pub meta: StoredDocMeta,
115 pub truncated: bool,
116 pub resumable: bool,
117}
118
119#[derive(PartialEq, Eq)]
120enum Flow {
121 Continue,
122 Stop,
123}
124
125pub struct Ingested {
126 cfg: Ingester,
127 resumed: Option<bool>,
128 state: Option<State>,
129 steps_hash: String,
130}
131
132impl Ingester {
133 pub fn run(self, rx: &mut mpsc::Receiver<Bytes>) -> io::Result<Ingested> {
134 let received = Arc::clone(&self.received);
135 let mut ingested = Ingested {
136 steps_hash: steps_hash(&self.steps),
137 cfg: self,
138 resumed: None,
139 state: None,
140 };
141 while let Some(chunk) = rx.blocking_recv() {
142 received.fetch_add(chunk.len() as u64, Ordering::Relaxed);
143 if ingested.start()?.push(&chunk)? == Flow::Stop {
144 break;
145 }
146 }
147 if !ingested.decide() {
148 ingested.start()?;
149 }
150 if let Some(state) = ingested.state.as_mut() {
151 state.seal();
152 if state.truncated {
153 rx.close();
154 }
155 }
156
157 Ok(ingested)
158 }
159}
160
161impl Ingested {
162 pub fn resumed(&self) -> bool {
163 self.resumed == Some(true)
164 }
165
166 pub fn truncated(&self) -> bool {
167 self.state.as_ref().is_some_and(|state| state.truncated)
168 }
169
170 pub fn unchanged(&self, read: &LogRead) -> bool {
171 let Some(base) = &self.cfg.base else {
172 return false;
173 };
174 let Some(prior) = &base.checkpoint else {
175 return false;
176 };
177 if base.document.meta().complete != self.cfg.complete || prior.steps_hash != self.steps_hash
178 {
179 return false;
180 }
181 let state = self.state.as_ref();
182 match read.mode {
183 LogReadMode::Append => {
184 state.is_none_or(|state| state.new_bytes == 0)
185 && read.cursor.as_ref().is_some_and(|cursor| {
186 prior
187 .resume
188 .as_ref()
189 .is_some_and(|resume| &resume.cursor == cursor)
190 })
191 }
192 LogReadMode::Replace => {
193 let hash = state.and_then(|state| state.raw_hash.as_deref());
194 hash.is_some() && hash == prior.raw_hash.as_deref()
195 }
196 }
197 }
198
199 pub fn finish(mut self, cursor: Option<LogCursor>) -> io::Result<IngestOutcome> {
200 self.start()?;
201 let Self {
202 cfg,
203 state,
204 steps_hash,
205 ..
206 } = self;
207 let mut state = state.ok_or_else(not_started)?;
208 let had_cursor = cursor.is_some();
209 let mut keep =
210 !cfg.complete && !state.truncated && (had_cursor || state.raw_hash.is_some());
211 if keep {
212 state.write_checkpoint(&cfg.dir, cursor, &steps_hash)?;
213 }
214 state.render_tail()?;
215 if keep && state.truncated {
216 checkpoint::remove(&cfg.dir);
217 keep = false;
218 }
219
220 state.finish(cfg, keep && had_cursor)
221 }
222
223 fn decide(&mut self) -> bool {
224 let Self { cfg, resumed, .. } = self;
225 *resumed.get_or_insert_with(|| {
226 cfg.base
227 .as_ref()
228 .is_some_and(|base| base.cursor(&cfg.steps).is_some())
229 && !cfg.restart.load(Ordering::SeqCst)
230 })
231 }
232
233 fn start(&mut self) -> io::Result<&mut State> {
234 if self.state.is_none() {
235 let start = if self.decide() {
236 Start::resume(&mut self.cfg)?
237 } else {
238 Start::fresh(&mut self.cfg)?
239 };
240 self.state = Some(State::new(&self.cfg, start));
241 }
242
243 self.state.as_mut().ok_or_else(not_started)
244 }
245}
246
247fn not_started() -> io::Error {
248 io::Error::other("The log ingest has not started")
249}
250
251struct Start {
252 outline: Outline,
253 styles: StyleTable,
254 blocks: File,
255 index: File,
256 blocks_len: u64,
257 index_len: u64,
258 line_count: u32,
259 text_bytes: u64,
260 first_ts: Option<i64>,
261 pending: Vec<u8>,
262 previous: Option<(i64, u32)>,
263}
264
265impl Start {
266 fn fresh(cfg: &mut Ingester) -> io::Result<Self> {
267 Ok(Self {
268 outline: Outline::new(std::mem::take(&mut cfg.steps), &cfg.dir),
269 styles: StyleTable::default(),
270 blocks: File::create(cfg.dir.join(BLOCKS_FILE))?,
271 index: File::create(cfg.dir.join(INDEX_FILE))?,
272 blocks_len: 0,
273 index_len: 0,
274 line_count: 0,
275 text_bytes: 0,
276 first_ts: None,
277 pending: Vec::new(),
278 previous: None,
279 })
280 }
281
282 fn resume(cfg: &mut Ingester) -> io::Result<Self> {
283 let missing = || {
284 io::Error::new(
285 io::ErrorKind::InvalidInput,
286 "There is no log checkpoint to resume from",
287 )
288 };
289 let Base {
290 revision,
291 document,
292 checkpoint,
293 ..
294 } = cfg.base.as_mut().ok_or_else(missing)?;
295 let resume = checkpoint
296 .as_mut()
297 .and_then(|checkpoint| checkpoint.resume.as_mut())
298 .ok_or_else(missing)?;
299 let from = document.dir();
300 let pending = match fs::read(from.join(TAIL_FILE)) {
301 Ok(tail) => tail,
302 Err(e) if e.kind() == io::ErrorKind::NotFound => Vec::new(),
303 Err(e) => return Err(e),
304 };
305
306 Ok(Self {
307 blocks: copy_prefix(
308 &from.join(BLOCKS_FILE),
309 &cfg.dir.join(BLOCKS_FILE),
310 resume.blocks_len,
311 )?,
312 index: copy_prefix(
313 &from.join(INDEX_FILE),
314 &cfg.dir.join(INDEX_FILE),
315 resume.index_len,
316 )?,
317 outline: std::mem::take(&mut resume.outline).resume(
318 std::mem::take(&mut cfg.steps),
319 from,
320 &cfg.dir,
321 )?,
322 styles: StyleTable::from_list(document.meta().styles[..resume.styles].to_vec()),
323 blocks_len: resume.blocks_len,
324 index_len: resume.index_len,
325 line_count: resume.line_count,
326 text_bytes: resume.text_bytes,
327 first_ts: resume.first_ts,
328 pending,
329 previous: Some((*revision, resume.line_count)),
330 })
331 }
332}
333
334struct State {
335 dialect: &'static dyn LogDialect,
336 infer_errors: bool,
337 max_doc_text_bytes: u64,
338 styles: StyleTable,
339 outline: Outline,
340 writer: BlockWriter,
341 blocks: BufWriter<File>,
342 index: BufWriter<File>,
343 blocks_len: u64,
344 index_len: u64,
345 text: String,
346 spans: Vec<Span>,
347 line_count: u32,
348 text_bytes: u64,
349 first_ts: Option<i64>,
350 truncated: bool,
351 pending: Vec<u8>,
352 hasher: Option<Sha256>,
353 raw_hash: Option<String>,
354 new_bytes: u64,
355 previous: Option<(i64, u32)>,
356}
357
358fn char_cut(buf: &[u8]) -> usize {
359 let mut cut = MAX_LINE_BYTES;
360 while cut > 0 && buf[cut] & 0xC0 == 0x80 {
361 cut -= 1;
362 }
363 if cut == 0 {
364 MAX_LINE_BYTES
365 } else {
366 cut
367 }
368}
369
370fn clean_carriage_returns(raw: &[u8]) -> &[u8] {
371 let raw = raw.strip_suffix(b"\r").unwrap_or(raw);
372 if memchr(b'\r', raw).is_none() {
373 return raw;
374 }
375 raw.rsplit(|&b| b == b'\r')
376 .find(|segment| !segment.is_empty())
377 .unwrap_or_default()
378}
379
380fn kind_of(marker: LineMarker) -> LineKind {
381 match marker {
382 LineMarker::GroupStart | LineMarker::SectionStart { .. } => LineKind::Group,
383 LineMarker::Error => LineKind::Error,
384 LineMarker::Warning => LineKind::Warning,
385 LineMarker::Notice => LineKind::Notice,
386 LineMarker::Debug => LineKind::Debug,
387 LineMarker::Command => LineKind::Command,
388 LineMarker::Plain | LineMarker::GroupEnd | LineMarker::OpenSection => LineKind::Plain,
389 }
390}
391
392impl State {
393 fn new(cfg: &Ingester, start: Start) -> Self {
394 Self {
395 dialect: cfg.dialect,
396 infer_errors: cfg.dialect.infers_errors(),
397 max_doc_text_bytes: cfg.limits.max_doc_text_bytes,
398 styles: start.styles,
399 outline: start.outline,
400 writer: BlockWriter::new(cfg.limits.lines, cfg.limits.target_bytes),
401 blocks: BufWriter::with_capacity(BLOCKS_BUFFER_BYTES, start.blocks),
402 index: BufWriter::new(start.index),
403 blocks_len: start.blocks_len,
404 index_len: start.index_len,
405 text: String::new(),
406 spans: Vec::new(),
407 line_count: start.line_count,
408 text_bytes: start.text_bytes,
409 first_ts: start.first_ts,
410 truncated: false,
411 pending: start.pending,
412 hasher: (!cfg.complete && start.previous.is_none()).then(Sha256::new),
413 raw_hash: None,
414 new_bytes: 0,
415 previous: start.previous,
416 }
417 }
418
419 fn push(&mut self, chunk: &[u8]) -> io::Result<Flow> {
420 self.new_bytes += chunk.len() as u64;
421 if let Some(hasher) = self.hasher.as_mut() {
422 hasher.update(chunk);
423 }
424 let mut start = 0;
425 for end in memchr_iter(b'\n', chunk) {
426 let flow = if self.pending.is_empty() {
427 self.complete_line(&chunk[start..end])?
428 } else {
429 let mut pending = std::mem::take(&mut self.pending);
430 pending.extend_from_slice(&chunk[start..end]);
431 let flow = self.complete_line(&pending)?;
432 pending.clear();
433 self.pending = pending;
434 flow
435 };
436 if flow == Flow::Stop {
437 return Ok(Flow::Stop);
438 }
439 start = end + 1;
440 }
441 self.pending.extend_from_slice(&chunk[start..]);
442 if self.pending.len() > MAX_LINE_BYTES {
443 let mut pending = std::mem::take(&mut self.pending);
444 let (consumed, flow) = self.split_long(&pending)?;
445 pending.drain(..consumed);
446 self.pending = pending;
447 return Ok(flow);
448 }
449
450 Ok(Flow::Continue)
451 }
452
453 fn seal(&mut self) {
454 self.raw_hash = self
455 .hasher
456 .take()
457 .map(|hasher| short_hex(&hasher.finalize()));
458 }
459
460 fn render_tail(&mut self) -> io::Result<()> {
461 if self.truncated || self.pending.is_empty() {
462 return Ok(());
463 }
464 let pending = std::mem::take(&mut self.pending);
465 self.complete_line(&pending).map(drop)
466 }
467
468 fn write_checkpoint(
469 &mut self, dir: &Path, cursor: Option<LogCursor>, steps_hash: &str,
470 ) -> io::Result<()> {
471 let resume = match cursor {
472 Some(cursor) => {
473 if self.writer.lines() > 0 {
474 self.flush_block()?;
475 }
476 Some(Resume {
477 cursor,
478 line_count: self.line_count,
479 text_bytes: self.text_bytes,
480 first_ts: self.first_ts,
481 blocks_len: self.blocks_len,
482 index_len: self.index_len,
483 styles: self.styles.len(),
484 outline: &self.outline,
485 })
486 }
487 None => None,
488 };
489 let tail: &[u8] = if resume.is_some() { &self.pending } else { &[] };
490
491 Checkpoint {
492 steps_hash: steps_hash.to_string(),
493 raw_hash: self.raw_hash.clone(),
494 resume,
495 }
496 .write(dir, tail)
497 }
498
499 fn split_long(&mut self, raw: &[u8]) -> io::Result<(usize, Flow)> {
500 let mut offset = 0;
501 while raw.len() - offset > MAX_LINE_BYTES {
502 let cut = char_cut(&raw[offset..]);
503 let flow = self.line(&raw[offset..offset + cut])?;
504 offset += cut;
505 if flow == Flow::Stop {
506 return Ok((offset, Flow::Stop));
507 }
508 }
509 Ok((offset, Flow::Continue))
510 }
511
512 fn complete_line(&mut self, raw: &[u8]) -> io::Result<Flow> {
513 let (offset, flow) = self.split_long(raw)?;
514 if flow == Flow::Stop {
515 return Ok(Flow::Stop);
516 }
517 self.line(&raw[offset..])
518 }
519
520 fn line(&mut self, raw: &[u8]) -> io::Result<Flow> {
521 let visible = clean_carriage_returns(raw);
522 let mut parsed = self.dialect.parse(visible);
523 let mut timestamp = parsed.timestamp;
524 if visible.as_ptr() != raw.as_ptr() {
525 let whole = self.dialect.parse(raw);
526 if parsed.marker == LineMarker::Plain && whole.marker != LineMarker::Plain {
527 parsed = whole;
528 }
529 timestamp = timestamp.or(whole.timestamp);
530 }
531 let ts = timestamp.map(|t| t.timestamp_millis());
532 match parsed.marker {
533 LineMarker::GroupEnd => {
534 self.outline.group_end(self.line_count, ts);
535 if parsed.body.trim_ascii().is_empty() {
536 return Ok(Flow::Continue);
537 }
538 }
539 LineMarker::OpenSection => {
540 self.render(parsed.body);
541 self.outline.open_current_group(&self.text);
542 return Ok(Flow::Continue);
543 }
544 LineMarker::SectionStart { .. } => self.outline.end_groups(self.line_count),
545 _ => {}
546 }
547 self.render(parsed.body);
548 let added = self.text.len() as u64 + 1;
549 if self.text_bytes + added > self.max_doc_text_bytes {
550 self.truncated = true;
551 return Ok(Flow::Stop);
552 }
553 let kind = match kind_of(parsed.marker) {
554 LineKind::Plain if self.infer_errors && failure::is_error(&self.text) => {
555 LineKind::Error
556 }
557 kind => kind,
558 };
559 let step_start = self.dialect.is_step_start(&parsed);
560 self.outline
561 .line(self.line_count, kind, ts, &self.text, step_start)?;
562 if parsed.marker == (LineMarker::SectionStart { open: true }) {
563 self.outline.open_current_group("");
564 }
565 if self.first_ts.is_none() {
566 self.first_ts = ts;
567 }
568 if self
569 .writer
570 .would_overflow(self.text.len(), self.spans.len())
571 {
572 self.flush_block()?;
573 }
574 self.writer.push(kind, ts, &self.text, &self.spans);
575 self.text_bytes += added;
576 self.line_count += 1;
577 Ok(Flow::Continue)
578 }
579
580 fn render(&mut self, body: &[u8]) {
581 self.text.clear();
582 self.spans.clear();
583 ansi::render(body, &mut self.styles, &mut self.text, &mut self.spans);
584 }
585
586 fn flush_block(&mut self) -> io::Result<()> {
587 let first_line = self.line_count - self.writer.lines() as u32;
588 let data = self.writer.finish();
589 let compressed_len = u32::try_from(data.len())
590 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Log block is too large"))?;
591 let record = BlockRef {
592 offset: self.blocks_len,
593 compressed_len,
594 first_line,
595 };
596 self.blocks.write_all(&data)?;
597 self.index.write_all(&record.to_bytes())?;
598 self.blocks_len += data.len() as u64;
599 self.index_len += super::block::INDEX_RECORD_BYTES as u64;
600 Ok(())
601 }
602
603 fn finish(mut self, cfg: Ingester, resumable: bool) -> io::Result<IngestOutcome> {
604 if self.writer.lines() > 0 {
605 self.flush_block()?;
606 }
607 let outline = self.outline.finish(self.line_count)?;
608 if !resumable {
609 self.outline.discard_spool()?;
610 }
611 self.blocks
612 .into_inner()
613 .map_err(|e| e.into_error())?
614 .sync_all()?;
615 self.index
616 .into_inner()
617 .map_err(|e| e.into_error())?
618 .sync_all()?;
619
620 let (previous_revision, stable_lines) = match self.previous {
621 Some((revision, lines)) => (Some(revision), lines),
622 None => (None, 0),
623 };
624 let meta = StoredDocMeta {
625 format: STORED_FORMAT,
626 key: cfg.key,
627 revision: cfg.revision,
628 complete: cfg.complete,
629 truncated: self.truncated,
630 line_count: self.line_count,
631 text_bytes: self.text_bytes,
632 disk_bytes: self.blocks_len + self.index_len,
633 first_timestamp_ms: self.first_ts,
634 styles: self.styles.into_list(),
635 sections: outline.sections,
636 error_lines: outline.error_lines,
637 warning_lines: outline.warning_lines,
638 error_total: outline.error_total,
639 warning_total: outline.warning_total,
640 focus_line: outline.focus_line,
641 updated_at_ms: Utc::now().timestamp_millis(),
642 stable_lines,
643 previous_revision,
644 };
645 let file = File::create(cfg.dir.join(META_FILE))?;
646 let mut writer = BufWriter::new(file);
647 serde_json::to_writer(&mut writer, &meta).map_err(io::Error::from)?;
648 writer
649 .into_inner()
650 .map_err(|e| e.into_error())?
651 .sync_all()?;
652 Ok(IngestOutcome {
653 truncated: self.truncated,
654 resumable,
655 meta,
656 })
657 }
658}
659
660#[cfg(test)]
661mod tests {
662 use pipedash_plugin_api::PlainDialect;
663 use pipedash_plugin_azure_devops::AzureDevOpsLogDialect;
664 use pipedash_plugin_bitbucket::BitbucketLogDialect;
665 use pipedash_plugin_buildkite::BuildkiteLogDialect;
666 use pipedash_plugin_gitea::GiteaLogDialect;
667 use pipedash_plugin_gitlab::GitlabLogDialect;
668 use pipedash_plugin_jenkins::JenkinsLogDialect;
669 use pipedash_plugin_tekton::TektonLogDialect;
670
671 use super::super::document::Document;
672 use super::super::testing::{
673 doc_key,
674 ingester,
675 };
676 use super::*;
677
678 static BUILDKITE: BuildkiteLogDialect = BuildkiteLogDialect;
679
680 const BUILDKITE_LOG: &str = concat!(
681 "\x1b_bk;t=1745322209921\x07~~~ Running global environment hook\r\n",
682 "\x1b_bk;t=1745322209921\x07\x1b[90m$\x1b[0m /buildkite/agent/hooks/environment\r\n",
683 "\x1b_bk;t=1745322210692\x07~~~ Running script\r\n",
684 "\x1b_bk;t=1745322210692\x07\x1b[90m$\x1b[0m ./script.sh\r\n",
685 "\x1b_bk;t=1745322210694\x07--- :package: Build\r\r\n",
686 "\x1b_bk;t=1745322210698\x07remote: Counting objects: 50%\x1b[K",
687 "\x1b_bk;t=1745322210699\x07\rremote: Counting objects: 100%, done.\x1b[K",
688 "\x1b_bk;t=1745322210700\x07\r\r\n",
689 "\x1b_bk;t=1745322210800\x07+++ Running tests\r\n",
690 "\x1b_bk;t=1745322210900\x07ok\r\n",
691 "\x1b_bk;t=1745322211000\x07--- Lint\r\n",
692 "\x1b_bk;t=1745322212000\x07\x1b[31m🚨 Error: The command exited with status 1\x1b[0m\r\n",
693 "\x1b_bk;t=1745322212000\x07^^^ +++\r\n",
694 "\x1b_bk;t=1745322212100\x07--- Upload\r\n",
695 "\x1b_bk;t=1745322212200\x07^^^ +++ Upload failed\r\n",
696 );
697
698 #[test]
699 fn buildkite_groups_become_flat_sections() {
700 let dir = tempfile::tempdir().expect("tempdir");
701 let mut ingester = ingester(dir.path(), doc_key(), 1, Vec::new(), BlockLimits::default());
702 ingester.dialect = &BUILDKITE;
703 let (tx, mut rx) = mpsc::channel(1);
704 tx.try_send(Bytes::from_static(BUILDKITE_LOG.as_bytes()))
705 .expect("channel has room");
706 drop(tx);
707
708 let meta = ingester
709 .run(&mut rx)
710 .expect("ingest")
711 .finish(None)
712 .expect("finish")
713 .meta;
714
715 let sections: Vec<_> = meta
716 .sections
717 .iter()
718 .map(|s| {
719 (
720 s.title.as_str(),
721 s.parent,
722 s.header_line,
723 s.end_line,
724 s.open,
725 )
726 })
727 .collect();
728 assert_eq!(
729 sections,
730 vec![
731 ("Running global environment hook", None, Some(0), 2, false),
732 ("Running script", None, Some(2), 4, false),
733 (":package: Build", None, Some(4), 6, false),
734 ("Running tests", None, Some(6), 8, true),
735 ("Lint", None, Some(8), 10, true),
736 ("Upload failed", None, Some(10), 11, true),
737 ]
738 );
739 assert_eq!(meta.sections[3].duration_ms, Some(100));
740 assert_eq!(meta.line_count, 11);
741 assert_eq!(meta.error_lines, vec![9]);
742 assert_eq!(meta.focus_line, Some(9));
743 assert_eq!(meta.first_timestamp_ms, Some(1_745_322_209_921));
744
745 let text = Document::open(dir.path())
746 .expect("open")
747 .text_block(0)
748 .expect("block");
749 let text = std::str::from_utf8(&text).expect("utf8");
750 let lines: Vec<&str> = text.lines().collect();
751 assert_eq!(lines.len(), 11);
752 assert_eq!(
753 lines[5],
754 "2025-04-22T11:43:30.698Z remote: Counting objects: 100%, done."
755 );
756 assert!(lines[9].ends_with(" 🚨 Error: The command exited with status 1"));
757 assert!(!text.contains("bk;t=") && !text.contains("^^^"));
758 }
759
760 static GITLAB: GitlabLogDialect = GitlabLogDialect;
761
762 const GITLAB_LOG: &str = concat!(
763 "Running with gitlab-runner 18.4.0 (139a0ac0)\n",
764 " on code-europa-eu-runner fkXhbgRUT, system ID: s_5247efa18f3c\n",
765 "section_start:1790320399:prepare_executor\r\x1b[0K\x1b[0K\x1b[36;1mPreparing the \"docker+machine\" executor\x1b[0;m\x1b[0;m\n",
766 "\x1b[0KUsing Docker executor with image python:3.11 ...\x1b[0;m\n",
767 "section_end:1790320401:prepare_executor\r\x1b[0Ksection_start:1790320401:prepare_script\r\x1b[0K\x1b[0K\x1b[36;1mPreparing environment\x1b[0;m\x1b[0;m\n",
768 "Running on runner-fkxhbgrut-project-1344-concurrent-0\n",
769 "section_end:1790320402:prepare_script\r\x1b[0Ksection_start:1790320402:get_sources[collapsed=true]\r\x1b[0K\x1b[0K\x1b[36;1mGetting source from Git repository\x1b[0;m\x1b[0;m\n",
770 "\x1b[32;1mFetching changes with git depth set to 20...\x1b[0;m\n",
771 "section_end:1790320403:get_sources\r\x1b[0Ksection_start:1790320403:step_script\r\x1b[0K\x1b[0K\x1b[36;1mExecuting \"step_script\" stage of the job script\x1b[0;m\x1b[0;m\n",
772 "\x1b[32;1m$ make test\x1b[0;m\n",
773 "test result: ok. 3 passed\n",
774 "section_end:1790320422:step_script\r\x1b[0Ksection_start:1790320422:cleanup_file_variables\r\x1b[0K\x1b[0K\x1b[36;1mCleaning up project directory and file based variables\x1b[0;m\x1b[0;m\n",
775 "section_end:1790320423:cleanup_file_variables\r\x1b[0K\x1b[32;1mJob succeeded\x1b[0;m\n",
776 );
777
778 const GITLAB_TIMESTAMPED_LOG: &str = concat!(
779 "2026-10-27T10:00:00.100000Z 00O Running with gitlab-runner 17.0.0\n",
780 "2026-10-27T10:00:00.100000Z 00O section_start:1793095200:prepare_executor\r\x1b[0K\n",
781 "2026-10-27T10:00:00.100100Z 00O+\x1b[0K\x1b[36;1mPreparing the \"shell\" executor\x1b[0;m\x1b[0;m\n",
782 "2026-10-27T10:00:01.200000Z 00O \x1b[0KUsing Shell (bash) executor...\x1b[0;m\n",
783 "2026-10-27T10:00:02.300000Z 00O section_end:1793095202:prepare_executor\r\x1b[0Ksection_start:1793095202:step_script\r\x1b[0K\n",
784 "2026-10-27T10:00:02.300100Z 00O+\x1b[0K\x1b[36;1mExecuting \"step_script\" stage of the job script\x1b[0;m\n",
785 "2026-10-27T10:00:03.400000Z 00O progress 10%\rprogress 100%\n",
786 "2026-10-27T10:00:04.500000Z 00O section_end:1793095204:step_script\r\x1b[0K\n",
787 "2026-10-27T10:00:04.500000Z 00O Job succeeded\n",
788 );
789
790 type SectionRow = (String, Option<u32>, Option<u32>, u32, bool);
791
792 fn ingest_gitlab(log: &str) -> (Vec<SectionRow>, Vec<String>) {
793 let dir = tempfile::tempdir().expect("tempdir");
794 let mut ingester = ingester(dir.path(), doc_key(), 1, Vec::new(), BlockLimits::default());
795 ingester.dialect = &GITLAB;
796 let (tx, mut rx) = mpsc::channel(1);
797 tx.try_send(Bytes::copy_from_slice(log.as_bytes()))
798 .expect("channel has room");
799 drop(tx);
800
801 let meta = ingester
802 .run(&mut rx)
803 .expect("ingest")
804 .finish(None)
805 .expect("finish")
806 .meta;
807
808 let sections = meta
809 .sections
810 .iter()
811 .map(|s| (s.title.clone(), s.parent, s.header_line, s.end_line, s.open))
812 .collect();
813 let text = Document::open(dir.path())
814 .expect("open")
815 .text_block(0)
816 .expect("block");
817 let lines = std::str::from_utf8(&text)
818 .expect("utf8")
819 .lines()
820 .map(str::to_string)
821 .collect();
822 (sections, lines)
823 }
824
825 #[test]
826 fn gitlab_sections_survive_ingest() {
827 let (sections, lines) = ingest_gitlab(GITLAB_LOG);
828
829 let row = |title: &str, header: u32, end: u32, open: bool| -> SectionRow {
830 (title.to_string(), None, Some(header), end, open)
831 };
832 assert_eq!(
833 sections,
834 vec![
835 row("Preparing the \"docker+machine\" executor", 2, 4, true),
836 row("Preparing environment", 4, 6, true),
837 row("Getting source from Git repository", 6, 8, false),
838 row(
839 "Executing \"step_script\" stage of the job script",
840 8,
841 11,
842 true
843 ),
844 row(
845 "Cleaning up project directory and file based variables",
846 11,
847 12,
848 true
849 ),
850 ]
851 );
852 assert_eq!(lines.len(), 13);
853 assert!(lines[12].ends_with("Job succeeded"));
854 assert!(lines.iter().all(|line| !line.contains("section_")));
855 }
856
857 #[test]
858 fn gitlab_text_after_a_section_end_follows_the_section() {
859 let (sections, lines) = ingest_gitlab(concat!(
860 "section_start:1700000000:build[collapsed=true]\r\x1b[0KBuild\n",
861 "compiling\n",
862 "section_end:1700000010:build\r\x1b[0KDone\n",
863 "after the section\n",
864 ));
865
866 assert_eq!(
867 sections,
868 vec![("Build".to_string(), None, Some(0), 2, false)]
869 );
870 assert_eq!(
871 lines,
872 vec![
873 "2023-11-14T22:13:20.000Z Build",
874 "compiling",
875 "2023-11-14T22:13:30.000Z Done",
876 "after the section",
877 ]
878 );
879 }
880
881 #[test]
882 fn gitlab_runner_timestamps_and_continued_headers_survive_ingest() {
883 let (sections, lines) = ingest_gitlab(GITLAB_TIMESTAMPED_LOG);
884
885 assert_eq!(
886 sections,
887 vec![
888 ("prepare_executor".to_string(), None, Some(1), 4, true),
889 ("step_script".to_string(), None, Some(4), 7, true),
890 ]
891 );
892 assert_eq!(lines.len(), 8);
893 assert_eq!(
894 lines[0],
895 "2026-10-27T10:00:00.100Z Running with gitlab-runner 17.0.0"
896 );
897 assert_eq!(lines[6], "2026-10-27T10:00:03.400Z progress 100%");
898 assert_eq!(lines[7], "2026-10-27T10:00:04.500Z Job succeeded");
899 assert!(lines.iter().all(|line| !line.contains("00O")));
900 }
901
902 static GITEA: GiteaLogDialect = GiteaLogDialect;
903 static BITBUCKET: BitbucketLogDialect = BitbucketLogDialect;
904 static TEKTON: TektonLogDialect = TektonLogDialect;
905 static JENKINS: JenkinsLogDialect = JenkinsLogDialect;
906 static AZURE: AzureDevOpsLogDialect = AzureDevOpsLogDialect;
907 static PLAIN: PlainDialect = PlainDialect;
908
909 fn error_texts(dialect: &'static dyn LogDialect, log: &str) -> Vec<String> {
910 let dir = tempfile::tempdir().expect("tempdir");
911 let mut ingester = ingester(dir.path(), doc_key(), 1, Vec::new(), BlockLimits::default());
912 ingester.dialect = dialect;
913 let (tx, mut rx) = mpsc::channel(1);
914 tx.try_send(Bytes::copy_from_slice(log.as_bytes()))
915 .expect("channel has room");
916 drop(tx);
917
918 let meta = ingester
919 .run(&mut rx)
920 .expect("ingest")
921 .finish(None)
922 .expect("finish")
923 .meta;
924
925 let text = Document::open(dir.path())
926 .expect("open")
927 .text_block(0)
928 .expect("block");
929 let lines: Vec<&str> = std::str::from_utf8(&text).expect("utf8").lines().collect();
930 assert_eq!(meta.error_lines.len() as u32, meta.error_total);
931 meta.error_lines
932 .iter()
933 .map(|&line| lines[line as usize].to_string())
934 .collect()
935 }
936
937 fn assert_errors(found: &[String], expected: &[&str]) {
938 assert_eq!(found.len(), expected.len(), "{found:#?}");
939 for (line, suffix) in found.iter().zip(expected) {
940 assert!(
941 line.ends_with(suffix),
942 "{line:?} does not end with {suffix:?}"
943 );
944 }
945 }
946
947 #[test]
948 fn jenkins_console_notes_are_dropped_before_search_and_error_inference() {
949 let dir = tempfile::tempdir().expect("tempdir");
950 let mut ingester = ingester(dir.path(), doc_key(), 1, Vec::new(), BlockLimits::default());
951 ingester.dialect = &JENKINS;
952 let (tx, mut rx) = mpsc::channel(1);
953 tx.try_send(Bytes::from_static(
954 concat!(
955 "\x1b[8mha:////4IkPeNOtKgYQ==\x1b[0m[Pipeline] Start of Pipeline\n",
956 "Started by user \x1b[8mha:////4O3qAAAA=\x1b[0madmin\n",
957 "\x1b[8mha:////4IkPeNOtKgYQ==\x1b[0mERROR: Job failed: exit code 1\n",
958 )
959 .as_bytes(),
960 ))
961 .expect("channel has room");
962 drop(tx);
963
964 let meta = ingester
965 .run(&mut rx)
966 .expect("ingest")
967 .finish(None)
968 .expect("finish")
969 .meta;
970
971 let text = Document::open(dir.path())
972 .expect("open")
973 .text_block(0)
974 .expect("block");
975 assert_eq!(
976 std::str::from_utf8(&text).expect("utf8"),
977 "[Pipeline] Start of Pipeline\nStarted by user admin\nERROR: Job failed: exit code 1\n"
978 );
979 assert!(meta.error_lines.is_empty());
980
981 let found = error_texts(
982 &PLAIN,
983 "\x1b[8mha:////4IkPeNOtKgYQ==\x1b[0mERROR: Job failed: exit code 1\n",
984 );
985 assert_eq!(found, ["ERROR: Job failed: exit code 1"]);
986 }
987
988 #[test]
989 fn gitlab_failed_job_flags_the_runner_verdict_and_tool_failures() {
990 let found = error_texts(
991 &GITLAB,
992 concat!(
993 "Running with gitlab-runner 18.4.0 (139a0ac0)\n",
994 "section_start:1790320399:step_script\r\x1b[0K\x1b[0K\x1b[36;1mExecuting \"step_script\" stage of the job script\x1b[0;m\x1b[0;m\n",
995 "\x1b[32;1m$ npm test\x1b[0;m\n",
996 "FAIL src/login.test.ts\n",
997 "npm ERR! Test failed. See above for more details.\n",
998 "section_end:1790320422:step_script\r\x1b[0Ksection_start:1790320422:cleanup_file_variables\r\x1b[0K\x1b[0K\x1b[36;1mCleaning up project directory and file based variables\x1b[0;m\x1b[0;m\n",
999 "section_end:1790320423:cleanup_file_variables\r\x1b[0K\x1b[31;1mERROR: Job failed: exit code 1\x1b[0;m\n",
1000 ),
1001 );
1002
1003 assert_errors(
1004 &found,
1005 &[
1006 "FAIL src/login.test.ts",
1007 "npm ERR! Test failed. See above for more details.",
1008 "ERROR: Job failed: exit code 1",
1009 ],
1010 );
1011 }
1012
1013 #[test]
1014 fn gitea_failed_job_flags_the_runner_verdict_and_tool_failures() {
1015 let found = error_texts(
1016 &GITEA,
1017 concat!(
1018 "2026-10-28T10:15:30.0000000Z ⭐ Run Main npm test\n",
1019 "2026-10-28T10:15:31.0000000Z > app@1.0.0 test\n",
1020 "2026-10-28T10:15:31.5000000Z FAIL src/login.test.ts\n",
1021 "2026-10-28T10:15:32.0000000Z npm ERR! Test failed. See above for more details.\n",
1022 "2026-10-28T10:15:33.0000000Z ❌ Failure - Main npm test\n",
1023 "2026-10-28T10:15:33.0000000Z exitcode '1': failure\n",
1024 ),
1025 );
1026
1027 assert_errors(
1028 &found,
1029 &[
1030 "FAIL src/login.test.ts",
1031 "npm ERR! Test failed. See above for more details.",
1032 "❌ Failure - Main npm test",
1033 "exitcode '1': failure",
1034 ],
1035 );
1036 }
1037
1038 #[test]
1039 fn bitbucket_failed_step_flags_test_and_build_failures() {
1040 let found = error_texts(
1041 &BITBUCKET,
1042 concat!(
1043 "+ npm ci\n",
1044 "added 214 packages\n",
1045 "+ python manage.py test\n",
1046 "Traceback (most recent call last):\n",
1047 " File \"manage.py\", line 12, in <module>\n",
1048 "ValueError: invalid literal for int() with base 10: 'x'\n",
1049 "+ make build\n",
1050 "make: *** [Makefile:12: build] Error 2\n",
1051 "Searching for test report files in directories named [test-results]\n",
1052 ),
1053 );
1054
1055 assert_errors(
1056 &found,
1057 &[
1058 "Traceback (most recent call last):",
1059 "ValueError: invalid literal for int() with base 10: 'x'",
1060 "make: *** [Makefile:12: build] Error 2",
1061 ],
1062 );
1063 }
1064
1065 #[test]
1066 fn tekton_failed_step_flags_container_output_failures() {
1067 let found = error_texts(
1068 &TEKTON,
1069 concat!(
1070 "\x1e2026-10-28T10:01:01Z Step run-tests\n",
1071 "2026-10-28T10:01:20Z running 3 tests\n",
1072 "2026-10-28T10:01:30Z test login::rejects_bad_password ... FAILED\n",
1073 "2026-10-28T10:01:30Z test result: FAILED. 2 passed; 1 failed\n",
1074 "2026-10-28T10:01:31Z {\"level\":\"error\",\"msg\":\"tests failed\"}\n",
1075 "2026-10-28T10:01:31Z time=\"2026-10-28T10:01:31Z\" level=fatal msg=\"exit status 1\"\n",
1076 ),
1077 );
1078
1079 assert_errors(
1080 &found,
1081 &[
1082 "test login::rejects_bad_password ... FAILED",
1083 "test result: FAILED. 2 passed; 1 failed",
1084 "{\"level\":\"error\",\"msg\":\"tests failed\"}",
1085 "time=\"2026-10-28T10:01:31Z\" level=fatal msg=\"exit status 1\"",
1086 ],
1087 );
1088 }
1089
1090 #[test]
1091 fn jenkins_unstable_build_is_an_error_like_its_failed_status() {
1092 let found = error_texts(
1093 &JENKINS,
1094 concat!(
1095 "Started by user admin\n",
1096 "[app] $ /bin/sh -xe /tmp/jenkins123.sh\n",
1097 "+ make test\n",
1098 "Tests run: 4, Failures: 1\n",
1099 "Finished: UNSTABLE\n",
1100 ),
1101 );
1102
1103 assert_errors(&found, &["Finished: UNSTABLE"]);
1104 }
1105
1106 #[test]
1107 fn providers_with_native_error_markers_do_not_infer_more() {
1108 let jenkins = error_texts(
1109 &JENKINS,
1110 concat!(
1111 "[Pipeline] sh\n",
1112 "+ make test\n",
1113 "test result: FAILED. 0 passed; 1 failed\n",
1114 "[Pipeline] }\n",
1115 "[Pipeline] // stage\n",
1116 "[Pipeline] error\n",
1117 "ERROR: script returned exit code 2\n",
1118 "Finished: FAILURE\n",
1119 ),
1120 );
1121 assert_errors(
1122 &jenkins,
1123 &["ERROR: script returned exit code 2", "Finished: FAILURE"],
1124 );
1125
1126 let azure = error_texts(
1127 &AZURE,
1128 concat!(
1129 "2026-10-28T10:00:00.0000000Z ##[section]Starting: Bash\n",
1130 "2026-10-28T10:00:01.0000000Z npm ERR! Test failed.\n",
1131 "2026-10-28T10:00:02.0000000Z ##[error]Bash exited with code '1'.\n",
1132 "2026-10-28T10:00:02.0000000Z ##[section]Finishing: Bash\n",
1133 ),
1134 );
1135 assert_errors(&azure, &["Bash exited with code '1'."]);
1136 }
1137
1138 #[test]
1139 fn ordinary_output_is_not_flagged_as_an_error() {
1140 let found = error_texts(
1141 &PLAIN,
1142 concat!(
1143 "0 errors\n",
1144 "error_handler.rs\n",
1145 "cargo test --fail-fast\n",
1146 "errors: none\n",
1147 "test errors::parse_error ... ok\n",
1148 "PASS src/error.test.ts\n",
1149 "✓ handles error responses\n",
1150 "Compiling error-chain v0.12.0\n",
1151 "ERROR_COUNT=0\n",
1152 "No errors found\n",
1153 "0 FAILED\n",
1154 "Process exited with code 0\n",
1155 "exit status 0\n",
1156 "log_level=error\n",
1157 "Error handling guide\n",
1158 "level=info msg=\"retrying after error\"\n",
1159 "Finished in 0.5s (0 failures)\n",
1160 "exit code 2 is documented below\n",
1161 ),
1162 );
1163
1164 assert_errors(&found, &[]);
1165 }
1166}