+ "once_cell",

+version = "0.1.3"

+ "serde 1.0.147",

+ "sigpipe",

+ "thiserror",

+name = "sigpipe"

+version = "0.1.3"

+source = "registry+https://github.com/rust-lang/crates.io-index"

+checksum = "5584bfb3e0d348139d8210285e39f6d2f8a1902ac06de343e06357d1d763d8e6"

+dependencies = [

+ "libc",

+]

+

+[[package]]

+tokio = { version="1", features = ["full"] }

+reqwest = { version="0.11", features=["json"]}

+clap = { version="4", features=["string", "cargo"] }

+thiserror = "1.0"

+serde = "1.0"

+sigpipe = "0.1.3"

+## How to install

+

+1. clone this repository

+2. run `cargo install --path=.`

+

+or download prebuild binaries for `amd64`

+[here](https://gitlab.com/imn1/glp/-/packages/).

+

+$ echo "456" > .glp; glp --trace=-1 # fetches traces for all failed jobs from the most recent pipeline

+$ echo "456" > .glp; glp --trace=789 | less # fetches traces for all failed jobs from pipeline 789

+### 0.1.3

+- tracing added (--trace) - possibility to fetch failed jobs traces (logs)

+- inner refactoring

+- much better error hangling

+use crate::{

+ job::{Job, JobError},

+ pipeline::{Pipeline, PipelineError},

+ stage::Stage,

+};

+use std::{error::Error, sync::Arc};

+

+use crate::{config::Config, ui::Ui};

+

+pub struct App {

+ config: Arc<Config>,

+ ui: Ui,

+}

+

+impl App {

+ pub fn new(config: Arc<Config>, ui: Ui) -> Self {

+ Self { config, ui }

+ }

+

+ pub async fn run(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {

+ let result;

+

+ if let Some(trace) = &self.config.trace {

+ // 1. Test trace case.

+ if trace.starts_with('-') {

+ result = self

+ .trace_relatively(trace.chars().skip(1).collect::<String>().parse::<u8>()?)

+ .await;

+ } else {

+ result = self.trace_by_id(trace.to_string()).await;

+ }

+ } else if false {

+ // Other future cases.

+ unimplemented!()

+ } else {

+ // 2. Default behavior

+ result = self.run_without_arguments().await;

+ }

+

+ //...

+ match result {

+ Ok(_) => Ok(()),

+ Err(e) => {

+ self.ui.print_error(&e);

+ Err(e)

+ }

+ }

+ }

+

+ async fn run_without_arguments(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {

+ self.ui.start_spinner();

+

+ // 2. Fetch jobs for each running pipeline.

+ let pips = Pipeline::fetch_last(self.config.clone(), Some(self.config.limit)).await?;

+

+ // 3. print tree.

+ self.ui.stop_spinner();

+ self.ui.print_tree(pips);

+

+ Ok(())

+ }

+

+ async fn trace_relatively(&mut self, trace: u8) -> Result<(), Box<dyn Error + Send + Sync>> {

+ self.ui.start_spinner();

+ // Fetch last N (see trace param) pipelines and get the last one's stages.

+ let stages = match Pipeline::fetch_last(self.config.clone(), Some(trace))

+ .await?

+ .iter()

+ .last()

+ {

+ Some(pip) => pip.to_owned(),

+ None => return Err(Box::new(PipelineError::NotFound)),

+ }

+ .stages;

+ self.ui.stop_spinner();

+ self.print_job_traces(self.fetch_failed_jobs(stages).await?);

+

+ Ok(())

+ }

+

+ async fn trace_by_id(&self, pipeline_id: String) -> Result<(), Box<dyn Error + Send + Sync>> {

+ let stages = Pipeline::fetch_stages(self.config.clone(), pipeline_id).await?;

+

+ self.print_job_traces(self.fetch_failed_jobs(stages).await?);

+ Ok(())

+ }

+

+ async fn fetch_failed_jobs(

+ &self,

+ stages: Vec<Stage>,

+ ) -> Result<Vec<Job>, Box<dyn Error + Send + Sync>> {

+ let mut failed_jobs: Vec<Job> = vec![];

+

+ for stage in stages {

+ failed_jobs.append(

+ &mut self

+ .filter_failed_jobs_and_fetch_traces(stage.clone())

+ .await?,

+ );

+ }

+

+ if failed_jobs.is_empty() {

+ return Err(Box::new(PipelineError::NoFailedJobs));

+ }

+

+ Ok(failed_jobs)

+ }

+

+ async fn filter_failed_jobs_and_fetch_traces(

+ &self,

+ stage: Stage,

+ ) -> Result<Vec<Job>, JobError> {

+ // Get all failed jobs from all the stages and fetch trace logs.

+ let filtered_failed_jobs = stage

+ .jobs

+ .iter()

+ .filter(|job| job.is_failed())

+ .collect::<Vec<&Job>>();

+ let mut failed_jobs: Vec<Job> = vec![];

+

+ for job in filtered_failed_jobs {

+ let mut xjob = job.clone();

+ xjob.fetch_trace(self.config.clone()).await?;

+ failed_jobs.push(xjob);

+ }

+

+ Ok(failed_jobs)

+ }

+

+ fn print_job_traces(&self, jobs: Vec<Job>) {

+ for job in jobs {

+ self.ui.print_job_trace(job);

+ }

+ }

+}

+use clap::{crate_version, value_parser, Arg, ArgAction, ArgMatches, Command};

+

+pub const DEFAULT_LIMIT: u8 = 3;

+ .version(crate_version!())

+ .value_parser(value_parser!(String))

+ .help("GitLab project ID."),

+ .default_value(DEFAULT_LIMIT.to_string())

+ .help("Limit number of fetched pipelines."),

+ .action(ArgAction::SetTrue)

+ .help("Care only about finished pipelines."),

+ )

+ .arg(

+ Arg::new("trace")

+ .long("trace")

+ .action(ArgAction::Set)

+ .value_parser(value_parser!(String))

+ .value_name("pipeline")

+ .conflicts_with("limit")

+ .help("Fetches traces (logs) from all failed pipeline jobs. Specify pipeline ID or it's relative number with '-' prefix."),

+use clap::ArgMatches;

+use std::env;

+use tokio::fs;

+

+pub struct Config {

+ pub project_id: String,

+ pub private_token: String,

+ pub limit: u8,

+ pub finished: bool,

+ pub trace: Option<String>,

+ _args: ArgMatches,

+}

+

+impl Config {

+ pub async fn new(args: ArgMatches) -> Self {

+ // Project ID.

+ let project_id = match args.get_one::<String>("project") {

+ Some(id) => id.to_owned(),

+ None => fs::read_to_string(".glp")

+ .await

+ .expect("No project ID (no parameter nor .glp file."),

+ };

+

+ let finished = args.get_one::<bool>("finished").unwrap().to_owned();

+ let private_token = env::var("GLP_PRIVATE_TOKEN")

+ .expect("No Gitlab private token found - set GLP_PRIVATE_TOKEN environment variable.");

+ let limit = args.get_one::<u8>("limit").unwrap().to_owned();

+ let trace = args.get_one::<String>("trace").cloned();

+

+ Self {

+ project_id,

+ private_token,

+ limit,

+ finished,

+ trace,

+ _args: args,

+ }

+ }

+}

+use crate::config::Config;

+use std::sync::Arc;

+use thiserror::Error;

+#[derive(Error, Debug)]

+pub enum JobError {

+ // #[error("Job not found.")]

+ // NotFound,

+ #[error("Fetching error: {0:?}")]

+ FetchingError(#[source] reqwest::Error),

+ #[error("Parsing error: {0:?}.")]

+ ParsingError(#[source] reqwest::Error),

+}

+

+#[derive(Debug, Clone, Default)]

+ pub trace: Option<String>,

+

+impl Job {

+ pub fn is_failed(&self) -> bool {

+ self.status == "failed"

+ }

+

+ pub async fn fetch_trace(&mut self, config: Arc<Config>) -> Result<(), JobError> {

+ // GET /projects/:id/jobs/:job_id/trace

+

+ let client = reqwest::Client::new();

+ self.trace = Some(

+ client

+ .get(format!(

+ "https://gitlab.com/api/v4/projects/{}/jobs/{}/trace",

+ &config.project_id, &self.id

+ ))

+ .header("PRIVATE-TOKEN", &config.private_token)

+ .send()

+ .await

+ .map_err(JobError::FetchingError)?

+ .text()

+ .await

+ .map_err(JobError::ParsingError)?,

+ );

+

+ Ok(())

+ }

+}

+

+mod app;

+mod config;

+use crate::app::App;

+use crate::config::Config;

+#[derive(Debug, Clone, Default)]

+#[tokio::main]

+pub async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {

+ // Fixes "failed printing to stdout: Broken pipe" known as sigpipe issue

+ // https://github.com/rust-lang/rust/issues/97889

+ sigpipe::reset();

+ // Create config and UI instances.

+ let config = Arc::new(Config::new(args::parse()).await);

+ let ui = Ui::from_args(config.clone());

+ // Run the app.

+ App::new(config, ui).run().await

+use crate::args::DEFAULT_LIMIT;

+use crate::config::Config;

+use crate::job::Job;

+use futures::future::try_join_all;

+use std::cmp::Ordering;

+use std::collections::HashMap;

+use std::sync::Arc;

+use thiserror::Error;

+use tokio::sync::Semaphore;

+

+const SEMAPHORE_LIMIT: usize = 10;

+

+#[derive(Error, Debug)]

+pub enum PipelineError {

+ #[error("Pipeline not found.")]

+ NotFound,

+ #[error("No failed jobs found.")]

+ NoFailedJobs,

+ #[error("Fetching error.")]

+ FetchingError,

+ #[error("Parsing error.")]

+ ParsingError,

+}

+ /// Fetches last N (see limit param) pipelines.

+ async fn fetch_last_pipelines(

+ private_token: &str,

+ project_id: &str,

+ limit: &u8,

+ ) -> Result<JsonValue, PipelineError> {

+ // Fetch pipelines.

+ let client = reqwest::Client::new();

+ json::parse(

+ client

+ .get(format!(

+ "https://gitlab.com/api/v4/projects/{}/pipelines?per_page={}",

+ project_id, limit

+ ))

+ .header("PRIVATE-TOKEN", private_token)

+ .send()

+ .await

+ .map_err(|_| PipelineError::FetchingError)?

+ .text()

+ .await

+ .map_err(|_| PipelineError::FetchingError)?

+ .as_str(),

+ )

+ .map_err(|_| PipelineError::ParsingError)

+ }

+

+ // Fetch stages + jobs for the given pipeline.

+ pub async fn fetch_stages(

+ config: Arc<Config>,

+ pipeline_id: String,

+ ) -> Result<Vec<Stage>, PipelineError> {

+ // Fetch data.

+ let client = reqwest::Client::new();

+ let jobs = json::parse(

+ client

+ .get(format!(

+ "https://gitlab.com/api/v4/projects/{}/pipelines/{}/jobs",

+ &config.project_id, &pipeline_id

+ ))

+ .header("PRIVATE-TOKEN", &config.private_token)

+ .send()

+ .await

+ .map_err(|_| PipelineError::FetchingError)?

+ .text()

+ .await

+ .map_err(|_| PipelineError::FetchingError)?

+ .as_str(),

+ )

+ .map_err(|_| PipelineError::FetchingError)?;

+

+ // Check for errors.

+ if jobs.has_key("error") {

+ return Err(PipelineError::NotFound);

+ }

+

+ // Process JSON response.

+ // Construct jobs.

+ let mut stages: HashMap<String, Vec<Job>> = HashMap::new();

+

+ for j in 0..jobs.len() {

+ let job = &jobs[j];

+

+ let pip_job = Job {

+ id: job["id"].as_usize().unwrap().to_string(),

+ name: Label(job["name"].as_str().unwrap().to_string()),

+ status: job["status"].as_str().unwrap().to_string(),

+ web_url: job["web_url"].as_str().unwrap().to_string(),

+ stage: job["stage"].as_str().unwrap().to_string(),

+ started_at: match job["started_at"].is_null() {

+ true => None,

+ false => Some(job["started_at"].as_str().unwrap().to_string()),

+ },

+ duration: match job["duration"].is_null() {

+ true => None,

+ false => Some(Duration::from_secs_f64(job["duration"].as_f64().unwrap())),

+ },

+ ..Default::default()

+ };

+

+ if stages.contains_key(&pip_job.stage) {

+ stages.get_mut(&pip_job.stage).unwrap().push(pip_job);

+ } else {

+ stages.insert(pip_job.stage.clone(), vec![pip_job]);

+ }

+ }

+

+ let mut pip_stages = vec![];

+

+ // Convert hashmap to vec of stages

+ for (stage, jobs) in stages.into_iter() {

+ pip_stages.push(Stage {

+ name: Label(stage),

+ jobs,

+ });

+ }

+

+ // Sort stages by job "started_at" times.

+ // None are always classified as "greater"

+ // so they end up as "last".

+ pip_stages.sort_by(|a, b| {

+ let mut a_jobs = a

+ .jobs

+ .iter()

+ .filter(|j| j.started_at.is_some())

+ .collect::<Vec<&Job>>();

+ a_jobs.sort_by_key(|j| j.started_at.clone());

+

+ let mut b_jobs = b

+ .jobs

+ .iter()

+ .filter(|j| j.started_at.is_some())

+ .collect::<Vec<&Job>>();

+ b_jobs.sort_by_key(|j| j.started_at.clone());

+

+ let a_started_at = match a_jobs.get(0) {

+ Some(j) => j.started_at.clone(),

+ _ => None,

+ };

+ let b_started_at = match b_jobs.get(0) {

+ Some(j) => j.started_at.clone(),

+ _ => None,

+ };

+

+ if a_started_at.is_none() {

+ return Ordering::Greater;

+ }

+ if b_started_at.is_none() {

+ return Ordering::Less;

+ }

+

+ a_started_at.partial_cmp(&b_started_at).unwrap()

+ });

+

+ Ok(pip_stages)

+ }

+

+ /// ...

+ pub async fn fetch_last(

+ config: Arc<Config>,

+ limit: Option<u8>,

+ ) -> Result<Vec<Self>, PipelineError> {

+ let pipelines = Self::fetch_last_pipelines(

+ &config.private_token,

+ &config.project_id,

+ &limit.unwrap_or(DEFAULT_LIMIT),

+ )

+ .await?;

+

+ // 2. Fetch jobs for each running pipeline.

+ let mut tasks = vec![];

+ let semaphore = Arc::new(Semaphore::new(SEMAPHORE_LIMIT));

+

+ for i in 0..pipelines.len() {

+ let pip = pipelines[i].clone();

+

+ // Acquire semaphore lock.

+ let semaphore_permit = semaphore.clone().acquire_owned().await.unwrap();

+ let xconfig = config.clone();

+

+ tasks.push(tokio::spawn(async move {

+ let pip_stages =

+ Self::fetch_stages(xconfig.clone(), pip["id"].clone().to_string()).await?;

+

+ let mut pip = Pipeline {

+ id: Label(pip["id"].as_usize().unwrap().to_string()),

+ git_ref: pip["ref"].as_str().unwrap().to_string(),

+ status: pip["status"].as_str().unwrap().to_string(),

+ stages: pip_stages,

+ show_finished: xconfig.finished,

+ details: None,

+ };

+

+ // Fetch details only if needed.

+ if xconfig.finished {

+ pip.fetch_details(&xconfig.private_token, &xconfig.project_id)

+ .await?;

+ }

+

+ // Free acquired semaphore lock.

+ drop(semaphore_permit);

+

+ // We need to explicitly tell the return types here.

+ Ok::<Pipeline, PipelineError>(pip)

+ }));

+ }

+

+ // Result gets compiled from collect (FromIterator).

+ // Once any task contains an Err() instead of Ok()

+ // the whole collect fails (and maps to PipelineError from

+ // the method definition).

+ try_join_all(tasks)

+ .await

+ .map_err(|_| PipelineError::FetchingError)?

+ .into_iter()

+ .collect()

+ }

+

+ pub async fn fetch_details(

+ &mut self,

+ private_token: &str,

+ project_id: &str,

+ ) -> Result<(), PipelineError> {

+ self.details = Some(

+ json::parse(

+ client

+ .get(format!(

+ "https://gitlab.com/api/v4/projects/{}/pipelines/{}",

+ project_id, &self.id.0

+ ))

+ .header("PRIVATE-TOKEN", private_token)

+ .send()

+ .await

+ .map_err(|_| PipelineError::FetchingError)?

+ .text()

+ .await

+ .map_err(|_| PipelineError::FetchingError)?

+ .as_str(),

+ )

+ .map_err(|_| PipelineError::ParsingError)?,

+ );

+ Ok(())

+use std::error::Error;

+use std::sync::Arc;

+use crate::config::Config;

+use crate::job::Job;

+ _config: Arc<Config>,

+ pub fn from_args(config: Arc<Config>) -> Self {

+ _config: config,

+ if let Some(spinner) = self.spinner.as_mut() {

+ spinner.stop();

+ print!("\x1b[2K\r");

+

+ /// Prints nicely job trace (log) with a header.

+ pub fn print_job_trace(&self, job: Job) {

+ let no_color_title = format!("Job: {} ID: {} URL: {}", &job.name.0, &job.id, &job.web_url);

+ let title = format!(

+ "Job: {} ID: {} URL: {}",

+ &job.name.to_string(&job.status),

+ &job.id,

+ &job.web_url

+ );

+ let size = no_color_title.chars().count();

+ println!("{}", "-".repeat(size));

+ println!("{}", title);

+ println!("{}", "-".repeat(size));

+ println!("{}", &job.trace.unwrap_or("No trace (yet) ...".to_string()));

+ println!("{}", "-".repeat(size));

+ }

+

+ pub fn print_error(&self, err: &Box<dyn Error + Send + Sync>) {

+ eprintln!("Error: {}", err);

+

+ if let Some(source) = err.source() {

+ eprintln!("Original error: {}", source);

+ }

+ }