pipedash_plugin_api/
logs.rs

1use std::collections::HashMap;
2use std::time::Duration;
3
4use bytes::Bytes;
5use chrono::{
6    DateTime,
7    Utc,
8};
9use serde::{
10    Deserialize,
11    Serialize,
12};
13use tokio::sync::mpsc;
14
15use crate::error::{
16    PluginError,
17    PluginResult,
18};
19use crate::http::map_send_error;
20use crate::types::PipelineStatus;
21
22pub const LOG_CHUNK_STALL_TIMEOUT: Duration = Duration::from_secs(60);
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(rename_all = "lowercase")]
26pub enum LogNodeKind {
27    Stage,
28    Job,
29    Step,
30}
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "lowercase")]
34pub enum LogAvailability {
35    Ready,
36    Partial,
37    Pending,
38}
39
40#[derive(Debug, Clone, Serialize, Deserialize)]
41pub struct LogNode {
42    pub id: String,
43    pub parent_id: Option<String>,
44    pub kind: LogNodeKind,
45    pub name: String,
46    pub status: PipelineStatus,
47    pub started_at: Option<DateTime<Utc>>,
48    pub finished_at: Option<DateTime<Utc>>,
49    pub log: Option<LogAvailability>,
50    pub web_url: Option<String>,
51}
52
53pub struct RunRef<'a> {
54    pub pipeline_id: &'a str,
55    pub run_number: i64,
56    pub run_id: &'a str,
57    pub metadata: &'a HashMap<String, serde_json::Value>,
58}
59
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61pub enum LineMarker {
62    Plain,
63    GroupStart,
64    GroupEnd,
65    Error,
66    Warning,
67    Notice,
68    Debug,
69    Command,
70}
71
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct ParsedLine<'a> {
74    pub timestamp: Option<DateTime<Utc>>,
75    pub marker: LineMarker,
76    pub body: &'a [u8],
77}
78
79pub trait LogDialect: Send + Sync {
80    fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a>;
81
82    fn is_step_start(&self, _line: &ParsedLine<'_>) -> bool {
83        false
84    }
85}
86
87pub struct PlainDialect;
88
89impl LogDialect for PlainDialect {
90    fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a> {
91        ParsedLine {
92            timestamp: None,
93            marker: LineMarker::Plain,
94            body: line,
95        }
96    }
97}
98
99pub struct LogSink {
100    tx: mpsc::Sender<Bytes>,
101}
102
103impl LogSink {
104    pub fn new(tx: mpsc::Sender<Bytes>) -> Self {
105        Self { tx }
106    }
107
108    pub async fn write(&mut self, chunk: Bytes) -> PluginResult<()> {
109        if chunk.is_empty() {
110            return Ok(());
111        }
112        self.tx
113            .send(chunk)
114            .await
115            .map_err(|_| PluginError::Internal("Log reader stopped".into()))
116    }
117
118    pub async fn pipe(&mut self, mut response: reqwest::Response) -> PluginResult<()> {
119        let url = response.url().clone();
120        loop {
121            match tokio::time::timeout(LOG_CHUNK_STALL_TIMEOUT, response.chunk()).await {
122                Err(_) => {
123                    return Err(PluginError::NetworkError(
124                        "Log download stalled for 60 seconds".into(),
125                    ))
126                }
127                Ok(Err(e)) => return Err(map_send_error(e, &url)),
128                Ok(Ok(None)) => return Ok(()),
129                Ok(Ok(Some(chunk))) => self.write(chunk).await?,
130            }
131        }
132    }
133}
134
135#[cfg(test)]
136mod tests {
137    use std::io::{
138        Read,
139        Write,
140    };
141    use std::net::TcpListener;
142
143    use super::*;
144
145    #[tokio::test]
146    async fn pipe_delivers_chunked_bytes_exactly() {
147        let _ = rustls::crypto::ring::default_provider().install_default();
148        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
149        let addr = listener.local_addr().unwrap();
150        std::thread::spawn(move || {
151            if let Ok((mut stream, _)) = listener.accept() {
152                let mut buf = [0u8; 4096];
153                let _ = stream.read(&mut buf);
154                let _ = stream.write_all(
155                    b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\
156                      5\r\nhello\r\n7\r\n world\n\r\n3\r\nend\r\n0\r\n\r\n",
157                );
158            }
159        });
160        let response = reqwest::Client::new()
161            .get(format!("http://{addr}/log"))
162            .send()
163            .await
164            .unwrap();
165        let (tx, mut rx) = mpsc::channel(16);
166        let mut sink = LogSink::new(tx);
167        sink.pipe(response).await.unwrap();
168        drop(sink);
169        let mut got = Vec::new();
170        while let Some(chunk) = rx.recv().await {
171            got.extend_from_slice(&chunk);
172        }
173        assert_eq!(got, b"hello world\nend");
174    }
175
176    #[tokio::test]
177    async fn write_after_reader_dropped_is_internal() {
178        let (tx, rx) = mpsc::channel(1);
179        drop(rx);
180        let mut sink = LogSink::new(tx);
181        let err = sink.write(Bytes::from_static(b"x")).await.unwrap_err();
182        assert!(matches!(err, PluginError::Internal(_)));
183    }
184}
pipedash_plugin_api/logs.rs

pipedash_plugin_api/
logs.rs
