302 lines
15 KiB
Rust
302 lines
15 KiB
Rust
//! `GET /metrics`, a resource of the gateway's table: it reads strictly as the text format; a
|
|
//! refused request, a WHIP session and a WHEP session move the requests, the refusals, the
|
|
//! sessions running and opened and the records signed, as each session opens and as it ends; an
|
|
//! RTMP connection counts as a session once its publish is admitted, and not before; an SRT
|
|
//! session counts while it runs; the energy marker says no meter backs the gateway; `HEAD` is
|
|
//! answered as `GET` is, without the text; every resource of the gateway's table, asked once, is
|
|
//! counted by its pattern; through a front the path is one the gateway does not have, to a request
|
|
//! with a grant too; and no room, principal, grant, session or address of those requests is in it.
|
|
|
|
mod common;
|
|
|
|
use common::rtmp::{stream_key, Client};
|
|
use common::*;
|
|
use ed25519_dalek::SigningKey;
|
|
use joule_contrib::srt::{self, CallConfig};
|
|
use joule_moq_ingest::frame::Codec;
|
|
use std::collections::BTreeSet;
|
|
use std::time::{Duration, Instant};
|
|
use str0m::media::Direction;
|
|
use tokio::net::TcpStream;
|
|
use tokio::time::timeout;
|
|
use wai_metrics::read::{identifier_like, label_values, parse, sum, value, Family};
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn metrics_move_with_a_session_and_name_no_party() {
|
|
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn scrape(w: &World) -> (String, Vec<Family>) {
|
|
let r = http("GET", &format!("{}/metrics", w.base()), &[], b"").await;
|
|
assert_eq!(r.status, 200, "{}", r.text());
|
|
assert_eq!(r.header("content-type"), Some(wai_metrics::CONTENT_TYPE));
|
|
assert!(r.header("access-control-allow-origin").is_none(), "no page of another origin reads it");
|
|
let text = r.text();
|
|
let f = parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
|
|
(text, f)
|
|
}
|
|
|
|
fn requests(f: &[Family], route: &str, method: &str, code: &str) -> Option<f64> {
|
|
value(f, "wai_ingest_requests_total", &[("route", route), ("method", method), ("code", code)])
|
|
}
|
|
|
|
/// The sessions of `protocol` running, as the gauge states them.
|
|
fn running(f: &[Family], protocol: &str) -> Option<f64> {
|
|
value(f, "wai_ingest_sessions", &[("protocol", protocol)])
|
|
}
|
|
|
|
/// The requests answered with a client or a server error.
|
|
fn errors(f: &[Family]) -> f64 {
|
|
f.iter().flat_map(|x| &x.samples).filter(|s| s.name == "wai_ingest_requests_total" && s.labels.iter().any(|(k, v)| k == "code" && (v.starts_with('4') || v.starts_with('5')))).map(|s| s.value).sum()
|
|
}
|
|
|
|
async fn run() {
|
|
let w = World::start(|_| {}).await;
|
|
let (_, f) = scrape(&w).await;
|
|
assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whip")]), Some(0.0));
|
|
assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whip")]), Some(0.0));
|
|
assert_eq!(value(&f, "wai_ingest_energy_unmetered", &[]), Some(1.0));
|
|
assert_eq!(value(&f, "wai_ingest_records_total", &[]), Some(0.0));
|
|
// A scrape that carries an Authorization field is not read as a grant.
|
|
let r = http("GET", &format!("{}/metrics", w.base()), &[("authorization", "Bearer bm90LWEtZ3JhbnQ")], b"").await;
|
|
assert_eq!(r.status, 200, "{}", r.text());
|
|
assert!(r.header("wai-room-challenge").is_none());
|
|
let r = http("POST", &format!("{}/metrics", w.base()), &[], b"").await;
|
|
assert_eq!((r.status, r.header("allow")), (405, Some("GET, HEAD")));
|
|
let r = http("HEAD", &format!("{}/metrics", w.base()), &[], b"").await;
|
|
assert_eq!((r.status, r.body.is_empty()), (200, true));
|
|
assert_eq!(r.header("content-type"), Some(wai_metrics::CONTENT_TYPE));
|
|
// Through a front, the scrape is no resource here.
|
|
let r = http("GET", &format!("{}/metrics", w.base()), &[("x-forwarded-for", "198.51.100.9")], b"").await;
|
|
assert_eq!(r.status, 404, "{}", r.text());
|
|
assert!(!r.text().contains("wai_ingest"), "{}", r.text());
|
|
assert!(r.header("access-control-allow-origin").is_some(), "as any other resource the gateway does not have");
|
|
// And with a grant's field, as at a path the gateway does not have: read, and answered so.
|
|
let grant = ("authorization", "Bearer bm90LWEtZ3JhbnQ");
|
|
let absent = http("GET", &format!("{}/no-such-resource", w.base()), &[grant], b"").await;
|
|
let r = http("GET", &format!("{}/metrics", w.base()), &[grant, ("x-forwarded-for", "198.51.100.9")], b"").await;
|
|
assert_eq!(r.status, absent.status, "{}", r.text());
|
|
assert_eq!(r.text(), absent.text());
|
|
assert_eq!(r.header("wai-room-challenge").is_some(), absent.header("wai-room-challenge").is_some(), "and each takes the next challenge alike");
|
|
|
|
// A viewer's grant may not publish: refused.
|
|
let whip = format!("{}/whip", w.base());
|
|
let viewer = SigningKey::from_bytes(&[0x61; 32]);
|
|
let va = w.auth(&viewer, 1);
|
|
let o = offer(&[Codec::Avc, Codec::Opus], Direction::SendOnly, true, true).await;
|
|
assert_eq!(va.send("POST", &whip, &[("content-type", "application/sdp")], o.offer.as_bytes()).await.status, 403);
|
|
|
|
// A publisher's session, while it runs.
|
|
let a = w.auth(&w.publisher, 0);
|
|
let p = publish(&w, &a).await;
|
|
let (text, f) = scrape(&w).await;
|
|
assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whip")]), Some(1.0));
|
|
assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whep")]), Some(0.0));
|
|
assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whip")]), Some(1.0));
|
|
assert_eq!((requests(&f, "/whip", "POST", "201"), requests(&f, "/whip", "POST", "403")), (Some(1.0), Some(1.0)));
|
|
assert_eq!((requests(&f, "/metrics", "GET", "200"), requests(&f, "/metrics", "POST", "405")), (Some(2.0), Some(1.0)));
|
|
assert_eq!(requests(&f, "/metrics", "HEAD", "200"), Some(1.0));
|
|
assert_eq!(value(&f, "wai_ingest_refused_total", &[]), Some(errors(&f)), "every refusal is a request answered with an error");
|
|
let mut texts = vec![text];
|
|
|
|
// A player's session, once the publisher's catalog is there: the gateway's fan-out.
|
|
let mut sent = Sent::default();
|
|
send(&p, &mut sent, 1, 15).await;
|
|
let deadline = Instant::now() + STEP;
|
|
let (o, resp) = loop {
|
|
let o = offer(&[Codec::Avc, Codec::Opus], Direction::RecvOnly, true, true).await;
|
|
let r = va.send("POST", &w.whep_url(), &[("content-type", "application/sdp")], o.offer.as_bytes()).await;
|
|
if r.status == 201 {
|
|
break (o, r);
|
|
}
|
|
assert_eq!(r.status, 409, "{}", r.text());
|
|
assert!(Instant::now() < deadline, "the catalog arrives");
|
|
tokio::time::sleep(Duration::from_millis(200)).await;
|
|
};
|
|
let player = o.start(&resp);
|
|
player.connected().await;
|
|
let (text, f) = scrape(&w).await;
|
|
assert_eq!((running(&f, "whep"), running(&f, "whip")), (Some(1.0), Some(1.0)));
|
|
assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whep")]), Some(1.0));
|
|
assert_eq!(requests(&f, "/whep/{room}/{owner}", "POST", "201"), Some(1.0));
|
|
assert_eq!(value(&f, "wai_ingest_refused_total", &[]), Some(errors(&f)), "a 409 while the catalog is awaited is a refusal too");
|
|
texts.push(text);
|
|
|
|
// The player leaves: the fan-out falls, what was opened stays counted, and its record is signed.
|
|
assert_eq!(delete(&w, &player.location, &va).await.status, 200);
|
|
assert!(until(STEP, || w.gateway.gateway.records().len() == 1).await, "the player's record");
|
|
assert!(until(STEP, || w.gateway.gateway.whep.lock().unwrap().is_empty()).await, "the player's session is gone");
|
|
let (text, f) = scrape(&w).await;
|
|
assert_eq!((running(&f, "whep"), running(&f, "whip")), (Some(0.0), Some(1.0)));
|
|
assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whep")]), Some(1.0));
|
|
assert_eq!(requests(&f, "/whep/s/{id}", "DELETE", "200"), Some(1.0));
|
|
texts.push(text);
|
|
|
|
// The publisher's session ends: its record is signed.
|
|
assert_eq!(delete(&w, &p.location, &a).await.status, 200);
|
|
assert!(until(STEP, || w.gateway.gateway.records().len() == 2).await, "the session's record");
|
|
assert!(until(STEP, || w.gateway.gateway.whip.lock().unwrap().is_empty()).await, "the session is gone");
|
|
let r = http("GET", &format!("{}/whep/s/{}", w.base(), "ab".repeat(16)), &[], b"").await;
|
|
assert_eq!(r.status, 404);
|
|
assert!(r.header("access-control-allow-origin").is_some(), "a page reads the gateway's other answers");
|
|
let (text, f) = scrape(&w).await;
|
|
assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whip")]), Some(0.0));
|
|
assert_eq!(value(&f, "wai_ingest_records_total", &[]), Some(2.0));
|
|
assert_eq!(requests(&f, "/whip/s/{id}", "DELETE", "200"), Some(1.0));
|
|
assert_eq!(requests(&f, "/whep/s/{id}", "GET", "404"), Some(1.0), "a session's resource is counted by its pattern");
|
|
assert_eq!(value(&f, "wai_ingest_refused_total", &[]), Some(errors(&f)));
|
|
assert!(sum(&f, "wai_ingest_requests_total", &[]) >= 8.0);
|
|
texts.push(text);
|
|
|
|
// No party and no thing of one: not a label, and nowhere in the text.
|
|
let session = p.location.rsplit('/').next().unwrap().to_string();
|
|
let played = player.location.rsplit('/').next().unwrap().to_string();
|
|
let ids = [
|
|
hex::encode(w.room),
|
|
w.owner(),
|
|
hex::encode(viewer.verifying_key().to_bytes()),
|
|
hex::encode(w.gateway_key.verifying_key().to_bytes()),
|
|
hex::encode(w.authority.verifying_key().to_bytes()),
|
|
a.token(),
|
|
va.token(),
|
|
session,
|
|
played,
|
|
"ab".repeat(16),
|
|
"127.0.0.1".to_string(),
|
|
];
|
|
for (text, f) in texts.iter().map(|t| (t, parse(t).unwrap())) {
|
|
for l in label_values(&f) {
|
|
assert!(!identifier_like(l), "a label value {l:?}");
|
|
}
|
|
for id in &ids {
|
|
assert!(!text.contains(id.as_str()), "{id} is in the metrics");
|
|
}
|
|
}
|
|
p.close();
|
|
player.close();
|
|
w.stop().await;
|
|
}
|
|
|
|
/// The gateway holds an RTMP connection from its first byte. One that says nothing, and one that
|
|
/// connects and presents no stream key, are no session; a publish is one once it is admitted,
|
|
/// beside them, and not after it ends.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_rtmp_connection_is_a_session_once_its_publish_is_admitted() {
|
|
timeout(Duration::from_secs(120), async {
|
|
let w = World::start(|c| {
|
|
c.rtmp.listen = Some("127.0.0.1:0".parse().unwrap());
|
|
// The idle connections below outlast the test on a loaded host.
|
|
c.rtmp.connect_timeout_ms = 100_000;
|
|
})
|
|
.await;
|
|
let addr = w.gateway.rtmp_addr.expect("an RTMP listener");
|
|
let connections = || w.gateway.gateway.rtmp.lock().unwrap().len();
|
|
|
|
let silent = TcpStream::connect(addr).await.unwrap();
|
|
let mut idle = Client::start(TcpStream::connect(addr).await.unwrap()).await;
|
|
idle.connect(0).await;
|
|
assert!(until(STEP, || connections() == 2).await, "both connections are held");
|
|
let (_, f) = scrape(&w).await;
|
|
assert_eq!(running(&f, "rtmp"), Some(0.0), "neither presented a stream key");
|
|
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let key = stream_key(&w, &auth).await;
|
|
let mut c = Client::start(TcpStream::connect(addr).await.unwrap()).await;
|
|
c.connect(0).await;
|
|
let s = c.publish(&key).await;
|
|
assert_eq!(s.code, "NetStream.Publish.Start", "{}", s.description);
|
|
assert_eq!(connections(), 3);
|
|
let (_, f) = scrape(&w).await;
|
|
assert_eq!(running(&f, "rtmp"), Some(1.0), "the admitted publish, and neither idle connection");
|
|
|
|
c.delete_stream().await;
|
|
c.shutdown().await;
|
|
assert!(until(STEP, || connections() == 2).await, "the publish's connection is gone");
|
|
let (_, f) = scrape(&w).await;
|
|
assert_eq!(running(&f, "rtmp"), Some(0.0));
|
|
|
|
drop((silent, idle));
|
|
w.stop().await;
|
|
})
|
|
.await
|
|
.expect("the test finishes in time");
|
|
}
|
|
|
|
/// An SRT session counts from its call being admitted until it ends, and not while it waits.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_contribution_session_counts_while_it_runs() {
|
|
timeout(Duration::from_secs(120), async {
|
|
let w = World::start(|c| {
|
|
c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
|
|
c.contrib.srt.latency_ms = 250;
|
|
})
|
|
.await;
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], b"").await;
|
|
assert_eq!(r.status, 201, "POST /srt: {}", r.text());
|
|
let s: serde_json::Value = serde_json::from_slice(&r.body).unwrap();
|
|
let (_, f) = scrape(&w).await;
|
|
assert_eq!(running(&f, "contribution"), Some(0.0), "opened, and not called yet");
|
|
|
|
let call = CallConfig {
|
|
stream_id: s["stream_id"].as_str().map(str::to_string),
|
|
passphrase: s["passphrase"].as_str().map(str::to_string),
|
|
keylen: s["pbkeylen"].as_u64().unwrap() as usize,
|
|
latency_ms: 300,
|
|
..CallConfig::default()
|
|
};
|
|
let link = timeout(STEP, srt::call(w.gateway.srt_addr.unwrap(), call)).await.unwrap().expect("the call is admitted");
|
|
assert!(contrib_live(&w, s["session"].as_str().unwrap()).await, "the session publishes its tracks");
|
|
let (_, f) = scrape(&w).await;
|
|
assert_eq!(running(&f, "contribution"), Some(1.0));
|
|
|
|
link.close();
|
|
assert!(until(STEP, || w.gateway.gateway.contrib.sessions.lock().unwrap().is_empty()).await, "the session is gone");
|
|
let (_, f) = scrape(&w).await;
|
|
assert_eq!(running(&f, "contribution"), Some(0.0));
|
|
w.stop().await;
|
|
})
|
|
.await
|
|
.expect("the test finishes in time");
|
|
}
|
|
|
|
/// A path of each resource `Gateway::answer` matches, read from its table: each slice pattern
|
|
/// there, every alternative of a segment taken, any other segment a sample one. A resource added
|
|
/// to the table is asked with no test changed.
|
|
fn resources() -> Vec<String> {
|
|
let source = include_str!("../src/gateway.rs");
|
|
let table = source.split_once("let r = match (seg.as_slice(), &m) {").expect("the table").1;
|
|
let table = table.split_once("_ => problem(StatusCode::NOT_FOUND").expect("its end").0;
|
|
let mut paths = BTreeSet::new();
|
|
for arm in table.split("([").skip(1) {
|
|
let pattern = arm.split_once(']').expect("a slice pattern").0;
|
|
let mut found = vec![String::new()];
|
|
for segment in pattern.split(", ") {
|
|
let literals: Vec<&str> = segment.split('"').skip(1).step_by(2).collect();
|
|
let choices = if literals.is_empty() { vec!["0"] } else { literals };
|
|
found = found.iter().flat_map(|p| choices.iter().map(move |c| format!("{p}/{c}"))).collect();
|
|
}
|
|
paths.extend(found);
|
|
}
|
|
paths.into_iter().collect()
|
|
}
|
|
|
|
/// Every resource of the gateway's table, asked once, is counted by its pattern: one left out of
|
|
/// `route_of` would be counted as `other`.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn every_resource_of_the_table_is_counted_by_its_pattern() {
|
|
timeout(Duration::from_secs(120), async {
|
|
let w = World::start(|_| {}).await;
|
|
let paths = resources();
|
|
assert!(paths.len() >= 10, "{paths:?}");
|
|
for p in &paths {
|
|
http("GET", &format!("{}{p}", w.base()), &[], b"").await;
|
|
}
|
|
let (text, f) = scrape(&w).await;
|
|
assert_eq!(sum(&f, "wai_ingest_requests_total", &[("route", "other")]), 0.0, "{text}");
|
|
assert!(sum(&f, "wai_ingest_requests_total", &[]) >= paths.len() as f64, "{text}");
|
|
w.stop().await;
|
|
})
|
|
.await
|
|
.expect("the test finishes in time");
|
|
}
|