split ott-balancer-bin into 2 crates (#1190)

* rename `ott-balancer-bin` to `ott-balancer`

* add new `ott-balancer-bin` that just calls in to the `ott-balancer` crate

Author: dyc3

Cargo.lock

@@ -17,6 +17,7 @@ hyper-util = { git = "https://github.com/hyperium/hyper-util.git", rev = "f89801
 http-body-util = "0.1.0-rc.2"
 once_cell = "1.17.1"
 ott-common = { path = "crates/ott-common" }
+ott-balancer = { path = "crates/ott-balancer" }
 ott-balancer-protocol = { path = "crates/ott-balancer-protocol" }
 pin-project = "1.0.12"
 prometheus = { version = "0.13.3", features = ["process"] }

Cargo.toml

@@ -3,35 +3,7 @@ name = "ott-balancer-bin"
 version = "0.7.0"
 edition = "2021"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
 anyhow.workspace = true
-async-trait.workspace = true
-bytes.workspace = true
-clap.workspace = true
-console-subscriber.workspace = true
-figment.workspace = true
-futures-util.workspace = true
-hyper.workspace = true
-hyper-util.workspace = true
-http-body-util.workspace = true
-rand.workspace = true
-reqwest.workspace = true
-serde.workspace = true
-serde_json.workspace = true
-tracing.workspace = true
-tracing-subscriber.workspace = true
+ott-balancer.workspace = true
 tokio.workspace = true
-tokio-tungstenite.workspace = true
-tokio-util.workspace = true
-tungstenite.workspace = true
-uuid.workspace = true
-url.workspace = true
-ott-common.workspace = true
-ott-balancer-protocol.workspace = true
-route-recognizer = "0.3.1"
-once_cell.workspace = true
-pin-project.workspace = true
-prometheus.workspace = true
-trust-dns-resolver = { version = "0.22.0", features = ["system-config"] }

crates/ott-balancer-bin/Cargo.toml

@@ -1,126 +1,4 @@
-use std::net::Ipv6Addr;
-use std::{net::SocketAddr, sync::Arc};
-
-use balancer::{start_dispatcher, Balancer, BalancerContext};
-use clap::Parser;
-use hyper::server::conn::http1;
-use tokio::net::TcpListener;
-use tokio::sync::RwLock;
-use tracing::{error, info};
-use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt;
-use tracing_subscriber::util::SubscriberInitExt;
-use tracing_subscriber::EnvFilter;
-
-use crate::config::{BalancerConfig, DiscoveryConfig};
-use crate::discovery::start_discovery_task;
-use crate::service::BalancerService;
-
-mod balancer;
-mod client;
-mod config;
-mod connection;
-mod discovery;
-mod messages;
-mod monolith;
-mod room;
-mod service;
-
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let args = config::Cli::parse();
-
-    BalancerConfig::load(&args.config_path)?;
-    let config = BalancerConfig::get();
-
-    let console_layer = console_subscriber::spawn();
-    let fmt_layer = tracing_subscriber::fmt::layer();
-    let filter = args.build_tracing_filter();
-    let filter_layer = EnvFilter::try_from_default_env().or_else(|_| EnvFilter::try_new(filter))?;
-    tracing_subscriber::registry()
-        .with(console_layer)
-        .with(filter_layer)
-        .with(fmt_layer)
-        .init();
-    info!("Loaded config: {:?}", config);
-
-    let (discovery_tx, discovery_rx) = tokio::sync::mpsc::channel(2);
-
-    info!("Starting balancer");
-    let ctx = Arc::new(RwLock::new(BalancerContext::new()));
-    let balancer = Balancer::new(ctx.clone());
-    let service_link = balancer.new_link();
-    let conman_link = balancer.new_link();
-    let _dispatcher_handle = start_dispatcher(balancer)?;
-    info!("Dispatcher started");
-
-    info!("Starting monolith discovery");
-    let _discovery_handle = match &config.discovery {
-        DiscoveryConfig::Fly(config) => {
-            let discovery = discovery::FlyMonolithDiscoverer::new(config.clone());
-            start_discovery_task(discovery, discovery_tx)
-        }
-        DiscoveryConfig::Manual(config) => {
-            let discovery = discovery::ManualMonolithDiscoverer::new(config.clone());
-            start_discovery_task(discovery, discovery_tx)
-        }
-        DiscoveryConfig::Harness(config) => {
-            let discovery = discovery::HarnessMonolithDiscoverer::new(config.clone());
-            start_discovery_task(discovery, discovery_tx)
-        }
-    };
-    info!("Monolith discovery started");
-
-    info!("Starting connection manager");
-    let mut conman = connection::MonolithConnectionManager::new(discovery_rx, conman_link);
-    let _conman_handle = tokio::task::Builder::new()
-        .name("connection manager")
-        .spawn(async move {
-            loop {
-                if let Err(err) = conman.do_connection_job().await {
-                    error!("Error in connection manager: {:?}", err);
-                }
-            }
-        });
-
-    let bind_addr6: SocketAddr =
-        SocketAddr::new(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0).into(), config.port);
-
-    let service = BalancerService {
-        ctx,
-        link: service_link,
-        addr: bind_addr6,
-    };
-
-    // on linux, binding ipv6 will also bind ipv4
-    let listener6 = TcpListener::bind(bind_addr6).await?;
-
-    info!("Serving on {}", bind_addr6);
-    loop {
-        let (stream, addr) = tokio::select! {
-            stream = listener6.accept() => {
-                let (stream, addr) = stream?;
-                (stream, addr)
-            }
-        };
-
-        let mut service = service.clone();
-        service.addr = addr;
-
-        let io = hyper_util::rt::TokioIo::new(stream);
-
-        // Spawn a tokio task to serve multiple connections concurrently
-        let result = tokio::task::Builder::new()
-            .name("serve http")
-            .spawn(async move {
-                let conn = http1::Builder::new()
-                    .serve_connection(io, service)
-                    .with_upgrades();
-                if let Err(err) = conn.await {
-                    error!("Error serving connection: {:?}", err);
-                }
-            });
-        if let Err(err) = result {
-            error!("Error spawning task to serve http: {:?}", err);
-        }
-    }
+    ott_balancer::run().await
 }

