mirror of
https://github.com/deuxfleurs-org/garage.git
synced 2026-08-10 22:36:53 +00:00
wip: split garage_api into garage_api_{common,s3,k2v,admin}
This commit is contained in:
Alex Auvolat
@@ -0,0 +1,378 @@
|
||||
use std::convert::Infallible;
|
||||
use std::fs::{self, Permissions};
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use futures::future::Future;
|
||||
use futures::stream::{futures_unordered::FuturesUnordered, StreamExt};
|
||||
|
||||
use http_body_util::BodyExt;
|
||||
use hyper::header::HeaderValue;
|
||||
use hyper::server::conn::http1;
|
||||
use hyper::service::service_fn;
|
||||
use hyper::{body::Incoming as IncomingBody, Request, Response};
|
||||
use hyper::{HeaderMap, StatusCode};
|
||||
use hyper_util::rt::TokioIo;
|
||||
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
|
||||
use tokio::sync::watch;
|
||||
use tokio::time::{sleep_until, Instant};
|
||||
|
||||
use opentelemetry::{
|
||||
global,
|
||||
metrics::{Counter, ValueRecorder},
|
||||
trace::{FutureExt, SpanRef, TraceContextExt, Tracer},
|
||||
Context, KeyValue,
|
||||
};
|
||||
|
||||
use garage_util::error::Error as GarageError;
|
||||
use garage_util::forwarded_headers;
|
||||
use garage_util::metrics::{gen_trace_id, RecordDuration};
|
||||
use garage_util::socket_address::UnixOrTCPSocketAddress;
|
||||
|
||||
use crate::helpers::{BoxBody, ErrorBody};
|
||||
|
||||
pub trait ApiEndpoint: Send + Sync + 'static {
|
||||
fn name(&self) -> &'static str;
|
||||
fn add_span_attributes(&self, span: SpanRef<'_>);
|
||||
}
|
||||
|
||||
pub trait ApiError: std::error::Error + Send + Sync + 'static {
|
||||
fn http_status_code(&self) -> StatusCode;
|
||||
fn add_http_headers(&self, header_map: &mut HeaderMap<HeaderValue>);
|
||||
fn http_body(&self, garage_region: &str, path: &str) -> ErrorBody;
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait ApiHandler: Send + Sync + 'static {
|
||||
const API_NAME: &'static str;
|
||||
const API_NAME_DISPLAY: &'static str;
|
||||
|
||||
type Endpoint: ApiEndpoint;
|
||||
type Error: ApiError;
|
||||
|
||||
fn parse_endpoint(&self, r: &Request<IncomingBody>) -> Result<Self::Endpoint, Self::Error>;
|
||||
async fn handle(
|
||||
&self,
|
||||
req: Request<IncomingBody>,
|
||||
endpoint: Self::Endpoint,
|
||||
) -> Result<Response<BoxBody<Self::Error>>, Self::Error>;
|
||||
}
|
||||
|
||||
pub struct ApiServer<A: ApiHandler> {
|
||||
region: String,
|
||||
api_handler: A,
|
||||
|
||||
// Metrics
|
||||
request_counter: Counter<u64>,
|
||||
error_counter: Counter<u64>,
|
||||
request_duration: ValueRecorder<f64>,
|
||||
}
|
||||
|
||||
impl<A: ApiHandler> ApiServer<A> {
|
||||
pub fn new(region: String, api_handler: A) -> Arc<Self> {
|
||||
let meter = global::meter("garage/api");
|
||||
Arc::new(Self {
|
||||
region,
|
||||
api_handler,
|
||||
request_counter: meter
|
||||
.u64_counter(format!("api.{}.request_counter", A::API_NAME))
|
||||
.with_description(format!(
|
||||
"Number of API calls to the various {} API endpoints",
|
||||
A::API_NAME_DISPLAY
|
||||
))
|
||||
.init(),
|
||||
error_counter: meter
|
||||
.u64_counter(format!("api.{}.error_counter", A::API_NAME))
|
||||
.with_description(format!(
|
||||
"Number of API calls to the various {} API endpoints that resulted in errors",
|
||||
A::API_NAME_DISPLAY
|
||||
))
|
||||
.init(),
|
||||
request_duration: meter
|
||||
.f64_value_recorder(format!("api.{}.request_duration", A::API_NAME))
|
||||
.with_description(format!(
|
||||
"Duration of API calls to the various {} API endpoints",
|
||||
A::API_NAME_DISPLAY
|
||||
))
|
||||
.init(),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn run_server(
|
||||
self: Arc<Self>,
|
||||
bind_addr: UnixOrTCPSocketAddress,
|
||||
unix_bind_addr_mode: Option<u32>,
|
||||
must_exit: watch::Receiver<bool>,
|
||||
) -> Result<(), GarageError> {
|
||||
let server_name = format!("{} API", A::API_NAME_DISPLAY);
|
||||
info!("{} server listening on {}", server_name, bind_addr);
|
||||
|
||||
match bind_addr {
|
||||
UnixOrTCPSocketAddress::TCPSocket(addr) => {
|
||||
let listener = TcpListener::bind(addr).await?;
|
||||
|
||||
let handler = move |request, socketaddr| self.clone().handler(request, socketaddr);
|
||||
server_loop(server_name, listener, handler, must_exit).await
|
||||
}
|
||||
UnixOrTCPSocketAddress::UnixSocket(ref path) => {
|
||||
if path.exists() {
|
||||
fs::remove_file(path)?
|
||||
}
|
||||
|
||||
let listener = UnixListener::bind(path)?;
|
||||
let listener = UnixListenerOn(listener, path.display().to_string());
|
||||
|
||||
fs::set_permissions(
|
||||
path,
|
||||
Permissions::from_mode(unix_bind_addr_mode.unwrap_or(0o222)),
|
||||
)?;
|
||||
|
||||
let handler = move |request, socketaddr| self.clone().handler(request, socketaddr);
|
||||
server_loop(server_name, listener, handler, must_exit).await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handler(
|
||||
self: Arc<Self>,
|
||||
req: Request<IncomingBody>,
|
||||
addr: String,
|
||||
) -> Result<Response<BoxBody<A::Error>>, http::Error> {
|
||||
let uri = req.uri().clone();
|
||||
|
||||
if let Ok(forwarded_for_ip_addr) =
|
||||
forwarded_headers::handle_forwarded_for_headers(req.headers())
|
||||
{
|
||||
info!(
|
||||
"{} (via {}) {} {}",
|
||||
forwarded_for_ip_addr,
|
||||
addr,
|
||||
req.method(),
|
||||
uri
|
||||
);
|
||||
} else {
|
||||
info!("{} {} {}", addr, req.method(), uri);
|
||||
}
|
||||
debug!("{:?}", req);
|
||||
|
||||
let tracer = opentelemetry::global::tracer("garage");
|
||||
let span = tracer
|
||||
.span_builder(format!("{} API call (unknown)", A::API_NAME_DISPLAY))
|
||||
.with_trace_id(gen_trace_id())
|
||||
.with_attributes(vec![
|
||||
KeyValue::new("method", format!("{}", req.method())),
|
||||
KeyValue::new("uri", req.uri().to_string()),
|
||||
])
|
||||
.start(&tracer);
|
||||
|
||||
let res = self
|
||||
.handler_stage2(req)
|
||||
.with_context(Context::current_with_span(span))
|
||||
.await;
|
||||
|
||||
match res {
|
||||
Ok(x) => {
|
||||
debug!("{} {:?}", x.status(), x.headers());
|
||||
Ok(x)
|
||||
}
|
||||
Err(e) => {
|
||||
let body = e.http_body(&self.region, uri.path());
|
||||
let mut http_error_builder = Response::builder().status(e.http_status_code());
|
||||
|
||||
if let Some(header_map) = http_error_builder.headers_mut() {
|
||||
e.add_http_headers(header_map)
|
||||
}
|
||||
|
||||
let http_error = http_error_builder.body(body)?;
|
||||
|
||||
if e.http_status_code().is_server_error() {
|
||||
warn!("Response: error {}, {}", e.http_status_code(), e);
|
||||
} else {
|
||||
info!("Response: error {}, {}", e.http_status_code(), e);
|
||||
}
|
||||
Ok(http_error
|
||||
.map(|body| BoxBody::new(body.map_err(|_: Infallible| unreachable!()))))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handler_stage2(
|
||||
&self,
|
||||
req: Request<IncomingBody>,
|
||||
) -> Result<Response<BoxBody<A::Error>>, A::Error> {
|
||||
let endpoint = self.api_handler.parse_endpoint(&req)?;
|
||||
debug!("Endpoint: {}", endpoint.name());
|
||||
|
||||
let current_context = Context::current();
|
||||
let current_span = current_context.span();
|
||||
current_span.update_name::<String>(format!(
|
||||
"{} API {}",
|
||||
A::API_NAME_DISPLAY,
|
||||
endpoint.name()
|
||||
));
|
||||
current_span.set_attribute(KeyValue::new("endpoint", endpoint.name()));
|
||||
endpoint.add_span_attributes(current_span);
|
||||
|
||||
let metrics_tags = &[KeyValue::new("api_endpoint", endpoint.name())];
|
||||
|
||||
let res = self
|
||||
.api_handler
|
||||
.handle(req, endpoint)
|
||||
.record_duration(&self.request_duration, &metrics_tags[..])
|
||||
.await;
|
||||
|
||||
self.request_counter.add(1, &metrics_tags[..]);
|
||||
|
||||
let status_code = match &res {
|
||||
Ok(r) => r.status(),
|
||||
Err(e) => e.http_status_code(),
|
||||
};
|
||||
if status_code.is_client_error() || status_code.is_server_error() {
|
||||
self.error_counter.add(
|
||||
1,
|
||||
&[
|
||||
metrics_tags[0].clone(),
|
||||
KeyValue::new("status_code", status_code.as_str().to_string()),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
res
|
||||
}
|
||||
}
|
||||
|
||||
// ==== helper functions ====
|
||||
|
||||
#[async_trait]
|
||||
pub trait Accept: Send + Sync + 'static {
|
||||
type Stream: AsyncRead + AsyncWrite + Send + Sync + 'static;
|
||||
async fn accept(&self) -> std::io::Result<(Self::Stream, String)>;
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Accept for TcpListener {
|
||||
type Stream = TcpStream;
|
||||
async fn accept(&self) -> std::io::Result<(Self::Stream, String)> {
|
||||
self.accept()
|
||||
.await
|
||||
.map(|(stream, addr)| (stream, addr.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
pub struct UnixListenerOn(pub UnixListener, pub String);
|
||||
|
||||
#[async_trait]
|
||||
impl Accept for UnixListenerOn {
|
||||
type Stream = UnixStream;
|
||||
async fn accept(&self) -> std::io::Result<(Self::Stream, String)> {
|
||||
self.0
|
||||
.accept()
|
||||
.await
|
||||
.map(|(stream, _addr)| (stream, self.1.clone()))
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn server_loop<A, H, F, E>(
|
||||
server_name: String,
|
||||
listener: A,
|
||||
handler: H,
|
||||
mut must_exit: watch::Receiver<bool>,
|
||||
) -> Result<(), GarageError>
|
||||
where
|
||||
A: Accept,
|
||||
H: Fn(Request<IncomingBody>, String) -> F + Send + Sync + Clone + 'static,
|
||||
F: Future<Output = Result<Response<BoxBody<E>>, http::Error>> + Send + 'static,
|
||||
E: Send + Sync + std::error::Error + 'static,
|
||||
{
|
||||
let (conn_in, mut conn_out) = tokio::sync::mpsc::unbounded_channel();
|
||||
let connection_collector = tokio::spawn({
|
||||
let server_name = server_name.clone();
|
||||
async move {
|
||||
let mut connections = FuturesUnordered::<tokio::task::JoinHandle<()>>::new();
|
||||
loop {
|
||||
let collect_next = async {
|
||||
if connections.is_empty() {
|
||||
futures::future::pending().await
|
||||
} else {
|
||||
connections.next().await
|
||||
}
|
||||
};
|
||||
tokio::select! {
|
||||
result = collect_next => {
|
||||
trace!("{} server: HTTP connection finished: {:?}", server_name, result);
|
||||
}
|
||||
new_fut = conn_out.recv() => {
|
||||
match new_fut {
|
||||
Some(f) => connections.push(f),
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
while !connections.is_empty() {
|
||||
info!(
|
||||
"{} server: {} connections still open, deadline in {:.2}s",
|
||||
server_name,
|
||||
connections.len(),
|
||||
(deadline - Instant::now()).as_secs_f32(),
|
||||
);
|
||||
tokio::select! {
|
||||
conn_res = connections.next() => {
|
||||
trace!(
|
||||
"{} server: HTTP connection finished: {:?}",
|
||||
server_name,
|
||||
conn_res.unwrap(),
|
||||
);
|
||||
}
|
||||
_ = sleep_until(deadline) => {
|
||||
warn!("{} server: exit deadline reached with {} connections still open, killing them now",
|
||||
server_name,
|
||||
connections.len());
|
||||
for conn in connections.iter() {
|
||||
conn.abort();
|
||||
}
|
||||
for conn in connections {
|
||||
assert!(conn.await.unwrap_err().is_cancelled());
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
while !*must_exit.borrow() {
|
||||
let (stream, client_addr) = tokio::select! {
|
||||
acc = listener.accept() => acc?,
|
||||
_ = must_exit.changed() => continue,
|
||||
};
|
||||
|
||||
let io = TokioIo::new(stream);
|
||||
|
||||
let handler = handler.clone();
|
||||
let serve = move |req: Request<IncomingBody>| handler(req, client_addr.clone());
|
||||
|
||||
let fut = tokio::task::spawn(async move {
|
||||
let io = Box::pin(io);
|
||||
if let Err(e) = http1::Builder::new()
|
||||
.serve_connection(io, service_fn(serve))
|
||||
.await
|
||||
{
|
||||
debug!("Error handling HTTP connection: {}", e);
|
||||
}
|
||||
});
|
||||
conn_in.send(fut)?;
|
||||
}
|
||||
|
||||
info!("{} server exiting", server_name);
|
||||
drop(conn_in);
|
||||
connection_collector.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user