Runners: Connectivity and Status apart, a heartbeat, a filterable queue e9b9619cd857

        .route("/flow/runners/heartbeat", post(crate::flow::heartbeat))

    /// When it started running.

    pub started_ms: Option<i64>,

        "SELECT j.id AS job_id, j.run_id, r.repo_id, p.owner, p.name, p.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox, j.started_ms \

    /// Whether the hub has heard from it lately (oxygenhub#243): any call it

    /// makes counts, a busy runner's heartbeat included.

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

        if self.online() {

            "online"

        } else {

            "offline"

        }

    }

/// `POST /api/v1/flow/runners/heartbeat` (oxygenhub#243): a runner whose

/// job slots are all full claims nothing, so it says it is there.

pub async fn heartbeat(State(app): State<Shared>, headers: HeaderMap) -> Result<StatusCode, ApiError> {

    let me = runner_auth(&app, &headers).await?;

    db::runners::seen(&app.db, me.id, None, None).await?;

    Ok(StatusCode::NO_CONTENT)

}

            started_ms: None,

    /// The jobs it runs now: (repository, run, job, started).

    pub jobs: Vec<(String, i64, String, Option<i64>)>,

impl RunnerRow {

    /// What it is doing (oxygenhub#243): `disabled`, `draining`, `idle`,

    /// or `running N of M`.

    pub fn doing(&self) -> String {

        match self.r.state.as_str() {

            "disabled" => "disabled".into(),

            "draining" if self.running > 0 => format!("draining, {} left", self.running),

            "draining" => "draining".into(),

            _ if self.running > 0 => format!("running {} of {}", self.running, self.r.max_jobs),

            _ => "idle".into(),

        }

    }

    /// The jobs it runs, for the status's tooltip.

    pub fn jobs_line(&self) -> String {

        let now = crate::db::now_ms();

        self.jobs

            .iter()

            .map(|(repo, run, job, started)| match started {

                Some(t) => format!("{repo} #{run} {job}, {} min", (now - t).max(0) / 60_000),

                None => format!("{repo} #{run} {job}"),

            })

            .collect::<Vec<_>>()

            .join("; ")

    }

}

/// The queue's filters (oxygenhub#243), kept in the URL.

#[derive(Deserialize, Default, Clone)]

pub struct QueueFilter {

    #[serde(default)]

    pub repo: String,

    #[serde(default)]

    pub status: String,

    #[serde(default)]

    pub runner: String,

}

impl QueueFilter {

    fn keeps(&self, q: &db::flow::QueueEntry) -> bool {

        (self.repo.is_empty() || format!("{}/{}", q.owner, q.name) == self.repo)

            && (self.status.is_empty() || q.status == self.status)

            && (self.runner.is_empty() || q.runner == self.runner)

    }

    pub fn any(&self) -> bool {

        !(self.repo.is_empty() && self.status.is_empty() && self.runner.is_empty())

    }

}

    /// Running jobs, then queued ones in the order runners claim them,

    /// as the filter keeps them.

    /// How many there are, unfiltered.

    queue_total: usize,

    filter: QueueFilter,

    /// The repositories and runners in the queue, for the filter's choices.

    queue_repos: Vec<String>,

    queue_runners: Vec<String>,

    page_filtered(app, shell, new_token, notice, error, QueueFilter::default()).await

}

async fn page_filtered(

    app: &Shared,

    shell: Shell,

    new_token: Option<String>,

    notice: Option<String>,

    error: Option<String>,

    filter: QueueFilter,

) -> Result<Page<RunnersPage>, Error> {

    let all = db::flow::queue(&app.db).await.map_err(e)?;

        let jobs = all

            .iter()

            .filter(|q| q.status == "running" && q.runner == r.name)

            .map(|q| (format!("{}/{}", q.owner, q.name), q.run_number, q.job.clone(), q.started_ms))

            .collect();

        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, swarm, jobs, r });

    let mut queue_repos: Vec<String> = all.iter().map(|q| format!("{}/{}", q.owner, q.name)).collect();

    queue_repos.sort();

    queue_repos.dedup();

    let mut queue_runners: Vec<String> = all.iter().filter(|q| !q.runner.is_empty()).map(|q| q.runner.clone()).collect();

    queue_runners.sort();

    queue_runners.dedup();

    let queue_total = all.len();

    let queue = all.into_iter().filter(|q| filter.keeps(q)).collect();

    Ok(Page(RunnersPage {

        hub: app.config.base_url().to_string(),

        shell,

        runners: rows,

        pending,

        new_token,

        notice,

        error,

        queue,

        queue_total,

        filter,

        queue_repos,

        queue_runners,

    }))

