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    SectionStart { open: bool },
66    OpenSection,
67    Error,
68    Warning,
69    Notice,
70    Debug,
71    Command,
72}
73
74#[derive(Debug, Clone, Copy, PartialEq, Eq)]
75pub struct ParsedLine<'a> {
76    pub timestamp: Option<DateTime<Utc>>,
77    pub marker: LineMarker,
78    pub body: &'a [u8],
79}
80
81pub trait LogDialect: Send + Sync {
82    fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a>;
83
84    fn is_step_start(&self, _line: &ParsedLine<'_>) -> bool {
85        false
86    }
87}
88
89pub struct PlainDialect;
90
91impl LogDialect for PlainDialect {
92    fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a> {
93        ParsedLine {
94            timestamp: None,
95            marker: LineMarker::Plain,
96            body: line,
97        }
98    }
99}
100
101pub struct LogSink {
102    tx: mpsc::Sender<Bytes>,
103}
104
105impl LogSink {
106    pub fn new(tx: mpsc::Sender<Bytes>) -> Self {
107        Self { tx }
108    }
109
110    pub async fn write(&mut self, chunk: Bytes) -> PluginResult<()> {
111        if chunk.is_empty() {
112            return Ok(());
113        }
114        self.tx
115            .send(chunk)
116            .await
117            .map_err(|_| PluginError::Internal("Log reader stopped".into()))
118    }
119
120    pub async fn pipe(&mut self, mut response: reqwest::Response) -> PluginResult<()> {
121        let url = response.url().clone();
122        loop {
123            match tokio::time::timeout(LOG_CHUNK_STALL_TIMEOUT, response.chunk()).await {
124                Err(_) => {
125                    return Err(PluginError::NetworkError(
126                        "Log download stalled for 60 seconds".into(),
127                    ))
128                }
129                Ok(Err(e)) => return Err(map_send_error(e, &url)),
130                Ok(Ok(None)) => return Ok(()),
131                Ok(Ok(Some(chunk))) => self.write(chunk).await?,
132            }
133        }
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use std::io::{
140        Read,
141        Write,
142    };
143    use std::net::TcpListener;
144
145    use super::*;
146
147    #[tokio::test]
148    async fn pipe_delivers_chunked_bytes_exactly() {
149        let _ = rustls::crypto::ring::default_provider().install_default();
150        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
151        let addr = listener.local_addr().unwrap();
152        std::thread::spawn(move || {
153            if let Ok((mut stream, _)) = listener.accept() {
154                let mut buf = [0u8; 4096];
155                let _ = stream.read(&mut buf);
156                let _ = stream.write_all(
157                    b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\
158                      5\r\nhello\r\n7\r\n world\n\r\n3\r\nend\r\n0\r\n\r\n",
159                );
160            }
161        });
162        let response = reqwest::Client::new()
163            .get(format!("http://{addr}/log"))
164            .send()
165            .await
166            .unwrap();
167        let (tx, mut rx) = mpsc::channel(16);
168        let mut sink = LogSink::new(tx);
169        sink.pipe(response).await.unwrap();
170        drop(sink);
171        let mut got = Vec::new();
172        while let Some(chunk) = rx.recv().await {
173            got.extend_from_slice(&chunk);
174        }
175        assert_eq!(got, b"hello world\nend");
176    }
177
178    #[tokio::test]
179    async fn write_after_reader_dropped_is_internal() {
180        let (tx, rx) = mpsc::channel(1);
181        drop(rx);
182        let mut sink = LogSink::new(tx);
183        let err = sink.write(Bytes::from_static(b"x")).await.unwrap_err();
184        assert!(matches!(err, PluginError::Internal(_)));
185    }
186}
pipedash_plugin_api/logs.rs

pipedash_plugin_api/
logs.rs
