feat: bootstrap api project (#8)

API-3

Cargo.toml

@@ -1,7 +1,8 @@
 [workspace]
 
 members = [
-    "apps/event-api",
+    "apps/*",
+    "shared",
 ]
 
 resolver = "2"
@@ -20,3 +21,6 @@ lto = "fat"
 inherits = "release"
 debug = true
 strip = false
+
+[workspace.dependencies]
+shared = { path = "shared" }

apps/api/CC_APACHE2_LICENSE

@@ -0,0 +1 @@
+../../licenses/CC_APACHE2_LICENSE

apps/api/Cargo.toml

@@ -0,0 +1,47 @@
+[package]
+name = "api"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0 with Commons Clause"
+
+[dependencies]
+tokio = { version = "1.35", features = ["full"] }
+scuffle-config = "0.0.1"
+scuffle-utils = { version = "0.0.5", features = ["http", "signal", "context", "prelude"] }
+serde = { version = "1.0", features = ["derive"] }
+tracing = "0.1"
+hyper = { version = "1", features = ["full"] }
+async-nats = "0.33.0"
+anyhow = "1"
+http-body-util = "0.1"
+hyper-util = { version = "0.1", features = ["tokio", "server", "server-auto"] }
+rustls-pemfile = "2"
+rustls = "0.22"
+tokio-rustls = "0.25"
+hyper-tungstenite = "0.13"
+http-body = "1.0"
+tokio-stream = { version = "0.1", features = ["sync"] }
+futures-util = "0.3"
+ulid = "1"
+hex = "0.4"
+serde_json = "1.0"
+chrono = { version = "0.4", features = ["serde"] }
+sha2 = "0.10"
+urlencoding = "2"
+url = "2"
+futures = "0.3"
+reqwest = { version = "0.11", features = ["json", "rustls"], default-features = false }
+prometheus-client = "0.22"
+scopeguard = "1.1"
+rand = "0.8"
+memory-stats = "1"
+cap = "0.1"
+tikv-jemallocator = { version = "0.5", features = ["background_threads"] }
+const-lru = "1"
+fnv = "1"
+pin-project = "1"
+
+shared = { workspace = true }
+
+[features]
+default = []

apps/api/Dockerfile

@@ -0,0 +1,25 @@
+FROM debian:bookworm-slim
+
+WORKDIR /app
+
+# Copy the binary from the host machine
+# This assumes that you have already built the binary using the
+# `cargo build --profile release-lto` command.
+RUN --mount=type=bind,src=target/release-lto/api,dst=/mount/api \
+    cp /mount/api /app/api && \
+    chmod +x /app/api
+
+STOPSIGNAL SIGTERM
+
+USER 1000
+
+# This is our default configuration for jemalloc
+# See https://github.com/jemalloc/jemalloc/blob/dev/TUNING.md for more information
+# on the various options.
+ARG JEMALLOC_CONF="background_thread:true,tcache_max:4096,metadata_thp:always,dirty_decay_ms:3000,muzzy_decay_ms:3000,abort_conf:true"
+
+# Set the environment variables for jemalloc
+ENV _RJEM_MALLOC_CONF=${JEMALLOC_CONF} \
+    MALLOC_CONF=${JEMALLOC_CONF}
+
+ENTRYPOINT ["/app/api"]

apps/api/src/config.rs

@@ -0,0 +1,34 @@
+use std::net::SocketAddr;
+
+use serde::Deserialize;
+use shared::config::TlsConfig;
+
+#[derive(Debug, Deserialize, config::Config)]
+#[serde(default)]
+pub struct Api {
+	/// API bind
+	pub bind: SocketAddr,
+	/// Max Listen Conn
+	pub listen_backlog: u32,
+	/// TLS configuration
+	pub tls: Option<TlsConfig>,
+}
+
+impl Default for Api {
+	fn default() -> Self {
+		Self {
+			bind: SocketAddr::new([0, 0, 0, 0].into(), 3000),
+			listen_backlog: 128,
+			tls: None,
+		}
+	}
+}
+
+#[derive(Default, Debug, Deserialize, config::Config)]
+#[serde(default)]
+pub struct Extra {
+	/// API configuration
+	pub api: Api,
+}
+
+pub type Config = shared::config::Config<Extra>;

apps/api/src/global.rs

@@ -0,0 +1,92 @@
+use anyhow::Context as _;
+use async_nats::connection::State;
+use scuffle_utils::context::Context;
+
+use crate::config::Config;
+use crate::metrics;
+
+pub struct Global {
+	ctx: Context,
+	nats: async_nats::Client,
+	config: Config,
+	http_client: reqwest::Client,
+	metrics: metrics::Metrics,
+}
+
+impl Global {
+	pub async fn new(ctx: Context, config: Config) -> anyhow::Result<Self> {
+		let nats = async_nats::connect(&config.nats.url).await.context("nats connect")?;
+
+		Ok(Self {
+			metrics: metrics::Metrics::new(
+				config
+					.monitoring
+					.labels
+					.iter()
+					.map(|x| (x.key.clone(), x.value.clone()))
+					.collect(),
+			),
+			ctx,
+			nats,
+			config,
+			http_client: reqwest::Client::new(),
+		})
+	}
+
+	/// The global context.
+	pub fn ctx(&self) -> &Context {
+		&self.ctx
+	}
+
+	/// The NATS client.
+	pub fn nats(&self) -> &async_nats::Client {
+		&self.nats
+	}
+
+	/// The configuration.
+	pub fn config(&self) -> &Config {
+		&self.config
+	}
+
+	/// Global HTTP client.
+	pub fn http_client(&self) -> &reqwest::Client {
+		&self.http_client
+	}
+
+	/// Global metrics.
+	pub fn metrics(&self) -> &metrics::Metrics {
+		&self.metrics
+	}
+}
+
+impl shared::metrics::MetricsProvider for Global {
+	fn ctx(&self) -> &scuffle_utils::context::Context {
+		&self.ctx
+	}
+
+	fn bind(&self) -> std::net::SocketAddr {
+		self.config.monitoring.bind
+	}
+
+	fn registry(&self) -> &prometheus_client::registry::Registry {
+		self.metrics.registry()
+	}
+
+	fn pre_hook(&self) {
+		self.metrics.observe_memory()
+	}
+}
+
+impl shared::health::HealthProvider for Global {
+	fn bind(&self) -> std::net::SocketAddr {
+		self.config.health.bind
+	}
+
+	fn ctx(&self) -> &scuffle_utils::context::Context {
+		&self.ctx
+	}
+
+	fn healthy(&self, _path: &str) -> bool {
+		matches!(self.nats.connection_state(), State::Connected)
+	}
+}

apps/api/src/main.rs

@@ -0,0 +1,61 @@
+use std::sync::Arc;
+
+use cap::Cap;
+use scuffle_utils::context::Context;
+use scuffle_utils::prelude::FutureTimeout;
+use tokio::signal::unix::SignalKind;
+
+mod config;
+mod global;
+mod metrics;
+
+#[global_allocator]
+static ALLOCATOR: Cap<tikv_jemallocator::Jemalloc> = Cap::new(tikv_jemallocator::Jemalloc, usize::max_value());
+
+#[tokio::main]
+async fn main() {
+	let config = shared::config::parse(true, Some("config".into())).expect("failed to parse config");
+	shared::logging::init(&config.logging.level, config.logging.mode).expect("failed to initialize logging");
+
+	if let Some(path) = config.config_file.as_ref() {
+		tracing::info!("using config file: {path}");
+	}
+
+	if let Some(limit) = config.memory.limit {
+		tracing::info!("setting memory limit to {limit} bytes");
+		ALLOCATOR.set_limit(limit).expect("failed to set memory limit");
+	}
+
+	tracing::info!("starting event-api");
+
+	let (ctx, handler) = Context::new();
+
+	let global = Arc::new(global::Global::new(ctx, config).await.expect("failed to initialize global"));
+
+	let mut signal = scuffle_utils::signal::SignalHandler::new()
+		.with_signal(SignalKind::interrupt())
+		.with_signal(SignalKind::terminate());
+
+	let health_handle = tokio::spawn(shared::health::run(global.clone()));
+	let metrics_handle = tokio::spawn(shared::metrics::run(global.clone()));
+
+	tokio::select! {
+		_ = signal.recv() => tracing::info!("received shutdown signal"),
+		r = health_handle => tracing::warn!("health server exited: {:?}", r),
+		r = metrics_handle => tracing::warn!("metrics server exited: {:?}", r),
+	}
+
+	drop(global);
+
+	tokio::select! {
+		_ = signal.recv() => tracing::info!("received second shutdown signal, forcing exit"),
+		r = handler.cancel().timeout(std::time::Duration::from_secs(60)) => {
+			if r.is_err() {
+				tracing::warn!("failed to cancel context in time, force exit");
+			}
+		}
+	}
+
+	tracing::info!("stopping event-api");
+	std::process::exit(0);
+}

apps/api/src/metrics/labels.rs

@@ -0,0 +1,18 @@
+use prometheus_client::encoding::EncodeLabelSet;
+
+#[derive(Debug, Clone, Hash, Copy, Eq, PartialEq, EncodeLabelSet)]
+/// Memory labels.
+pub struct Memory {
+	kind: &'static str,
+}
+
+impl Memory {
+	/// Allocated memory.
+	pub const ALLOCATED: Self = Self { kind: "allocated" };
+	/// Free memory.
+	pub const REMAINING: Self = Self { kind: "remaining" };
+	/// Virtual memory.
+	pub const RESIDENT: Self = Self { kind: "resident" };
+	/// Virtual memory.
+	pub const VIRTUAL: Self = Self { kind: "virtual" };
+}