pipedash_plugin_api/
logs.rs

1pub mod fetch;
2pub mod rendered;
3pub mod tagged;
4
5use std::collections::HashMap;
6use std::future::Future;
7use std::sync::atomic::{
8    AtomicBool,
9    Ordering,
10};
11use std::sync::Arc;
12use std::time::Duration;
13
14use bytes::Bytes;
15use chrono::{
16    DateTime,
17    Utc,
18};
19use reqwest::header::{
20    HeaderMap,
21    CONTENT_RANGE,
22};
23use serde::de::DeserializeOwned;
24use serde::{
25    Deserialize,
26    Serialize,
27};
28use tokio::sync::mpsc;
29
30use crate::error::{
31    PluginError,
32    PluginResult,
33};
34use crate::http::{
35    check_response,
36    map_send_error,
37};
38use crate::types::PipelineStatus;
39
40pub const LOG_CHUNK_STALL_TIMEOUT: Duration = Duration::from_secs(60);
41
42pub async fn next_log_chunk(response: &mut reqwest::Response) -> PluginResult<Option<Bytes>> {
43    match tokio::time::timeout(LOG_CHUNK_STALL_TIMEOUT, response.chunk()).await {
44        Err(_) => Err(PluginError::NetworkError(
45            "Log download stalled for 60 seconds".into(),
46        )),
47        Ok(chunk) => chunk.map_err(|e| map_send_error(e, response.url())),
48    }
49}
50
51const LOG_CHUNK_BYTES: usize = 64 * 1024;
52
53pub const CONTENT_HASH_SEED: u64 = 0xcbf2_9ce4_8422_2325;
54
55pub fn content_hash(bytes: &[u8]) -> u64 {
56    extend_content_hash(CONTENT_HASH_SEED, bytes)
57}
58
59pub fn extend_content_hash(hash: u64, bytes: &[u8]) -> u64 {
60    bytes.iter().fold(hash, |hash, &byte| {
61        (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3)
62    })
63}
64
65#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
66#[serde(rename_all = "lowercase")]
67pub enum LogNodeKind {
68    Stage,
69    Job,
70    Step,
71}
72
73#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
74#[serde(rename_all = "lowercase")]
75pub enum LogAvailability {
76    Ready,
77    Partial,
78    Pending,
79}
80
81#[derive(Debug, Clone, Serialize, Deserialize)]
82pub struct LogNode {
83    pub id: String,
84    pub parent_id: Option<String>,
85    pub kind: LogNodeKind,
86    pub name: String,
87    pub status: PipelineStatus,
88    pub started_at: Option<DateTime<Utc>>,
89    pub finished_at: Option<DateTime<Utc>>,
90    pub log: Option<LogAvailability>,
91    pub web_url: Option<String>,
92}
93
94impl LogNode {
95    pub fn step(
96        job_id: &str, number: impl std::fmt::Display, name: String, status: PipelineStatus,
97        started_at: Option<DateTime<Utc>>, finished_at: Option<DateTime<Utc>>,
98    ) -> Self {
99        Self {
100            id: format!("{job_id}:{number}"),
101            parent_id: Some(job_id.to_string()),
102            kind: LogNodeKind::Step,
103            name,
104            status,
105            started_at,
106            finished_at,
107            log: None,
108            web_url: None,
109        }
110    }
111}
112
113pub struct RunRef<'a> {
114    pub pipeline_id: &'a str,
115    pub run_number: i64,
116    pub run_id: &'a str,
117    pub metadata: &'a HashMap<String, serde_json::Value>,
118}
119
120#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
121#[serde(transparent)]
122pub struct LogCursor(serde_json::Value);
123
124impl LogCursor {
125    pub fn new<T: Serialize>(position: &T) -> PluginResult<Self> {
126        serde_json::to_value(position)
127            .map(Self)
128            .map_err(|error| PluginError::SerializationError(error.to_string()))
129    }
130
131    pub fn parse<T: DeserializeOwned>(&self) -> Option<T> {
132        serde_json::from_value(self.0.clone()).ok()
133    }
134}
135
136#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
137#[serde(rename_all = "lowercase")]
138pub enum LogReadMode {
139    Append,
140    Replace,
141}
142
143#[derive(Debug, Clone, PartialEq, Eq)]
144pub struct LogRead {
145    pub mode: LogReadMode,
146    pub cursor: Option<LogCursor>,
147}
148
149impl LogRead {
150    pub fn replace(cursor: Option<LogCursor>) -> Self {
151        Self {
152            mode: LogReadMode::Replace,
153            cursor,
154        }
155    }
156
157    pub fn append(cursor: LogCursor) -> Self {
158        Self {
159            mode: LogReadMode::Append,
160            cursor: Some(cursor),
161        }
162    }
163}
164
165#[derive(Debug, Clone, Copy, PartialEq, Eq)]
166pub enum LineMarker {
167    Plain,
168    GroupStart,
169    GroupEnd,
170    SectionStart { open: bool },
171    OpenSection,
172    Error,
173    Warning,
174    Notice,
175    Debug,
176    Command,
177}
178
179#[derive(Debug, Clone, Copy, PartialEq, Eq)]
180pub struct ParsedLine<'a> {
181    pub timestamp: Option<DateTime<Utc>>,
182    pub marker: LineMarker,
183    pub body: &'a [u8],
184}
185
186impl<'a> ParsedLine<'a> {
187    pub fn plain(line: &'a [u8]) -> Self {
188        Self {
189            timestamp: None,
190            marker: LineMarker::Plain,
191            body: line,
192        }
193    }
194}
195
196pub trait LogDialect: Send + Sync {
197    fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a>;
198
199    fn is_step_start(&self, _line: &ParsedLine<'_>) -> bool {
200        false
201    }
202
203    fn infers_errors(&self) -> bool {
204        true
205    }
206}
207
208pub struct PlainDialect;
209
210impl LogDialect for PlainDialect {
211    fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a> {
212        ParsedLine::plain(line)
213    }
214}
215
216#[derive(Debug, Clone, Copy, PartialEq, Eq)]
217pub struct ContentRange {
218    pub span: Option<(u64, u64)>,
219    pub total: Option<u64>,
220}
221
222impl ContentRange {
223    pub fn parse(headers: &HeaderMap) -> Option<Self> {
224        let value = headers
225            .get(CONTENT_RANGE)?
226            .to_str()
227            .ok()?
228            .strip_prefix("bytes ")?;
229        let (span, total) = value.split_once('/')?;
230        let span = match span.trim() {
231            "*" => None,
232            span => {
233                let (start, end) = span.split_once('-')?;
234                Some((start.parse().ok()?, end.parse().ok()?))
235            }
236        };
237        let total = match total.trim() {
238            "*" => None,
239            total => Some(total.parse().ok()?),
240        };
241        Some(Self { span, total })
242    }
243}
244
245#[derive(Debug, Clone, Copy, PartialEq, Eq)]
246pub enum RangedRead {
247    Appended(u64),
248    Replaced(u64),
249    Refetch,
250}
251
252pub struct LogSink {
253    tx: mpsc::Sender<Bytes>,
254    bytes: u64,
255    restarted: Arc<AtomicBool>,
256}
257
258impl LogSink {
259    pub fn new(tx: mpsc::Sender<Bytes>) -> Self {
260        Self {
261            tx,
262            bytes: 0,
263            restarted: Arc::new(AtomicBool::new(false)),
264        }
265    }
266
267    pub fn restart_signal(&self) -> Arc<AtomicBool> {
268        Arc::clone(&self.restarted)
269    }
270
271    pub fn restart(&mut self) -> PluginResult<()> {
272        if self.bytes > 0 {
273            return Err(PluginError::Internal(
274                "A log read can only restart before it writes".into(),
275            ));
276        }
277        self.restarted.store(true, Ordering::SeqCst);
278        Ok(())
279    }
280
281    pub fn restarted(&self) -> bool {
282        self.restarted.load(Ordering::SeqCst)
283    }
284
285    pub fn bytes_written(&self) -> u64 {
286        self.bytes
287    }
288
289    pub async fn write_chunks(&mut self, mut bytes: Bytes) -> PluginResult<()> {
290        while !bytes.is_empty() {
291            self.write(bytes.split_to(bytes.len().min(LOG_CHUNK_BYTES)))
292                .await?;
293        }
294        Ok(())
295    }
296
297    pub async fn write(&mut self, chunk: Bytes) -> PluginResult<()> {
298        if chunk.is_empty() {
299            return Ok(());
300        }
301        self.bytes += chunk.len() as u64;
302        self.tx
303            .send(chunk)
304            .await
305            .map_err(|_| PluginError::Internal("Log reader stopped".into()))
306    }
307
308    pub async fn pipe(&mut self, response: reqwest::Response) -> PluginResult<()> {
309        self.pipe_after(response, 0).await
310    }
311
312    pub async fn pipe_after(
313        &mut self, mut response: reqwest::Response, mut skip: u64,
314    ) -> PluginResult<()> {
315        while let Some(chunk) = next_log_chunk(&mut response).await? {
316            let cut = usize::try_from(skip).map_or(chunk.len(), |s| s.min(chunk.len()));
317            skip -= cut as u64;
318            self.write(chunk.slice(cut..)).await?;
319        }
320        Ok(())
321    }
322
323    pub async fn pipe_range(
324        &mut self, response: reqwest::Response, offset: u64,
325    ) -> PluginResult<RangedRead> {
326        let range = ContentRange::parse(response.headers());
327        match response.status().as_u16() {
328            206 if range
329                .and_then(|r| r.span)
330                .is_some_and(|(start, _)| start == offset) =>
331            {
332                self.pipe(response).await?;
333                Ok(RangedRead::Appended(offset + self.bytes))
334            }
335            416 if range.is_some_and(|r| r.span.is_none() && r.total == Some(offset)) => {
336                Ok(RangedRead::Appended(offset))
337            }
338            200 => {
339                self.restart()?;
340                self.pipe(response).await?;
341                Ok(RangedRead::Replaced(self.bytes))
342            }
343            206 | 416 => Ok(RangedRead::Refetch),
344            _ => {
345                check_response(response).await?;
346                Ok(RangedRead::Refetch)
347            }
348        }
349    }
350
351    pub async fn pipe_resumed<F, Fut>(
352        &mut self, offset: Option<u64>, mut fetch: F,
353    ) -> PluginResult<(LogReadMode, u64)>
354    where
355        F: FnMut(Option<u64>) -> Fut,
356        Fut: Future<Output = PluginResult<reqwest::Response>>,
357    {
358        if let Some(offset) = offset {
359            match self.pipe_range(fetch(Some(offset)).await?, offset).await? {
360                RangedRead::Appended(next) => return Ok((LogReadMode::Append, next)),
361                RangedRead::Replaced(next) => return Ok((LogReadMode::Replace, next)),
362                RangedRead::Refetch => self.restart()?,
363            }
364        }
365        self.pipe(check_response(fetch(None).await?).await?).await?;
366        Ok((LogReadMode::Replace, self.bytes))
367    }
368}
369
370#[cfg(test)]
371mod tests;
pipedash_plugin_api/logs.rs

pipedash_plugin_api/
logs.rs
