Coverage / crates/hub/src/insights.rs 94.5%

    pub fn parse(s: &str) -> Range {

        match s {

            "1h" => Range::Hour,

            "7d" => Range::Week,

            _ => Range::Day,

    }

    pub fn key(self) -> &'static str {

        match self {

            Range::Hour => "1h",

            Range::Day => "24h",

            Range::Week => "7d",

    }

    pub fn ms(self) -> i64 {

        match self {

            Range::Hour => 3_600_000,

            Range::Day => 86_400_000,

            Range::Week => 7 * 86_400_000,

    }

    pub fn buckets(self) -> usize {

        match self {

            Range::Hour => 60,

            Range::Day => 48,

            Range::Week => 56,

    }

    pub fn bucket_ms(self) -> i64 {

        self.ms() / self.buckets() as i64

    }

    fn pickup(&self) -> Option<f64> {

        Some((self.started_ms? - self.ready_ms.or(self.queued_ms)?).max(0) as f64 / 1000.0)

    }

    fn duration(&self) -> Option<f64> {

        Some((self.finished_ms? - self.started_ms?).max(0) as f64 / 1000.0)

    }

    fn left_queue(&self) -> Option<i64> {

        self.started_ms.or(self.finished_ms)

    }

    fn ended(&self) -> bool {

        !matches!(self.status.as_str(), "queued" | "running")

    }

pub fn percentile(v: &mut [f64], p: f64) -> Option<f64> {

    if v.is_empty() {

        return None;

    }

    v.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));

    let rank = ((p * v.len() as f64).ceil() as usize).clamp(1, v.len());

    Some(v[rank - 1])

}

async fn rows(pool: &Pool, scope: &Scope, since: i64) -> sqlx::Result<Vec<JobRow>> {

    sqlx::query_as(

        "SELECT j.status, j.runner, p.owner, p.name AS repo, COALESCE(j.queued_ms, r.created_ms) AS queued_ms, j.ready_ms,

                j.started_ms, j.finished_ms

         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id

         WHERE (j.finished_ms >= $1 OR j.status IN ('queued', 'running'))

           AND ($2 = '' OR p.owner = $3) AND ($4 IS NULL OR p.id = $5) AND ($6 = '' OR j.runner = $7)",

    )

    .bind(since)

    .bind(scope.owner.as_deref().unwrap_or(""))

    .bind(scope.owner.as_deref().unwrap_or(""))

    .bind(scope.repo_id)

    .bind(scope.repo_id)

    .bind(scope.runner.as_deref().unwrap_or(""))

    .bind(scope.runner.as_deref().unwrap_or(""))

    .fetch_all(pool)

    .await

}

pub async fn gather(pool: &Pool, scope: &Scope, range: Range) -> sqlx::Result<Insights> {

    let now = db::now_ms();

    let start = now - range.ms();

    let jobs = rows(pool, scope, start).await?;

    let runners = db::runners::list(pool).await?;

    let n = range.buckets();

    let width = range.bucket_ms();

    let bucket = |t: i64| -> Option<usize> { (t >= start && t <= now).then(|| (((t - start) / width) as usize).min(n - 1)) };

    let mut pickups: Vec<Vec<f64>> = vec![Vec::new(); n];

    let mut durations: Vec<Vec<f64>> = vec![Vec::new(); n];

    let (mut passed, mut failed, mut other) = (vec![0f64; n], vec![0f64; n], vec![0f64; n]);

    let (mut all_pickups, mut all_durations) = (Vec::new(), Vec::new());

    for j in &jobs {

        if let (Some(p), Some(b)) = (j.pickup(), j.started_ms.and_then(bucket)) {

            pickups[b].push(p);

            all_pickups.push(p);

        }

        if let (Some(f), true) = (j.finished_ms, j.ended()) {

            if let Some(b) = bucket(f) {

                match j.status.as_str() {

                    "passed" => passed[b] += 1.0,

                if let Some(d) = j.duration() {

                    durations[b].push(d);

                    all_durations.push(d);

                }

    let (mut queued_at, mut running_at) = (vec![0f64; n], vec![0f64; n]);

    for (i, (q, r)) in queued_at.iter_mut().zip(running_at.iter_mut()).enumerate() {

        let t = (start + (i as i64 + 1) * width).min(now);

        for j in &jobs {

            let entered = j.queued_ms.unwrap_or(i64::MAX);

            if entered <= t && j.left_queue().is_none_or(|l| l > t) && !(j.ended() && j.finished_ms.is_none()) {

            }

            if j.started_ms.is_some_and(|s| s <= t) && j.finished_ms.is_none_or(|f| f > t) && !(j.ended() && j.finished_ms.is_none()) {

            }

    let serving: Vec<&db::runners::Runner> =

        runners.iter().filter(|r| scope.runner.as_deref().is_none_or(|n| n == r.name)).filter(|r| r.state != "removed").collect();

    let slots_total: i64 = serving.iter().filter(|r| r.online()).map(|r| r.max_jobs.max(1)).sum();

    let utilisation: Vec<Option<f64>> =

        running_at.iter().map(|r| (slots_total > 0).then(|| (r / slots_total as f64 * 100.0).min(100.0))).collect();

    let quantiles = |sets: &mut Vec<Vec<f64>>, p: f64| -> Vec<Option<f64>> { sets.iter_mut().map(|v| percentile(v, p)).collect() };

    let pickup = vec![

        Series { name: "median", class: "s-main", values: quantiles(&mut pickups, 0.5) },

        Series { name: "95th percentile", class: "s-warn", values: quantiles(&mut pickups, 0.95) },

    let durations_series = vec![

        Series { name: "median", class: "s-main", values: quantiles(&mut durations, 0.5) },

        Series { name: "95th percentile", class: "s-warn", values: quantiles(&mut durations, 0.95) },

    let some = |v: Vec<f64>| v.into_iter().map(Some).collect::<Vec<_>>();

    let ended_total = |v: &[f64]| v.iter().sum::<f64>() as i64;

    let (p_total, f_total, o_total) = (ended_total(&passed), ended_total(&failed), ended_total(&other));

    let mut runner_rows: Vec<RunnerRow> = serving

        .iter()

        .filter(|r| scope.owner.is_none() && scope.repo_id.is_none() || jobs.iter().any(|j| j.runner == r.name))

        .map(|r| {

            let mine: Vec<&JobRow> = jobs.iter().filter(|j| j.runner == r.name).collect();

            let mut p: Vec<f64> = mine.iter().filter(|j| j.started_ms.is_some_and(|s| s >= start)).filter_map(|j| j.pickup()).collect();

            let mut d: Vec<f64> = mine.iter().filter(|j| j.ended()).filter_map(|j| j.duration()).collect();

                name: r.name.clone(),

                status: r.status(),

                running: mine.iter().filter(|j| j.status == "running").count() as i64,

                max_jobs: r.max_jobs.max(1),

                jobs: mine.iter().filter(|j| j.ended()).count() as i64,

                failed: mine.iter().filter(|j| j.status == "failed").count() as i64,

                pickup_median: percentile(&mut p, 0.5),

                duration_median: percentile(&mut d, 0.5),

                last_seen_ms: r.last_seen_ms,

        })

        .collect();

    runner_rows.sort_by(|a, b| b.jobs.cmp(&a.jobs).then_with(|| a.name.cmp(&b.name)));

    let mut by_repo: std::collections::BTreeMap<String, (i64, i64, Vec<f64>, i64)> = Default::default();

    for j in &jobs {

        let e = by_repo.entry(format!("{}/{}", j.owner, j.repo)).or_default();

        if j.ended() && j.finished_ms.is_some_and(|f| f >= start) {

            e.0 += 1;

            if j.status == "failed" {

            }

        if let Some(p) = j.pickup().filter(|_| j.started_ms.is_some_and(|s| s >= start)) {

            e.2.push(p);

        }

        if j.status == "queued" {

        }

    let mut repos: Vec<RepoRow> = by_repo

        .into_iter()

        .map(|(slug, (jobs, failed, mut p, queued))| RepoRow {

            slug,

            jobs,

            failed,

            pickup_median: percentile(&mut p, 0.5),

            pickup_p95: percentile(&mut p, 0.95),

            queued,

        })

        .collect();

    repos.sort_by(|a, b| (b.jobs + b.queued).cmp(&(a.jobs + a.queued)).then_with(|| a.slug.cmp(&b.slug)));

    repos.truncate(15);

        range,

        start_ms: start,

        now_ms: now,

        pickup,

        results: vec![

            Series { name: "passed", class: "s-ok", values: some(passed) },

            Series { name: "failed", class: "s-bad", values: some(failed) },

            Series { name: "other (error, cancelled, skipped)", class: "s-mute", values: some(other) },

        durations: durations_series,

        load: vec![

            Series { name: "queued", class: "s-warn", values: some(queued_at) },

            Series { name: "running", class: "s-main", values: some(running_at) },

        utilisation: vec![Series { name: "busy slots", class: "s-main", values: utilisation }],

        ended: p_total + f_total + o_total,

        passed: p_total,

        failed: f_total,

        pickup_median: percentile(&mut all_pickups, 0.5),

        pickup_p95: percentile(&mut all_pickups, 0.95),

        duration_median: percentile(&mut all_durations, 0.5),

        queued_now: jobs.iter().filter(|j| j.status == "queued").count() as i64,

        running_now: jobs.iter().filter(|j| j.status == "running").count() as i64,

        runners_online: serving.iter().filter(|r| r.online()).count() as i64,

        runners_total: serving.len() as i64,

        slots_busy: jobs.iter().filter(|j| j.status == "running").count() as i64,

        slots_total,

        runners: runner_rows,

        repos,

}

pub fn seconds(s: f64) -> String {

    if s < 60.0 {

        format!("{s:.0} s")

    } else if s < 3600.0 {

        let m = s / 60.0;

        if m < 10.0 {

            format!("{m:.1} min")

        format!("{:.1} h", s / 3600.0)

}

    fn percentiles_are_nearest_rank() {

        let mut v = vec![5.0, 1.0, 3.0, 2.0, 4.0];

        assert_eq!(percentile(&mut v, 0.5), Some(3.0));

        assert_eq!(percentile(&mut v, 0.95), Some(5.0));

        assert_eq!(percentile(&mut [], 0.5), None);

        assert_eq!(seconds(42.0), "42 s");

        assert_eq!(seconds(210.0), "3.5 min");

        assert_eq!(seconds(4320.0), "1.2 h");

    }