crabka-replicator 0.3.8

Cross-cluster geo-replication service for Crabka (MirrorMaker-2 equivalent)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

use anyhow::Context as _;
use clap::Parser;
use crabka_replicator::config::ReplicatorConfig;
use crabka_replicator::supervisor::FlowSupervisor;
use tracing_subscriber::layer::SubscriberExt as _;
use tracing_subscriber::util::SubscriberInitExt as _;

#[derive(Debug, Parser)]
#[command(name = "crabka-replicator", version, about)]
struct Cli {
    /// Path to the replicator YAML config.
    #[arg(long, env = "CRABKA_REPLICATOR_CONFIG")]
    config: std::path::PathBuf,
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> anyhow::Result<()> {
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();
    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| "crabka_replicator=info,info".into());
    tracing_subscriber::registry()
        .with(crabka_logfmt::layer(filter, std::io::stdout))
        .init();

    let cli = Cli::parse();
    let yaml = std::fs::read_to_string(&cli.config)
        .with_context(|| format!("reading config {}", cli.config.display()))?;
    let config = ReplicatorConfig::from_yaml(&yaml)?;
    config.validate()?;

    let supervisor = FlowSupervisor::run(config).await?;
    tracing::info!("crabka-replicator running; send SIGINT/ctrl-c to stop");
    tokio::signal::ctrl_c().await?;
    tracing::info!("shutdown requested; draining flows");
    supervisor.shutdown().await;
    Ok(())
}