garage/src/api/admin/cluster.rs

199 lines
5 KiB
Rust
Raw Normal View History

2022-05-05 11:40:31 +00:00
use std::collections::HashMap;
2022-05-05 11:51:23 +00:00
use std::net::SocketAddr;
use std::sync::Arc;
2022-05-05 11:40:31 +00:00
use hyper::{Body, Request, Response, StatusCode};
use serde::{Deserialize, Serialize};
2022-05-05 11:40:31 +00:00
use garage_util::crdt::*;
use garage_util::data::*;
use garage_util::error::Error as GarageError;
2022-05-05 11:40:31 +00:00
use garage_rpc::layout::*;
use garage_model::garage::Garage;
use crate::admin::error::*;
2022-05-13 13:21:32 +00:00
use crate::helpers::parse_json_body;
2022-05-05 11:40:31 +00:00
2022-05-05 11:51:23 +00:00
pub async fn handle_get_cluster_status(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
2022-05-05 11:40:31 +00:00
let res = GetClusterStatusResponse {
node: hex::encode(garage.system.id),
garage_version: garage.system.garage_version(),
2022-05-05 11:51:23 +00:00
known_nodes: garage
.system
.get_known_nodes()
2022-05-05 11:40:31 +00:00
.into_iter()
2022-05-05 11:51:23 +00:00
.map(|i| {
(
hex::encode(i.id),
KnownNodeResp {
addr: i.addr,
is_up: i.is_up,
last_seen_secs_ago: i.last_seen_secs_ago,
hostname: i.status.hostname,
},
)
})
2022-05-05 11:40:31 +00:00
.collect(),
2022-05-05 11:51:23 +00:00
layout: get_cluster_layout(garage),
};
let resp_json = serde_json::to_string_pretty(&res).map_err(GarageError::from)?;
Ok(Response::builder()
.status(StatusCode::OK)
.body(Body::from(resp_json))?)
}
2022-05-17 16:43:47 +00:00
pub async fn handle_connect_cluster_nodes(
garage: &Arc<Garage>,
req: Request<Body>,
) -> Result<Response<Body>, Error> {
let req = parse_json_body::<Vec<String>>(req).await?;
let res = futures::future::join_all(req.iter().map(|node| garage.system.connect(node)))
.await
.into_iter()
.map(|r| match r {
Ok(()) => ConnectClusterNodesResponse {
success: true,
error: None,
},
Err(e) => ConnectClusterNodesResponse {
success: false,
error: Some(format!("{}", e)),
},
})
.collect::<Vec<_>>();
let resp_json = serde_json::to_string_pretty(&res).map_err(GarageError::from)?;
Ok(Response::builder()
.status(StatusCode::OK)
.body(Body::from(resp_json))?)
}
2022-05-05 11:51:23 +00:00
pub async fn handle_get_cluster_layout(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
let res = get_cluster_layout(garage);
let resp_json = serde_json::to_string_pretty(&res).map_err(GarageError::from)?;
Ok(Response::builder()
.status(StatusCode::OK)
.body(Body::from(resp_json))?)
}
fn get_cluster_layout(garage: &Arc<Garage>) -> GetClusterLayoutResponse {
let layout = garage.system.get_cluster_layout();
GetClusterLayoutResponse {
version: layout.version,
2022-05-05 11:51:23 +00:00
roles: layout
.roles
.items()
2022-05-05 11:40:31 +00:00
.iter()
.filter(|(_, _, v)| v.0.is_some())
.map(|(k, _, v)| (hex::encode(k), v.0.clone()))
.collect(),
2022-05-05 11:51:23 +00:00
staged_role_changes: layout
.staging
.items()
2022-05-05 11:40:31 +00:00
.iter()
.filter(|(k, _, v)| layout.roles.get(k) != Some(v))
.map(|(k, _, v)| (hex::encode(k), v.0.clone()))
.collect(),
2022-05-05 11:51:23 +00:00
}
2022-05-05 11:40:31 +00:00
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
2022-05-05 11:40:31 +00:00
struct GetClusterStatusResponse {
node: String,
garage_version: &'static str,
2022-05-05 11:40:31 +00:00
known_nodes: HashMap<String, KnownNodeResp>,
2022-05-05 11:51:23 +00:00
layout: GetClusterLayoutResponse,
}
2022-05-17 16:43:47 +00:00
#[derive(Serialize)]
struct ConnectClusterNodesResponse {
success: bool,
error: Option<String>,
}
2022-05-05 11:51:23 +00:00
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
2022-05-05 11:51:23 +00:00
struct GetClusterLayoutResponse {
version: u64,
2022-05-05 11:40:31 +00:00
roles: HashMap<String, Option<NodeRole>>,
staged_role_changes: HashMap<String, Option<NodeRole>>,
}
#[derive(Serialize)]
struct KnownNodeResp {
addr: SocketAddr,
is_up: bool,
last_seen_secs_ago: Option<u64>,
hostname: String,
}
pub async fn handle_update_cluster_layout(
garage: &Arc<Garage>,
req: Request<Body>,
) -> Result<Response<Body>, Error> {
let updates = parse_json_body::<UpdateClusterLayoutRequest>(req).await?;
let mut layout = garage.system.get_cluster_layout();
let mut roles = layout.roles.clone();
roles.merge(&layout.staging);
for (node, role) in updates {
let node = hex::decode(node).ok_or_bad_request("Invalid node identifier")?;
let node = Uuid::try_from(&node).ok_or_bad_request("Invalid node identifier")?;
layout
.staging
.merge(&roles.update_mutator(node, NodeRoleV(role)));
}
garage.system.update_cluster_layout(&layout).await?;
Ok(Response::builder()
.status(StatusCode::OK)
.body(Body::empty())?)
}
pub async fn handle_apply_cluster_layout(
garage: &Arc<Garage>,
req: Request<Body>,
) -> Result<Response<Body>, Error> {
let param = parse_json_body::<ApplyRevertLayoutRequest>(req).await?;
let layout = garage.system.get_cluster_layout();
let layout = layout.apply_staged_changes(Some(param.version))?;
garage.system.update_cluster_layout(&layout).await?;
Ok(Response::builder()
.status(StatusCode::OK)
.body(Body::empty())?)
}
pub async fn handle_revert_cluster_layout(
garage: &Arc<Garage>,
req: Request<Body>,
) -> Result<Response<Body>, Error> {
let param = parse_json_body::<ApplyRevertLayoutRequest>(req).await?;
let layout = garage.system.get_cluster_layout();
let layout = layout.revert_staged_changes(Some(param.version))?;
garage.system.update_cluster_layout(&layout).await?;
Ok(Response::builder()
.status(StatusCode::OK)
.body(Body::empty())?)
}
type UpdateClusterLayoutRequest = HashMap<String, Option<NodeRole>>;
#[derive(Deserialize)]
struct ApplyRevertLayoutRequest {
version: u64,
}