From e28122326c51d15d3efb788a15e26f5d332e2cf4 Mon Sep 17 00:00:00 2001 From: John Spray Date: Wed, 8 May 2024 17:01:37 +0100 Subject: [PATCH] pageserver: add "top N" API --- libs/pageserver_api/src/models.rs | 49 +++++++++++ libs/pageserver_api/src/shard.rs | 2 +- pageserver/client/src/mgmt_api.rs | 12 +++ pageserver/src/http/routes.rs | 83 +++++++++++++++++++ pageserver/src/metrics.rs | 8 +- pageserver/src/tenant.rs | 42 ++++++++++ .../src/tenant/remote_timeline_client.rs | 2 +- 7 files changed, 192 insertions(+), 6 deletions(-) diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs index 1df5820fb9..123283dd70 100644 --- a/libs/pageserver_api/src/models.rs +++ b/libs/pageserver_api/src/models.rs @@ -817,6 +817,55 @@ pub struct TenantScanRemoteStorageResponse { pub shards: Vec, } +#[derive(Serialize, Deserialize, Debug, Clone)] +#[serde(rename_all = "lowercase")] +pub enum TopNSorting { + ResidentSize, + MaxLogicalSize, +} + +impl Default for TopNSorting { + fn default() -> Self { + Self::ResidentSize + } +} + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct TopNTenantShardsRequest { + // How would you like to sort the tenants? + pub order_by: TopNSorting, + + // How many results? + pub limit: usize, + + // Omit tenants with more than this many shards (e.g. if this is the max number of shards + // that the caller would ever split to) + pub where_shards_lt: Option, + + // Omit tenants where the ordering metric is less than this (this is an optimization to + // let us quickly exclude numerous tiny shards) + pub where_gt: Option, +} + +#[derive(Serialize, Deserialize, Debug)] +pub struct TopNTenantShardItem { + pub id: TenantShardId, + + /// Total size of layers on local disk for all timelines in this tenant + pub resident_size: u64, + + /// Total size of layers in remote storage for all timelines in this tenant + pub physical_size: u64, + + /// The largest logical size of a timeline within this tenant + pub max_logical_size: u64, +} + +#[derive(Serialize, Deserialize, Debug, Default)] +pub struct TopNTenantShardsResponse { + pub shards: Vec, +} + pub mod virtual_file { #[derive( Copy, diff --git a/libs/pageserver_api/src/shard.rs b/libs/pageserver_api/src/shard.rs index ff6d3d91b6..43d9b2e48c 100644 --- a/libs/pageserver_api/src/shard.rs +++ b/libs/pageserver_api/src/shard.rs @@ -125,7 +125,7 @@ impl ShardCount { /// `v` may be zero, or the number of shards in the tenant. `v` is what /// [`Self::literal`] would return. - pub fn new(val: u8) -> Self { + pub const fn new(val: u8) -> Self { Self(val) } } diff --git a/pageserver/client/src/mgmt_api.rs b/pageserver/client/src/mgmt_api.rs index 6df8b2170d..d4470bd25f 100644 --- a/pageserver/client/src/mgmt_api.rs +++ b/pageserver/client/src/mgmt_api.rs @@ -486,6 +486,18 @@ impl Client { .map_err(Error::ReceiveBody) } + pub async fn top_n_tenant_shards( + &self, + request: TopNTenantShardsRequest, + ) -> Result { + let uri = format!("{}/v1/top_n_tenants", self.mgmt_api_endpoint); + self.request(Method::POST, uri, request) + .await? + .json() + .await + .map_err(Error::ReceiveBody) + } + pub async fn layer_map_info( &self, tenant_shard_id: TenantShardId, diff --git a/pageserver/src/http/routes.rs b/pageserver/src/http/routes.rs index 2370561756..3640b30ccc 100644 --- a/pageserver/src/http/routes.rs +++ b/pageserver/src/http/routes.rs @@ -25,6 +25,10 @@ use pageserver_api::models::TenantShardLocation; use pageserver_api::models::TenantShardSplitRequest; use pageserver_api::models::TenantShardSplitResponse; use pageserver_api::models::TenantState; +use pageserver_api::models::TopNSorting; +use pageserver_api::models::TopNTenantShardItem; +use pageserver_api::models::TopNTenantShardsRequest; +use pageserver_api::models::TopNTenantShardsResponse; use pageserver_api::models::{ DownloadRemoteLayersTaskSpawnRequest, LocationConfigMode, TenantAttachRequest, TenantLoadRequest, TenantLocationConfigRequest, @@ -2353,6 +2357,84 @@ async fn get_utilization( .map_err(ApiError::InternalServerError) } +/// Report on the largest tenants on this pageserver, for the storage controller to identify +/// candidates for splitting +async fn post_top_n_tenants( + mut r: Request, + _cancel: CancellationToken, +) -> Result, ApiError> { + check_permission(&r, None)?; + let request: TopNTenantShardsRequest = json_request(&mut r).await?; + let state = get_state(&r); + + fn get_size_metric(sizes: &TopNTenantShardItem, order_by: &TopNSorting) -> u64 { + match order_by { + TopNSorting::ResidentSize => sizes.resident_size, + TopNSorting::MaxLogicalSize => sizes.max_logical_size, + } + } + + let mut top_n = Vec::new(); + + // FIXME: this is a lot of clones to take this tenant list + for (tenant_shard_id, tenant_slot) in state.tenant_manager.list() { + if let Some(shards_lt) = request.where_shards_lt { + // Ignore tenants which already have >= this many shards + if tenant_shard_id.shard_count >= shards_lt { + continue; + } + } + + let sizes = match tenant_slot { + TenantSlot::Attached(tenant) => tenant.get_sizes(), + TenantSlot::Secondary(_) | TenantSlot::InProgress(_) => { + continue; + } + }; + let metric = get_size_metric(&sizes, &request.order_by); + + if let Some(gt) = request.where_gt { + // Ignore tenants whose metric is <= the lower size threshold, to do less sorting work + if metric <= gt { + continue; + } + }; + + // Insert to vector keeping it sorted + match top_n.last() { + None => { + // Top N list is empty: candidate becomes first member + top_n.push((metric, sizes)); + } + Some(last) if last.0 > metric && top_n.len() < request.limit => { + // Lowest item in list is greater than our candidate, but we aren't at limit yet: push to end + top_n.push((metric, sizes)); + } + Some(last) if last.0 > metric => { + // List is at limit and lowest value is greater than our candidate, drop it. + } + Some(_) => { + // Find location to insert + for i in 0..top_n.len() { + if top_n[i].0 < metric { + top_n.insert(i, (metric, sizes)); + break; + } + } + + top_n.truncate(request.limit); + } + } + } + + json_response( + StatusCode::OK, + TopNTenantShardsResponse { + shards: top_n.into_iter().map(|i| i.1).collect(), + }, + ) +} + /// Common functionality of all the HTTP API handlers. /// /// - Adds a tracing span to each request (by `request_span`) @@ -2639,5 +2721,6 @@ pub fn make_router( ) .put("/v1/io_engine", |r| api_handler(r, put_io_engine_handler)) .get("/v1/utilization", |r| api_handler(r, get_utilization)) + .post("/v1/top_n_tenants", |r| api_handler(r, post_top_n_tenants)) .any(handler_404)) } diff --git a/pageserver/src/metrics.rs b/pageserver/src/metrics.rs index b27bfb43b0..b8c919a520 100644 --- a/pageserver/src/metrics.rs +++ b/pageserver/src/metrics.rs @@ -2121,7 +2121,7 @@ pub(crate) struct TimelineMetrics { pub garbage_collect_histo: StorageTimeMetrics, pub find_gc_cutoffs_histo: StorageTimeMetrics, pub last_record_gauge: IntGauge, - resident_physical_size_gauge: UIntGauge, + pub resident_physical_size_gauge: UIntGauge, /// copy of LayeredTimeline.current_logical_size pub current_logical_size_gauge: UIntGauge, pub aux_file_size_gauge: IntGauge, @@ -2344,7 +2344,7 @@ use crate::task_mgr::TaskKind; use crate::tenant::mgr::TenantSlot; /// Maintain a per timeline gauge in addition to the global gauge. -struct PerTimelineRemotePhysicalSizeGauge { +pub(crate) struct PerTimelineRemotePhysicalSizeGauge { last_set: u64, gauge: UIntGauge, } @@ -2365,7 +2365,7 @@ impl PerTimelineRemotePhysicalSizeGauge { }; self.last_set = sz; } - fn get(&self) -> u64 { + pub(crate) fn get(&self) -> u64 { self.gauge.get() } } @@ -2380,7 +2380,7 @@ pub(crate) struct RemoteTimelineClientMetrics { tenant_id: String, shard_id: String, timeline_id: String, - remote_physical_size_gauge: Mutex>, + pub(crate) remote_physical_size_gauge: Mutex>, calls: Mutex>, bytes_started_counter: Mutex>, bytes_finished_counter: Mutex>, diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index 80d354d79e..ab55c37fe5 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -21,6 +21,7 @@ use futures::FutureExt; use futures::StreamExt; use pageserver_api::models; use pageserver_api::models::TimelineState; +use pageserver_api::models::TopNTenantShardItem; use pageserver_api::models::WalRedoManagerStatus; use pageserver_api::shard::ShardIdentity; use pageserver_api::shard::ShardStripeSize; @@ -2212,6 +2213,47 @@ impl Tenant { Ok(()) } + + pub(crate) fn get_sizes(&self) -> TopNTenantShardItem { + let mut result = TopNTenantShardItem { + id: self.tenant_shard_id, + resident_size: 0, + physical_size: 0, + max_logical_size: 0, + }; + + let timelines = self + .timelines + .lock() + .unwrap() + .values() + .cloned() + .collect::>(); + + for timeline in timelines { + result.resident_size += timeline.metrics.resident_physical_size_gauge.get(); + + // FIXME: it's too expensive to take a mutex in this loop + result.physical_size += timeline + .remote_client + .as_ref() + .and_then(|c| { + c.metrics + .remote_physical_size_gauge + .lock() + .unwrap() + .as_ref() + .map(|g| g.get()) + }) + .unwrap_or(0); + result.max_logical_size = std::cmp::max( + result.max_logical_size, + timeline.metrics.current_logical_size_gauge.get(), + ); + } + + result + } } /// Given a Vec of timelines and their ancestors (timeline_id, ancestor_id), diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs index 630ade5c13..698e9d1f0e 100644 --- a/pageserver/src/tenant/remote_timeline_client.rs +++ b/pageserver/src/tenant/remote_timeline_client.rs @@ -317,7 +317,7 @@ pub struct RemoteTimelineClient { upload_queue: Mutex, - metrics: Arc, + pub(crate) metrics: Arc, storage_impl: GenericRemoteStorage,