pijul/nest - Change 7L26BC7TAXMWKGQ33Y2PGTGF6PET2XSOSVY4XAFSOUFGKF6Y2EEAC

job logs: proper sync using standard tools

Created by pmeunier yesterday at 15:23

7L26BC7TAXMWKGQ33Y2PGTGF6PET2XSOSVY4XAFSOUFGKF6Y2EEAC

Dependencies

In channels

main

Change contents

Replacement in ci/src/main.rs at line 4 [2.65]
∅:D[6.50] → [4.4474:4496]
B:BD[4.4474] → [4.4474:4496]
```
    http::StatusCode,
```
[6.50]
[4.4496]
```
    http::{HeaderMap, StatusCode, header},
```
Deletion in ci/src/main.rs at line 8 [2.65]
B:BD[4.4565] → [4.4565:4603]
∅:D[4.4603] → [2.83:104]
B:BD[2.83] → [2.83:104]
B:BD[2.104] → [4.4604:4687]
```
// use ci::Message;
use clap::Parser;
use serde_derive::*;
// use std::io::Read;
use axum_extra::TypedHeader;
use axum_extra::headers::Range;
```
Insertion in ci/src/main.rs at line 9 [2.65]
[4.4698]
[4.4698]
```
use clap::Parser;
```
Insertion in ci/src/main.rs at line 11 [2.65]
[4.4723]
[4.4723]
```
use serde_derive::*;
```
Insertion in ci/src/main.rs at line 20 [2.65]
[4.4924]
[6.51]
```
use tower_http::services::ServeDir;
```

Replacement in ci/src/main.rs at line 78 [2.65]

∅:D[6.129] → [4.5745:5835]

B:BD[4.5745] → [4.5745:5835]

        .route("/stdout/{job}", get(stdout))
        .route("/stderr/{job}", get(stderr))

[6.129]

[4.5835]

        .nest_service("/logs", ServeDir::new(&config.config.ci_path))

Deletion in ci/src/main.rs at line 118 [2.65]

∅:D[5.271] → [2.2720:2722]

∅:D[4.6235] → [2.2720:2722]

B:BD[2.2720] → [2.2720:2722]

∅:D[4.6287] → [2.2910:2911]

B:BD[2.2910] → [2.2910:2911]

B:BD[2.2911] → [4.6288:6705]

B:BD[4.6705] → [7.0:728]

∅:D[7.728] → [4.6817:6974]

B:BD[4.6817] → [4.6817:6974]

∅:D[4.6974] → [2.3010:3019]

B:BD[2.3010] → [2.3010:3019]

B:BD[2.3019] → [4.6975:7392]

B:BD[4.7392] → [7.729:1452]

∅:D[7.1452] → [4.7504:7661]

B:BD[4.7504] → [4.7504:7661]

∅:D[4.7661] → [2.5066:5072]

