LCOV - code coverage report
Current view: top level - storage_controller/src - http.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 1.6 % 1633 26
Test Date: 2025-03-12 16:10:49 Functions: 0.7 % 441 3

            Line data    Source code
       1              : use std::str::FromStr;
       2              : use std::sync::{Arc, LazyLock};
       3              : use std::time::{Duration, Instant};
       4              : 
       5              : use anyhow::Context;
       6              : use control_plane::storage_controller::{AttachHookRequest, InspectRequest};
       7              : use futures::Future;
       8              : use http_utils::endpoint::{
       9              :     self, auth_middleware, check_permission_with, profile_cpu_handler, profile_heap_handler,
      10              :     request_span,
      11              : };
      12              : use http_utils::error::ApiError;
      13              : use http_utils::failpoints::failpoints_handler;
      14              : use http_utils::json::{json_request, json_response};
      15              : use http_utils::request::{must_get_query_param, parse_query_param, parse_request_param};
      16              : use http_utils::{RequestExt, RouterBuilder};
      17              : use hyper::header::CONTENT_TYPE;
      18              : use hyper::{Body, Request, Response, StatusCode, Uri};
      19              : use metrics::{BuildInfo, NeonMetrics};
      20              : use pageserver_api::controller_api::{
      21              :     MetadataHealthListOutdatedRequest, MetadataHealthListOutdatedResponse,
      22              :     MetadataHealthListUnhealthyResponse, MetadataHealthUpdateRequest, MetadataHealthUpdateResponse,
      23              :     NodeAvailability, NodeConfigureRequest, NodeRegisterRequest, SafekeeperSchedulingPolicyRequest,
      24              :     ShardsPreferredAzsRequest, TenantCreateRequest, TenantPolicyRequest, TenantShardMigrateRequest,
      25              : };
      26              : use pageserver_api::models::{
      27              :     TenantConfigPatchRequest, TenantConfigRequest, TenantLocationConfigRequest,
      28              :     TenantShardSplitRequest, TenantTimeTravelRequest, TimelineArchivalConfigRequest,
      29              :     TimelineCreateRequest,
      30              : };
      31              : use pageserver_api::shard::TenantShardId;
      32              : use pageserver_api::upcall_api::{ReAttachRequest, ValidateRequest};
      33              : use pageserver_client::{BlockUnblock, mgmt_api};
      34              : use routerify::Middleware;
      35              : use tokio_util::sync::CancellationToken;
      36              : use tracing::warn;
      37              : use utils::auth::{Scope, SwappableJwtAuth};
      38              : use utils::id::{NodeId, TenantId, TimelineId};
      39              : 
      40              : use crate::http;
      41              : use crate::metrics::{
      42              :     HttpRequestLatencyLabelGroup, HttpRequestStatusLabelGroup, METRICS_REGISTRY,
      43              :     PageserverRequestLabelGroup,
      44              : };
      45              : use crate::persistence::SafekeeperUpsert;
      46              : use crate::reconciler::ReconcileError;
      47              : use crate::service::{LeadershipStatus, RECONCILE_TIMEOUT, STARTUP_RECONCILE_TIMEOUT, Service};
      48              : 
      49              : /// State available to HTTP request handlers
      50              : pub struct HttpState {
      51              :     service: Arc<crate::service::Service>,
      52              :     auth: Option<Arc<SwappableJwtAuth>>,
      53              :     rate_limiter: governor::DefaultKeyedRateLimiter<TenantId>,
      54              :     neon_metrics: NeonMetrics,
      55              :     allowlist_routes: &'static [&'static str],
      56              : }
      57              : 
      58              : impl HttpState {
      59            0 :     pub fn new(
      60            0 :         service: Arc<crate::service::Service>,
      61            0 :         auth: Option<Arc<SwappableJwtAuth>>,
      62            0 :         build_info: BuildInfo,
      63            0 :     ) -> Self {
      64            0 :         let quota = governor::Quota::per_second(service.get_config().tenant_rate_limit);
      65            0 :         Self {
      66            0 :             service,
      67            0 :             auth,
      68            0 :             rate_limiter: governor::RateLimiter::keyed(quota),
      69            0 :             neon_metrics: NeonMetrics::new(build_info),
      70            0 :             allowlist_routes: &[
      71            0 :                 "/status",
      72            0 :                 "/ready",
      73            0 :                 "/metrics",
      74            0 :                 "/profile/cpu",
      75            0 :                 "/profile/heap",
      76            0 :             ],
      77            0 :         }
      78            0 :     }
      79              : }
      80              : 
      81              : #[inline(always)]
      82            0 : fn get_state(request: &Request<Body>) -> &HttpState {
      83            0 :     request
      84            0 :         .data::<Arc<HttpState>>()
      85            0 :         .expect("unknown state type")
      86            0 :         .as_ref()
      87            0 : }
      88              : 
      89              : /// Rate limits tenant requests.
      90              : ///
      91              : /// TODO: this should be a request middleware, but requires us to extract the tenant ID from
      92              : /// different URLs in a systematic way.
      93              : ///
      94              : /// TODO: consider returning a 429 response if these start piling up.
      95            0 : async fn maybe_rate_limit(request: &Request<Body>, tenant_id: TenantId) {
      96            0 :     // Check if the tenant should be rate-limited.
      97            0 :     let rate_limiter = &get_state(request).rate_limiter;
      98            0 :     if rate_limiter.check_key(&tenant_id).is_ok() {
      99            0 :         return;
     100            0 :     }
     101            0 : 
     102            0 :     // Measure the rate limiting delay.
     103            0 :     let _timer = METRICS_REGISTRY
     104            0 :         .metrics_group
     105            0 :         .storage_controller_http_request_rate_limited
     106            0 :         .start_timer();
     107              : 
     108              :     // Log rate limited tenants once every 10 seconds.
     109              :     static LOG_RATE_LIMITER: LazyLock<governor::DefaultKeyedRateLimiter<TenantId>> =
     110            0 :         LazyLock::new(|| {
     111            0 :             let quota = governor::Quota::with_period(Duration::from_secs(10)).unwrap();
     112            0 :             governor::RateLimiter::keyed(quota)
     113            0 :         });
     114              : 
     115            0 :     if LOG_RATE_LIMITER.check_key(&tenant_id).is_ok() {
     116            0 :         warn!("tenant {tenant_id} is rate limited")
     117            0 :     }
     118              : 
     119              :     // Wait for quota.
     120            0 :     rate_limiter.until_key_ready(&tenant_id).await;
     121            0 : }
     122              : 
     123              : /// Pageserver calls into this on startup, to learn which tenants it should attach
     124            0 : async fn handle_re_attach(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     125            0 :     check_permissions(&req, Scope::GenerationsApi)?;
     126              : 
     127            0 :     let mut req = match maybe_forward(req).await {
     128            0 :         ForwardOutcome::Forwarded(res) => {
     129            0 :             return res;
     130              :         }
     131            0 :         ForwardOutcome::NotForwarded(req) => req,
     132              :     };
     133              : 
     134            0 :     let reattach_req = json_request::<ReAttachRequest>(&mut req).await?;
     135            0 :     let state = get_state(&req);
     136            0 :     json_response(StatusCode::OK, state.service.re_attach(reattach_req).await?)
     137            0 : }
     138              : 
     139              : /// Pageserver calls into this before doing deletions, to confirm that it still
     140              : /// holds the latest generation for the tenants with deletions enqueued
     141            0 : async fn handle_validate(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     142            0 :     check_permissions(&req, Scope::GenerationsApi)?;
     143              : 
     144            0 :     let mut req = match maybe_forward(req).await {
     145            0 :         ForwardOutcome::Forwarded(res) => {
     146            0 :             return res;
     147              :         }
     148            0 :         ForwardOutcome::NotForwarded(req) => req,
     149              :     };
     150              : 
     151            0 :     let validate_req = json_request::<ValidateRequest>(&mut req).await?;
     152            0 :     let state = get_state(&req);
     153            0 :     json_response(StatusCode::OK, state.service.validate(validate_req).await?)
     154            0 : }
     155              : 
     156              : /// Call into this before attaching a tenant to a pageserver, to acquire a generation number
     157              : /// (in the real control plane this is unnecessary, because the same program is managing
     158              : ///  generation numbers and doing attachments).
     159            0 : async fn handle_attach_hook(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     160            0 :     check_permissions(&req, Scope::Admin)?;
     161              : 
     162            0 :     let mut req = match maybe_forward(req).await {
     163            0 :         ForwardOutcome::Forwarded(res) => {
     164            0 :             return res;
     165              :         }
     166            0 :         ForwardOutcome::NotForwarded(req) => req,
     167              :     };
     168              : 
     169            0 :     let attach_req = json_request::<AttachHookRequest>(&mut req).await?;
     170            0 :     let state = get_state(&req);
     171            0 : 
     172            0 :     json_response(
     173            0 :         StatusCode::OK,
     174            0 :         state
     175            0 :             .service
     176            0 :             .attach_hook(attach_req)
     177            0 :             .await
     178            0 :             .map_err(ApiError::InternalServerError)?,
     179              :     )
     180            0 : }
     181              : 
     182            0 : async fn handle_inspect(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     183            0 :     check_permissions(&req, Scope::Admin)?;
     184              : 
     185            0 :     let mut req = match maybe_forward(req).await {
     186            0 :         ForwardOutcome::Forwarded(res) => {
     187            0 :             return res;
     188              :         }
     189            0 :         ForwardOutcome::NotForwarded(req) => req,
     190              :     };
     191              : 
     192            0 :     let inspect_req = json_request::<InspectRequest>(&mut req).await?;
     193              : 
     194            0 :     let state = get_state(&req);
     195            0 : 
     196            0 :     json_response(StatusCode::OK, state.service.inspect(inspect_req))
     197            0 : }
     198              : 
     199            0 : async fn handle_tenant_create(
     200            0 :     service: Arc<Service>,
     201            0 :     req: Request<Body>,
     202            0 : ) -> Result<Response<Body>, ApiError> {
     203            0 :     check_permissions(&req, Scope::PageServerApi)?;
     204              : 
     205            0 :     let mut req = match maybe_forward(req).await {
     206            0 :         ForwardOutcome::Forwarded(res) => {
     207            0 :             return res;
     208              :         }
     209            0 :         ForwardOutcome::NotForwarded(req) => req,
     210              :     };
     211              : 
     212            0 :     let create_req = json_request::<TenantCreateRequest>(&mut req).await?;
     213              : 
     214              :     json_response(
     215              :         StatusCode::CREATED,
     216            0 :         service.tenant_create(create_req).await?,
     217              :     )
     218            0 : }
     219              : 
     220            0 : async fn handle_tenant_location_config(
     221            0 :     service: Arc<Service>,
     222            0 :     req: Request<Body>,
     223            0 : ) -> Result<Response<Body>, ApiError> {
     224            0 :     let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
     225            0 :     check_permissions(&req, Scope::PageServerApi)?;
     226              : 
     227            0 :     let mut req = match maybe_forward(req).await {
     228            0 :         ForwardOutcome::Forwarded(res) => {
     229            0 :             return res;
     230              :         }
     231            0 :         ForwardOutcome::NotForwarded(req) => req,
     232              :     };
     233              : 
     234            0 :     let config_req = json_request::<TenantLocationConfigRequest>(&mut req).await?;
     235              :     json_response(
     236              :         StatusCode::OK,
     237            0 :         service
     238            0 :             .tenant_location_config(tenant_shard_id, config_req)
     239            0 :             .await?,
     240              :     )
     241            0 : }
     242              : 
     243            0 : async fn handle_tenant_config_patch(
     244            0 :     service: Arc<Service>,
     245            0 :     req: Request<Body>,
     246            0 : ) -> Result<Response<Body>, ApiError> {
     247            0 :     check_permissions(&req, Scope::PageServerApi)?;
     248              : 
     249            0 :     let mut req = match maybe_forward(req).await {
     250            0 :         ForwardOutcome::Forwarded(res) => {
     251            0 :             return res;
     252              :         }
     253            0 :         ForwardOutcome::NotForwarded(req) => req,
     254              :     };
     255              : 
     256            0 :     let config_req = json_request::<TenantConfigPatchRequest>(&mut req).await?;
     257              : 
     258              :     json_response(
     259              :         StatusCode::OK,
     260            0 :         service.tenant_config_patch(config_req).await?,
     261              :     )
     262            0 : }
     263              : 
     264            0 : async fn handle_tenant_config_set(
     265            0 :     service: Arc<Service>,
     266            0 :     req: Request<Body>,
     267            0 : ) -> Result<Response<Body>, ApiError> {
     268            0 :     check_permissions(&req, Scope::PageServerApi)?;
     269              : 
     270            0 :     let mut req = match maybe_forward(req).await {
     271            0 :         ForwardOutcome::Forwarded(res) => {
     272            0 :             return res;
     273              :         }
     274            0 :         ForwardOutcome::NotForwarded(req) => req,
     275              :     };
     276              : 
     277            0 :     let config_req = json_request::<TenantConfigRequest>(&mut req).await?;
     278              : 
     279            0 :     json_response(StatusCode::OK, service.tenant_config_set(config_req).await?)
     280            0 : }
     281              : 
     282            0 : async fn handle_tenant_config_get(
     283            0 :     service: Arc<Service>,
     284            0 :     req: Request<Body>,
     285            0 : ) -> Result<Response<Body>, ApiError> {
     286            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     287            0 :     check_permissions(&req, Scope::PageServerApi)?;
     288            0 :     maybe_rate_limit(&req, tenant_id).await;
     289              : 
     290            0 :     match maybe_forward(req).await {
     291            0 :         ForwardOutcome::Forwarded(res) => {
     292            0 :             return res;
     293              :         }
     294            0 :         ForwardOutcome::NotForwarded(_req) => {}
     295            0 :     };
     296            0 : 
     297            0 :     json_response(StatusCode::OK, service.tenant_config_get(tenant_id)?)
     298            0 : }
     299              : 
     300            0 : async fn handle_tenant_time_travel_remote_storage(
     301            0 :     service: Arc<Service>,
     302            0 :     req: Request<Body>,
     303            0 : ) -> Result<Response<Body>, ApiError> {
     304            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     305            0 :     check_permissions(&req, Scope::PageServerApi)?;
     306            0 :     maybe_rate_limit(&req, tenant_id).await;
     307              : 
     308            0 :     let mut req = match maybe_forward(req).await {
     309            0 :         ForwardOutcome::Forwarded(res) => {
     310            0 :             return res;
     311              :         }
     312            0 :         ForwardOutcome::NotForwarded(req) => req,
     313              :     };
     314              : 
     315            0 :     let time_travel_req = json_request::<TenantTimeTravelRequest>(&mut req).await?;
     316              : 
     317            0 :     let timestamp_raw = must_get_query_param(&req, "travel_to")?;
     318            0 :     let _timestamp = humantime::parse_rfc3339(&timestamp_raw).map_err(|_e| {
     319            0 :         ApiError::BadRequest(anyhow::anyhow!(
     320            0 :             "Invalid time for travel_to: {timestamp_raw:?}"
     321            0 :         ))
     322            0 :     })?;
     323              : 
     324            0 :     let done_if_after_raw = must_get_query_param(&req, "done_if_after")?;
     325            0 :     let _done_if_after = humantime::parse_rfc3339(&done_if_after_raw).map_err(|_e| {
     326            0 :         ApiError::BadRequest(anyhow::anyhow!(
     327            0 :             "Invalid time for done_if_after: {done_if_after_raw:?}"
     328            0 :         ))
     329            0 :     })?;
     330              : 
     331            0 :     service
     332            0 :         .tenant_time_travel_remote_storage(
     333            0 :             &time_travel_req,
     334            0 :             tenant_id,
     335            0 :             timestamp_raw,
     336            0 :             done_if_after_raw,
     337            0 :         )
     338            0 :         .await?;
     339            0 :     json_response(StatusCode::OK, ())
     340            0 : }
     341              : 
     342            0 : fn map_reqwest_hyper_status(status: reqwest::StatusCode) -> Result<hyper::StatusCode, ApiError> {
     343            0 :     hyper::StatusCode::from_u16(status.as_u16())
     344            0 :         .context("invalid status code")
     345            0 :         .map_err(ApiError::InternalServerError)
     346            0 : }
     347              : 
     348            0 : async fn handle_tenant_secondary_download(
     349            0 :     service: Arc<Service>,
     350            0 :     req: Request<Body>,
     351            0 : ) -> Result<Response<Body>, ApiError> {
     352            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     353            0 :     let wait = parse_query_param(&req, "wait_ms")?.map(Duration::from_millis);
     354            0 :     maybe_rate_limit(&req, tenant_id).await;
     355              : 
     356            0 :     match maybe_forward(req).await {
     357            0 :         ForwardOutcome::Forwarded(res) => {
     358            0 :             return res;
     359              :         }
     360            0 :         ForwardOutcome::NotForwarded(_req) => {}
     361              :     };
     362              : 
     363            0 :     let (status, progress) = service.tenant_secondary_download(tenant_id, wait).await?;
     364            0 :     json_response(map_reqwest_hyper_status(status)?, progress)
     365            0 : }
     366              : 
     367            0 : async fn handle_tenant_delete(
     368            0 :     service: Arc<Service>,
     369            0 :     req: Request<Body>,
     370            0 : ) -> Result<Response<Body>, ApiError> {
     371            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     372            0 :     check_permissions(&req, Scope::PageServerApi)?;
     373            0 :     maybe_rate_limit(&req, tenant_id).await;
     374              : 
     375            0 :     match maybe_forward(req).await {
     376            0 :         ForwardOutcome::Forwarded(res) => {
     377            0 :             return res;
     378              :         }
     379            0 :         ForwardOutcome::NotForwarded(_req) => {}
     380              :     };
     381              : 
     382            0 :     let status_code = service
     383            0 :         .tenant_delete(tenant_id)
     384            0 :         .await
     385            0 :         .and_then(map_reqwest_hyper_status)?;
     386              : 
     387            0 :     if status_code == StatusCode::NOT_FOUND {
     388              :         // The pageserver uses 404 for successful deletion, but we use 200
     389            0 :         json_response(StatusCode::OK, ())
     390              :     } else {
     391            0 :         json_response(status_code, ())
     392              :     }
     393            0 : }
     394              : 
     395            0 : async fn handle_tenant_timeline_create(
     396            0 :     service: Arc<Service>,
     397            0 :     req: Request<Body>,
     398            0 : ) -> Result<Response<Body>, ApiError> {
     399            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     400            0 :     check_permissions(&req, Scope::PageServerApi)?;
     401            0 :     maybe_rate_limit(&req, tenant_id).await;
     402              : 
     403            0 :     let mut req = match maybe_forward(req).await {
     404            0 :         ForwardOutcome::Forwarded(res) => {
     405            0 :             return res;
     406              :         }
     407            0 :         ForwardOutcome::NotForwarded(req) => req,
     408              :     };
     409              : 
     410            0 :     let create_req = json_request::<TimelineCreateRequest>(&mut req).await?;
     411              :     json_response(
     412              :         StatusCode::CREATED,
     413            0 :         service
     414            0 :             .tenant_timeline_create(tenant_id, create_req)
     415            0 :             .await?,
     416              :     )
     417            0 : }
     418              : 
     419            0 : async fn handle_tenant_timeline_delete(
     420            0 :     service: Arc<Service>,
     421            0 :     req: Request<Body>,
     422            0 : ) -> Result<Response<Body>, ApiError> {
     423            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     424            0 :     let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
     425              : 
     426            0 :     check_permissions(&req, Scope::PageServerApi)?;
     427            0 :     maybe_rate_limit(&req, tenant_id).await;
     428              : 
     429            0 :     match maybe_forward(req).await {
     430            0 :         ForwardOutcome::Forwarded(res) => {
     431            0 :             return res;
     432              :         }
     433            0 :         ForwardOutcome::NotForwarded(_req) => {}
     434              :     };
     435              : 
     436              :     // For timeline deletions, which both implement an "initially return 202, then 404 once
     437              :     // we're done" semantic, we wrap with a retry loop to expose a simpler API upstream.
     438            0 :     async fn deletion_wrapper<R, F>(service: Arc<Service>, f: F) -> Result<Response<Body>, ApiError>
     439            0 :     where
     440            0 :         R: std::future::Future<Output = Result<StatusCode, ApiError>> + Send + 'static,
     441            0 :         F: Fn(Arc<Service>) -> R + Send + Sync + 'static,
     442            0 :     {
     443              :         // On subsequent retries, wait longer.
     444              :         // Enable callers with a 25 second request timeout to reliably get a response
     445              :         const MAX_WAIT: Duration = Duration::from_secs(25);
     446              :         const MAX_RETRY_PERIOD: Duration = Duration::from_secs(5);
     447              : 
     448            0 :         let started_at = Instant::now();
     449            0 : 
     450            0 :         // To keep deletion reasonably snappy for small tenants, initially check after 1 second if deletion
     451            0 :         // completed.
     452            0 :         let mut retry_period = Duration::from_secs(1);
     453              : 
     454              :         loop {
     455            0 :             let status = f(service.clone()).await?;
     456            0 :             match status {
     457              :                 StatusCode::ACCEPTED => {
     458            0 :                     tracing::info!("Deletion accepted, waiting to try again...");
     459            0 :                     tokio::time::sleep(retry_period).await;
     460            0 :                     retry_period = MAX_RETRY_PERIOD;
     461              :                 }
     462              :                 StatusCode::CONFLICT => {
     463            0 :                     tracing::info!("Deletion already in progress, waiting to try again...");
     464            0 :                     tokio::time::sleep(retry_period).await;
     465              :                 }
     466              :                 StatusCode::NOT_FOUND => {
     467            0 :                     tracing::info!("Deletion complete");
     468            0 :                     return json_response(StatusCode::OK, ());
     469              :                 }
     470              :                 _ => {
     471            0 :                     tracing::warn!("Unexpected status {status}");
     472            0 :                     return json_response(status, ());
     473              :                 }
     474              :             }
     475              : 
     476            0 :             let now = Instant::now();
     477            0 :             if now + retry_period > started_at + MAX_WAIT {
     478            0 :                 tracing::info!("Deletion timed out waiting for 404");
     479              :                 // REQUEST_TIMEOUT would be more appropriate, but CONFLICT is already part of
     480              :                 // the pageserver's swagger definition for this endpoint, and has the same desired
     481              :                 // effect of causing the control plane to retry later.
     482            0 :                 return json_response(StatusCode::CONFLICT, ());
     483            0 :             }
     484              :         }
     485            0 :     }
     486              : 
     487            0 :     deletion_wrapper(service, move |service| async move {
     488            0 :         service
     489            0 :             .tenant_timeline_delete(tenant_id, timeline_id)
     490            0 :             .await
     491            0 :             .and_then(map_reqwest_hyper_status)
     492            0 :     })
     493            0 :     .await
     494            0 : }
     495              : 
     496            0 : async fn handle_tenant_timeline_archival_config(
     497            0 :     service: Arc<Service>,
     498            0 :     req: Request<Body>,
     499            0 : ) -> Result<Response<Body>, ApiError> {
     500            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     501            0 :     let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
     502              : 
     503            0 :     check_permissions(&req, Scope::PageServerApi)?;
     504            0 :     maybe_rate_limit(&req, tenant_id).await;
     505              : 
     506            0 :     let mut req = match maybe_forward(req).await {
     507            0 :         ForwardOutcome::Forwarded(res) => {
     508            0 :             return res;
     509              :         }
     510            0 :         ForwardOutcome::NotForwarded(req) => req,
     511              :     };
     512              : 
     513            0 :     let create_req = json_request::<TimelineArchivalConfigRequest>(&mut req).await?;
     514              : 
     515            0 :     service
     516            0 :         .tenant_timeline_archival_config(tenant_id, timeline_id, create_req)
     517            0 :         .await?;
     518              : 
     519            0 :     json_response(StatusCode::OK, ())
     520            0 : }
     521              : 
     522            0 : async fn handle_tenant_timeline_detach_ancestor(
     523            0 :     service: Arc<Service>,
     524            0 :     req: Request<Body>,
     525            0 : ) -> Result<Response<Body>, ApiError> {
     526            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     527            0 :     let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
     528              : 
     529            0 :     check_permissions(&req, Scope::PageServerApi)?;
     530            0 :     maybe_rate_limit(&req, tenant_id).await;
     531              : 
     532            0 :     match maybe_forward(req).await {
     533            0 :         ForwardOutcome::Forwarded(res) => {
     534            0 :             return res;
     535              :         }
     536            0 :         ForwardOutcome::NotForwarded(_req) => {}
     537              :     };
     538              : 
     539            0 :     let res = service
     540            0 :         .tenant_timeline_detach_ancestor(tenant_id, timeline_id)
     541            0 :         .await?;
     542              : 
     543            0 :     json_response(StatusCode::OK, res)
     544            0 : }
     545              : 
     546            0 : async fn handle_tenant_timeline_block_unblock_gc(
     547            0 :     service: Arc<Service>,
     548            0 :     req: Request<Body>,
     549            0 :     dir: BlockUnblock,
     550            0 : ) -> Result<Response<Body>, ApiError> {
     551            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     552            0 :     check_permissions(&req, Scope::PageServerApi)?;
     553            0 :     maybe_rate_limit(&req, tenant_id).await;
     554              : 
     555            0 :     let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
     556              : 
     557            0 :     service
     558            0 :         .tenant_timeline_block_unblock_gc(tenant_id, timeline_id, dir)
     559            0 :         .await?;
     560              : 
     561            0 :     json_response(StatusCode::OK, ())
     562            0 : }
     563              : 
     564            0 : async fn handle_tenant_timeline_download_heatmap_layers(
     565            0 :     service: Arc<Service>,
     566            0 :     req: Request<Body>,
     567            0 : ) -> Result<Response<Body>, ApiError> {
     568            0 :     let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
     569              : 
     570            0 :     check_permissions(&req, Scope::PageServerApi)?;
     571            0 :     maybe_rate_limit(&req, tenant_shard_id.tenant_id).await;
     572              : 
     573            0 :     let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
     574            0 :     let concurrency: Option<usize> = parse_query_param(&req, "concurrency")?;
     575            0 :     let recurse = parse_query_param(&req, "recurse")?.unwrap_or(false);
     576            0 : 
     577            0 :     service
     578            0 :         .tenant_timeline_download_heatmap_layers(tenant_shard_id, timeline_id, concurrency, recurse)
     579            0 :         .await?;
     580              : 
     581            0 :     json_response(StatusCode::OK, ())
     582            0 : }
     583              : 
     584              : // For metric labels where we would like to include the approximate path, but exclude high-cardinality fields like query parameters
     585              : // and tenant/timeline IDs.  Since we are proxying to arbitrary paths, we don't have routing templates to
     586              : // compare to, so we can just filter out our well known ID format with regexes.
     587            3 : fn path_without_ids(path: &str) -> String {
     588              :     static ID_REGEX: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
     589            3 :     ID_REGEX
     590            3 :         .get_or_init(|| regex::Regex::new(r"([0-9a-fA-F]{32}(-[0-9]{4})?|\?.*)").unwrap())
     591            3 :         .replace_all(path, "")
     592            3 :         .to_string()
     593            3 : }
     594              : 
     595            0 : async fn handle_tenant_timeline_passthrough(
     596            0 :     service: Arc<Service>,
     597            0 :     req: Request<Body>,
     598            0 : ) -> Result<Response<Body>, ApiError> {
     599            0 :     let tenant_or_shard_id: TenantShardId = parse_request_param(&req, "tenant_id")?;
     600            0 :     check_permissions(&req, Scope::PageServerApi)?;
     601            0 :     maybe_rate_limit(&req, tenant_or_shard_id.tenant_id).await;
     602              : 
     603            0 :     let req = match maybe_forward(req).await {
     604            0 :         ForwardOutcome::Forwarded(res) => {
     605            0 :             return res;
     606              :         }
     607            0 :         ForwardOutcome::NotForwarded(req) => req,
     608              :     };
     609              : 
     610            0 :     let Some(path) = req.uri().path_and_query() else {
     611              :         // This should never happen, our request router only calls us if there is a path
     612            0 :         return Err(ApiError::BadRequest(anyhow::anyhow!("Missing path")));
     613              :     };
     614              : 
     615            0 :     tracing::info!(
     616            0 :         "Proxying request for tenant {} ({})",
     617              :         tenant_or_shard_id.tenant_id,
     618              :         path
     619              :     );
     620              : 
     621              :     // Find the node that holds shard zero
     622            0 :     let (node, tenant_shard_id) = if tenant_or_shard_id.is_unsharded() {
     623            0 :         service
     624            0 :             .tenant_shard0_node(tenant_or_shard_id.tenant_id)
     625            0 :             .await?
     626              :     } else {
     627              :         (
     628            0 :             service.tenant_shard_node(tenant_or_shard_id).await?,
     629            0 :             tenant_or_shard_id,
     630              :         )
     631              :     };
     632              : 
     633              :     // Callers will always pass an unsharded tenant ID.  Before proxying, we must
     634              :     // rewrite this to a shard-aware shard zero ID.
     635            0 :     let path = format!("{}", path);
     636            0 :     let tenant_str = tenant_or_shard_id.tenant_id.to_string();
     637            0 :     let tenant_shard_str = format!("{}", tenant_shard_id);
     638            0 :     let path = path.replace(&tenant_str, &tenant_shard_str);
     639            0 : 
     640            0 :     let latency = &METRICS_REGISTRY
     641            0 :         .metrics_group
     642            0 :         .storage_controller_passthrough_request_latency;
     643            0 : 
     644            0 :     let path_label = path_without_ids(&path)
     645            0 :         .split('/')
     646            0 :         .filter(|token| !token.is_empty())
     647            0 :         .collect::<Vec<_>>()
     648            0 :         .join("_");
     649            0 :     let labels = PageserverRequestLabelGroup {
     650            0 :         pageserver_id: &node.get_id().to_string(),
     651            0 :         path: &path_label,
     652            0 :         method: crate::metrics::Method::Get,
     653            0 :     };
     654            0 : 
     655            0 :     let _timer = latency.start_timer(labels.clone());
     656              : 
     657            0 :     let client = mgmt_api::Client::new(
     658            0 :         node.base_url(),
     659            0 :         service.get_config().pageserver_jwt_token.as_deref(),
     660            0 :         service.get_config().ssl_ca_cert.clone(),
     661            0 :     )
     662            0 :     .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
     663            0 :     let resp = client.get_raw(path).await.map_err(|e|
     664              :         // We return 503 here because if we can't successfully send a request to the pageserver,
     665              :         // either we aren't available or the pageserver is unavailable.
     666            0 :         ApiError::ResourceUnavailable(format!("Error sending pageserver API request to {node}: {e}").into()))?;
     667              : 
     668            0 :     if !resp.status().is_success() {
     669            0 :         let error_counter = &METRICS_REGISTRY
     670            0 :             .metrics_group
     671            0 :             .storage_controller_passthrough_request_error;
     672            0 :         error_counter.inc(labels);
     673            0 :     }
     674              : 
     675              :     // Transform 404 into 503 if we raced with a migration
     676            0 :     if resp.status() == reqwest::StatusCode::NOT_FOUND {
     677              :         // Look up node again: if we migrated it will be different
     678            0 :         let new_node = service.tenant_shard_node(tenant_shard_id).await?;
     679            0 :         if new_node.get_id() != node.get_id() {
     680              :             // Rather than retry here, send the client a 503 to prompt a retry: this matches
     681              :             // the pageserver's use of 503, and all clients calling this API should retry on 503.
     682            0 :             return Err(ApiError::ResourceUnavailable(
     683            0 :                 format!("Pageserver {node} returned 404, was migrated to {new_node}").into(),
     684            0 :             ));
     685            0 :         }
     686            0 :     }
     687              : 
     688              :     // We have a reqest::Response, would like a http::Response
     689            0 :     let mut builder = hyper::Response::builder().status(map_reqwest_hyper_status(resp.status())?);
     690            0 :     for (k, v) in resp.headers() {
     691            0 :         builder = builder.header(k.as_str(), v.as_bytes());
     692            0 :     }
     693              : 
     694            0 :     let response = builder
     695            0 :         .body(Body::wrap_stream(resp.bytes_stream()))
     696            0 :         .map_err(|e| ApiError::InternalServerError(e.into()))?;
     697              : 
     698            0 :     Ok(response)
     699            0 : }
     700              : 
     701            0 : async fn handle_tenant_locate(
     702            0 :     service: Arc<Service>,
     703            0 :     req: Request<Body>,
     704            0 : ) -> Result<Response<Body>, ApiError> {
     705            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     706              : 
     707            0 :     check_permissions(&req, Scope::Admin)?;
     708              :     // NB: don't rate limit: admin operation.
     709              : 
     710            0 :     match maybe_forward(req).await {
     711            0 :         ForwardOutcome::Forwarded(res) => {
     712            0 :             return res;
     713              :         }
     714            0 :         ForwardOutcome::NotForwarded(_req) => {}
     715            0 :     };
     716            0 : 
     717            0 :     json_response(StatusCode::OK, service.tenant_locate(tenant_id)?)
     718            0 : }
     719              : 
     720            0 : async fn handle_tenant_describe(
     721            0 :     service: Arc<Service>,
     722            0 :     req: Request<Body>,
     723            0 : ) -> Result<Response<Body>, ApiError> {
     724            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     725            0 :     check_permissions(&req, Scope::Scrubber)?;
     726              :     // NB: don't rate limit: scrubber operation.
     727              : 
     728            0 :     match maybe_forward(req).await {
     729            0 :         ForwardOutcome::Forwarded(res) => {
     730            0 :             return res;
     731              :         }
     732            0 :         ForwardOutcome::NotForwarded(_req) => {}
     733            0 :     };
     734            0 : 
     735            0 :     json_response(StatusCode::OK, service.tenant_describe(tenant_id)?)
     736            0 : }
     737              : 
     738            0 : async fn handle_tenant_list(
     739            0 :     service: Arc<Service>,
     740            0 :     req: Request<Body>,
     741            0 : ) -> Result<Response<Body>, ApiError> {
     742            0 :     check_permissions(&req, Scope::Admin)?;
     743              : 
     744            0 :     let limit: Option<usize> = parse_query_param(&req, "limit")?;
     745            0 :     let start_after: Option<TenantId> = parse_query_param(&req, "start_after")?;
     746            0 :     tracing::info!("start_after: {:?}", start_after);
     747              : 
     748            0 :     match maybe_forward(req).await {
     749            0 :         ForwardOutcome::Forwarded(res) => {
     750            0 :             return res;
     751              :         }
     752            0 :         ForwardOutcome::NotForwarded(_req) => {}
     753            0 :     };
     754            0 : 
     755            0 :     json_response(StatusCode::OK, service.tenant_list(limit, start_after))
     756            0 : }
     757              : 
     758            0 : async fn handle_node_register(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     759            0 :     check_permissions(&req, Scope::Infra)?;
     760              : 
     761            0 :     let mut req = match maybe_forward(req).await {
     762            0 :         ForwardOutcome::Forwarded(res) => {
     763            0 :             return res;
     764              :         }
     765            0 :         ForwardOutcome::NotForwarded(req) => req,
     766              :     };
     767              : 
     768            0 :     let register_req = json_request::<NodeRegisterRequest>(&mut req).await?;
     769            0 :     let state = get_state(&req);
     770            0 :     state.service.node_register(register_req).await?;
     771            0 :     json_response(StatusCode::OK, ())
     772            0 : }
     773              : 
     774            0 : async fn handle_node_list(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     775            0 :     check_permissions(&req, Scope::Infra)?;
     776              : 
     777            0 :     let req = match maybe_forward(req).await {
     778            0 :         ForwardOutcome::Forwarded(res) => {
     779            0 :             return res;
     780              :         }
     781            0 :         ForwardOutcome::NotForwarded(req) => req,
     782            0 :     };
     783            0 : 
     784            0 :     let state = get_state(&req);
     785            0 :     let mut nodes = state.service.node_list().await?;
     786            0 :     nodes.sort_by_key(|n| n.get_id());
     787            0 :     let api_nodes = nodes.into_iter().map(|n| n.describe()).collect::<Vec<_>>();
     788            0 : 
     789            0 :     json_response(StatusCode::OK, api_nodes)
     790            0 : }
     791              : 
     792            0 : async fn handle_node_drop(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     793            0 :     check_permissions(&req, Scope::Admin)?;
     794              : 
     795            0 :     let req = match maybe_forward(req).await {
     796            0 :         ForwardOutcome::Forwarded(res) => {
     797            0 :             return res;
     798              :         }
     799            0 :         ForwardOutcome::NotForwarded(req) => req,
     800            0 :     };
     801            0 : 
     802            0 :     let state = get_state(&req);
     803            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     804            0 :     json_response(StatusCode::OK, state.service.node_drop(node_id).await?)
     805            0 : }
     806              : 
     807            0 : async fn handle_node_delete(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     808            0 :     check_permissions(&req, Scope::Admin)?;
     809              : 
     810            0 :     let req = match maybe_forward(req).await {
     811            0 :         ForwardOutcome::Forwarded(res) => {
     812            0 :             return res;
     813              :         }
     814            0 :         ForwardOutcome::NotForwarded(req) => req,
     815            0 :     };
     816            0 : 
     817            0 :     let state = get_state(&req);
     818            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     819            0 :     json_response(StatusCode::OK, state.service.node_delete(node_id).await?)
     820            0 : }
     821              : 
     822            0 : async fn handle_node_configure(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     823            0 :     check_permissions(&req, Scope::Admin)?;
     824              : 
     825            0 :     let mut req = match maybe_forward(req).await {
     826            0 :         ForwardOutcome::Forwarded(res) => {
     827            0 :             return res;
     828              :         }
     829            0 :         ForwardOutcome::NotForwarded(req) => req,
     830              :     };
     831              : 
     832            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     833            0 :     let config_req = json_request::<NodeConfigureRequest>(&mut req).await?;
     834            0 :     if node_id != config_req.node_id {
     835            0 :         return Err(ApiError::BadRequest(anyhow::anyhow!(
     836            0 :             "Path and body node_id differ"
     837            0 :         )));
     838            0 :     }
     839            0 :     let state = get_state(&req);
     840            0 : 
     841            0 :     json_response(
     842            0 :         StatusCode::OK,
     843            0 :         state
     844            0 :             .service
     845            0 :             .external_node_configure(
     846            0 :                 config_req.node_id,
     847            0 :                 config_req.availability.map(NodeAvailability::from),
     848            0 :                 config_req.scheduling,
     849            0 :             )
     850            0 :             .await?,
     851              :     )
     852            0 : }
     853              : 
     854            0 : async fn handle_node_status(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     855            0 :     check_permissions(&req, Scope::Infra)?;
     856              : 
     857            0 :     let req = match maybe_forward(req).await {
     858            0 :         ForwardOutcome::Forwarded(res) => {
     859            0 :             return res;
     860              :         }
     861            0 :         ForwardOutcome::NotForwarded(req) => req,
     862            0 :     };
     863            0 : 
     864            0 :     let state = get_state(&req);
     865            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     866              : 
     867            0 :     let node_status = state.service.get_node(node_id).await?;
     868              : 
     869            0 :     json_response(StatusCode::OK, node_status)
     870            0 : }
     871              : 
     872            0 : async fn handle_node_shards(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     873            0 :     check_permissions(&req, Scope::Admin)?;
     874              : 
     875            0 :     let state = get_state(&req);
     876            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     877              : 
     878            0 :     let node_status = state.service.get_node_shards(node_id).await?;
     879              : 
     880            0 :     json_response(StatusCode::OK, node_status)
     881            0 : }
     882              : 
     883            0 : async fn handle_get_leader(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     884            0 :     check_permissions(&req, Scope::Admin)?;
     885              : 
     886            0 :     let req = match maybe_forward(req).await {
     887            0 :         ForwardOutcome::Forwarded(res) => {
     888            0 :             return res;
     889              :         }
     890            0 :         ForwardOutcome::NotForwarded(req) => req,
     891            0 :     };
     892            0 : 
     893            0 :     let state = get_state(&req);
     894            0 :     let leader = state.service.get_leader().await.map_err(|err| {
     895            0 :         ApiError::InternalServerError(anyhow::anyhow!(
     896            0 :             "Failed to read leader from database: {err}"
     897            0 :         ))
     898            0 :     })?;
     899              : 
     900            0 :     json_response(StatusCode::OK, leader)
     901            0 : }
     902              : 
     903            0 : async fn handle_node_drain(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     904            0 :     check_permissions(&req, Scope::Infra)?;
     905              : 
     906            0 :     let req = match maybe_forward(req).await {
     907            0 :         ForwardOutcome::Forwarded(res) => {
     908            0 :             return res;
     909              :         }
     910            0 :         ForwardOutcome::NotForwarded(req) => req,
     911            0 :     };
     912            0 : 
     913            0 :     let state = get_state(&req);
     914            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     915              : 
     916            0 :     state.service.start_node_drain(node_id).await?;
     917              : 
     918            0 :     json_response(StatusCode::ACCEPTED, ())
     919            0 : }
     920              : 
     921            0 : async fn handle_cancel_node_drain(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     922            0 :     check_permissions(&req, Scope::Infra)?;
     923              : 
     924            0 :     let req = match maybe_forward(req).await {
     925            0 :         ForwardOutcome::Forwarded(res) => {
     926            0 :             return res;
     927              :         }
     928            0 :         ForwardOutcome::NotForwarded(req) => req,
     929            0 :     };
     930            0 : 
     931            0 :     let state = get_state(&req);
     932            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     933              : 
     934            0 :     state.service.cancel_node_drain(node_id).await?;
     935              : 
     936            0 :     json_response(StatusCode::ACCEPTED, ())
     937            0 : }
     938              : 
     939            0 : async fn handle_node_fill(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     940            0 :     check_permissions(&req, Scope::Infra)?;
     941              : 
     942            0 :     let req = match maybe_forward(req).await {
     943            0 :         ForwardOutcome::Forwarded(res) => {
     944            0 :             return res;
     945              :         }
     946            0 :         ForwardOutcome::NotForwarded(req) => req,
     947            0 :     };
     948            0 : 
     949            0 :     let state = get_state(&req);
     950            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     951              : 
     952            0 :     state.service.start_node_fill(node_id).await?;
     953              : 
     954            0 :     json_response(StatusCode::ACCEPTED, ())
     955            0 : }
     956              : 
     957            0 : async fn handle_cancel_node_fill(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     958            0 :     check_permissions(&req, Scope::Infra)?;
     959              : 
     960            0 :     let req = match maybe_forward(req).await {
     961            0 :         ForwardOutcome::Forwarded(res) => {
     962            0 :             return res;
     963              :         }
     964            0 :         ForwardOutcome::NotForwarded(req) => req,
     965            0 :     };
     966            0 : 
     967            0 :     let state = get_state(&req);
     968            0 :     let node_id: NodeId = parse_request_param(&req, "node_id")?;
     969              : 
     970            0 :     state.service.cancel_node_fill(node_id).await?;
     971              : 
     972            0 :     json_response(StatusCode::ACCEPTED, ())
     973            0 : }
     974              : 
     975            0 : async fn handle_safekeeper_list(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     976            0 :     check_permissions(&req, Scope::Infra)?;
     977              : 
     978            0 :     let req = match maybe_forward(req).await {
     979            0 :         ForwardOutcome::Forwarded(res) => {
     980            0 :             return res;
     981              :         }
     982            0 :         ForwardOutcome::NotForwarded(req) => req,
     983            0 :     };
     984            0 : 
     985            0 :     let state = get_state(&req);
     986            0 :     let safekeepers = state.service.safekeepers_list().await?;
     987            0 :     json_response(StatusCode::OK, safekeepers)
     988            0 : }
     989              : 
     990            0 : async fn handle_metadata_health_update(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     991            0 :     check_permissions(&req, Scope::Scrubber)?;
     992              : 
     993            0 :     let mut req = match maybe_forward(req).await {
     994            0 :         ForwardOutcome::Forwarded(res) => {
     995            0 :             return res;
     996              :         }
     997            0 :         ForwardOutcome::NotForwarded(req) => req,
     998              :     };
     999              : 
    1000            0 :     let update_req = json_request::<MetadataHealthUpdateRequest>(&mut req).await?;
    1001            0 :     let state = get_state(&req);
    1002            0 : 
    1003            0 :     state.service.metadata_health_update(update_req).await?;
    1004              : 
    1005            0 :     json_response(StatusCode::OK, MetadataHealthUpdateResponse {})
    1006            0 : }
    1007              : 
    1008            0 : async fn handle_metadata_health_list_unhealthy(
    1009            0 :     req: Request<Body>,
    1010            0 : ) -> Result<Response<Body>, ApiError> {
    1011            0 :     check_permissions(&req, Scope::Admin)?;
    1012              : 
    1013            0 :     let req = match maybe_forward(req).await {
    1014            0 :         ForwardOutcome::Forwarded(res) => {
    1015            0 :             return res;
    1016              :         }
    1017            0 :         ForwardOutcome::NotForwarded(req) => req,
    1018            0 :     };
    1019            0 : 
    1020            0 :     let state = get_state(&req);
    1021            0 :     let unhealthy_tenant_shards = state.service.metadata_health_list_unhealthy().await?;
    1022              : 
    1023            0 :     json_response(
    1024            0 :         StatusCode::OK,
    1025            0 :         MetadataHealthListUnhealthyResponse {
    1026            0 :             unhealthy_tenant_shards,
    1027            0 :         },
    1028            0 :     )
    1029            0 : }
    1030              : 
    1031            0 : async fn handle_metadata_health_list_outdated(
    1032            0 :     req: Request<Body>,
    1033            0 : ) -> Result<Response<Body>, ApiError> {
    1034            0 :     check_permissions(&req, Scope::Admin)?;
    1035              : 
    1036            0 :     let mut req = match maybe_forward(req).await {
    1037            0 :         ForwardOutcome::Forwarded(res) => {
    1038            0 :             return res;
    1039              :         }
    1040            0 :         ForwardOutcome::NotForwarded(req) => req,
    1041              :     };
    1042              : 
    1043            0 :     let list_outdated_req = json_request::<MetadataHealthListOutdatedRequest>(&mut req).await?;
    1044            0 :     let state = get_state(&req);
    1045            0 :     let health_records = state
    1046            0 :         .service
    1047            0 :         .metadata_health_list_outdated(list_outdated_req.not_scrubbed_for)
    1048            0 :         .await?;
    1049              : 
    1050            0 :     json_response(
    1051            0 :         StatusCode::OK,
    1052            0 :         MetadataHealthListOutdatedResponse { health_records },
    1053            0 :     )
    1054            0 : }
    1055              : 
    1056            0 : async fn handle_tenant_shard_split(
    1057            0 :     service: Arc<Service>,
    1058            0 :     req: Request<Body>,
    1059            0 : ) -> Result<Response<Body>, ApiError> {
    1060            0 :     check_permissions(&req, Scope::Admin)?;
    1061              :     // NB: don't rate limit: admin operation.
    1062              : 
    1063            0 :     let mut req = match maybe_forward(req).await {
    1064            0 :         ForwardOutcome::Forwarded(res) => {
    1065            0 :             return res;
    1066              :         }
    1067            0 :         ForwardOutcome::NotForwarded(req) => req,
    1068              :     };
    1069              : 
    1070            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
    1071            0 :     let split_req = json_request::<TenantShardSplitRequest>(&mut req).await?;
    1072              : 
    1073              :     json_response(
    1074              :         StatusCode::OK,
    1075            0 :         service.tenant_shard_split(tenant_id, split_req).await?,
    1076              :     )
    1077            0 : }
    1078              : 
    1079            0 : async fn handle_tenant_shard_migrate(
    1080            0 :     service: Arc<Service>,
    1081            0 :     req: Request<Body>,
    1082            0 : ) -> Result<Response<Body>, ApiError> {
    1083            0 :     check_permissions(&req, Scope::Admin)?;
    1084              :     // NB: don't rate limit: admin operation.
    1085              : 
    1086            0 :     let mut req = match maybe_forward(req).await {
    1087            0 :         ForwardOutcome::Forwarded(res) => {
    1088            0 :             return res;
    1089              :         }
    1090            0 :         ForwardOutcome::NotForwarded(req) => req,
    1091              :     };
    1092              : 
    1093            0 :     let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
    1094            0 :     let migrate_req = json_request::<TenantShardMigrateRequest>(&mut req).await?;
    1095              :     json_response(
    1096              :         StatusCode::OK,
    1097            0 :         service
    1098            0 :             .tenant_shard_migrate(tenant_shard_id, migrate_req)
    1099            0 :             .await?,
    1100              :     )
    1101            0 : }
    1102              : 
    1103            0 : async fn handle_tenant_shard_migrate_secondary(
    1104            0 :     service: Arc<Service>,
    1105            0 :     req: Request<Body>,
    1106            0 : ) -> Result<Response<Body>, ApiError> {
    1107            0 :     check_permissions(&req, Scope::Admin)?;
    1108              :     // NB: don't rate limit: admin operation.
    1109              : 
    1110            0 :     let mut req = match maybe_forward(req).await {
    1111            0 :         ForwardOutcome::Forwarded(res) => {
    1112            0 :             return res;
    1113              :         }
    1114            0 :         ForwardOutcome::NotForwarded(req) => req,
    1115              :     };
    1116              : 
    1117            0 :     let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
    1118            0 :     let migrate_req = json_request::<TenantShardMigrateRequest>(&mut req).await?;
    1119              :     json_response(
    1120              :         StatusCode::OK,
    1121            0 :         service
    1122            0 :             .tenant_shard_migrate_secondary(tenant_shard_id, migrate_req)
    1123            0 :             .await?,
    1124              :     )
    1125            0 : }
    1126              : 
    1127            0 : async fn handle_tenant_shard_cancel_reconcile(
    1128            0 :     service: Arc<Service>,
    1129            0 :     req: Request<Body>,
    1130            0 : ) -> Result<Response<Body>, ApiError> {
    1131            0 :     check_permissions(&req, Scope::Admin)?;
    1132              :     // NB: don't rate limit: admin operation.
    1133              : 
    1134            0 :     let req = match maybe_forward(req).await {
    1135            0 :         ForwardOutcome::Forwarded(res) => {
    1136            0 :             return res;
    1137              :         }
    1138            0 :         ForwardOutcome::NotForwarded(req) => req,
    1139              :     };
    1140              : 
    1141            0 :     let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
    1142              :     json_response(
    1143              :         StatusCode::OK,
    1144            0 :         service
    1145            0 :             .tenant_shard_cancel_reconcile(tenant_shard_id)
    1146            0 :             .await?,
    1147              :     )
    1148            0 : }
    1149              : 
    1150            0 : async fn handle_tenant_update_policy(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1151            0 :     check_permissions(&req, Scope::Admin)?;
    1152              :     // NB: don't rate limit: admin operation.
    1153              : 
    1154            0 :     let mut req = match maybe_forward(req).await {
    1155            0 :         ForwardOutcome::Forwarded(res) => {
    1156            0 :             return res;
    1157              :         }
    1158            0 :         ForwardOutcome::NotForwarded(req) => req,
    1159              :     };
    1160              : 
    1161            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
    1162            0 :     let update_req = json_request::<TenantPolicyRequest>(&mut req).await?;
    1163            0 :     let state = get_state(&req);
    1164            0 : 
    1165            0 :     json_response(
    1166            0 :         StatusCode::OK,
    1167            0 :         state
    1168            0 :             .service
    1169            0 :             .tenant_update_policy(tenant_id, update_req)
    1170            0 :             .await?,
    1171              :     )
    1172            0 : }
    1173              : 
    1174            0 : async fn handle_update_preferred_azs(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1175            0 :     check_permissions(&req, Scope::Admin)?;
    1176              : 
    1177            0 :     let mut req = match maybe_forward(req).await {
    1178            0 :         ForwardOutcome::Forwarded(res) => {
    1179            0 :             return res;
    1180              :         }
    1181            0 :         ForwardOutcome::NotForwarded(req) => req,
    1182              :     };
    1183              : 
    1184            0 :     let azs_req = json_request::<ShardsPreferredAzsRequest>(&mut req).await?;
    1185            0 :     let state = get_state(&req);
    1186            0 : 
    1187            0 :     json_response(
    1188            0 :         StatusCode::OK,
    1189            0 :         state.service.update_shards_preferred_azs(azs_req).await?,
    1190              :     )
    1191            0 : }
    1192              : 
    1193            0 : async fn handle_step_down(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1194            0 :     check_permissions(&req, Scope::ControllerPeer)?;
    1195              : 
    1196            0 :     let req = match maybe_forward(req).await {
    1197            0 :         ForwardOutcome::Forwarded(res) => {
    1198            0 :             return res;
    1199              :         }
    1200            0 :         ForwardOutcome::NotForwarded(req) => req,
    1201            0 :     };
    1202            0 : 
    1203            0 :     let state = get_state(&req);
    1204            0 :     json_response(StatusCode::OK, state.service.step_down().await)
    1205            0 : }
    1206              : 
    1207            0 : async fn handle_tenant_drop(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1208            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
    1209            0 :     check_permissions(&req, Scope::PageServerApi)?;
    1210            0 :     maybe_rate_limit(&req, tenant_id).await;
    1211              : 
    1212            0 :     let req = match maybe_forward(req).await {
    1213            0 :         ForwardOutcome::Forwarded(res) => {
    1214            0 :             return res;
    1215              :         }
    1216            0 :         ForwardOutcome::NotForwarded(req) => req,
    1217            0 :     };
    1218            0 : 
    1219            0 :     let state = get_state(&req);
    1220            0 : 
    1221            0 :     json_response(StatusCode::OK, state.service.tenant_drop(tenant_id).await?)
    1222            0 : }
    1223              : 
    1224            0 : async fn handle_tenant_import(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1225            0 :     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
    1226            0 :     check_permissions(&req, Scope::PageServerApi)?;
    1227            0 :     maybe_rate_limit(&req, tenant_id).await;
    1228              : 
    1229            0 :     let req = match maybe_forward(req).await {
    1230            0 :         ForwardOutcome::Forwarded(res) => {
    1231            0 :             return res;
    1232              :         }
    1233            0 :         ForwardOutcome::NotForwarded(req) => req,
    1234            0 :     };
    1235            0 : 
    1236            0 :     let state = get_state(&req);
    1237            0 : 
    1238            0 :     json_response(
    1239            0 :         StatusCode::OK,
    1240            0 :         state.service.tenant_import(tenant_id).await?,
    1241              :     )
    1242            0 : }
    1243              : 
    1244            0 : async fn handle_tenants_dump(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1245            0 :     check_permissions(&req, Scope::Admin)?;
    1246              : 
    1247            0 :     let req = match maybe_forward(req).await {
    1248            0 :         ForwardOutcome::Forwarded(res) => {
    1249            0 :             return res;
    1250              :         }
    1251            0 :         ForwardOutcome::NotForwarded(req) => req,
    1252            0 :     };
    1253            0 : 
    1254            0 :     let state = get_state(&req);
    1255            0 :     state.service.tenants_dump()
    1256            0 : }
    1257              : 
    1258            0 : async fn handle_scheduler_dump(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1259            0 :     check_permissions(&req, Scope::Admin)?;
    1260              : 
    1261            0 :     let req = match maybe_forward(req).await {
    1262            0 :         ForwardOutcome::Forwarded(res) => {
    1263            0 :             return res;
    1264              :         }
    1265            0 :         ForwardOutcome::NotForwarded(req) => req,
    1266            0 :     };
    1267            0 : 
    1268            0 :     let state = get_state(&req);
    1269            0 :     state.service.scheduler_dump()
    1270            0 : }
    1271              : 
    1272            0 : async fn handle_consistency_check(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1273            0 :     check_permissions(&req, Scope::Admin)?;
    1274              : 
    1275            0 :     let req = match maybe_forward(req).await {
    1276            0 :         ForwardOutcome::Forwarded(res) => {
    1277            0 :             return res;
    1278              :         }
    1279            0 :         ForwardOutcome::NotForwarded(req) => req,
    1280            0 :     };
    1281            0 : 
    1282            0 :     let state = get_state(&req);
    1283            0 : 
    1284            0 :     json_response(StatusCode::OK, state.service.consistency_check().await?)
    1285            0 : }
    1286              : 
    1287            0 : async fn handle_reconcile_all(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1288            0 :     check_permissions(&req, Scope::Admin)?;
    1289              : 
    1290            0 :     let req = match maybe_forward(req).await {
    1291            0 :         ForwardOutcome::Forwarded(res) => {
    1292            0 :             return res;
    1293              :         }
    1294            0 :         ForwardOutcome::NotForwarded(req) => req,
    1295            0 :     };
    1296            0 : 
    1297            0 :     let state = get_state(&req);
    1298            0 : 
    1299            0 :     json_response(StatusCode::OK, state.service.reconcile_all_now().await?)
    1300            0 : }
    1301              : 
    1302              : /// Status endpoint is just used for checking that our HTTP listener is up
    1303            0 : async fn handle_status(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1304            0 :     match maybe_forward(req).await {
    1305            0 :         ForwardOutcome::Forwarded(res) => {
    1306            0 :             return res;
    1307              :         }
    1308            0 :         ForwardOutcome::NotForwarded(_req) => {}
    1309            0 :     };
    1310            0 : 
    1311            0 :     json_response(StatusCode::OK, ())
    1312            0 : }
    1313              : 
    1314              : /// Readiness endpoint indicates when we're done doing startup I/O (e.g. reconciling
    1315              : /// with remote pageserver nodes).  This is intended for use as a kubernetes readiness probe.
    1316            0 : async fn handle_ready(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1317            0 :     let req = match maybe_forward(req).await {
    1318            0 :         ForwardOutcome::Forwarded(res) => {
    1319            0 :             return res;
    1320              :         }
    1321            0 :         ForwardOutcome::NotForwarded(req) => req,
    1322            0 :     };
    1323            0 : 
    1324            0 :     let state = get_state(&req);
    1325            0 :     if state.service.startup_complete.is_ready() {
    1326            0 :         json_response(StatusCode::OK, ())
    1327              :     } else {
    1328            0 :         json_response(StatusCode::SERVICE_UNAVAILABLE, ())
    1329              :     }
    1330            0 : }
    1331              : 
    1332              : impl From<ReconcileError> for ApiError {
    1333            0 :     fn from(value: ReconcileError) -> Self {
    1334            0 :         ApiError::Conflict(format!("Reconciliation error: {}", value))
    1335            0 :     }
    1336              : }
    1337              : 
    1338              : /// Return the safekeeper record by instance id, or 404.
    1339              : ///
    1340              : /// Not used by anything except manual testing.
    1341            0 : async fn handle_get_safekeeper(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1342            0 :     check_permissions(&req, Scope::Infra)?;
    1343              : 
    1344            0 :     let id = parse_request_param::<i64>(&req, "id")?;
    1345              : 
    1346            0 :     let req = match maybe_forward(req).await {
    1347            0 :         ForwardOutcome::Forwarded(res) => {
    1348            0 :             return res;
    1349              :         }
    1350            0 :         ForwardOutcome::NotForwarded(req) => req,
    1351            0 :     };
    1352            0 : 
    1353            0 :     let state = get_state(&req);
    1354              : 
    1355            0 :     let res = state.service.get_safekeeper(id).await;
    1356              : 
    1357            0 :     match res {
    1358            0 :         Ok(b) => json_response(StatusCode::OK, b),
    1359              :         Err(crate::persistence::DatabaseError::Query(diesel::result::Error::NotFound)) => {
    1360            0 :             Err(ApiError::NotFound("unknown instance id".into()))
    1361              :         }
    1362            0 :         Err(other) => Err(other.into()),
    1363              :     }
    1364            0 : }
    1365              : 
    1366              : /// Used as part of deployment scripts.
    1367              : ///
    1368              : /// Assumes information is only relayed to storage controller after first selecting an unique id on
    1369              : /// control plane database, which means we have an id field in the request and payload.
    1370            0 : async fn handle_upsert_safekeeper(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1371            0 :     check_permissions(&req, Scope::Infra)?;
    1372              : 
    1373            0 :     let body = json_request::<SafekeeperUpsert>(&mut req).await?;
    1374            0 :     let id = parse_request_param::<i64>(&req, "id")?;
    1375              : 
    1376            0 :     if id != body.id {
    1377              :         // it should be repeated
    1378            0 :         return Err(ApiError::BadRequest(anyhow::anyhow!(
    1379            0 :             "id mismatch: url={id:?}, body={:?}",
    1380            0 :             body.id
    1381            0 :         )));
    1382            0 :     }
    1383              : 
    1384            0 :     let req = match maybe_forward(req).await {
    1385            0 :         ForwardOutcome::Forwarded(res) => {
    1386            0 :             return res;
    1387              :         }
    1388            0 :         ForwardOutcome::NotForwarded(req) => req,
    1389            0 :     };
    1390            0 : 
    1391            0 :     let state = get_state(&req);
    1392            0 : 
    1393            0 :     state.service.upsert_safekeeper(body).await?;
    1394              : 
    1395            0 :     Ok(Response::builder()
    1396            0 :         .status(StatusCode::NO_CONTENT)
    1397            0 :         .body(Body::empty())
    1398            0 :         .unwrap())
    1399            0 : }
    1400              : 
    1401              : /// Sets the scheduling policy of the specified safekeeper
    1402            0 : async fn handle_safekeeper_scheduling_policy(
    1403            0 :     mut req: Request<Body>,
    1404            0 : ) -> Result<Response<Body>, ApiError> {
    1405            0 :     check_permissions(&req, Scope::Admin)?;
    1406              : 
    1407            0 :     let body = json_request::<SafekeeperSchedulingPolicyRequest>(&mut req).await?;
    1408            0 :     let id = parse_request_param::<i64>(&req, "id")?;
    1409              : 
    1410            0 :     let req = match maybe_forward(req).await {
    1411            0 :         ForwardOutcome::Forwarded(res) => {
    1412            0 :             return res;
    1413              :         }
    1414            0 :         ForwardOutcome::NotForwarded(req) => req,
    1415            0 :     };
    1416            0 : 
    1417            0 :     let state = get_state(&req);
    1418            0 : 
    1419            0 :     state
    1420            0 :         .service
    1421            0 :         .set_safekeeper_scheduling_policy(id, body.scheduling_policy)
    1422            0 :         .await?;
    1423              : 
    1424            0 :     json_response(StatusCode::OK, ())
    1425            0 : }
    1426              : 
    1427              : /// Common wrapper for request handlers that call into Service and will operate on tenants: they must only
    1428              : /// be allowed to run if Service has finished its initial reconciliation.
    1429            0 : async fn tenant_service_handler<R, H>(
    1430            0 :     request: Request<Body>,
    1431            0 :     handler: H,
    1432            0 :     request_name: RequestName,
    1433            0 : ) -> R::Output
    1434            0 : where
    1435            0 :     R: std::future::Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
    1436            0 :     H: FnOnce(Arc<Service>, Request<Body>) -> R + Send + Sync + 'static,
    1437            0 : {
    1438            0 :     let state = get_state(&request);
    1439            0 :     let service = state.service.clone();
    1440            0 : 
    1441            0 :     let startup_complete = service.startup_complete.clone();
    1442            0 :     if tokio::time::timeout(STARTUP_RECONCILE_TIMEOUT, startup_complete.wait())
    1443            0 :         .await
    1444            0 :         .is_err()
    1445              :     {
    1446              :         // This shouldn't happen: it is the responsibilty of [`Service::startup_reconcile`] to use appropriate
    1447              :         // timeouts around its remote calls, to bound its runtime.
    1448            0 :         return Err(ApiError::Timeout(
    1449            0 :             "Timed out waiting for service readiness".into(),
    1450            0 :         ));
    1451            0 :     }
    1452            0 : 
    1453            0 :     named_request_span(
    1454            0 :         request,
    1455            0 :         |request| async move { handler(service, request).await },
    1456            0 :         request_name,
    1457            0 :     )
    1458            0 :     .await
    1459            0 : }
    1460              : 
    1461              : /// Check if the required scope is held in the request's token, or if the request has
    1462              : /// a token with 'admin' scope then always permit it.
    1463            0 : fn check_permissions(request: &Request<Body>, required_scope: Scope) -> Result<(), ApiError> {
    1464            0 :     check_permission_with(request, |claims| {
    1465            0 :         match crate::auth::check_permission(claims, required_scope) {
    1466            0 :             Err(e) => match crate::auth::check_permission(claims, Scope::Admin) {
    1467            0 :                 Ok(()) => Ok(()),
    1468            0 :                 Err(_) => Err(e),
    1469              :             },
    1470            0 :             Ok(()) => Ok(()),
    1471              :         }
    1472            0 :     })
    1473            0 : }
    1474              : 
    1475              : #[derive(Clone, Debug)]
    1476              : struct RequestMeta {
    1477              :     method: hyper::http::Method,
    1478              :     at: Instant,
    1479              : }
    1480              : 
    1481            0 : pub fn prologue_leadership_status_check_middleware<
    1482            0 :     B: hyper::body::HttpBody + Send + Sync + 'static,
    1483            0 : >() -> Middleware<B, ApiError> {
    1484            0 :     Middleware::pre(move |req| async move {
    1485            0 :         let state = get_state(&req);
    1486            0 :         let leadership_status = state.service.get_leadership_status();
    1487              : 
    1488              :         enum AllowedRoutes {
    1489              :             All,
    1490              :             Some(&'static [&'static str]),
    1491              :         }
    1492              : 
    1493            0 :         let allowed_routes = match leadership_status {
    1494            0 :             LeadershipStatus::Leader => AllowedRoutes::All,
    1495            0 :             LeadershipStatus::SteppedDown => AllowedRoutes::All,
    1496            0 :             LeadershipStatus::Candidate => AllowedRoutes::Some(&[
    1497            0 :                 "/ready",
    1498            0 :                 "/status",
    1499            0 :                 "/metrics",
    1500            0 :                 "/profile/cpu",
    1501            0 :                 "/profile/heap",
    1502            0 :             ]),
    1503              :         };
    1504              : 
    1505            0 :         match allowed_routes {
    1506            0 :             AllowedRoutes::All => Ok(req),
    1507            0 :             AllowedRoutes::Some(allowed) if allowed.contains(&req.uri().path()) => Ok(req),
    1508              :             _ => {
    1509            0 :                 tracing::info!(
    1510            0 :                     "Request {} not allowed due to current leadership state",
    1511            0 :                     req.uri()
    1512              :                 );
    1513              : 
    1514            0 :                 Err(ApiError::ResourceUnavailable(
    1515            0 :                     format!("Current leadership status is {leadership_status}").into(),
    1516            0 :                 ))
    1517              :             }
    1518              :         }
    1519            0 :     })
    1520            0 : }
    1521              : 
    1522            0 : fn prologue_metrics_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>()
    1523            0 : -> Middleware<B, ApiError> {
    1524            0 :     Middleware::pre(move |req| async move {
    1525            0 :         let meta = RequestMeta {
    1526            0 :             method: req.method().clone(),
    1527            0 :             at: Instant::now(),
    1528            0 :         };
    1529            0 : 
    1530            0 :         req.set_context(meta);
    1531            0 : 
    1532            0 :         Ok(req)
    1533            0 :     })
    1534            0 : }
    1535              : 
    1536            0 : fn epilogue_metrics_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>()
    1537            0 : -> Middleware<B, ApiError> {
    1538            0 :     Middleware::post_with_info(move |resp, req_info| async move {
    1539            0 :         let request_name = match req_info.context::<RequestName>() {
    1540            0 :             Some(name) => name,
    1541              :             None => {
    1542            0 :                 return Ok(resp);
    1543              :             }
    1544              :         };
    1545              : 
    1546            0 :         if let Some(meta) = req_info.context::<RequestMeta>() {
    1547            0 :             let status = &crate::metrics::METRICS_REGISTRY
    1548            0 :                 .metrics_group
    1549            0 :                 .storage_controller_http_request_status;
    1550            0 :             let latency = &crate::metrics::METRICS_REGISTRY
    1551            0 :                 .metrics_group
    1552            0 :                 .storage_controller_http_request_latency;
    1553            0 : 
    1554            0 :             status.inc(HttpRequestStatusLabelGroup {
    1555            0 :                 path: request_name.0,
    1556            0 :                 method: meta.method.clone().into(),
    1557            0 :                 status: crate::metrics::StatusCode(resp.status()),
    1558            0 :             });
    1559            0 : 
    1560            0 :             latency.observe(
    1561            0 :                 HttpRequestLatencyLabelGroup {
    1562            0 :                     path: request_name.0,
    1563            0 :                     method: meta.method.into(),
    1564            0 :                 },
    1565            0 :                 meta.at.elapsed().as_secs_f64(),
    1566            0 :             );
    1567            0 :         }
    1568            0 :         Ok(resp)
    1569            0 :     })
    1570            0 : }
    1571              : 
    1572            0 : pub async fn measured_metrics_handler(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    1573              :     pub const TEXT_FORMAT: &str = "text/plain; version=0.0.4";
    1574              : 
    1575            0 :     let req = match maybe_forward(req).await {
    1576            0 :         ForwardOutcome::Forwarded(res) => {
    1577            0 :             return res;
    1578              :         }
    1579            0 :         ForwardOutcome::NotForwarded(req) => req,
    1580            0 :     };
    1581            0 : 
    1582            0 :     let state = get_state(&req);
    1583            0 :     let payload = crate::metrics::METRICS_REGISTRY.encode(&state.neon_metrics);
    1584            0 :     let response = Response::builder()
    1585            0 :         .status(200)
    1586            0 :         .header(CONTENT_TYPE, TEXT_FORMAT)
    1587            0 :         .body(payload.into())
    1588            0 :         .unwrap();
    1589            0 : 
    1590            0 :     Ok(response)
    1591            0 : }
    1592              : 
    1593              : #[derive(Clone)]
    1594              : struct RequestName(&'static str);
    1595              : 
    1596            0 : async fn named_request_span<R, H>(
    1597            0 :     request: Request<Body>,
    1598            0 :     handler: H,
    1599            0 :     name: RequestName,
    1600            0 : ) -> R::Output
    1601            0 : where
    1602            0 :     R: Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
    1603            0 :     H: FnOnce(Request<Body>) -> R + Send + Sync + 'static,
    1604            0 : {
    1605            0 :     request.set_context(name);
    1606            0 :     request_span(request, handler).await
    1607            0 : }
    1608              : 
    1609              : enum ForwardOutcome {
    1610              :     Forwarded(Result<Response<Body>, ApiError>),
    1611              :     NotForwarded(Request<Body>),
    1612              : }
    1613              : 
    1614              : /// Potentially forward the request to the current storage controler leader.
    1615              : /// More specifically we forward when:
    1616              : /// 1. Request is not one of:
    1617              : ///    ["/control/v1/step_down", "/status", "/ready", "/metrics", "/profile/cpu", "/profile/heap"]
    1618              : /// 2. Current instance is in [`LeadershipStatus::SteppedDown`] state
    1619              : /// 3. There is a leader in the database to forward to
    1620              : /// 4. Leader from step (3) is not the current instance
    1621              : ///
    1622              : /// Why forward?
    1623              : /// It turns out that we can't rely on external orchestration to promptly route trafic to the
    1624              : /// new leader. This is downtime inducing. Forwarding provides a safe way out.
    1625              : ///
    1626              : /// Why is it safe?
    1627              : /// If a storcon instance is persisted in the database, then we know that it is the current leader.
    1628              : /// There's one exception: time between handling step-down request and the new leader updating the
    1629              : /// database.
    1630              : ///
    1631              : /// Let's treat the happy case first. The stepped down node does not produce any side effects,
    1632              : /// since all request handling happens on the leader.
    1633              : ///
    1634              : /// As for the edge case, we are guaranteed to always have a maximum of two running instances.
    1635              : /// Hence, if we are in the edge case scenario the leader persisted in the database is the
    1636              : /// stepped down instance that received the request. Condition (4) above covers this scenario.
    1637            0 : async fn maybe_forward(req: Request<Body>) -> ForwardOutcome {
    1638              :     const NOT_FOR_FORWARD: &[&str] = &[
    1639              :         "/control/v1/step_down",
    1640              :         "/status",
    1641              :         "/ready",
    1642              :         "/metrics",
    1643              :         "/profile/cpu",
    1644              :         "/profile/heap",
    1645              :     ];
    1646              : 
    1647            0 :     let uri = req.uri();
    1648            0 :     let uri_for_forward = !NOT_FOR_FORWARD.contains(&uri.path());
    1649            0 : 
    1650            0 :     // Fast return before trying to take any Service locks, if we will never forward anyway
    1651            0 :     if !uri_for_forward {
    1652            0 :         return ForwardOutcome::NotForwarded(req);
    1653            0 :     }
    1654            0 : 
    1655            0 :     let state = get_state(&req);
    1656            0 :     let leadership_status = state.service.get_leadership_status();
    1657            0 : 
    1658            0 :     if leadership_status != LeadershipStatus::SteppedDown {
    1659            0 :         return ForwardOutcome::NotForwarded(req);
    1660            0 :     }
    1661              : 
    1662            0 :     let leader = state.service.get_leader().await;
    1663            0 :     let leader = {
    1664            0 :         match leader {
    1665            0 :             Ok(Some(leader)) => leader,
    1666              :             Ok(None) => {
    1667            0 :                 return ForwardOutcome::Forwarded(Err(ApiError::ResourceUnavailable(
    1668            0 :                     "No leader to forward to while in stepped down state".into(),
    1669            0 :                 )));
    1670              :             }
    1671            0 :             Err(err) => {
    1672            0 :                 return ForwardOutcome::Forwarded(Err(ApiError::InternalServerError(
    1673            0 :                     anyhow::anyhow!(
    1674            0 :                         "Failed to get leader for forwarding while in stepped down state: {err}"
    1675            0 :                     ),
    1676            0 :                 )));
    1677              :             }
    1678              :         }
    1679              :     };
    1680              : 
    1681            0 :     let cfg = state.service.get_config();
    1682            0 :     if let Some(ref self_addr) = cfg.address_for_peers {
    1683            0 :         let leader_addr = match Uri::from_str(leader.address.as_str()) {
    1684            0 :             Ok(uri) => uri,
    1685            0 :             Err(err) => {
    1686            0 :                 return ForwardOutcome::Forwarded(Err(ApiError::InternalServerError(
    1687            0 :                     anyhow::anyhow!(
    1688            0 :                         "Failed to parse leader uri for forwarding while in stepped down state: {err}"
    1689            0 :                     ),
    1690            0 :                 )));
    1691              :             }
    1692              :         };
    1693              : 
    1694            0 :         if *self_addr == leader_addr {
    1695            0 :             return ForwardOutcome::Forwarded(Err(ApiError::InternalServerError(anyhow::anyhow!(
    1696            0 :                 "Leader is stepped down instance"
    1697            0 :             ))));
    1698            0 :         }
    1699            0 :     }
    1700              : 
    1701            0 :     tracing::info!("Forwarding {} to leader at {}", uri, leader.address);
    1702              : 
    1703              :     // Use [`RECONCILE_TIMEOUT`] as the max amount of time a request should block for and
    1704              :     // include some leeway to get the timeout for proxied requests.
    1705              :     const PROXIED_REQUEST_TIMEOUT: Duration = Duration::from_secs(RECONCILE_TIMEOUT.as_secs() + 10);
    1706            0 :     let client = reqwest::ClientBuilder::new()
    1707            0 :         .timeout(PROXIED_REQUEST_TIMEOUT)
    1708            0 :         .build();
    1709            0 :     let client = match client {
    1710            0 :         Ok(client) => client,
    1711            0 :         Err(err) => {
    1712            0 :             return ForwardOutcome::Forwarded(Err(ApiError::InternalServerError(anyhow::anyhow!(
    1713            0 :                 "Failed to build leader client for forwarding while in stepped down state: {err}"
    1714            0 :             ))));
    1715              :         }
    1716              :     };
    1717              : 
    1718            0 :     let request: reqwest::Request = match convert_request(req, &client, leader.address).await {
    1719            0 :         Ok(r) => r,
    1720            0 :         Err(err) => {
    1721            0 :             return ForwardOutcome::Forwarded(Err(ApiError::InternalServerError(anyhow::anyhow!(
    1722            0 :                 "Failed to convert request for forwarding while in stepped down state: {err}"
    1723            0 :             ))));
    1724              :         }
    1725              :     };
    1726              : 
    1727            0 :     let response = match client.execute(request).await {
    1728            0 :         Ok(r) => r,
    1729            0 :         Err(err) => {
    1730            0 :             return ForwardOutcome::Forwarded(Err(ApiError::InternalServerError(anyhow::anyhow!(
    1731            0 :                 "Failed to forward while in stepped down state: {err}"
    1732            0 :             ))));
    1733              :         }
    1734              :     };
    1735              : 
    1736            0 :     ForwardOutcome::Forwarded(convert_response(response).await)
    1737            0 : }
    1738              : 
    1739              : /// Convert a [`reqwest::Response`] to a [hyper::Response`] by passing through
    1740              : /// a stable representation (string, bytes or integer)
    1741              : ///
    1742              : /// Ideally, we would not have to do this since both types use the http crate
    1743              : /// under the hood. However, they use different versions of the crate and keeping
    1744              : /// second order dependencies in sync is difficult.
    1745            0 : async fn convert_response(resp: reqwest::Response) -> Result<hyper::Response<Body>, ApiError> {
    1746              :     use std::str::FromStr;
    1747              : 
    1748            0 :     let mut builder = hyper::Response::builder().status(resp.status().as_u16());
    1749            0 :     for (key, value) in resp.headers().into_iter() {
    1750            0 :         let key = hyper::header::HeaderName::from_str(key.as_str()).map_err(|err| {
    1751            0 :             ApiError::InternalServerError(anyhow::anyhow!("Response conversion failed: {err}"))
    1752            0 :         })?;
    1753              : 
    1754            0 :         let value = hyper::header::HeaderValue::from_bytes(value.as_bytes()).map_err(|err| {
    1755            0 :             ApiError::InternalServerError(anyhow::anyhow!("Response conversion failed: {err}"))
    1756            0 :         })?;
    1757              : 
    1758            0 :         builder = builder.header(key, value);
    1759              :     }
    1760              : 
    1761            0 :     let body = http::Body::wrap_stream(resp.bytes_stream());
    1762            0 : 
    1763            0 :     builder.body(body).map_err(|err| {
    1764            0 :         ApiError::InternalServerError(anyhow::anyhow!("Response conversion failed: {err}"))
    1765            0 :     })
    1766            0 : }
    1767              : 
    1768              : /// Convert a [`reqwest::Request`] to a [hyper::Request`] by passing through
    1769              : /// a stable representation (string, bytes or integer)
    1770              : ///
    1771              : /// See [`convert_response`] for why we are doing it this way.
    1772            0 : async fn convert_request(
    1773            0 :     req: hyper::Request<Body>,
    1774            0 :     client: &reqwest::Client,
    1775            0 :     to_address: String,
    1776            0 : ) -> Result<reqwest::Request, ApiError> {
    1777              :     use std::str::FromStr;
    1778              : 
    1779            0 :     let (parts, body) = req.into_parts();
    1780            0 :     let method = reqwest::Method::from_str(parts.method.as_str()).map_err(|err| {
    1781            0 :         ApiError::InternalServerError(anyhow::anyhow!("Request conversion failed: {err}"))
    1782            0 :     })?;
    1783              : 
    1784            0 :     let path_and_query = parts.uri.path_and_query().ok_or_else(|| {
    1785            0 :         ApiError::InternalServerError(anyhow::anyhow!(
    1786            0 :             "Request conversion failed: no path and query"
    1787            0 :         ))
    1788            0 :     })?;
    1789              : 
    1790            0 :     let uri = reqwest::Url::from_str(
    1791            0 :         format!(
    1792            0 :             "{}{}",
    1793            0 :             to_address.trim_end_matches("/"),
    1794            0 :             path_and_query.as_str()
    1795            0 :         )
    1796            0 :         .as_str(),
    1797            0 :     )
    1798            0 :     .map_err(|err| {
    1799            0 :         ApiError::InternalServerError(anyhow::anyhow!("Request conversion failed: {err}"))
    1800            0 :     })?;
    1801              : 
    1802            0 :     let mut headers = reqwest::header::HeaderMap::new();
    1803            0 :     for (key, value) in parts.headers.into_iter() {
    1804            0 :         let key = match key {
    1805            0 :             Some(k) => k,
    1806              :             None => {
    1807            0 :                 continue;
    1808              :             }
    1809              :         };
    1810              : 
    1811            0 :         let key = reqwest::header::HeaderName::from_str(key.as_str()).map_err(|err| {
    1812            0 :             ApiError::InternalServerError(anyhow::anyhow!("Request conversion failed: {err}"))
    1813            0 :         })?;
    1814              : 
    1815            0 :         let value = reqwest::header::HeaderValue::from_bytes(value.as_bytes()).map_err(|err| {
    1816            0 :             ApiError::InternalServerError(anyhow::anyhow!("Request conversion failed: {err}"))
    1817            0 :         })?;
    1818              : 
    1819            0 :         headers.insert(key, value);
    1820              :     }
    1821              : 
    1822            0 :     let body = hyper::body::to_bytes(body).await.map_err(|err| {
    1823            0 :         ApiError::InternalServerError(anyhow::anyhow!("Request conversion failed: {err}"))
    1824            0 :     })?;
    1825              : 
    1826            0 :     client
    1827            0 :         .request(method, uri)
    1828            0 :         .headers(headers)
    1829            0 :         .body(body)
    1830            0 :         .build()
    1831            0 :         .map_err(|err| {
    1832            0 :             ApiError::InternalServerError(anyhow::anyhow!("Request conversion failed: {err}"))
    1833            0 :         })
    1834            0 : }
    1835              : 
    1836            0 : pub fn make_router(
    1837            0 :     service: Arc<Service>,
    1838            0 :     auth: Option<Arc<SwappableJwtAuth>>,
    1839            0 :     build_info: BuildInfo,
    1840            0 : ) -> RouterBuilder<hyper::Body, ApiError> {
    1841            0 :     let mut router = endpoint::make_router()
    1842            0 :         .middleware(prologue_leadership_status_check_middleware())
    1843            0 :         .middleware(prologue_metrics_middleware())
    1844            0 :         .middleware(epilogue_metrics_middleware());
    1845            0 :     if auth.is_some() {
    1846            0 :         router = router.middleware(auth_middleware(|request| {
    1847            0 :             let state = get_state(request);
    1848            0 :             if state.allowlist_routes.contains(&request.uri().path()) {
    1849            0 :                 None
    1850              :             } else {
    1851            0 :                 state.auth.as_deref()
    1852              :             }
    1853            0 :         }));
    1854            0 :     }
    1855              : 
    1856            0 :     router
    1857            0 :         .data(Arc::new(HttpState::new(service, auth, build_info)))
    1858            0 :         .get("/metrics", |r| {
    1859            0 :             named_request_span(r, measured_metrics_handler, RequestName("metrics"))
    1860            0 :         })
    1861            0 :         // Non-prefixed generic endpoints (status, metrics, profiling)
    1862            0 :         .get("/status", |r| {
    1863            0 :             named_request_span(r, handle_status, RequestName("status"))
    1864            0 :         })
    1865            0 :         .get("/ready", |r| {
    1866            0 :             named_request_span(r, handle_ready, RequestName("ready"))
    1867            0 :         })
    1868            0 :         .get("/profile/cpu", |r| {
    1869            0 :             named_request_span(r, profile_cpu_handler, RequestName("profile_cpu"))
    1870            0 :         })
    1871            0 :         .get("/profile/heap", |r| {
    1872            0 :             named_request_span(r, profile_heap_handler, RequestName("profile_heap"))
    1873            0 :         })
    1874            0 :         // Upcalls for the pageserver: point the pageserver's `control_plane_api` config to this prefix
    1875            0 :         .post("/upcall/v1/re-attach", |r| {
    1876            0 :             named_request_span(r, handle_re_attach, RequestName("upcall_v1_reattach"))
    1877            0 :         })
    1878            0 :         .post("/upcall/v1/validate", |r| {
    1879            0 :             named_request_span(r, handle_validate, RequestName("upcall_v1_validate"))
    1880            0 :         })
    1881            0 :         // Test/dev/debug endpoints
    1882            0 :         .post("/debug/v1/attach-hook", |r| {
    1883            0 :             named_request_span(r, handle_attach_hook, RequestName("debug_v1_attach_hook"))
    1884            0 :         })
    1885            0 :         .post("/debug/v1/inspect", |r| {
    1886            0 :             named_request_span(r, handle_inspect, RequestName("debug_v1_inspect"))
    1887            0 :         })
    1888            0 :         .post("/debug/v1/tenant/:tenant_id/drop", |r| {
    1889            0 :             named_request_span(r, handle_tenant_drop, RequestName("debug_v1_tenant_drop"))
    1890            0 :         })
    1891            0 :         .post("/debug/v1/node/:node_id/drop", |r| {
    1892            0 :             named_request_span(r, handle_node_drop, RequestName("debug_v1_node_drop"))
    1893            0 :         })
    1894            0 :         .post("/debug/v1/tenant/:tenant_id/import", |r| {
    1895            0 :             named_request_span(
    1896            0 :                 r,
    1897            0 :                 handle_tenant_import,
    1898            0 :                 RequestName("debug_v1_tenant_import"),
    1899            0 :             )
    1900            0 :         })
    1901            0 :         .get("/debug/v1/tenant", |r| {
    1902            0 :             named_request_span(r, handle_tenants_dump, RequestName("debug_v1_tenant"))
    1903            0 :         })
    1904            0 :         .get("/debug/v1/tenant/:tenant_id/locate", |r| {
    1905            0 :             tenant_service_handler(
    1906            0 :                 r,
    1907            0 :                 handle_tenant_locate,
    1908            0 :                 RequestName("debug_v1_tenant_locate"),
    1909            0 :             )
    1910            0 :         })
    1911            0 :         .get("/debug/v1/scheduler", |r| {
    1912            0 :             named_request_span(r, handle_scheduler_dump, RequestName("debug_v1_scheduler"))
    1913            0 :         })
    1914            0 :         .post("/debug/v1/consistency_check", |r| {
    1915            0 :             named_request_span(
    1916            0 :                 r,
    1917            0 :                 handle_consistency_check,
    1918            0 :                 RequestName("debug_v1_consistency_check"),
    1919            0 :             )
    1920            0 :         })
    1921            0 :         .post("/debug/v1/reconcile_all", |r| {
    1922            0 :             request_span(r, handle_reconcile_all)
    1923            0 :         })
    1924            0 :         .put("/debug/v1/failpoints", |r| {
    1925            0 :             request_span(r, |r| failpoints_handler(r, CancellationToken::new()))
    1926            0 :         })
    1927            0 :         // Node operations
    1928            0 :         .post("/control/v1/node", |r| {
    1929            0 :             named_request_span(r, handle_node_register, RequestName("control_v1_node"))
    1930            0 :         })
    1931            0 :         .delete("/control/v1/node/:node_id", |r| {
    1932            0 :             named_request_span(r, handle_node_delete, RequestName("control_v1_node_delete"))
    1933            0 :         })
    1934            0 :         .get("/control/v1/node", |r| {
    1935            0 :             named_request_span(r, handle_node_list, RequestName("control_v1_node"))
    1936            0 :         })
    1937            0 :         .put("/control/v1/node/:node_id/config", |r| {
    1938            0 :             named_request_span(
    1939            0 :                 r,
    1940            0 :                 handle_node_configure,
    1941            0 :                 RequestName("control_v1_node_config"),
    1942            0 :             )
    1943            0 :         })
    1944            0 :         .get("/control/v1/node/:node_id", |r| {
    1945            0 :             named_request_span(r, handle_node_status, RequestName("control_v1_node_status"))
    1946            0 :         })
    1947            0 :         .get("/control/v1/node/:node_id/shards", |r| {
    1948            0 :             named_request_span(
    1949            0 :                 r,
    1950            0 :                 handle_node_shards,
    1951            0 :                 RequestName("control_v1_node_describe"),
    1952            0 :             )
    1953            0 :         })
    1954            0 :         .get("/control/v1/leader", |r| {
    1955            0 :             named_request_span(r, handle_get_leader, RequestName("control_v1_get_leader"))
    1956            0 :         })
    1957            0 :         .put("/control/v1/node/:node_id/drain", |r| {
    1958            0 :             named_request_span(r, handle_node_drain, RequestName("control_v1_node_drain"))
    1959            0 :         })
    1960            0 :         .delete("/control/v1/node/:node_id/drain", |r| {
    1961            0 :             named_request_span(
    1962            0 :                 r,
    1963            0 :                 handle_cancel_node_drain,
    1964            0 :                 RequestName("control_v1_cancel_node_drain"),
    1965            0 :             )
    1966            0 :         })
    1967            0 :         .put("/control/v1/node/:node_id/fill", |r| {
    1968            0 :             named_request_span(r, handle_node_fill, RequestName("control_v1_node_fill"))
    1969            0 :         })
    1970            0 :         .delete("/control/v1/node/:node_id/fill", |r| {
    1971            0 :             named_request_span(
    1972            0 :                 r,
    1973            0 :                 handle_cancel_node_fill,
    1974            0 :                 RequestName("control_v1_cancel_node_fill"),
    1975            0 :             )
    1976            0 :         })
    1977            0 :         // Metadata health operations
    1978            0 :         .post("/control/v1/metadata_health/update", |r| {
    1979            0 :             named_request_span(
    1980            0 :                 r,
    1981            0 :                 handle_metadata_health_update,
    1982            0 :                 RequestName("control_v1_metadata_health_update"),
    1983            0 :             )
    1984            0 :         })
    1985            0 :         .get("/control/v1/metadata_health/unhealthy", |r| {
    1986            0 :             named_request_span(
    1987            0 :                 r,
    1988            0 :                 handle_metadata_health_list_unhealthy,
    1989            0 :                 RequestName("control_v1_metadata_health_list_unhealthy"),
    1990            0 :             )
    1991            0 :         })
    1992            0 :         .post("/control/v1/metadata_health/outdated", |r| {
    1993            0 :             named_request_span(
    1994            0 :                 r,
    1995            0 :                 handle_metadata_health_list_outdated,
    1996            0 :                 RequestName("control_v1_metadata_health_list_outdated"),
    1997            0 :             )
    1998            0 :         })
    1999            0 :         // Safekeepers
    2000            0 :         .get("/control/v1/safekeeper", |r| {
    2001            0 :             named_request_span(
    2002            0 :                 r,
    2003            0 :                 handle_safekeeper_list,
    2004            0 :                 RequestName("control_v1_safekeeper_list"),
    2005            0 :             )
    2006            0 :         })
    2007            0 :         .get("/control/v1/safekeeper/:id", |r| {
    2008            0 :             named_request_span(r, handle_get_safekeeper, RequestName("v1_safekeeper"))
    2009            0 :         })
    2010            0 :         .post("/control/v1/safekeeper/:id", |r| {
    2011            0 :             // id is in the body
    2012            0 :             named_request_span(
    2013            0 :                 r,
    2014            0 :                 handle_upsert_safekeeper,
    2015            0 :                 RequestName("v1_safekeeper_post"),
    2016            0 :             )
    2017            0 :         })
    2018            0 :         .post("/control/v1/safekeeper/:id/scheduling_policy", |r| {
    2019            0 :             named_request_span(
    2020            0 :                 r,
    2021            0 :                 handle_safekeeper_scheduling_policy,
    2022            0 :                 RequestName("v1_safekeeper_status"),
    2023            0 :             )
    2024            0 :         })
    2025            0 :         // Tenant Shard operations
    2026            0 :         .put("/control/v1/tenant/:tenant_shard_id/migrate", |r| {
    2027            0 :             tenant_service_handler(
    2028            0 :                 r,
    2029            0 :                 handle_tenant_shard_migrate,
    2030            0 :                 RequestName("control_v1_tenant_migrate"),
    2031            0 :             )
    2032            0 :         })
    2033            0 :         .put(
    2034            0 :             "/control/v1/tenant/:tenant_shard_id/migrate_secondary",
    2035            0 :             |r| {
    2036            0 :                 tenant_service_handler(
    2037            0 :                     r,
    2038            0 :                     handle_tenant_shard_migrate_secondary,
    2039            0 :                     RequestName("control_v1_tenant_migrate_secondary"),
    2040            0 :                 )
    2041            0 :             },
    2042            0 :         )
    2043            0 :         .put(
    2044            0 :             "/control/v1/tenant/:tenant_shard_id/cancel_reconcile",
    2045            0 :             |r| {
    2046            0 :                 tenant_service_handler(
    2047            0 :                     r,
    2048            0 :                     handle_tenant_shard_cancel_reconcile,
    2049            0 :                     RequestName("control_v1_tenant_cancel_reconcile"),
    2050            0 :                 )
    2051            0 :             },
    2052            0 :         )
    2053            0 :         .put("/control/v1/tenant/:tenant_id/shard_split", |r| {
    2054            0 :             tenant_service_handler(
    2055            0 :                 r,
    2056            0 :                 handle_tenant_shard_split,
    2057            0 :                 RequestName("control_v1_tenant_shard_split"),
    2058            0 :             )
    2059            0 :         })
    2060            0 :         .get("/control/v1/tenant/:tenant_id", |r| {
    2061            0 :             tenant_service_handler(
    2062            0 :                 r,
    2063            0 :                 handle_tenant_describe,
    2064            0 :                 RequestName("control_v1_tenant_describe"),
    2065            0 :             )
    2066            0 :         })
    2067            0 :         .get("/control/v1/tenant", |r| {
    2068            0 :             tenant_service_handler(r, handle_tenant_list, RequestName("control_v1_tenant_list"))
    2069            0 :         })
    2070            0 :         .put("/control/v1/tenant/:tenant_id/policy", |r| {
    2071            0 :             named_request_span(
    2072            0 :                 r,
    2073            0 :                 handle_tenant_update_policy,
    2074            0 :                 RequestName("control_v1_tenant_policy"),
    2075            0 :             )
    2076            0 :         })
    2077            0 :         .put("/control/v1/preferred_azs", |r| {
    2078            0 :             named_request_span(
    2079            0 :                 r,
    2080            0 :                 handle_update_preferred_azs,
    2081            0 :                 RequestName("control_v1_preferred_azs"),
    2082            0 :             )
    2083            0 :         })
    2084            0 :         .put("/control/v1/step_down", |r| {
    2085            0 :             named_request_span(r, handle_step_down, RequestName("control_v1_step_down"))
    2086            0 :         })
    2087            0 :         // Tenant operations
    2088            0 :         // The ^/v1/ endpoints act as a "Virtual Pageserver", enabling shard-naive clients to call into
    2089            0 :         // this service to manage tenants that actually consist of many tenant shards, as if they are a single entity.
    2090            0 :         .post("/v1/tenant", |r| {
    2091            0 :             tenant_service_handler(r, handle_tenant_create, RequestName("v1_tenant"))
    2092            0 :         })
    2093            0 :         .delete("/v1/tenant/:tenant_id", |r| {
    2094            0 :             tenant_service_handler(r, handle_tenant_delete, RequestName("v1_tenant"))
    2095            0 :         })
    2096            0 :         .patch("/v1/tenant/config", |r| {
    2097            0 :             tenant_service_handler(
    2098            0 :                 r,
    2099            0 :                 handle_tenant_config_patch,
    2100            0 :                 RequestName("v1_tenant_config"),
    2101            0 :             )
    2102            0 :         })
    2103            0 :         .put("/v1/tenant/config", |r| {
    2104            0 :             tenant_service_handler(r, handle_tenant_config_set, RequestName("v1_tenant_config"))
    2105            0 :         })
    2106            0 :         .get("/v1/tenant/:tenant_id/config", |r| {
    2107            0 :             tenant_service_handler(r, handle_tenant_config_get, RequestName("v1_tenant_config"))
    2108            0 :         })
    2109            0 :         .put("/v1/tenant/:tenant_shard_id/location_config", |r| {
    2110            0 :             tenant_service_handler(
    2111            0 :                 r,
    2112            0 :                 handle_tenant_location_config,
    2113            0 :                 RequestName("v1_tenant_location_config"),
    2114            0 :             )
    2115            0 :         })
    2116            0 :         .put("/v1/tenant/:tenant_id/time_travel_remote_storage", |r| {
    2117            0 :             tenant_service_handler(
    2118            0 :                 r,
    2119            0 :                 handle_tenant_time_travel_remote_storage,
    2120            0 :                 RequestName("v1_tenant_time_travel_remote_storage"),
    2121            0 :             )
    2122            0 :         })
    2123            0 :         .post("/v1/tenant/:tenant_id/secondary/download", |r| {
    2124            0 :             tenant_service_handler(
    2125            0 :                 r,
    2126            0 :                 handle_tenant_secondary_download,
    2127            0 :                 RequestName("v1_tenant_secondary_download"),
    2128            0 :             )
    2129            0 :         })
    2130            0 :         // Timeline operations
    2131            0 :         .delete("/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
    2132            0 :             tenant_service_handler(
    2133            0 :                 r,
    2134            0 :                 handle_tenant_timeline_delete,
    2135            0 :                 RequestName("v1_tenant_timeline"),
    2136            0 :             )
    2137            0 :         })
    2138            0 :         .post("/v1/tenant/:tenant_id/timeline", |r| {
    2139            0 :             tenant_service_handler(
    2140            0 :                 r,
    2141            0 :                 handle_tenant_timeline_create,
    2142            0 :                 RequestName("v1_tenant_timeline"),
    2143            0 :             )
    2144            0 :         })
    2145            0 :         .put(
    2146            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/archival_config",
    2147            0 :             |r| {
    2148            0 :                 tenant_service_handler(
    2149            0 :                     r,
    2150            0 :                     handle_tenant_timeline_archival_config,
    2151            0 :                     RequestName("v1_tenant_timeline_archival_config"),
    2152            0 :                 )
    2153            0 :             },
    2154            0 :         )
    2155            0 :         .put(
    2156            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/detach_ancestor",
    2157            0 :             |r| {
    2158            0 :                 tenant_service_handler(
    2159            0 :                     r,
    2160            0 :                     handle_tenant_timeline_detach_ancestor,
    2161            0 :                     RequestName("v1_tenant_timeline_detach_ancestor"),
    2162            0 :                 )
    2163            0 :             },
    2164            0 :         )
    2165            0 :         .post(
    2166            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/block_gc",
    2167            0 :             |r| {
    2168            0 :                 tenant_service_handler(
    2169            0 :                     r,
    2170            0 :                     |s, r| handle_tenant_timeline_block_unblock_gc(s, r, BlockUnblock::Block),
    2171            0 :                     RequestName("v1_tenant_timeline_block_unblock_gc"),
    2172            0 :                 )
    2173            0 :             },
    2174            0 :         )
    2175            0 :         .post(
    2176            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/unblock_gc",
    2177            0 :             |r| {
    2178            0 :                 tenant_service_handler(
    2179            0 :                     r,
    2180            0 :                     |s, r| handle_tenant_timeline_block_unblock_gc(s, r, BlockUnblock::Unblock),
    2181            0 :                     RequestName("v1_tenant_timeline_block_unblock_gc"),
    2182            0 :                 )
    2183            0 :             },
    2184            0 :         )
    2185            0 :         .post(
    2186            0 :             "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/download_heatmap_layers",
    2187            0 :             |r| {
    2188            0 :                 tenant_service_handler(
    2189            0 :                     r,
    2190            0 :                     handle_tenant_timeline_download_heatmap_layers,
    2191            0 :                     RequestName("v1_tenant_timeline_download_heatmap_layers"),
    2192            0 :                 )
    2193            0 :             },
    2194            0 :         )
    2195            0 :         // Tenant detail GET passthrough to shard zero:
    2196            0 :         .get("/v1/tenant/:tenant_id", |r| {
    2197            0 :             tenant_service_handler(
    2198            0 :                 r,
    2199            0 :                 handle_tenant_timeline_passthrough,
    2200            0 :                 RequestName("v1_tenant_passthrough"),
    2201            0 :             )
    2202            0 :         })
    2203            0 :         // The `*` in the  URL is a wildcard: any tenant/timeline GET APIs on the pageserver
    2204            0 :         // are implicitly exposed here.  This must be last in the list to avoid
    2205            0 :         // taking precedence over other GET methods we might implement by hand.
    2206            0 :         .get("/v1/tenant/:tenant_id/*", |r| {
    2207            0 :             tenant_service_handler(
    2208            0 :                 r,
    2209            0 :                 handle_tenant_timeline_passthrough,
    2210            0 :                 RequestName("v1_tenant_passthrough"),
    2211            0 :             )
    2212            0 :         })
    2213            0 : }
    2214              : 
    2215              : #[cfg(test)]
    2216              : mod test {
    2217              : 
    2218              :     use super::path_without_ids;
    2219              : 
    2220              :     #[test]
    2221            1 :     fn test_path_without_ids() {
    2222            1 :         assert_eq!(
    2223            1 :             path_without_ids(
    2224            1 :                 "/v1/tenant/1a2b3344556677881122334455667788/timeline/AA223344556677881122334455667788"
    2225            1 :             ),
    2226            1 :             "/v1/tenant//timeline/"
    2227            1 :         );
    2228            1 :         assert_eq!(
    2229            1 :             path_without_ids(
    2230            1 :                 "/v1/tenant/1a2b3344556677881122334455667788-0108/timeline/AA223344556677881122334455667788"
    2231            1 :             ),
    2232            1 :             "/v1/tenant//timeline/"
    2233            1 :         );
    2234            1 :         assert_eq!(
    2235            1 :             path_without_ids(
    2236            1 :                 "/v1/tenant/1a2b3344556677881122334455667788-0108/timeline/AA223344556677881122334455667788?parameter=foo"
    2237            1 :             ),
    2238            1 :             "/v1/tenant//timeline/"
    2239            1 :         );
    2240            1 :     }
    2241              : }
        

Generated by: LCOV version 2.1-beta