|
| 1 | +// Copyright Materialize, Inc. and contributors. All rights reserved. |
| 2 | +// |
| 3 | +// Use of this software is governed by the Business Source License |
| 4 | +// included in the LICENSE file. |
| 5 | +// |
| 6 | +// As of the Change Date specified in that file, in accordance with |
| 7 | +// the Business Source License, use of this software will be governed |
| 8 | +// by the Apache License, Version 2.0. |
| 9 | + |
| 10 | +//! Headless driver entry point for `mzcompose`. Connects to a running `clusterd`, |
| 11 | +//! hosts persist PubSub, and runs a JSON command script against it (see |
| 12 | +//! [`mz_clusterd_test_driver::script`]), exiting non-zero on assertion failure. |
| 13 | +//! |
| 14 | +//! The script source is the file named by `DRIVER_SCRIPT`, or stdin when that is |
| 15 | +//! unset. Connection and persist configuration come from the environment: |
| 16 | +//! `CLUSTERD_COMPUTE_ADDR`, `PERSIST_BLOB_URL`, `PERSIST_CONSENSUS_URL`, and |
| 17 | +//! `DRIVER_PUBSUB_BIND`. |
| 18 | +
|
| 19 | +use std::net::SocketAddr; |
| 20 | + |
| 21 | +use anyhow::Context; |
| 22 | +use mz_clusterd_test_driver::driver::Driver; |
| 23 | +use mz_clusterd_test_driver::persist_host::PersistHost; |
| 24 | +use mz_clusterd_test_driver::script; |
| 25 | +use mz_orchestrator_tracing::{StaticTracingConfig, TracingCliArgs}; |
| 26 | +use mz_ore::metrics::MetricsRegistry; |
| 27 | +use mz_persist_types::PersistLocation; |
| 28 | +use tokio::io::AsyncReadExt; |
| 29 | + |
| 30 | +/// Connect to `clusterd` and host persist PubSub, reading configuration from the |
| 31 | +/// environment. Returns the persist location (for dataflow imports) and a |
| 32 | +/// connected [`Driver`]. |
| 33 | +async fn setup() -> anyhow::Result<(PersistLocation, Driver)> { |
| 34 | + let compute_addr = |
| 35 | + std::env::var("CLUSTERD_COMPUTE_ADDR").unwrap_or_else(|_| "clusterd:2101".to_string()); |
| 36 | + let blob = std::env::var("PERSIST_BLOB_URL").expect("PERSIST_BLOB_URL"); |
| 37 | + let consensus = std::env::var("PERSIST_CONSENSUS_URL").expect("PERSIST_CONSENSUS_URL"); |
| 38 | + let pubsub_bind: SocketAddr = std::env::var("DRIVER_PUBSUB_BIND") |
| 39 | + .unwrap_or_else(|_| "0.0.0.0:6879".to_string()) |
| 40 | + .parse()?; |
| 41 | + |
| 42 | + let loc = PersistLocation { |
| 43 | + blob_uri: blob.parse()?, |
| 44 | + consensus_uri: consensus.parse()?, |
| 45 | + }; |
| 46 | + let host = PersistHost::start_on(pubsub_bind, loc.clone()).await?; |
| 47 | + let driver = Driver::connect(host, &compute_addr).await?; |
| 48 | + Ok((loc, driver)) |
| 49 | +} |
| 50 | + |
| 51 | +#[tokio::main] |
| 52 | +async fn main() -> anyhow::Result<()> { |
| 53 | + // Configure tracing so the driver emits structured logs like the real |
| 54 | + // Materialize binaries. |
| 55 | + let _tracing_handle = TracingCliArgs::default() |
| 56 | + .configure_tracing( |
| 57 | + StaticTracingConfig { |
| 58 | + service_name: "headless-driver", |
| 59 | + build_info: mz_persist_client::BUILD_INFO, |
| 60 | + }, |
| 61 | + MetricsRegistry::new(), |
| 62 | + ) |
| 63 | + .await?; |
| 64 | + |
| 65 | + let (loc, driver) = setup().await?; |
| 66 | + |
| 67 | + // Read the script from `DRIVER_SCRIPT` if set, else stdin. The path is passed |
| 68 | + // through so a `REWRITE` run can rewrite the file in place. |
| 69 | + match std::env::var("DRIVER_SCRIPT") { |
| 70 | + Ok(path) => { |
| 71 | + let content = tokio::fs::read_to_string(&path) |
| 72 | + .await |
| 73 | + .with_context(|| format!("reading DRIVER_SCRIPT {path}"))?; |
| 74 | + script::run(driver, loc, &content, Some(std::path::Path::new(&path))).await |
| 75 | + } |
| 76 | + Err(_) => { |
| 77 | + let mut content = String::new(); |
| 78 | + tokio::io::stdin().read_to_string(&mut content).await?; |
| 79 | + script::run(driver, loc, &content, None).await |
| 80 | + } |
| 81 | + } |
| 82 | +} |
0 commit comments