pub async fn list(State(app): State<Shared>, viewer: Viewer, Query(filter): Query<QueueFilter>) -> Result<Page<RunnersPage>, Error> {

    page_filtered(&app, shell, None, None, None, filter).await

    assert!(page.contains("id=\"queue\"") && page.contains(">plex</a></td>"), "{page}");

    // Connectivity and what it is doing, kept apart (#243); its status

    // opens the queue filtered to it, and the filters keep and drop jobs.

    assert!(page.contains("running 1 of ") && page.contains("/admin/runners?runner=plex#queue"), "{page}");

    let (_, only) = web("GET", "/admin/runners?runner=plex".into(), None).await;

    assert!(only.contains(">plex</a></td>") && only.contains(" of ") && only.contains(">all</a>"), "{only}");

    let (_, none) = web("GET", "/admin/runners?runner=nobody&status=running".into(), None).await;

    assert!(none.contains("Nothing matches."), "{none}");

    // A busy runner claims nothing; its heartbeat keeps it online.

    sqlx::query("UPDATE flow_runners SET last_seen_ms = 0 WHERE id = $1").bind(plex.id).execute(&h.app.db).await.unwrap();

    let (_, page) = web("GET", "/admin/runners".into(), None).await;

    assert!(page.contains("st-offline"), "quiet for long: offline");

    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/heartbeat", Some(&credential), None).await.0, StatusCode::NO_CONTENT);

    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/heartbeat", Some("flowrunner_nope"), None).await.0, StatusCode::UNAUTHORIZED);

    let plex_now = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex").unwrap();

    assert_eq!(plex_now.connectivity(), "online");

  <tr><th>Runner</th><th>Connectivity</th><th>Status</th><th>Serves</th><th>Signing key</th><th>Machine</th><th>Labels</th><th>Capacity</th><th>Last seen</th><th></th></tr>

    <td><span class="runner-status st-{{ r.connectivity() }}">{{ r.connectivity() }}</span></td>

    <td class="small"><a href="/admin/runners?runner={{ r.name|urlencode }}#queue" title="{% if row.jobs.is_empty() %}Its jobs in the queue{% else %}{{ row.jobs_line() }}{% endif %}">{{ row.doing() }}</a></td>

<section class="panel" id="queue">

  <h2>Queue <small>{% if filter.any() %}{{ queue.len() }} of {{ queue_total }} · <a href="/admin/runners#queue">all</a>{% else %}{{ queue_total|plural("job", "jobs") }}{% endif %}</small></h2>

  <form class="filters" method="get" action="/admin/runners#queue">

    <select name="repo" aria-label="Repository"><option value="">every repository</option>{% for r in queue_repos %}<option{% if *r == filter.repo %} selected{% endif %}>{{ r }}</option>{% endfor %}</select>

    <select name="status" aria-label="Status"><option value="">running and queued</option><option value="running"{% if filter.status == "running" %} selected{% endif %}>running</option><option value="queued"{% if filter.status == "queued" %} selected{% endif %}>queued</option></select>

    <select name="runner" aria-label="Runner"><option value="">every runner</option>{% for r in queue_runners %}<option{% if *r == filter.runner %} selected{% endif %}>{{ r }}</option>{% endfor %}{% if !filter.runner.is_empty() && !queue_runners.contains(filter.runner) %}<option selected>{{ filter.runner }}</option>{% endif %}</select>

    <button>filter</button>

  </form>

  {% if queue.is_empty() %}<p class="mute small">{% if filter.any() %}Nothing matches.{% else %}Nothing running or waiting.{% endif %}</p>{% else %}

  <tr><th>Repository</th><th>Run</th><th>Job</th><th>Status</th><th>Started</th><th>Runner</th></tr>

  {% for q in queue %}<tr><td><a href="/{{ q.owner }}/{{ q.name }}">{{ q.owner }}/{{ q.name }}</a></td><td><a href="/{{ q.owner }}/{{ q.name }}/flow/{{ q.run_number }}">#{{ q.run_number }}</a></td><td>{{ q.job }}</td><td>{{ q.status }}</td><td class="small">{% if let Some(t) = q.started_ms %}<time title="{{ t|when_i }}">{{ t|ago_i }}</time>{% endif %}</td><td>{% if q.runner.is_empty() %}<span class="mute">waiting{% let needs = q.needs() %}{% if !needs.is_empty() %} for a runner with {% for l in needs %}<span class="lbl">{{ l }}</span> {% endfor %}{% endif %}{% if let Some(a) = q.arch() %} on <span class="lbl">{{ a }}</span>{% endif %}</span>{% else %}<a href="/admin/runners?runner={{ q.runner|urlencode }}#queue">{{ q.runner }}</a>{% endif %}</td></tr>{% endfor %}