diff --git a/readme.md b/readme.md index 6f1e0bf..9330e52 100644 --- a/readme.md +++ b/readme.md @@ -36,6 +36,19 @@ Allegedly can --experimental-write-upstream ``` +- Reverse-proxy to any PLC server, terminating TLS and forwarding writes upstream + + ```bash + sudo allegedly wrap \ + --wrap "http://127.0.0.1:3000" \ + --acme-ipv6 \ + --acme-cache-path ./acme-cache \ + --acme-domain "plc.wtf" \ + --experimental-acme-domain "experimental.plc.wtf" \ + --experimental-write-upstream \ + --upstream "https://plc.wtf" \ + ``` + add `--help` to any command for more info about it @@ -66,7 +79,7 @@ to install the latest from source: - monitoring of the various tasks - health check pings - expose metrics/tracing -- read-only flag for mirror wrapper +- [x] read-only flag for mirror wrapper - bundle: write directly to s3-compatible object storage - helpers for automating periodic `bundle` runs diff --git a/src/bin/allegedly.rs b/src/bin/allegedly.rs index a51148b..1c9eed7 100644 --- a/src/bin/allegedly.rs +++ b/src/bin/allegedly.rs @@ -49,6 +49,11 @@ enum Commands { #[command(flatten)] args: mirror::Args, }, + /// Wrap any did-method-plc server, without syncing upstream (read-only) + Wrap { + #[command(flatten)] + args: mirror::Args, + }, /// Poll an upstream PLC server and log new ops to stdout Tail { /// Begin tailing from a specific timestamp for replay or wait-until @@ -91,7 +96,8 @@ async fn main() -> anyhow::Result<()> { .await .expect("to write bundles to output files"); } - Commands::Mirror { args } => mirror::run(globals, args).await?, + Commands::Mirror { args } => mirror::run(globals, args, true).await?, + Commands::Wrap { args } => mirror::run(globals, args, false).await?, Commands::Tail { after } => { let mut url = globals.upstream; url.set_path("/export"); diff --git a/src/bin/mirror.rs b/src/bin/mirror.rs index c612efa..adb5f59 100644 --- a/src/bin/mirror.rs +++ b/src/bin/mirror.rs @@ -13,7 +13,7 @@ pub struct Args { wrap: Url, /// the wrapped did-method-plc server's database (write access required) #[arg(long, env = "ALLEGEDLY_WRAP_PG")] - wrap_pg: Url, + wrap_pg: Option, /// path to tls cert for the wrapped postgres db, if needed #[arg(long, env = "ALLEGEDLY_WRAP_PG_CERT")] wrap_pg_cert: Option, @@ -76,16 +76,8 @@ pub async fn run( experimental_acme_domain, experimental_write_upstream, }: Args, + sync: bool, ) -> anyhow::Result<()> { - let db = Db::new(wrap_pg.as_str(), wrap_pg_cert).await?; - - // TODO: allow starting up with polling backfill from beginning? - log::debug!("getting the latest op from the db..."); - let latest = db - .get_latest() - .await? - .expect("there to be at least one op in the db. did you backfill?"); - let listen_conf = match (bind, acme_domain.is_empty(), acme_cache_path) { (_, false, Some(cache_path)) => { create_dir_all(&cache_path).await?; @@ -112,14 +104,32 @@ pub async fn run( let mut tasks = JoinSet::new(); - let (send_page, recv_page) = mpsc::channel(8); + let db = if sync { + let wrap_pg = wrap_pg.ok_or(anyhow::anyhow!( + "a wrapped reference postgres must be provided to sync" + ))?; + let db = Db::new(wrap_pg.as_str(), wrap_pg_cert).await?; + + // TODO: allow starting up with polling backfill from beginning? + log::debug!("getting the latest op from the db..."); + let latest = db + .get_latest() + .await? + .expect("there to be at least one op in the db. did you backfill?"); - let mut poll_url = upstream.clone(); - poll_url.set_path("/export"); - let throttle = Duration::from_millis(upstream_throttle_ms); + let (send_page, recv_page) = mpsc::channel(8); + + let mut poll_url = upstream.clone(); + poll_url.set_path("/export"); + let throttle = Duration::from_millis(upstream_throttle_ms); + + tasks.spawn(poll_upstream(Some(latest), poll_url, throttle, send_page)); + tasks.spawn(pages_to_pg(db.clone(), recv_page)); + Some(db) + } else { + None + }; - tasks.spawn(poll_upstream(Some(latest), poll_url, throttle, send_page)); - tasks.spawn(pages_to_pg(db.clone(), recv_page)); tasks.spawn(serve( upstream, wrap, @@ -157,6 +167,9 @@ struct CliArgs { globals: GlobalArgs, #[command(flatten)] args: Args, + /// Run the mirror in wrap mode, no upstream synchronization (read-only) + #[arg(long, action)] + wrap_mode: bool, } #[allow(dead_code)] @@ -164,6 +177,6 @@ struct CliArgs { async fn main() -> anyhow::Result<()> { let args = CliArgs::parse(); bin_init("mirror"); - run(args.globals, args.args).await?; + run(args.globals, args.args, !args.wrap_mode).await?; Ok(()) } diff --git a/src/mirror.rs b/src/mirror.rs index 4dff7ea..1d23a7b 100644 --- a/src/mirror.rs +++ b/src/mirror.rs @@ -19,20 +19,56 @@ struct State { client: Client, plc: Url, upstream: Url, + sync_info: Option, + experimental: ExperimentalConf, +} + +/// server info that only applies in mirror (synchronizing) mode +#[derive(Clone)] +struct SyncInfo { latest_at: CachedValue, upstream_status: CachedValue, - experimental: ExperimentalConf, } #[handler] fn hello( Data(State { + sync_info, upstream, experimental: exp, .. }): Data<&State>, req: &Request, ) -> String { + // let mode = if sync_info.is_some() { "mirror" } else { "wrap" }; + let pre_info = if sync_info.is_some() { + format!( + r#" +This is a PLC[1] mirror running Allegedly in mirror mode. Mirror mode wraps and +synchronizes a local PLC reference server instance[2] (why?[3]). + + +Configured upstream: + + {upstream} + +"# + ) + } else { + format!( + r#" +This is a PLC[1] mirror running Allegedly in wrap mode. Wrap mode reverse- +proxies requests to a PLC server and can terminate TLS, like NGINX or Caddy. + + +Configured upstream (only used if experimental op forwarding is enabled): + + {upstream} + +"# + ) + }; + let post_info = match (exp.write_upstream, &exp.acme_domain, req.uri().host()) { (false, _, _) => " - POST /* Always rejected. This is a mirror.".to_string(), (_, None, _) => { @@ -49,15 +85,7 @@ fn hello( format!( r#"{} - -This is a PLC[1] mirror running Allegedly in mirror mode. Mirror mode wraps and -synchronizes a local PLC reference server instance[2] (why?[3]). - - -Configured upstream: - - {upstream} - +{pre_info} Available APIs: @@ -176,8 +204,7 @@ async fn health( Data(State { plc, client, - latest_at, - upstream_status, + sync_info, .. }): Data<&State>, ) -> impl IntoResponse { @@ -186,21 +213,38 @@ async fn health( if !ok { overall_status = StatusCode::BAD_GATEWAY; } - let (ok, upstream_status) = upstream_status.get().await.expect("plc_status infallible"); - if !ok { - overall_status = StatusCode::BAD_GATEWAY; + if let Some(SyncInfo { + latest_at, + upstream_status, + }) = sync_info + { + // mirror mode + let (ok, upstream_status) = upstream_status.get().await.expect("plc_status infallible"); + if !ok { + overall_status = StatusCode::BAD_GATEWAY; + } + let latest = latest_at.get().await.ok(); + ( + overall_status, + Json(serde_json::json!({ + "server": "allegedly (mirror)", + "version": env!("CARGO_PKG_VERSION"), + "wrapped_plc": wrapped_status, + "upstream_plc": upstream_status, + "latest_at": latest, + })), + ) + } else { + // wrap mode + ( + overall_status, + Json(serde_json::json!({ + "server": "allegedly (mirror)", + "version": env!("CARGO_PKG_VERSION"), + "wrapped_plc": wrapped_status, + })), + ) } - let latest = latest_at.get().await.ok(); - ( - overall_status, - Json(serde_json::json!({ - "server": "allegedly (mirror)", - "version": env!("CARGO_PKG_VERSION"), - "wrapped_plc": wrapped_status, - "upstream_plc": upstream_status, - "latest_at": latest, - })), - ) } fn proxy_response(res: reqwest::Response) -> Response { @@ -344,7 +388,7 @@ pub async fn serve( plc: Url, listen: ListenConf, experimental: ExperimentalConf, - db: Db, + db: Option, ) -> anyhow::Result<&'static str> { log::info!("starting server..."); @@ -355,18 +399,20 @@ pub async fn serve( .build() .expect("reqwest client to build"); - let latest_at = CachedValue::new(GetLatestAt(db), Duration::from_secs(2)); - let upstream_status = CachedValue::new( - CheckUpstream(upstream.clone(), client.clone()), - Duration::from_secs(6), - ); + // when `db` is None, we're running in wrap mode. no db access, no upstream sync + let sync_info = db.map(|db| SyncInfo { + latest_at: CachedValue::new(GetLatestAt(db), Duration::from_secs(2)), + upstream_status: CachedValue::new( + CheckUpstream(upstream.clone(), client.clone()), + Duration::from_secs(6), + ), + }); let state = State { client, plc, upstream: upstream.clone(), - latest_at, - upstream_status, + sync_info, experimental: experimental.clone(), };