mod support;
use actix_web::{http::StatusCode, web, App, HttpRequest, HttpResponse, HttpServer};
use futures::{stream, StreamExt};
use later::backend::Backend;
use serde::Deserialize;
use serde_json::json;
use std::{sync::Arc, time::Instant};
use support::{
start_server_with_workers, NoopJob, PartitionJob, PartitionLogEntry, PrintJob, Server,
PARTITION_COUNT,
};
use tokio::sync::Mutex;
const MAX_BULK_JOBS: usize = 10_000;
const SQLITE_BULK_CONCURRENCY: usize = 4;
const DEFAULT_BULK_CONCURRENCY: usize = 12;
const DEMO_PAGE: &str = include_str!("dashboard_demo.html");
#[derive(Clone, Copy)]
enum StateBackend {
Sqlite,
Postgres,
}
impl StateBackend {
fn name(self) -> &'static str {
match self {
Self::Sqlite => "sqlite",
Self::Postgres => "postgres",
}
}
}
struct Options {
backend: StateBackend,
namespace: String,
workers: u8,
help: bool,
}
impl Options {
fn parse() -> std::io::Result<Self> {
let mut options = Self {
backend: default_backend(),
namespace: "dashboard-example".to_string(),
workers: 6,
help: false,
};
let mut arguments = std::env::args().skip(1);
while let Some(argument) = arguments.next() {
match argument.as_str() {
"--backend" => {
let value = arguments
.next()
.ok_or_else(|| std::io::Error::other("--backend needs a value"))?;
options.backend = match value.as_str() {
"sqlite" => StateBackend::Sqlite,
"postgres" => StateBackend::Postgres,
_ => {
return Err(std::io::Error::other("backend must be sqlite or postgres"))
}
};
}
"--namespace" => {
options.namespace = arguments
.next()
.ok_or_else(|| std::io::Error::other("--namespace needs a value"))?;
}
"--workers" => {
let value = arguments
.next()
.ok_or_else(|| std::io::Error::other("--workers needs a value"))?;
options.workers = value.parse().map_err(|_| {
std::io::Error::other("workers must be a number from 1 to 255")
})?;
if options.workers == 0 {
return Err(std::io::Error::other("workers must be at least 1"));
}
}
"--help" | "-h" => options.help = true,
_ => {
return Err(std::io::Error::other(format!(
"unknown dashboard option: {argument}"
)))
}
}
}
Ok(options)
}
}
#[cfg(feature = "sqlite")]
fn default_backend() -> StateBackend {
StateBackend::Sqlite
}
#[cfg(all(not(feature = "sqlite"), feature = "postgres"))]
fn default_backend() -> StateBackend {
StateBackend::Postgres
}
#[cfg(not(any(feature = "sqlite", feature = "postgres")))]
fn default_backend() -> StateBackend {
StateBackend::Sqlite
}
fn fallible_job(name: String, failures_before_success: usize) -> PrintJob {
PrintJob {
message: format!("{name}-{}", later::generate_id()),
failures_before_success,
}
}
async fn demo_page() -> HttpResponse {
let page = DEMO_PAGE.replace("__PARTITION_COUNT__", &PARTITION_COUNT.to_string());
HttpResponse::Ok()
.content_type("text/html; charset=utf-8")
.body(page)
}
async fn enqueue_single(server: web::Data<Arc<Server>>) -> HttpResponse {
match server.enqueue(NoopJob).await {
Ok(id) => HttpResponse::Ok().json(json!({ "job_ids": [id] })),
Err(error) => {
HttpResponse::InternalServerError().json(json!({ "error": error.to_string() }))
}
}
}
async fn enqueue_continuation(server: web::Data<Arc<Server>>) -> HttpResponse {
let result = async {
let first = server.enqueue(NoopJob).await?;
let second = server.enqueue_continue(first.clone(), NoopJob).await?;
let third = server.enqueue_continue(second.clone(), NoopJob).await?;
Ok::<_, anyhow::Error>(vec![first, second, third])
}
.await;
match result {
Ok(ids) => HttpResponse::Ok().json(json!({ "job_ids": ids })),
Err(error) => {
HttpResponse::InternalServerError().json(json!({ "error": error.to_string() }))
}
}
}
async fn enqueue_retry(server: web::Data<Arc<Server>>) -> HttpResponse {
match server.enqueue(fallible_job("retry".to_string(), 1)).await {
Ok(id) => HttpResponse::Ok().json(json!({ "job_ids": [id] })),
Err(error) => {
HttpResponse::InternalServerError().json(json!({ "error": error.to_string() }))
}
}
}
#[derive(Deserialize)]
struct BulkQuery {
count: usize,
#[serde(default)]
failures_before_success: usize,
}
async fn enqueue_bulk(
server: web::Data<Arc<Server>>,
bulk_concurrency: web::Data<usize>,
query: web::Query<BulkQuery>,
) -> HttpResponse {
if !(1..=MAX_BULK_JOBS).contains(&query.count) {
return HttpResponse::BadRequest().json(json!({
"error": format!("count must be from 1 to {MAX_BULK_JOBS}")
}));
}
let started = Instant::now();
let count = query.count;
let failures_before_success = query.failures_before_success;
let results = stream::iter(0..count)
.map(|index| {
let server = server.clone();
async move {
if failures_before_success == 0 {
server.enqueue(NoopJob).await
} else {
server
.enqueue(fallible_job(
format!("bulk-{index}"),
failures_before_success,
))
.await
}
}
})
.buffer_unordered(**bulk_concurrency)
.collect::<Vec<_>>()
.await;
let elapsed = started.elapsed();
let succeeded = results.iter().filter(|result| result.is_ok()).count();
let errors = results
.iter()
.filter_map(|result| result.as_ref().err())
.take(5)
.map(ToString::to_string)
.collect::<Vec<_>>();
let sample_job_ids = results
.iter()
.filter_map(|result| result.as_ref().ok())
.take(10)
.collect::<Vec<_>>();
let enqueue_per_second = succeeded as f64 / elapsed.as_secs_f64().max(f64::EPSILON);
let body = json!({
"requested": count,
"enqueued": succeeded,
"failed": count - succeeded,
"elapsed_ms": elapsed.as_millis(),
"enqueue_per_second": enqueue_per_second,
"enqueue_concurrency": **bulk_concurrency,
"sample_job_ids": sample_job_ids,
"errors": errors,
});
if errors.is_empty() {
HttpResponse::Ok().json(body)
} else {
HttpResponse::InternalServerError().json(body)
}
}
async fn worker_count(server: web::Data<Arc<Server>>) -> HttpResponse {
HttpResponse::Ok().json(json!({ "workers": server.worker_count() }))
}
#[derive(Deserialize)]
struct EnqueuePartitionQuery {
key: String,
count: usize,
#[serde(default)]
delay_ms: u32,
}
const MAX_PARTITION_JOBS: usize = 500;
const MAX_PARTITION_KEY_LEN: usize = 200;
async fn enqueue_partition(
server: web::Data<Arc<Server>>,
query: web::Query<EnqueuePartitionQuery>,
) -> HttpResponse {
if query.key.is_empty() || query.key.len() > MAX_PARTITION_KEY_LEN {
return HttpResponse::BadRequest().json(json!({
"error": format!("key must be from 1 to {MAX_PARTITION_KEY_LEN} characters")
}));
}
if !(1..=MAX_PARTITION_JOBS).contains(&query.count) {
return HttpResponse::BadRequest().json(json!({
"error": format!("count must be from 1 to {MAX_PARTITION_JOBS}")
}));
}
let resolved_partition = later::topic::partition_for_key(
&later::topic::PartitionKey::from(query.key.clone()),
PARTITION_COUNT,
)
.0;
let mut job_ids = Vec::with_capacity(query.count);
for seq in 0..query.count as u32 {
match server
.enqueue_to_partition(PartitionJob {
key: query.key.clone(),
seq,
delay_ms: query.delay_ms,
})
.await
{
Ok(id) => job_ids.push(id),
Err(error) => {
return HttpResponse::InternalServerError().json(json!({
"error": error.to_string(),
"enqueued_before_error": job_ids.len(),
}))
}
}
}
HttpResponse::Ok().json(json!({
"enqueued": job_ids.len(),
"job_ids": job_ids,
"resolved_partition": resolved_partition,
}))
}
fn representative_key_per_partition() -> Vec<String> {
let mut keys: Vec<Option<String>> = vec![None; PARTITION_COUNT as usize];
let mut candidate: u64 = 0;
while keys.iter().any(Option::is_none) {
let key = format!("stress-{candidate}");
let partition = later::topic::partition_for_key(
&later::topic::PartitionKey::from(key.clone()),
PARTITION_COUNT,
)
.0 as usize;
if keys[partition].is_none() {
keys[partition] = Some(key);
}
candidate += 1;
}
keys.into_iter()
.map(|key| key.expect("every partition got a candidate before the loop above exits"))
.collect()
}
#[derive(Deserialize)]
struct EnqueuePartitionStressQuery {
jobs_per_partition: usize,
#[serde(default)]
delay_ms: u32,
}
const MAX_STRESS_JOBS_PER_PARTITION: usize = 200;
async fn enqueue_partition_stress(
server: web::Data<Arc<Server>>,
query: web::Query<EnqueuePartitionStressQuery>,
) -> HttpResponse {
if !(1..=MAX_STRESS_JOBS_PER_PARTITION).contains(&query.jobs_per_partition) {
return HttpResponse::BadRequest().json(json!({
"error": format!("jobs_per_partition must be from 1 to {MAX_STRESS_JOBS_PER_PARTITION}")
}));
}
let keys = representative_key_per_partition();
let mut job_ids = Vec::with_capacity(keys.len() * query.jobs_per_partition);
for key in &keys {
for seq in 0..query.jobs_per_partition as u32 {
match server
.enqueue_to_partition(PartitionJob {
key: key.clone(),
seq,
delay_ms: query.delay_ms,
})
.await
{
Ok(id) => job_ids.push(id),
Err(error) => {
return HttpResponse::InternalServerError().json(json!({
"error": error.to_string(),
"enqueued_before_error": job_ids.len(),
}))
}
}
}
}
HttpResponse::Ok().json(json!({
"enqueued": job_ids.len(),
"partitions": keys.len(),
"jobs_per_partition": query.jobs_per_partition,
}))
}
async fn get_partition_log(log: web::Data<Arc<Mutex<Vec<PartitionLogEntry>>>>) -> HttpResponse {
let entries = log.lock().await.clone();
HttpResponse::Ok().json(json!({ "entries": entries }))
}
async fn clear_partition_log(log: web::Data<Arc<Mutex<Vec<PartitionLogEntry>>>>) -> HttpResponse {
log.lock().await.clear();
HttpResponse::Ok().json(json!({ "cleared": true }))
}
async fn add_worker(server: web::Data<Arc<Server>>) -> HttpResponse {
match server.add_worker().await {
Ok(workers) => HttpResponse::Ok().json(json!({ "workers": workers })),
Err(error) => {
HttpResponse::InternalServerError().json(json!({ "error": error.to_string() }))
}
}
}
async fn remove_worker(server: web::Data<Arc<Server>>) -> HttpResponse {
match server.remove_worker().await {
Ok(workers) => HttpResponse::Ok().json(json!({ "workers": workers })),
Err(error) => HttpResponse::BadRequest().json(json!({ "error": error.to_string() })),
}
}
async fn dashboard(request: HttpRequest, server: web::Data<Arc<Server>>) -> HttpResponse {
match server
.get_dashboard("/dashboard".to_string(), request.query_string().to_string())
.await
{
Ok(response) => {
let status = StatusCode::from_u16(response.status_code)
.ok()
.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
let mut builder = HttpResponse::build(status);
for (name, value) in response.headers {
builder.append_header((name, value));
}
builder.body(response.body)
}
Err(error) => HttpResponse::InternalServerError().body(error.to_string()),
}
}
#[cfg(feature = "prometheus")]
async fn prometheus_metrics(server: web::Data<Arc<Server>>) -> HttpResponse {
match server.get_metrics() {
Ok(metrics) => HttpResponse::Ok()
.content_type("text/plain; version=0.0.4; charset=utf-8")
.body(metrics),
Err(error) => HttpResponse::InternalServerError().body(error.to_string()),
}
}
#[cfg(feature = "prometheus")]
fn write_prometheus_target(
address: std::net::SocketAddr,
options: &Options,
) -> std::io::Result<()> {
let Ok(path) = std::env::var("LATER_PROMETHEUS_TARGET_FILE") else {
return Ok(());
};
let host = std::env::var("LATER_PROMETHEUS_TARGET_HOST")
.unwrap_or_else(|_| "host.docker.internal".to_string());
let target = json!([{
"targets": [format!("{host}:{}", address.port())],
"labels": {
"backend": options.backend.name(),
"delivery": "sql",
"namespace": &options.namespace,
}
}]);
let target = serde_json::to_vec_pretty(&target).map_err(std::io::Error::other)?;
std::fs::write(path, target)
}
async fn create_backend(options: &Options) -> std::io::Result<Box<dyn Backend>> {
match options.backend {
StateBackend::Sqlite => create_sqlite_backend(options).await,
StateBackend::Postgres => create_postgres_backend(options).await,
}
}
#[cfg(feature = "sqlite")]
async fn create_sqlite_backend(options: &Options) -> std::io::Result<Box<dyn Backend>> {
let database_url = std::env::var("LATER_SQLITE_URL")
.unwrap_or_else(|_| "sqlite://later-dashboard-example.db".to_string());
let storage = later::storage::Sqlite::new(&database_url)
.await
.map_err(std::io::Error::other)?;
let backend = later::backend::SqliteBackend::new(options.namespace.clone(), storage)
.map_err(std::io::Error::other)?;
Ok(Box::new(backend))
}
#[cfg(not(feature = "sqlite"))]
async fn create_sqlite_backend(_options: &Options) -> std::io::Result<Box<dyn Backend>> {
Err(std::io::Error::other(
"SQLite needs the sqlite Cargo feature",
))
}
#[cfg(feature = "postgres")]
async fn create_postgres_backend(options: &Options) -> std::io::Result<Box<dyn Backend>> {
let database_url = std::env::var("DATABASE_URL").map_err(|_| {
std::io::Error::other("Postgres needs DATABASE_URL; run eval \"$(make env)\" first")
})?;
let storage = later::storage::Postgres::new(&database_url)
.await
.map_err(std::io::Error::other)?;
let backend = later::backend::PostgresBackend::new(options.namespace.clone(), storage)
.map_err(std::io::Error::other)?;
Ok(Box::new(backend))
}
#[cfg(not(feature = "postgres"))]
async fn create_postgres_backend(_options: &Options) -> std::io::Result<Box<dyn Backend>> {
Err(std::io::Error::other(
"Postgres needs the postgres Cargo feature",
))
}
fn open_browser(url: &str) {
let result = if cfg!(target_os = "macos") {
std::process::Command::new("open").arg(url).status()
} else if cfg!(target_os = "windows") {
std::process::Command::new("cmd")
.args(["/C", "start", "", url])
.status()
} else {
std::process::Command::new("xdg-open").arg(url).status()
};
if !matches!(result, Ok(status) if status.success()) {
println!("Could not open a browser automatically; open {url} manually.");
}
}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
let options = Options::parse()?;
if options.help {
println!("Usage: dashboard [--backend sqlite|postgres] [--workers N] [--namespace NAME]");
return Ok(());
}
let backend = create_backend(&options).await?;
let partition_log: Arc<Mutex<Vec<PartitionLogEntry>>> = Arc::new(Mutex::new(Vec::new()));
let server = start_server_with_workers(
"dashboard-server",
backend,
options.workers,
partition_log.clone(),
)
.await
.map_err(std::io::Error::other)?;
let server = Arc::new(server);
let bulk_concurrency = match options.backend {
StateBackend::Sqlite => SQLITE_BULK_CONCURRENCY,
StateBackend::Postgres => DEFAULT_BULK_CONCURRENCY,
};
let bind_host =
std::env::var("LATER_DASHBOARD_BIND").unwrap_or_else(|_| "127.0.0.1".to_string());
let bind_port: u16 = std::env::var("LATER_DASHBOARD_PORT")
.ok()
.and_then(|value| value.parse().ok())
.unwrap_or(0);
let http_server = HttpServer::new(move || {
let app = App::new()
.app_data(web::Data::new(server.clone()))
.app_data(web::Data::new(bulk_concurrency))
.app_data(web::Data::new(partition_log.clone()))
.route("/", web::get().to(demo_page))
.route("/enqueue/single", web::post().to(enqueue_single))
.route(
"/enqueue/continuation",
web::post().to(enqueue_continuation),
)
.route("/enqueue/retry", web::post().to(enqueue_retry))
.route("/enqueue/bulk", web::post().to(enqueue_bulk))
.route("/enqueue/partition", web::post().to(enqueue_partition))
.route(
"/enqueue/partition/stress",
web::post().to(enqueue_partition_stress),
)
.route("/partitions/log", web::get().to(get_partition_log))
.route("/partitions/log/clear", web::post().to(clear_partition_log))
.route("/workers", web::get().to(worker_count))
.route("/workers/add", web::post().to(add_worker))
.route("/workers/remove", web::post().to(remove_worker))
.route("/dashboard", web::get().to(dashboard));
#[cfg(feature = "prometheus")]
let app = app.route("/metrics", web::get().to(prometheus_metrics));
app
})
.bind((bind_host, bind_port))?;
let address = http_server
.addrs()
.first()
.copied()
.ok_or_else(|| std::io::Error::other("dashboard server has no listening address"))?;
#[cfg(feature = "prometheus")]
write_prometheus_target(address, &options)?;
let local_address = format!("127.0.0.1:{}", address.port());
println!(
"Backend: {}, delivery: sql, workers: {}, namespace: {}",
options.backend.name(),
options.workers,
options.namespace,
);
let demo_url = format!("http://{local_address}");
let dashboard_url = format!("http://{local_address}/dashboard");
let grafana_url = std::env::var("LATER_GRAFANA_URL").ok();
if let Some(grafana_url) = &grafana_url {
println!("Grafana: {grafana_url}");
}
println!("Demo: {demo_url}");
println!("Dashboard: {dashboard_url}");
#[cfg(feature = "prometheus")]
println!("Metrics: http://{local_address}/metrics");
if std::env::var("LATER_NO_BROWSER").is_err() {
println!("Opening your browser... switch to it now.");
if let Some(grafana_url) = &grafana_url {
open_browser(grafana_url);
}
open_browser(&dashboard_url);
open_browser(&demo_url);
}
http_server.run().await
}
#[cfg(test)]
mod tests {
use super::DEMO_PAGE;
#[test]
fn demo_page_has_a_stoppable_noop_stress_loop() {
assert!(DEMO_PAGE.contains("id=\"stress-toggle\""));
assert!(DEMO_PAGE.contains("while (!stressStopRequested)"));
assert!(DEMO_PAGE.contains("failures_before_success=${encodeURIComponent(failures)}"));
assert!(DEMO_PAGE.contains("average_enqueue_per_second"));
assert!(DEMO_PAGE.contains("Stopping after current batch"));
assert!(DEMO_PAGE.contains("id=\"add-worker\""));
assert!(DEMO_PAGE.contains("/workers/remove"));
}
#[test]
fn demo_page_lets_the_stress_loop_target_regular_and_sequential_jobs_independently() {
assert!(DEMO_PAGE.contains("id=\"stress-regular\""));
assert!(DEMO_PAGE.contains("id=\"stress-sequential\""));
assert!(DEMO_PAGE.contains("id=\"stress-partition-count\""));
assert!(DEMO_PAGE.contains("stressRegular.checked"));
assert!(DEMO_PAGE.contains("stressSequential.checked"));
assert!(DEMO_PAGE.contains("/enqueue/partition/stress"));
}
}