Merge remote-tracking branch 'origin/main' into optimal-layout

1 files changed, 139 insertions, 49 deletions

diff --git a/src/garage/server.rs b/src/garage/server.rs
index 58c9e782..d4099a97 100644
--- a/src/garage/server.rs
+++ b/src/garage/server.rs
@@ -6,13 +6,17 @@ use garage_util::background::*;
 use garage_util::config::*;
 use garage_util::error::Error;
 
-use garage_admin::metrics::*;
-use garage_admin::tracing_setup::*;
-use garage_api::run_api_server;
+use garage_api::admin::api_server::AdminApiServer;
+use garage_api::s3::api_server::S3ApiServer;
 use garage_model::garage::Garage;
-use garage_web::run_web_server;
+use garage_web::WebServer;
+
+#[cfg(feature = "k2v")]
+use garage_api::k2v::api_server::K2VApiServer;
 
 use crate::admin::*;
+#[cfg(feature = "telemetry-otlp")]
+use crate::tracing_setup::*;
 
 async fn wait_from(mut chan: watch::Receiver<bool>) {
 	while !*chan.borrow() {
@@ -24,79 +28,124 @@ async fn wait_from(mut chan: watch::Receiver<bool>) {
 
 pub async fn run_server(config_file: PathBuf) -> Result<(), Error> {
 	info!("Loading configuration...");
-	let config = read_config(config_file).expect("Unable to read config file");
+	let config = read_config(config_file)?;
 
-	info!("Opening database...");
-	let mut db_path = config.metadata_dir.clone();
-	db_path.push("db");
-	let db = sled::Config::default()
-		.path(&db_path)
-		.cache_capacity(config.sled_cache_capacity)
-		.flush_every_ms(Some(config.sled_flush_every_ms))
-		.open()
-		.expect("Unable to open sled DB");
+	// ---- Initialize Garage internals ----
 
-	info!("Initialize admin web server and metric backend...");
-	let admin_server_init = AdminServer::init();
+	#[cfg(feature = "metrics")]
+	let metrics_exporter = opentelemetry_prometheus::exporter().init();
 
 	info!("Initializing background runner...");
-	let watch_cancel = netapp::util::watch_ctrl_c();
+	let watch_cancel = watch_shutdown_signal();
 	let (background, await_background_done) = BackgroundRunner::new(16, watch_cancel.clone());
 
 	info!("Initializing Garage main data store...");
-	let garage = Garage::new(config.clone(), db, background);
+	let garage = Garage::new(config.clone(), background)?;
+
+	if config.admin.trace_sink.is_some() {
+		info!("Initialize tracing...");
 
-	info!("Initialize tracing...");
-	if let Some(export_to) = config.admin.trace_sink {
-		init_tracing(&export_to, garage.system.id)?;
+		#[cfg(feature = "telemetry-otlp")]
+		init_tracing(config.admin.trace_sink.as_ref().unwrap(), garage.system.id)?;
+
+		#[cfg(not(feature = "telemetry-otlp"))]
+		error!("Garage was built without OTLP exporter, admin.trace_sink is ignored.");
 	}
 
+	info!("Initialize Admin API server and metrics collector...");
+	let admin_server = AdminApiServer::new(
+		garage.clone(),
+		#[cfg(feature = "metrics")]
+		metrics_exporter,
+	);
+
+	info!("Launching internal Garage cluster communications...");
 	let run_system = tokio::spawn(garage.system.clone().run(watch_cancel.clone()));
 
 	info!("Create admin RPC handler...");
 	AdminRpcHandler::new(garage.clone());
 
-	info!("Initializing API server...");
-	let api_server = tokio::spawn(run_api_server(
-		garage.clone(),
-		wait_from(watch_cancel.clone()),
-	));
+	// ---- Launch public-facing API servers ----
+
+	let mut servers = vec![];
+
+	if let Some(s3_bind_addr) = &config.s3_api.api_bind_addr {
+		info!("Initializing S3 API server...");
+		servers.push((
+			"S3 API",
+			tokio::spawn(S3ApiServer::run(
+				garage.clone(),
+				*s3_bind_addr,
+				config.s3_api.s3_region.clone(),
+				wait_from(watch_cancel.clone()),
+			)),
+		));
+	}
 
-	info!("Initializing web server...");
-	let web_server = tokio::spawn(run_web_server(
-		garage.clone(),
-		wait_from(watch_cancel.clone()),
-	));
-
-	let admin_server = if let Some(admin_bind_addr) = config.admin.api_bind_addr {
-		info!("Configure and run admin web server...");
-		Some(tokio::spawn(
-			admin_server_init.run(admin_bind_addr, wait_from(watch_cancel.clone())),
-		))
-	} else {
-		None
-	};
+	if config.k2v_api.is_some() {
+		#[cfg(feature = "k2v")]
+		{
+			info!("Initializing K2V API server...");
+			servers.push((
+				"K2V API",
+				tokio::spawn(K2VApiServer::run(
+					garage.clone(),
+					config.k2v_api.as_ref().unwrap().api_bind_addr,
+					config.s3_api.s3_region.clone(),
+					wait_from(watch_cancel.clone()),
+				)),
+			));
+		}
+		#[cfg(not(feature = "k2v"))]
+		error!("K2V is not enabled in this build, cannot start K2V API server");
+	}
 
-	// Stuff runs
+	if let Some(web_config) = &config.s3_web {
+		info!("Initializing web server...");
+		servers.push((
+			"Web",
+			tokio::spawn(WebServer::run(
+				garage.clone(),
+				web_config.bind_addr,
+				web_config.root_domain.clone(),
+				wait_from(watch_cancel.clone()),
+			)),
+		));
+	}
 
-	// When a cancel signal is sent, stuff stops
-	if let Err(e) = api_server.await? {
-		warn!("API server exited with error: {}", e);
+	if let Some(admin_bind_addr) = &config.admin.api_bind_addr {
+		info!("Launching Admin API server...");
+		servers.push((
+			"Admin",
+			tokio::spawn(admin_server.run(*admin_bind_addr, wait_from(watch_cancel.clone()))),
+		));
 	}
-	if let Err(e) = web_server.await? {
-		warn!("Web server exited with error: {}", e);
+
+	#[cfg(not(feature = "metrics"))]
+	if config.admin.metrics_token.is_some() {
+		warn!("This Garage version is built without the metrics feature");
 	}
-	if let Some(a) = admin_server {
-		if let Err(e) = a.await? {
-			warn!("Admin web server exited with error: {}", e);
+
+	// Stuff runs
+
+	// When a cancel signal is sent, stuff stops
+
+	// Collect stuff
+	for (desc, join_handle) in servers {
+		if let Err(e) = join_handle.await? {
+			error!("{} server exited with error: {}", desc, e);
+		} else {
+			info!("{} server exited without error.", desc);
 		}
 	}
 
 	// Remove RPC handlers for system to break reference cycles
 	garage.system.netapp.drop_all_handlers();
+	opentelemetry::global::shutdown_tracer_provider();
 
 	// Await for netapp RPC system to end
 	run_system.await?;
+	info!("Netapp exited");
 
 	// Drop all references so that stuff can terminate properly
 	drop(garage);
@@ -108,3 +157,44 @@ pub async fn run_server(config_file: PathBuf) -> Result<(), Error> {
 
 	Ok(())
 }
+
+#[cfg(unix)]
+fn watch_shutdown_signal() -> watch::Receiver<bool> {
+	use tokio::signal::unix::*;
+
+	let (send_cancel, watch_cancel) = watch::channel(false);
+	tokio::spawn(async move {
+		let mut sigint = signal(SignalKind::interrupt()).expect("Failed to install SIGINT handler");
+		let mut sigterm =
+			signal(SignalKind::terminate()).expect("Failed to install SIGTERM handler");
+		let mut sighup = signal(SignalKind::hangup()).expect("Failed to install SIGHUP handler");
+		tokio::select! {
+			_ = sigint.recv() => info!("Received SIGINT, shutting down."),
+			_ = sigterm.recv() => info!("Received SIGTERM, shutting down."),
+			_ = sighup.recv() => info!("Received SIGHUP, shutting down."),
+		}
+		send_cancel.send(true).unwrap();
+	});
+	watch_cancel
+}
+
+#[cfg(windows)]
+fn watch_shutdown_signal() -> watch::Receiver<bool> {
+	use tokio::signal::windows::*;
+
+	let (send_cancel, watch_cancel) = watch::channel(false);
+	tokio::spawn(async move {
+		let mut sigint = ctrl_c().expect("Failed to install Ctrl-C handler");
+		let mut sigclose = ctrl_close().expect("Failed to install Ctrl-Close handler");
+		let mut siglogoff = ctrl_logoff().expect("Failed to install Ctrl-Logoff handler");
+		let mut sigsdown = ctrl_shutdown().expect("Failed to install Ctrl-Shutdown handler");
+		tokio::select! {
+			_ = sigint.recv() => info!("Received Ctrl-C, shutting down."),
+			_ = sigclose.recv() => info!("Received Ctrl-Close, shutting down."),
+			_ = siglogoff.recv() => info!("Received Ctrl-Logoff, shutting down."),
+			_ = sigsdown.recv() => info!("Received Ctrl-Shutdown, shutting down."),
+		}
+		send_cancel.send(true).unwrap();
+	});
+	watch_cancel
+}