Function run

src/persist-cli/src/open_loop.rs:107–167

(args: Args)

const MIB: u64 = 1024 * 1024;

pub async fn run(args: Args) -> Result<(), anyhow::Error> {
    let metrics_registry = MetricsRegistry::new();
    {
        let metrics_registry = metrics_registry.clone();
        info!(
            "serving internal HTTP server on http://{}/metrics",
            args.internal_http_listen_addr
        );
        let listener = TcpListener::bind(&args.internal_http_listen_addr)
            .await
            .expect("can bind");
        mz_ore::task::spawn(
            || "http_server",
            axum::serve(
                listener,
                axum::Router::new()
                    .route(
                        "/metrics",
                        axum::routing::get(move |headers: axum::http::HeaderMap| async move {
                            mz_http_util::handle_prometheus(&metrics_registry, headers).await
                        }),
                    )
                    .into_make_service(),
            )
            .into_future(),
        );
    }

    let location = PersistLocation {
        blob_uri: args.blob_uri.clone(),
        consensus_uri: args.consensus_uri.clone(),
    };
    let persist = PersistClientCache::new(
        PersistConfig::new_default_configs(&mz_persist_client::BUILD_INFO, SYSTEM_TIME.clone()),
        &metrics_registry,
        |_, _| PubSubClientConnection::noop(),
    )
    .open(location)
    .await?;

    let shard_id = match args.shard_id.clone() {
        Some(shard_id) => ShardId::from_str(&shard_id).map_err(anyhow::Error::msg)?,
        None => ShardId::new(),
    };

    let metrics = Arc::clone(persist.metrics());
    let (writers, readers) = match args.benchmark_type.clone() {
        BenchmarkType::RawWriter => {
            raw_persist_benchmark::setup_raw_persist(
                persist,
                shard_id,
                args.num_writers,
                args.num_readers,
            )
            .await?
        }
        BenchmarkType::MzSourceModel => panic!("source model"),
    };