B:BD[2.5066] → [2.5066:5072]

}
async fn stdout(
    State(config): State<Config>,
    Path(job): Path<uuid::Uuid>,
    range: Option<TypedHeader<Range>>,
) -> Response {
    debug!("stdout {:?} {:?}", job, range);
    let Ok(mut file) =
        tokio::fs::File::open(&config.config.ci_path.join(&format!("{}.stdout", job))).await
    else {
        return StatusCode::NOT_FOUND.into_response();
    };
    if let Some(TypedHeader(range)) = range {
        use tokio::io::AsyncSeekExt;
        match range.satisfiable_ranges(0).next() {
            Some((std::ops::Bound::Included(a), std::ops::Bound::Unbounded)) => {
                use tokio::io::AsyncReadExt;
                let mut v = Vec::new();
                file.seek(std::io::SeekFrom::Start(a)).await.unwrap();
                file.read_to_end(&mut v).await.unwrap();
                v.into_response()
            }
            None => {
                use tokio::io::AsyncReadExt;
                let mut v = Vec::new();
                file.read_to_end(&mut v).await.unwrap();
                v.into_response()
            }
            r => {
                panic!("range = {:?}", r)
            }
        }
    } else {
        use tokio::io::AsyncReadExt;
        let mut v = Vec::new();
        file.read_to_end(&mut v).await.unwrap();
        v.into_response()
    }
}
async fn stderr(
    State(config): State<Config>,
    Path(job): Path<uuid::Uuid>,
    range: Option<TypedHeader<Range>>,
) -> Response {
    debug!("stderr {:?} {:?}", job, range);
    let Ok(mut file) =
        tokio::fs::File::open(&config.config.ci_path.join(&format!("{}.stderr", job))).await
    else {
        return StatusCode::NOT_FOUND.into_response();
    };
    if let Some(TypedHeader(range)) = range {
        use tokio::io::AsyncSeekExt;
        match range.satisfiable_ranges(0).next() {
            Some((std::ops::Bound::Included(a), std::ops::Bound::Unbounded)) => {
                use tokio::io::AsyncReadExt;
                let mut v = Vec::new();
                file.seek(std::io::SeekFrom::Start(a)).await.unwrap();
                file.read_to_end(&mut v).await.unwrap();
                v.into_response()
            }
            None => {
                use tokio::io::AsyncReadExt;
                let mut v = Vec::new();
                file.read_to_end(&mut v).await.unwrap();
                v.into_response()
            }
            r => {
                panic!("r {:?}", r);
            }
        }
    } else {
        use tokio::io::AsyncReadExt;
        let mut v = Vec::new();
        file.read_to_end(&mut v).await.unwrap();
        v.into_response()
    }

Insertion in api/src/replication.rs at line 14 [8.431834]
[3.18441]
[3.18441]
```
use reqwest::StatusCode;
```

Insertion in api/src/replication.rs at line 429 [8.431834]

[4.44511]

[3.24870]

    let mut ticker = tokio::time::interval(std::time::Duration::from_secs(1));
    let (mut stdout_offset, mut stderr_offset) = if let Some((ref stdout, ref stderr)) = files {
        (
            stdout.metadata().await.map(|m| m.len()).unwrap_or(0),
            stderr.metadata().await.map(|m| m.len()).unwrap_or(0),
        )
    } else {
        (0, 0)
    };

Replacement in api/src/replication.rs at line 471 [8.431834]

B:BD[4.45513] → [4.45513:45652]

                    .get(format!("{}/stdout/{}", ci.url[ci_n], job))
                    .header("Range", format!("bytes={stdout_len}-*"))

[4.45513]

[4.45652]

                    .get(format!("{}/logs/{}.stdout", ci.url[ci_n], job))
                    .header(reqwest::header::RANGE, format!("bytes={stdout_len}-"))

Replacement in api/src/replication.rs at line 475 [8.431834]

B:BD[4.45704] → [4.45704:45843]

                    .get(format!("{}/stderr/{}", ci.url[ci_n], job))
                    .header("Range", format!("bytes={stderr_len}-*"))

[4.45704]

[4.45843]

                    .get(format!("{}/logs/{}.stderr", ci.url[ci_n], job))
                    .header(reqwest::header::RANGE, format!("bytes={stderr_len}-"))

Replacement in api/src/replication.rs at line 481 [8.431834]

B:BD[4.45960] → [4.45960:46221]

            debug!("{:?} {:?}", stdout_, stderr_);
            let bytes = stdout_.bytes().await.unwrap();
            stdout.write(&bytes).await.unwrap();
            let bytes = stderr_.bytes().await.unwrap();
            stderr.write(&bytes).await.unwrap();

[4.45960]

[3.30829]

            match stdout_.status() {
                StatusCode::PARTIAL_CONTENT => {
                    let bytes = stdout_.bytes().await.unwrap();
                    stdout.write_all(&bytes).await.unwrap();
                    stdout_offset += bytes.len() as u64;
                }
                other => eprintln!("unexpected status : {other}"),
            }
            match stderr_.status() {
                StatusCode::PARTIAL_CONTENT => {
                    let bytes = stderr_.bytes().await.unwrap();
                    stderr.write_all(&bytes).await.unwrap();
                    stderr_offset += bytes.len() as u64;
                }
                other => eprintln!("unexpected status : {other}"),
            }

Replacement in api/src/replication.rs at line 557 [8.431834]

B:BD[4.47502] → [4.47502:47570]

        tokio::time::sleep(std::time::Duration::from_secs(1)).await

[4.47502]

[3.30854]

        ticker.tick().await;

job logs: proper sync using standard tools

Dependencies

In channels

Change contents

Replacement in ci/src/main.rs at line 4 [2.65]

Deletion in ci/src/main.rs at line 8 [2.65]

Insertion in ci/src/main.rs at line 9 [2.65]

Insertion in ci/src/main.rs at line 11 [2.65]

Insertion in ci/src/main.rs at line 20 [2.65]