ore/cli: add env_prefix configuration option

Add an `env_prefix` option to our CLI parsing module. This option will
eventually come to clap itself (clap-rs/clap#3221), but for now we can
roll it ourselves using clap's reflection API.

This option will allow us to factor out common CLI options into their
own structs while still having the environment variable prefixes
determined by the root command.

src/billing-demo/src/bin/billing-demo/main.rs

@@ -27,6 +27,7 @@ use prost::Message;
 use tokio::time::{self, Duration};
 use tracing::{error, info, trace};
 
+use mz_ore::cli::{self, CliConfig};
 use mz_ore::task;
 use mz_test_util::kafka::kafka_client;
 use mz_test_util::mz_client;
@@ -47,7 +48,7 @@ async fn main() {
 }
 
 async fn run() -> Result<()> {
-    let config: Args = mz_ore::cli::parse_args();
+    let config: Args = cli::parse_args(CliConfig::default());
 
     let k_config = config.kafka_config();
     let mz_config = config.mz_config();

src/compute/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "mz-compute"
 description = "Materialize's compute layer."
-version = "0.0.0"
+version = "0.26.1-dev"
 edition = "2021"
 rust-version = "1.61.0"
 publish = false
@@ -30,6 +30,7 @@ mz-prof = { path = "../prof" }
 mz-repr = { path = "../repr" }
 mz-storage = { path = "../storage", default-features = false }
 mz-timely-util = { path = "../timely-util" }
+once_cell = "1.12.0"
 prometheus = { version = "0.13.1", default-features = false }
 rdkafka = { git = "https://github.com/fede1024/rust-rdkafka.git", features = ["cmake-build", "ssl-vendored", "libz-static", "zstd"] }
 scopeguard = "1.1.0"

src/compute/src/bin/computed.rs

@@ -14,17 +14,19 @@ use std::process;
 use anyhow::bail;
 use futures::sink::SinkExt;
 use futures::stream::TryStreamExt;
-use mz_build_info::{build_info, BuildInfo};
+use once_cell::sync::Lazy;
 use serde::de::DeserializeOwned;
 use serde::ser::Serialize;
 use tokio::net::TcpListener;
 use tokio::select;
 use tracing::info;
 use tracing_subscriber::filter::Targets;
 
+use mz_build_info::{build_info, BuildInfo};
 use mz_dataflow_types::client::{ComputeClient, GenericClient};
 use mz_dataflow_types::reconciliation::command::ComputeCommandReconcile;
 use mz_dataflow_types::ConnectorContext;
+use mz_ore::cli::{self, CliConfig};
 use mz_ore::metrics::MetricsRegistry;
 use mz_ore::now::SYSTEM_TIME;
 
@@ -45,34 +47,31 @@ static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 
 const BUILD_INFO: BuildInfo = build_info!();
 
+pub static VERSION: Lazy<String> = Lazy::new(|| BUILD_INFO.human_version());
+
 /// Independent dataflow server for Materialize.
 #[derive(clap::Parser)]
+#[clap(version = VERSION.as_str())]
 struct Args {
     /// The address on which to listen for a connection from the controller.
     #[clap(
         long,
-        env = "COMPUTED_LISTEN_ADDR",
+        env = "LISTEN_ADDR",
         value_name = "HOST:PORT",
         default_value = "127.0.0.1:2100"
     )]
     listen_addr: String,
     /// Number of dataflow worker threads.
-    #[clap(
-        short,
-        long,
-        env = "COMPUTED_WORKERS",
-        value_name = "W",
-        default_value = "1"
-    )]
+    #[clap(short, long, env = "WORKERS", value_name = "W", default_value = "1")]
     workers: usize,
     /// Number of this computed process.
-    #[clap(short = 'p', long, env = "COMPUTED_PROCESS", value_name = "P")]
+    #[clap(short = 'p', long, env = "PROCESS", value_name = "P")]
     process: Option<usize>,
     /// Total number of computed processes.
     #[clap(
         short = 'n',
         long,
-        env = "COMPUTED_PROCESSES",
+        env = "PROCESSES",
         value_name = "N",
         default_value = "1"
     )]
@@ -107,7 +106,7 @@ struct Args {
     /// The default value for this option is "info".
     #[clap(
         long,
-        env = "COMPUTED_LOG_FILTER",
+        env = "LOG_FILTER",
         value_name = "FILTER",
         default_value = "info"
     )]
@@ -120,7 +119,11 @@ struct Args {
 
 #[tokio::main]
 async fn main() {
-    if let Err(err) = run(mz_ore::cli::parse_args()).await {
+    let args = cli::parse_args(CliConfig {
+        env_prefix: Some("COMPUTED_"),
+        enable_version_flag: true,
+    });
+    if let Err(err) = run(args).await {
         eprintln!("computed: fatal: {:#}", err);
         process::exit(1);
     }

src/dataflow-bin/src/bin/dataflow-logview.rs

@@ -18,6 +18,8 @@ use timely::dataflow::operators::capture::Replay;
 use timely::dataflow::operators::Inspect;
 use timely::logging::{TimelyEvent, WorkerIdentifier};
 
+use mz_ore::cli::{self, CliConfig};
+
 /// Views Timely logs from a running dataflow server.
 ///
 /// Listens for incoming log connections from a Timely/Differential program running
@@ -40,7 +42,7 @@ struct Args {
 }
 
 fn main() {
-    let args: Args = mz_ore::cli::parse_args();
+    let args: Args = cli::parse_args(CliConfig::default());
 
     let listener =
         TcpListener::bind((&*args.listen_addr, args.port)).expect("binding tcp listener");

src/kafka-util/src/bin/kgen.rs

@@ -33,6 +33,7 @@ use mz_avro::schema::{SchemaNode, SchemaPiece, SchemaPieceOrNamed};
 use mz_avro::types::{DecimalValue, Value};
 use mz_avro::Schema;
 use mz_ore::cast::CastFrom;
+use mz_ore::cli::{self, CliConfig};
 use mz_ore::retry::Retry;
 
 trait Generator<R>: FnMut(&mut ThreadRng) -> R + Send + Sync {
@@ -583,7 +584,7 @@ struct Args {
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let args: Args = mz_ore::cli::parse_args();
+    let args: Args = cli::parse_args(CliConfig::default());
 
     let value_gen = match args.value_format {
         ValueFormat::Bytes => {