crates/ott-balancer-bin/src/main.rs

@@ -0,0 +1,37 @@
+[package]
+name = "ott-balancer"
+version = "0.7.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow.workspace = true
+async-trait.workspace = true
+bytes.workspace = true
+clap.workspace = true
+console-subscriber.workspace = true
+figment.workspace = true
+futures-util.workspace = true
+hyper.workspace = true
+hyper-util.workspace = true
+http-body-util.workspace = true
+rand.workspace = true
+reqwest.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+tokio.workspace = true
+tokio-tungstenite.workspace = true
+tokio-util.workspace = true
+tungstenite.workspace = true
+uuid.workspace = true
+url.workspace = true
+ott-common.workspace = true
+ott-balancer-protocol.workspace = true
+route-recognizer = "0.3.1"
+once_cell.workspace = true
+pin-project.workspace = true
+prometheus.workspace = true
+trust-dns-resolver = { version = "0.22.0", features = ["system-config"] }

crates/ott-balancer/Cargo.toml

@@ -0,0 +1,125 @@
+use std::net::Ipv6Addr;
+use std::{net::SocketAddr, sync::Arc};
+
+use balancer::{start_dispatcher, Balancer, BalancerContext};
+use clap::Parser;
+use hyper::server::conn::http1;
+use tokio::net::TcpListener;
+use tokio::sync::RwLock;
+use tracing::{error, info};
+use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::EnvFilter;
+
+use crate::config::{BalancerConfig, DiscoveryConfig};
+use crate::discovery::start_discovery_task;
+use crate::service::BalancerService;
+
+mod balancer;
+mod client;
+mod config;
+mod connection;
+mod discovery;
+mod messages;
+mod monolith;
+mod room;
+mod service;
+
+pub async fn run() -> anyhow::Result<()> {
+    let args = config::Cli::parse();
+
+    BalancerConfig::load(&args.config_path)?;
+    let config = BalancerConfig::get();
+
+    let console_layer = console_subscriber::spawn();
+    let fmt_layer = tracing_subscriber::fmt::layer();
+    let filter = args.build_tracing_filter();
+    let filter_layer = EnvFilter::try_from_default_env().or_else(|_| EnvFilter::try_new(filter))?;
+    tracing_subscriber::registry()
+        .with(console_layer)
+        .with(filter_layer)
+        .with(fmt_layer)
+        .init();
+    info!("Loaded config: {:?}", config);
+
+    let (discovery_tx, discovery_rx) = tokio::sync::mpsc::channel(2);
+
+    info!("Starting balancer");
+    let ctx = Arc::new(RwLock::new(BalancerContext::new()));
+    let balancer = Balancer::new(ctx.clone());
+    let service_link = balancer.new_link();
+    let conman_link = balancer.new_link();
+    let _dispatcher_handle = start_dispatcher(balancer)?;
+    info!("Dispatcher started");
+
+    info!("Starting monolith discovery");
+    let _discovery_handle = match &config.discovery {
+        DiscoveryConfig::Fly(config) => {
+            let discovery = discovery::FlyMonolithDiscoverer::new(config.clone());
+            start_discovery_task(discovery, discovery_tx)
+        }
+        DiscoveryConfig::Manual(config) => {
+            let discovery = discovery::ManualMonolithDiscoverer::new(config.clone());
+            start_discovery_task(discovery, discovery_tx)
+        }
+        DiscoveryConfig::Harness(config) => {
+            let discovery = discovery::HarnessMonolithDiscoverer::new(config.clone());
+            start_discovery_task(discovery, discovery_tx)
+        }
+    };
+    info!("Monolith discovery started");
+
+    info!("Starting connection manager");
+    let mut conman = connection::MonolithConnectionManager::new(discovery_rx, conman_link);
+    let _conman_handle = tokio::task::Builder::new()
+        .name("connection manager")
+        .spawn(async move {
+            loop {
+                if let Err(err) = conman.do_connection_job().await {
+                    error!("Error in connection manager: {:?}", err);
+                }
+            }
+        });
+
+    let bind_addr6: SocketAddr =
+        SocketAddr::new(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0).into(), config.port);
+
+    let service = BalancerService {
+        ctx,
+        link: service_link,
+        addr: bind_addr6,
+    };
+
+    // on linux, binding ipv6 will also bind ipv4
+    let listener6 = TcpListener::bind(bind_addr6).await?;
+
+    info!("Serving on {}", bind_addr6);
+    loop {
+        let (stream, addr) = tokio::select! {
+            stream = listener6.accept() => {
+                let (stream, addr) = stream?;
+                (stream, addr)
+            }
+        };
+
+        let mut service = service.clone();
+        service.addr = addr;
+
+        let io = hyper_util::rt::TokioIo::new(stream);
+
+        // Spawn a tokio task to serve multiple connections concurrently
+        let result = tokio::task::Builder::new()
+            .name("serve http")
+            .spawn(async move {
+                let conn = http1::Builder::new()
+                    .serve_connection(io, service)
+                    .with_upgrades();
+                if let Err(err) = conn.await {
+                    error!("Error serving connection: {:?}", err);
+                }
+            });
+        if let Err(err) = result {
+            error!("Error spawning task to serve http: {:?}", err);
+        }
+    }
+}