Line data Source code
1 : //!
2 : //! Management HTTP API
3 : //!
4 : use std::collections::HashMap;
5 : use std::str::FromStr;
6 : use std::sync::Arc;
7 : use std::time::Duration;
8 :
9 : use anyhow::{anyhow, Context, Result};
10 : use enumset::EnumSet;
11 : use futures::TryFutureExt;
12 : use humantime::format_rfc3339;
13 : use hyper::header;
14 : use hyper::StatusCode;
15 : use hyper::{Body, Request, Response, Uri};
16 : use metrics::launch_timestamp::LaunchTimestamp;
17 : use pageserver_api::models::LocationConfigListResponse;
18 : use pageserver_api::models::ShardParameters;
19 : use pageserver_api::models::TenantDetails;
20 : use pageserver_api::models::TenantLocationConfigResponse;
21 : use pageserver_api::models::TenantShardLocation;
22 : use pageserver_api::models::TenantShardSplitRequest;
23 : use pageserver_api::models::TenantShardSplitResponse;
24 : use pageserver_api::models::TenantState;
25 : use pageserver_api::models::{
26 : DownloadRemoteLayersTaskSpawnRequest, LocationConfigMode, TenantAttachRequest,
27 : TenantLoadRequest, TenantLocationConfigRequest,
28 : };
29 : use pageserver_api::shard::ShardCount;
30 : use pageserver_api::shard::TenantShardId;
31 : use remote_storage::GenericRemoteStorage;
32 : use remote_storage::TimeTravelError;
33 : use tenant_size_model::{SizeResult, StorageModel};
34 : use tokio_util::sync::CancellationToken;
35 : use tracing::*;
36 : use utils::auth::JwtAuth;
37 : use utils::failpoint_support::failpoints_handler;
38 : use utils::http::endpoint::request_span;
39 : use utils::http::json::json_request_or_empty_body;
40 : use utils::http::request::{get_request_param, must_get_query_param, parse_query_param};
41 :
42 : use crate::context::{DownloadBehavior, RequestContext};
43 : use crate::deletion_queue::DeletionQueueClient;
44 : use crate::metrics::{StorageTimeOperation, STORAGE_TIME_GLOBAL};
45 : use crate::pgdatadir_mapping::LsnForTimestamp;
46 : use crate::task_mgr::TaskKind;
47 : use crate::tenant::config::{LocationConf, TenantConfOpt};
48 : use crate::tenant::mgr::GetActiveTenantError;
49 : use crate::tenant::mgr::{
50 : GetTenantError, SetNewTenantConfigError, TenantManager, TenantMapError, TenantMapInsertError,
51 : TenantSlotError, TenantSlotUpsertError, TenantStateError,
52 : };
53 : use crate::tenant::mgr::{TenantSlot, UpsertLocationError};
54 : use crate::tenant::remote_timeline_client;
55 : use crate::tenant::secondary::SecondaryController;
56 : use crate::tenant::size::ModelInputs;
57 : use crate::tenant::storage_layer::LayerAccessStatsReset;
58 : use crate::tenant::timeline::CompactFlags;
59 : use crate::tenant::timeline::Timeline;
60 : use crate::tenant::SpawnMode;
61 : use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError};
62 : use crate::{config::PageServerConf, tenant::mgr};
63 : use crate::{disk_usage_eviction_task, tenant};
64 : use pageserver_api::models::{
65 : StatusResponse, TenantConfigRequest, TenantCreateRequest, TenantCreateResponse, TenantInfo,
66 : TimelineCreateRequest, TimelineGcRequest, TimelineInfo,
67 : };
68 : use utils::{
69 : auth::SwappableJwtAuth,
70 : generation::Generation,
71 : http::{
72 : endpoint::{self, attach_openapi_ui, auth_middleware, check_permission_with},
73 : error::{ApiError, HttpErrorBody},
74 : json::{json_request, json_response},
75 : request::parse_request_param,
76 : RequestExt, RouterBuilder,
77 : },
78 : id::{TenantId, TimelineId},
79 : lsn::Lsn,
80 : };
81 :
82 : // For APIs that require an Active tenant, how long should we block waiting for that state?
83 : // This is not functionally necessary (clients will retry), but avoids generating a lot of
84 : // failed API calls while tenants are activating.
85 : #[cfg(not(feature = "testing"))]
86 : pub(crate) const ACTIVE_TENANT_TIMEOUT: Duration = Duration::from_millis(5000);
87 :
88 : // Tests run on slow/oversubscribed nodes, and may need to wait much longer for tenants to
89 : // finish attaching, if calls to remote storage are slow.
90 : #[cfg(feature = "testing")]
91 : pub(crate) const ACTIVE_TENANT_TIMEOUT: Duration = Duration::from_millis(30000);
92 :
93 : pub struct State {
94 : conf: &'static PageServerConf,
95 : tenant_manager: Arc<TenantManager>,
96 : auth: Option<Arc<SwappableJwtAuth>>,
97 : allowlist_routes: Vec<Uri>,
98 : remote_storage: Option<GenericRemoteStorage>,
99 : broker_client: storage_broker::BrokerClientChannel,
100 : disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
101 : deletion_queue_client: DeletionQueueClient,
102 : secondary_controller: SecondaryController,
103 : latest_utilization: tokio::sync::Mutex<Option<(std::time::Instant, bytes::Bytes)>>,
104 : }
105 :
106 : impl State {
107 : #[allow(clippy::too_many_arguments)]
108 0 : pub fn new(
109 0 : conf: &'static PageServerConf,
110 0 : tenant_manager: Arc<TenantManager>,
111 0 : auth: Option<Arc<SwappableJwtAuth>>,
112 0 : remote_storage: Option<GenericRemoteStorage>,
113 0 : broker_client: storage_broker::BrokerClientChannel,
114 0 : disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
115 0 : deletion_queue_client: DeletionQueueClient,
116 0 : secondary_controller: SecondaryController,
117 0 : ) -> anyhow::Result<Self> {
118 0 : let allowlist_routes = ["/v1/status", "/v1/doc", "/swagger.yml", "/metrics"]
119 0 : .iter()
120 0 : .map(|v| v.parse().unwrap())
121 0 : .collect::<Vec<_>>();
122 0 : Ok(Self {
123 0 : conf,
124 0 : tenant_manager,
125 0 : auth,
126 0 : allowlist_routes,
127 0 : remote_storage,
128 0 : broker_client,
129 0 : disk_usage_eviction_state,
130 0 : deletion_queue_client,
131 0 : secondary_controller,
132 0 : latest_utilization: Default::default(),
133 0 : })
134 0 : }
135 : }
136 :
137 : #[inline(always)]
138 0 : fn get_state(request: &Request<Body>) -> &State {
139 0 : request
140 0 : .data::<Arc<State>>()
141 0 : .expect("unknown state type")
142 0 : .as_ref()
143 0 : }
144 :
145 : #[inline(always)]
146 0 : fn get_config(request: &Request<Body>) -> &'static PageServerConf {
147 0 : get_state(request).conf
148 0 : }
149 :
150 : /// Check that the requester is authorized to operate on given tenant
151 0 : fn check_permission(request: &Request<Body>, tenant_id: Option<TenantId>) -> Result<(), ApiError> {
152 0 : check_permission_with(request, |claims| {
153 0 : crate::auth::check_permission(claims, tenant_id)
154 0 : })
155 0 : }
156 :
157 : impl From<PageReconstructError> for ApiError {
158 0 : fn from(pre: PageReconstructError) -> ApiError {
159 0 : match pre {
160 0 : PageReconstructError::Other(pre) => ApiError::InternalServerError(pre),
161 : PageReconstructError::Cancelled => {
162 0 : ApiError::InternalServerError(anyhow::anyhow!("request was cancelled"))
163 : }
164 : PageReconstructError::AncestorStopping(_) => {
165 0 : ApiError::ResourceUnavailable(format!("{pre}").into())
166 : }
167 0 : PageReconstructError::AncestorLsnTimeout(e) => ApiError::Timeout(format!("{e}").into()),
168 0 : PageReconstructError::WalRedo(pre) => ApiError::InternalServerError(pre),
169 : }
170 0 : }
171 : }
172 :
173 : impl From<TenantMapInsertError> for ApiError {
174 0 : fn from(tmie: TenantMapInsertError) -> ApiError {
175 0 : match tmie {
176 0 : TenantMapInsertError::SlotError(e) => e.into(),
177 0 : TenantMapInsertError::SlotUpsertError(e) => e.into(),
178 0 : TenantMapInsertError::Other(e) => ApiError::InternalServerError(e),
179 : }
180 0 : }
181 : }
182 :
183 : impl From<TenantSlotError> for ApiError {
184 0 : fn from(e: TenantSlotError) -> ApiError {
185 0 : use TenantSlotError::*;
186 0 : match e {
187 0 : NotFound(tenant_id) => {
188 0 : ApiError::NotFound(anyhow::anyhow!("NotFound: tenant {tenant_id}").into())
189 : }
190 0 : e @ AlreadyExists(_, _) => ApiError::Conflict(format!("{e}")),
191 : InProgress => {
192 0 : ApiError::ResourceUnavailable("Tenant is being modified concurrently".into())
193 : }
194 0 : MapState(e) => e.into(),
195 : }
196 0 : }
197 : }
198 :
199 : impl From<TenantSlotUpsertError> for ApiError {
200 0 : fn from(e: TenantSlotUpsertError) -> ApiError {
201 0 : use TenantSlotUpsertError::*;
202 0 : match e {
203 0 : InternalError(e) => ApiError::InternalServerError(anyhow::anyhow!("{e}")),
204 0 : MapState(e) => e.into(),
205 0 : ShuttingDown(_) => ApiError::ShuttingDown,
206 : }
207 0 : }
208 : }
209 :
210 : impl From<UpsertLocationError> for ApiError {
211 0 : fn from(e: UpsertLocationError) -> ApiError {
212 0 : use UpsertLocationError::*;
213 0 : match e {
214 0 : BadRequest(e) => ApiError::BadRequest(e),
215 0 : Unavailable(_) => ApiError::ShuttingDown,
216 0 : e @ InProgress => ApiError::Conflict(format!("{e}")),
217 0 : Flush(e) | Other(e) => ApiError::InternalServerError(e),
218 : }
219 0 : }
220 : }
221 :
222 : impl From<TenantMapError> for ApiError {
223 0 : fn from(e: TenantMapError) -> ApiError {
224 0 : use TenantMapError::*;
225 0 : match e {
226 : StillInitializing | ShuttingDown => {
227 0 : ApiError::ResourceUnavailable(format!("{e}").into())
228 0 : }
229 0 : }
230 0 : }
231 : }
232 :
233 : impl From<TenantStateError> for ApiError {
234 0 : fn from(tse: TenantStateError) -> ApiError {
235 0 : match tse {
236 : TenantStateError::IsStopping(_) => {
237 0 : ApiError::ResourceUnavailable("Tenant is stopping".into())
238 : }
239 0 : TenantStateError::SlotError(e) => e.into(),
240 0 : TenantStateError::SlotUpsertError(e) => e.into(),
241 0 : TenantStateError::Other(e) => ApiError::InternalServerError(anyhow!(e)),
242 : }
243 0 : }
244 : }
245 :
246 : impl From<GetTenantError> for ApiError {
247 0 : fn from(tse: GetTenantError) -> ApiError {
248 0 : match tse {
249 0 : GetTenantError::NotFound(tid) => ApiError::NotFound(anyhow!("tenant {}", tid).into()),
250 0 : GetTenantError::Broken(reason) => {
251 0 : ApiError::InternalServerError(anyhow!("tenant is broken: {}", reason))
252 : }
253 : GetTenantError::NotActive(_) => {
254 : // Why is this not `ApiError::NotFound`?
255 : // Because we must be careful to never return 404 for a tenant if it does
256 : // in fact exist locally. If we did, the caller could draw the conclusion
257 : // that it can attach the tenant to another PS and we'd be in split-brain.
258 : //
259 : // (We can produce this variant only in `mgr::get_tenant(..., active=true)` calls).
260 0 : ApiError::ResourceUnavailable("Tenant not yet active".into())
261 : }
262 0 : GetTenantError::MapState(e) => ApiError::ResourceUnavailable(format!("{e}").into()),
263 : }
264 0 : }
265 : }
266 :
267 : impl From<GetActiveTenantError> for ApiError {
268 0 : fn from(e: GetActiveTenantError) -> ApiError {
269 0 : match e {
270 0 : GetActiveTenantError::WillNotBecomeActive(_) => ApiError::Conflict(format!("{}", e)),
271 0 : GetActiveTenantError::Cancelled => ApiError::ShuttingDown,
272 0 : GetActiveTenantError::NotFound(gte) => gte.into(),
273 : GetActiveTenantError::WaitForActiveTimeout { .. } => {
274 0 : ApiError::ResourceUnavailable(format!("{}", e).into())
275 : }
276 : }
277 0 : }
278 : }
279 :
280 : impl From<SetNewTenantConfigError> for ApiError {
281 0 : fn from(e: SetNewTenantConfigError) -> ApiError {
282 0 : match e {
283 0 : SetNewTenantConfigError::GetTenant(tid) => {
284 0 : ApiError::NotFound(anyhow!("tenant {}", tid).into())
285 : }
286 0 : e @ (SetNewTenantConfigError::Persist(_) | SetNewTenantConfigError::Other(_)) => {
287 0 : ApiError::InternalServerError(anyhow::Error::new(e))
288 : }
289 : }
290 0 : }
291 : }
292 :
293 : impl From<crate::tenant::DeleteTimelineError> for ApiError {
294 0 : fn from(value: crate::tenant::DeleteTimelineError) -> Self {
295 0 : use crate::tenant::DeleteTimelineError::*;
296 0 : match value {
297 0 : NotFound => ApiError::NotFound(anyhow::anyhow!("timeline not found").into()),
298 0 : HasChildren(children) => ApiError::PreconditionFailed(
299 0 : format!("Cannot delete timeline which has child timelines: {children:?}")
300 0 : .into_boxed_str(),
301 0 : ),
302 0 : a @ AlreadyInProgress(_) => ApiError::Conflict(a.to_string()),
303 0 : Other(e) => ApiError::InternalServerError(e),
304 : }
305 0 : }
306 : }
307 :
308 : impl From<crate::tenant::mgr::DeleteTimelineError> for ApiError {
309 0 : fn from(value: crate::tenant::mgr::DeleteTimelineError) -> Self {
310 : use crate::tenant::mgr::DeleteTimelineError::*;
311 0 : match value {
312 : // Report Precondition failed so client can distinguish between
313 : // "tenant is missing" case from "timeline is missing"
314 0 : Tenant(GetTenantError::NotFound(..)) => ApiError::PreconditionFailed(
315 0 : "Requested tenant is missing".to_owned().into_boxed_str(),
316 0 : ),
317 0 : Tenant(t) => ApiError::from(t),
318 0 : Timeline(t) => ApiError::from(t),
319 : }
320 0 : }
321 : }
322 :
323 : impl From<crate::tenant::delete::DeleteTenantError> for ApiError {
324 0 : fn from(value: crate::tenant::delete::DeleteTenantError) -> Self {
325 0 : use crate::tenant::delete::DeleteTenantError::*;
326 0 : match value {
327 0 : Get(g) => ApiError::from(g),
328 0 : e @ AlreadyInProgress => ApiError::Conflict(e.to_string()),
329 0 : Timeline(t) => ApiError::from(t),
330 0 : NotAttached => ApiError::NotFound(anyhow::anyhow!("Tenant is not attached").into()),
331 0 : SlotError(e) => e.into(),
332 0 : SlotUpsertError(e) => e.into(),
333 0 : Other(o) => ApiError::InternalServerError(o),
334 0 : e @ InvalidState(_) => ApiError::PreconditionFailed(e.to_string().into_boxed_str()),
335 0 : Cancelled => ApiError::ShuttingDown,
336 : }
337 0 : }
338 : }
339 :
340 : // Helper function to construct a TimelineInfo struct for a timeline
341 0 : async fn build_timeline_info(
342 0 : timeline: &Arc<Timeline>,
343 0 : include_non_incremental_logical_size: bool,
344 0 : force_await_initial_logical_size: bool,
345 0 : ctx: &RequestContext,
346 0 : ) -> anyhow::Result<TimelineInfo> {
347 0 : crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
348 0 :
349 0 : if force_await_initial_logical_size {
350 0 : timeline.clone().await_initial_logical_size().await
351 0 : }
352 :
353 0 : let mut info = build_timeline_info_common(
354 0 : timeline,
355 0 : ctx,
356 0 : tenant::timeline::GetLogicalSizePriority::Background,
357 0 : )
358 0 : .await?;
359 0 : if include_non_incremental_logical_size {
360 : // XXX we should be using spawn_ondemand_logical_size_calculation here.
361 : // Otherwise, if someone deletes the timeline / detaches the tenant while
362 : // we're executing this function, we will outlive the timeline on-disk state.
363 : info.current_logical_size_non_incremental = Some(
364 0 : timeline
365 0 : .get_current_logical_size_non_incremental(info.last_record_lsn, ctx)
366 0 : .await?,
367 : );
368 0 : }
369 0 : Ok(info)
370 0 : }
371 :
372 0 : async fn build_timeline_info_common(
373 0 : timeline: &Arc<Timeline>,
374 0 : ctx: &RequestContext,
375 0 : logical_size_task_priority: tenant::timeline::GetLogicalSizePriority,
376 0 : ) -> anyhow::Result<TimelineInfo> {
377 0 : crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
378 0 : let initdb_lsn = timeline.initdb_lsn;
379 0 : let last_record_lsn = timeline.get_last_record_lsn();
380 0 : let (wal_source_connstr, last_received_msg_lsn, last_received_msg_ts) = {
381 0 : let guard = timeline.last_received_wal.lock().unwrap();
382 0 : if let Some(info) = guard.as_ref() {
383 0 : (
384 0 : Some(format!("{:?}", info.wal_source_connconf)), // Password is hidden, but it's for statistics only.
385 0 : Some(info.last_received_msg_lsn),
386 0 : Some(info.last_received_msg_ts),
387 0 : )
388 : } else {
389 0 : (None, None, None)
390 : }
391 : };
392 :
393 0 : let ancestor_timeline_id = timeline.get_ancestor_timeline_id();
394 0 : let ancestor_lsn = match timeline.get_ancestor_lsn() {
395 0 : Lsn(0) => None,
396 0 : lsn @ Lsn(_) => Some(lsn),
397 : };
398 0 : let current_logical_size = timeline.get_current_logical_size(logical_size_task_priority, ctx);
399 0 : let current_physical_size = Some(timeline.layer_size_sum().await);
400 0 : let state = timeline.current_state();
401 0 : let remote_consistent_lsn_projected = timeline
402 0 : .get_remote_consistent_lsn_projected()
403 0 : .unwrap_or(Lsn(0));
404 0 : let remote_consistent_lsn_visible = timeline
405 0 : .get_remote_consistent_lsn_visible()
406 0 : .unwrap_or(Lsn(0));
407 0 :
408 0 : let walreceiver_status = timeline.walreceiver_status();
409 :
410 0 : let info = TimelineInfo {
411 0 : tenant_id: timeline.tenant_shard_id,
412 0 : timeline_id: timeline.timeline_id,
413 0 : ancestor_timeline_id,
414 0 : ancestor_lsn,
415 0 : disk_consistent_lsn: timeline.get_disk_consistent_lsn(),
416 0 : remote_consistent_lsn: remote_consistent_lsn_projected,
417 0 : remote_consistent_lsn_visible,
418 0 : initdb_lsn,
419 0 : last_record_lsn,
420 0 : prev_record_lsn: Some(timeline.get_prev_record_lsn()),
421 0 : latest_gc_cutoff_lsn: *timeline.get_latest_gc_cutoff_lsn(),
422 0 : current_logical_size: current_logical_size.size_dont_care_about_accuracy(),
423 0 : current_logical_size_is_accurate: match current_logical_size.accuracy() {
424 0 : tenant::timeline::logical_size::Accuracy::Approximate => false,
425 0 : tenant::timeline::logical_size::Accuracy::Exact => true,
426 : },
427 0 : directory_entries_counts: timeline.get_directory_metrics().to_vec(),
428 0 : current_physical_size,
429 0 : current_logical_size_non_incremental: None,
430 0 : timeline_dir_layer_file_size_sum: None,
431 0 : wal_source_connstr,
432 0 : last_received_msg_lsn,
433 0 : last_received_msg_ts,
434 0 : pg_version: timeline.pg_version,
435 0 :
436 0 : state,
437 0 :
438 0 : walreceiver_status,
439 0 : };
440 0 : Ok(info)
441 0 : }
442 :
443 : // healthcheck handler
444 0 : async fn status_handler(
445 0 : request: Request<Body>,
446 0 : _cancel: CancellationToken,
447 0 : ) -> Result<Response<Body>, ApiError> {
448 0 : check_permission(&request, None)?;
449 0 : let config = get_config(&request);
450 0 : json_response(StatusCode::OK, StatusResponse { id: config.id })
451 0 : }
452 :
453 0 : async fn reload_auth_validation_keys_handler(
454 0 : request: Request<Body>,
455 0 : _cancel: CancellationToken,
456 0 : ) -> Result<Response<Body>, ApiError> {
457 0 : check_permission(&request, None)?;
458 0 : let config = get_config(&request);
459 0 : let state = get_state(&request);
460 0 : let Some(shared_auth) = &state.auth else {
461 0 : return json_response(StatusCode::BAD_REQUEST, ());
462 : };
463 : // unwrap is ok because check is performed when creating config, so path is set and exists
464 0 : let key_path = config.auth_validation_public_key_path.as_ref().unwrap();
465 0 : info!("Reloading public key(s) for verifying JWT tokens from {key_path:?}");
466 :
467 0 : match JwtAuth::from_key_path(key_path) {
468 0 : Ok(new_auth) => {
469 0 : shared_auth.swap(new_auth);
470 0 : json_response(StatusCode::OK, ())
471 : }
472 0 : Err(e) => {
473 0 : warn!("Error reloading public keys from {key_path:?}: {e:}");
474 0 : json_response(StatusCode::INTERNAL_SERVER_ERROR, ())
475 : }
476 : }
477 0 : }
478 :
479 0 : async fn timeline_create_handler(
480 0 : mut request: Request<Body>,
481 0 : _cancel: CancellationToken,
482 0 : ) -> Result<Response<Body>, ApiError> {
483 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
484 0 : let request_data: TimelineCreateRequest = json_request(&mut request).await?;
485 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
486 :
487 0 : let new_timeline_id = request_data.new_timeline_id;
488 0 :
489 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
490 0 :
491 0 : let state = get_state(&request);
492 :
493 0 : async {
494 0 : let tenant = state
495 0 : .tenant_manager
496 0 : .get_attached_tenant_shard(tenant_shard_id, false)?;
497 :
498 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
499 :
500 0 : if let Some(ancestor_id) = request_data.ancestor_timeline_id.as_ref() {
501 0 : tracing::info!(%ancestor_id, "starting to branch");
502 : } else {
503 0 : tracing::info!("bootstrapping");
504 : }
505 :
506 0 : match tenant
507 0 : .create_timeline(
508 0 : new_timeline_id,
509 0 : request_data.ancestor_timeline_id,
510 0 : request_data.ancestor_start_lsn,
511 0 : request_data.pg_version.unwrap_or(crate::DEFAULT_PG_VERSION),
512 0 : request_data.existing_initdb_timeline_id,
513 0 : state.broker_client.clone(),
514 0 : &ctx,
515 0 : )
516 0 : .await
517 : {
518 0 : Ok(new_timeline) => {
519 : // Created. Construct a TimelineInfo for it.
520 0 : let timeline_info = build_timeline_info_common(
521 0 : &new_timeline,
522 0 : &ctx,
523 0 : tenant::timeline::GetLogicalSizePriority::User,
524 0 : )
525 0 : .await
526 0 : .map_err(ApiError::InternalServerError)?;
527 0 : json_response(StatusCode::CREATED, timeline_info)
528 : }
529 0 : Err(_) if tenant.cancel.is_cancelled() => {
530 0 : // In case we get some ugly error type during shutdown, cast it into a clean 503.
531 0 : json_response(
532 0 : StatusCode::SERVICE_UNAVAILABLE,
533 0 : HttpErrorBody::from_msg("Tenant shutting down".to_string()),
534 0 : )
535 : }
536 : Err(
537 : tenant::CreateTimelineError::Conflict
538 : | tenant::CreateTimelineError::AlreadyCreating,
539 0 : ) => json_response(StatusCode::CONFLICT, ()),
540 0 : Err(tenant::CreateTimelineError::AncestorLsn(err)) => json_response(
541 0 : StatusCode::NOT_ACCEPTABLE,
542 0 : HttpErrorBody::from_msg(format!("{err:#}")),
543 0 : ),
544 0 : Err(e @ tenant::CreateTimelineError::AncestorNotActive) => json_response(
545 0 : StatusCode::SERVICE_UNAVAILABLE,
546 0 : HttpErrorBody::from_msg(e.to_string()),
547 0 : ),
548 0 : Err(tenant::CreateTimelineError::ShuttingDown) => json_response(
549 0 : StatusCode::SERVICE_UNAVAILABLE,
550 0 : HttpErrorBody::from_msg("tenant shutting down".to_string()),
551 0 : ),
552 0 : Err(tenant::CreateTimelineError::Other(err)) => Err(ApiError::InternalServerError(err)),
553 : }
554 0 : }
555 : .instrument(info_span!("timeline_create",
556 : tenant_id = %tenant_shard_id.tenant_id,
557 0 : shard_id = %tenant_shard_id.shard_slug(),
558 : timeline_id = %new_timeline_id,
559 : lsn=?request_data.ancestor_start_lsn,
560 : pg_version=?request_data.pg_version
561 : ))
562 0 : .await
563 0 : }
564 :
565 0 : async fn timeline_list_handler(
566 0 : request: Request<Body>,
567 0 : _cancel: CancellationToken,
568 0 : ) -> Result<Response<Body>, ApiError> {
569 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
570 0 : let include_non_incremental_logical_size: Option<bool> =
571 0 : parse_query_param(&request, "include-non-incremental-logical-size")?;
572 0 : let force_await_initial_logical_size: Option<bool> =
573 0 : parse_query_param(&request, "force-await-initial-logical-size")?;
574 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
575 :
576 0 : let state = get_state(&request);
577 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
578 :
579 0 : let response_data = async {
580 0 : let tenant = state
581 0 : .tenant_manager
582 0 : .get_attached_tenant_shard(tenant_shard_id, false)?;
583 :
584 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
585 :
586 0 : let timelines = tenant.list_timelines();
587 0 :
588 0 : let mut response_data = Vec::with_capacity(timelines.len());
589 0 : for timeline in timelines {
590 0 : let timeline_info = build_timeline_info(
591 0 : &timeline,
592 0 : include_non_incremental_logical_size.unwrap_or(false),
593 0 : force_await_initial_logical_size.unwrap_or(false),
594 0 : &ctx,
595 0 : )
596 0 : .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
597 0 : .await
598 0 : .context("Failed to convert tenant timeline {timeline_id} into the local one: {e:?}")
599 0 : .map_err(ApiError::InternalServerError)?;
600 :
601 0 : response_data.push(timeline_info);
602 : }
603 0 : Ok::<Vec<TimelineInfo>, ApiError>(response_data)
604 0 : }
605 : .instrument(info_span!("timeline_list",
606 : tenant_id = %tenant_shard_id.tenant_id,
607 0 : shard_id = %tenant_shard_id.shard_slug()))
608 0 : .await?;
609 :
610 0 : json_response(StatusCode::OK, response_data)
611 0 : }
612 :
613 0 : async fn timeline_preserve_initdb_handler(
614 0 : request: Request<Body>,
615 0 : _cancel: CancellationToken,
616 0 : ) -> Result<Response<Body>, ApiError> {
617 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
618 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
619 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
620 :
621 : // Part of the process for disaster recovery from safekeeper-stored WAL:
622 : // If we don't recover into a new timeline but want to keep the timeline ID,
623 : // then the initdb archive is deleted. This endpoint copies it to a different
624 : // location where timeline recreation cand find it.
625 :
626 0 : async {
627 0 : let tenant = mgr::get_tenant(tenant_shard_id, false)?;
628 :
629 0 : let timeline = tenant
630 0 : .get_timeline(timeline_id, false)
631 0 : .map_err(|e| ApiError::NotFound(e.into()))?;
632 :
633 0 : timeline
634 0 : .preserve_initdb_archive()
635 0 : .await
636 0 : .context("preserving initdb archive")
637 0 : .map_err(ApiError::InternalServerError)?;
638 :
639 0 : Ok::<_, ApiError>(())
640 0 : }
641 : .instrument(info_span!("timeline_preserve_initdb_archive",
642 : tenant_id = %tenant_shard_id.tenant_id,
643 0 : shard_id = %tenant_shard_id.shard_slug(),
644 : %timeline_id))
645 0 : .await?;
646 :
647 0 : json_response(StatusCode::OK, ())
648 0 : }
649 :
650 0 : async fn timeline_detail_handler(
651 0 : request: Request<Body>,
652 0 : _cancel: CancellationToken,
653 0 : ) -> Result<Response<Body>, ApiError> {
654 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
655 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
656 0 : let include_non_incremental_logical_size: Option<bool> =
657 0 : parse_query_param(&request, "include-non-incremental-logical-size")?;
658 0 : let force_await_initial_logical_size: Option<bool> =
659 0 : parse_query_param(&request, "force-await-initial-logical-size")?;
660 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
661 :
662 : // Logical size calculation needs downloading.
663 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
664 0 : let state = get_state(&request);
665 :
666 0 : let timeline_info = async {
667 0 : let tenant = state
668 0 : .tenant_manager
669 0 : .get_attached_tenant_shard(tenant_shard_id, false)?;
670 :
671 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
672 :
673 0 : let timeline = tenant
674 0 : .get_timeline(timeline_id, false)
675 0 : .map_err(|e| ApiError::NotFound(e.into()))?;
676 :
677 0 : let timeline_info = build_timeline_info(
678 0 : &timeline,
679 0 : include_non_incremental_logical_size.unwrap_or(false),
680 0 : force_await_initial_logical_size.unwrap_or(false),
681 0 : &ctx,
682 0 : )
683 0 : .await
684 0 : .context("get local timeline info")
685 0 : .map_err(ApiError::InternalServerError)?;
686 :
687 0 : Ok::<_, ApiError>(timeline_info)
688 0 : }
689 : .instrument(info_span!("timeline_detail",
690 : tenant_id = %tenant_shard_id.tenant_id,
691 0 : shard_id = %tenant_shard_id.shard_slug(),
692 : %timeline_id))
693 0 : .await?;
694 :
695 0 : json_response(StatusCode::OK, timeline_info)
696 0 : }
697 :
698 0 : async fn get_lsn_by_timestamp_handler(
699 0 : request: Request<Body>,
700 0 : cancel: CancellationToken,
701 0 : ) -> Result<Response<Body>, ApiError> {
702 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
703 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
704 0 : let state = get_state(&request);
705 0 :
706 0 : if !tenant_shard_id.is_zero() {
707 : // Requires SLRU contents, which are only stored on shard zero
708 0 : return Err(ApiError::BadRequest(anyhow!(
709 0 : "Size calculations are only available on shard zero"
710 0 : )));
711 0 : }
712 :
713 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
714 0 : let timestamp_raw = must_get_query_param(&request, "timestamp")?;
715 0 : let timestamp = humantime::parse_rfc3339(×tamp_raw)
716 0 : .with_context(|| format!("Invalid time: {:?}", timestamp_raw))
717 0 : .map_err(ApiError::BadRequest)?;
718 0 : let timestamp_pg = postgres_ffi::to_pg_timestamp(timestamp);
719 0 :
720 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
721 :
722 0 : let timeline =
723 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
724 0 : .await?;
725 0 : let result = timeline
726 0 : .find_lsn_for_timestamp(timestamp_pg, &cancel, &ctx)
727 0 : .await?;
728 0 : #[derive(serde::Serialize, Debug)]
729 : struct Result {
730 : lsn: Lsn,
731 : kind: &'static str,
732 : }
733 0 : let (lsn, kind) = match result {
734 0 : LsnForTimestamp::Present(lsn) => (lsn, "present"),
735 0 : LsnForTimestamp::Future(lsn) => (lsn, "future"),
736 0 : LsnForTimestamp::Past(lsn) => (lsn, "past"),
737 0 : LsnForTimestamp::NoData(lsn) => (lsn, "nodata"),
738 : };
739 0 : let result = Result { lsn, kind };
740 0 : tracing::info!(
741 0 : lsn=?result.lsn,
742 0 : kind=%result.kind,
743 0 : timestamp=%timestamp_raw,
744 0 : "lsn_by_timestamp finished"
745 0 : );
746 0 : json_response(StatusCode::OK, result)
747 0 : }
748 :
749 0 : async fn get_timestamp_of_lsn_handler(
750 0 : request: Request<Body>,
751 0 : _cancel: CancellationToken,
752 0 : ) -> Result<Response<Body>, ApiError> {
753 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
754 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
755 0 : let state = get_state(&request);
756 0 :
757 0 : if !tenant_shard_id.is_zero() {
758 : // Requires SLRU contents, which are only stored on shard zero
759 0 : return Err(ApiError::BadRequest(anyhow!(
760 0 : "Size calculations are only available on shard zero"
761 0 : )));
762 0 : }
763 :
764 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
765 :
766 0 : let lsn_str = must_get_query_param(&request, "lsn")?;
767 0 : let lsn = Lsn::from_str(&lsn_str)
768 0 : .with_context(|| format!("Invalid LSN: {lsn_str:?}"))
769 0 : .map_err(ApiError::BadRequest)?;
770 :
771 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
772 0 : let timeline =
773 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
774 0 : .await?;
775 0 : let result = timeline.get_timestamp_for_lsn(lsn, &ctx).await?;
776 :
777 0 : match result {
778 0 : Some(time) => {
779 0 : let time = format_rfc3339(postgres_ffi::from_pg_timestamp(time)).to_string();
780 0 : json_response(StatusCode::OK, time)
781 : }
782 0 : None => json_response(StatusCode::NOT_FOUND, ()),
783 : }
784 0 : }
785 :
786 0 : async fn tenant_attach_handler(
787 0 : mut request: Request<Body>,
788 0 : _cancel: CancellationToken,
789 0 : ) -> Result<Response<Body>, ApiError> {
790 0 : let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
791 0 : check_permission(&request, Some(tenant_id))?;
792 :
793 0 : let maybe_body: Option<TenantAttachRequest> = json_request_or_empty_body(&mut request).await?;
794 0 : let tenant_conf = match &maybe_body {
795 0 : Some(request) => TenantConfOpt::try_from(&*request.config).map_err(ApiError::BadRequest)?,
796 0 : None => TenantConfOpt::default(),
797 : };
798 :
799 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
800 :
801 0 : info!("Handling tenant attach {tenant_id}");
802 :
803 0 : let state = get_state(&request);
804 :
805 0 : let generation = get_request_generation(state, maybe_body.as_ref().and_then(|r| r.generation))?;
806 :
807 0 : if state.remote_storage.is_none() {
808 0 : return Err(ApiError::BadRequest(anyhow!(
809 0 : "attach_tenant is not possible because pageserver was configured without remote storage"
810 0 : )));
811 0 : }
812 0 :
813 0 : let tenant_shard_id = TenantShardId::unsharded(tenant_id);
814 0 : let shard_params = ShardParameters::default();
815 0 : let location_conf = LocationConf::attached_single(tenant_conf, generation, &shard_params);
816 :
817 0 : let tenant = state
818 0 : .tenant_manager
819 0 : .upsert_location(
820 0 : tenant_shard_id,
821 0 : location_conf,
822 0 : None,
823 0 : SpawnMode::Normal,
824 0 : &ctx,
825 0 : )
826 0 : .await?;
827 :
828 0 : let Some(tenant) = tenant else {
829 : // This should never happen: indicates a bug in upsert_location
830 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!(
831 0 : "Upsert succeeded but didn't return tenant!"
832 0 : )));
833 : };
834 :
835 : // We might have successfully constructed a Tenant, but it could still
836 : // end up in a broken state:
837 : if let TenantState::Broken {
838 0 : reason,
839 : backtrace: _,
840 0 : } = tenant.current_state()
841 : {
842 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!(
843 0 : "Tenant state is Broken: {reason}"
844 0 : )));
845 0 : }
846 0 :
847 0 : json_response(StatusCode::ACCEPTED, ())
848 0 : }
849 :
850 0 : async fn timeline_delete_handler(
851 0 : request: Request<Body>,
852 0 : _cancel: CancellationToken,
853 0 : ) -> Result<Response<Body>, ApiError> {
854 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
855 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
856 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
857 :
858 0 : let state = get_state(&request);
859 :
860 0 : let tenant = state
861 0 : .tenant_manager
862 0 : .get_attached_tenant_shard(tenant_shard_id, false)
863 0 : .map_err(|e| {
864 0 : match e {
865 : // GetTenantError has a built-in conversion to ApiError, but in this context we don't
866 : // want to treat missing tenants as 404, to avoid ambiguity with successful deletions.
867 0 : GetTenantError::NotFound(_) => ApiError::PreconditionFailed(
868 0 : "Requested tenant is missing".to_string().into_boxed_str(),
869 0 : ),
870 0 : e => e.into(),
871 : }
872 0 : })?;
873 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
874 0 : tenant.delete_timeline(timeline_id).instrument(info_span!("timeline_delete", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(), %timeline_id))
875 0 : .await?;
876 :
877 0 : json_response(StatusCode::ACCEPTED, ())
878 0 : }
879 :
880 0 : async fn tenant_detach_handler(
881 0 : request: Request<Body>,
882 0 : _cancel: CancellationToken,
883 0 : ) -> Result<Response<Body>, ApiError> {
884 0 : let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
885 0 : check_permission(&request, Some(tenant_id))?;
886 0 : let detach_ignored: Option<bool> = parse_query_param(&request, "detach_ignored")?;
887 :
888 : // This is a legacy API (`/location_conf` is the replacement). It only supports unsharded tenants
889 0 : let tenant_shard_id = TenantShardId::unsharded(tenant_id);
890 0 :
891 0 : let state = get_state(&request);
892 0 : let conf = state.conf;
893 0 : mgr::detach_tenant(
894 0 : conf,
895 0 : tenant_shard_id,
896 0 : detach_ignored.unwrap_or(false),
897 0 : &state.deletion_queue_client,
898 0 : )
899 0 : .instrument(info_span!("tenant_detach", %tenant_id, shard_id=%tenant_shard_id.shard_slug()))
900 0 : .await?;
901 :
902 0 : json_response(StatusCode::OK, ())
903 0 : }
904 :
905 0 : async fn tenant_reset_handler(
906 0 : request: Request<Body>,
907 0 : _cancel: CancellationToken,
908 0 : ) -> Result<Response<Body>, ApiError> {
909 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
910 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
911 :
912 0 : let drop_cache: Option<bool> = parse_query_param(&request, "drop_cache")?;
913 :
914 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
915 0 : let state = get_state(&request);
916 0 : state
917 0 : .tenant_manager
918 0 : .reset_tenant(tenant_shard_id, drop_cache.unwrap_or(false), &ctx)
919 0 : .await
920 0 : .map_err(ApiError::InternalServerError)?;
921 :
922 0 : json_response(StatusCode::OK, ())
923 0 : }
924 :
925 0 : async fn tenant_load_handler(
926 0 : mut request: Request<Body>,
927 0 : _cancel: CancellationToken,
928 0 : ) -> Result<Response<Body>, ApiError> {
929 0 : let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
930 0 : check_permission(&request, Some(tenant_id))?;
931 :
932 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
933 :
934 0 : let maybe_body: Option<TenantLoadRequest> = json_request_or_empty_body(&mut request).await?;
935 :
936 0 : let state = get_state(&request);
937 :
938 : // The /load request is only usable when control_plane_api is not set. Once it is set, callers
939 : // should always use /attach instead.
940 0 : let generation = get_request_generation(state, maybe_body.as_ref().and_then(|r| r.generation))?;
941 :
942 0 : mgr::load_tenant(
943 0 : state.conf,
944 0 : tenant_id,
945 0 : generation,
946 0 : state.broker_client.clone(),
947 0 : state.remote_storage.clone(),
948 0 : state.deletion_queue_client.clone(),
949 0 : &ctx,
950 0 : )
951 0 : .instrument(info_span!("load", %tenant_id))
952 0 : .await?;
953 :
954 0 : json_response(StatusCode::ACCEPTED, ())
955 0 : }
956 :
957 0 : async fn tenant_ignore_handler(
958 0 : request: Request<Body>,
959 0 : _cancel: CancellationToken,
960 0 : ) -> Result<Response<Body>, ApiError> {
961 0 : let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
962 0 : check_permission(&request, Some(tenant_id))?;
963 :
964 0 : let state = get_state(&request);
965 0 : let conf = state.conf;
966 0 : mgr::ignore_tenant(conf, tenant_id)
967 0 : .instrument(info_span!("ignore_tenant", %tenant_id))
968 0 : .await?;
969 :
970 0 : json_response(StatusCode::OK, ())
971 0 : }
972 :
973 0 : async fn tenant_list_handler(
974 0 : request: Request<Body>,
975 0 : _cancel: CancellationToken,
976 0 : ) -> Result<Response<Body>, ApiError> {
977 0 : check_permission(&request, None)?;
978 :
979 0 : let response_data = mgr::list_tenants()
980 0 : .instrument(info_span!("tenant_list"))
981 0 : .await
982 0 : .map_err(|_| {
983 0 : ApiError::ResourceUnavailable("Tenant map is initializing or shutting down".into())
984 0 : })?
985 0 : .iter()
986 0 : .map(|(id, state, gen)| TenantInfo {
987 0 : id: *id,
988 0 : state: state.clone(),
989 0 : current_physical_size: None,
990 0 : attachment_status: state.attachment_status(),
991 0 : generation: (*gen).into(),
992 0 : })
993 0 : .collect::<Vec<TenantInfo>>();
994 0 :
995 0 : json_response(StatusCode::OK, response_data)
996 0 : }
997 :
998 0 : async fn tenant_status(
999 0 : request: Request<Body>,
1000 0 : _cancel: CancellationToken,
1001 0 : ) -> Result<Response<Body>, ApiError> {
1002 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1003 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1004 :
1005 0 : let tenant_info = async {
1006 0 : let tenant = mgr::get_tenant(tenant_shard_id, false)?;
1007 :
1008 : // Calculate total physical size of all timelines
1009 0 : let mut current_physical_size = 0;
1010 0 : for timeline in tenant.list_timelines().iter() {
1011 0 : current_physical_size += timeline.layer_size_sum().await;
1012 : }
1013 :
1014 0 : let state = tenant.current_state();
1015 0 : Result::<_, ApiError>::Ok(TenantDetails {
1016 0 : tenant_info: TenantInfo {
1017 0 : id: tenant_shard_id,
1018 0 : state: state.clone(),
1019 0 : current_physical_size: Some(current_physical_size),
1020 0 : attachment_status: state.attachment_status(),
1021 0 : generation: tenant.generation().into(),
1022 0 : },
1023 0 : walredo: tenant.wal_redo_manager_status(),
1024 0 : timelines: tenant.list_timeline_ids(),
1025 0 : })
1026 0 : }
1027 : .instrument(info_span!("tenant_status_handler",
1028 : tenant_id = %tenant_shard_id.tenant_id,
1029 0 : shard_id = %tenant_shard_id.shard_slug()))
1030 0 : .await?;
1031 :
1032 0 : json_response(StatusCode::OK, tenant_info)
1033 0 : }
1034 :
1035 0 : async fn tenant_delete_handler(
1036 0 : request: Request<Body>,
1037 0 : _cancel: CancellationToken,
1038 0 : ) -> Result<Response<Body>, ApiError> {
1039 : // TODO openapi spec
1040 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1041 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1042 :
1043 0 : let state = get_state(&request);
1044 0 :
1045 0 : state
1046 0 : .tenant_manager
1047 0 : .delete_tenant(tenant_shard_id, ACTIVE_TENANT_TIMEOUT)
1048 : .instrument(info_span!("tenant_delete_handler",
1049 : tenant_id = %tenant_shard_id.tenant_id,
1050 0 : shard_id = %tenant_shard_id.shard_slug()
1051 : ))
1052 0 : .await?;
1053 :
1054 0 : json_response(StatusCode::ACCEPTED, ())
1055 0 : }
1056 :
1057 : /// HTTP endpoint to query the current tenant_size of a tenant.
1058 : ///
1059 : /// This is not used by consumption metrics under [`crate::consumption_metrics`], but can be used
1060 : /// to debug any of the calculations. Requires `tenant_id` request parameter, supports
1061 : /// `inputs_only=true|false` (default false) which supports debugging failure to calculate model
1062 : /// values.
1063 : ///
1064 : /// 'retention_period' query parameter overrides the cutoff that is used to calculate the size
1065 : /// (only if it is shorter than the real cutoff).
1066 : ///
1067 : /// Note: we don't update the cached size and prometheus metric here.
1068 : /// The retention period might be different, and it's nice to have a method to just calculate it
1069 : /// without modifying anything anyway.
1070 0 : async fn tenant_size_handler(
1071 0 : request: Request<Body>,
1072 0 : cancel: CancellationToken,
1073 0 : ) -> Result<Response<Body>, ApiError> {
1074 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1075 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1076 0 : let inputs_only: Option<bool> = parse_query_param(&request, "inputs_only")?;
1077 0 : let retention_period: Option<u64> = parse_query_param(&request, "retention_period")?;
1078 0 : let headers = request.headers();
1079 0 :
1080 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1081 0 : let tenant = mgr::get_tenant(tenant_shard_id, true)?;
1082 :
1083 0 : if !tenant_shard_id.is_zero() {
1084 0 : return Err(ApiError::BadRequest(anyhow!(
1085 0 : "Size calculations are only available on shard zero"
1086 0 : )));
1087 0 : }
1088 :
1089 : // this can be long operation
1090 0 : let inputs = tenant
1091 0 : .gather_size_inputs(
1092 0 : retention_period,
1093 0 : LogicalSizeCalculationCause::TenantSizeHandler,
1094 0 : &cancel,
1095 0 : &ctx,
1096 0 : )
1097 0 : .await
1098 0 : .map_err(ApiError::InternalServerError)?;
1099 :
1100 0 : let mut sizes = None;
1101 0 : let accepts_html = headers
1102 0 : .get(header::ACCEPT)
1103 0 : .map(|v| v == "text/html")
1104 0 : .unwrap_or_default();
1105 0 : if !inputs_only.unwrap_or(false) {
1106 0 : let storage_model = inputs
1107 0 : .calculate_model()
1108 0 : .map_err(ApiError::InternalServerError)?;
1109 0 : let size = storage_model.calculate();
1110 0 :
1111 0 : // If request header expects html, return html
1112 0 : if accepts_html {
1113 0 : return synthetic_size_html_response(inputs, storage_model, size);
1114 0 : }
1115 0 : sizes = Some(size);
1116 0 : } else if accepts_html {
1117 0 : return Err(ApiError::BadRequest(anyhow!(
1118 0 : "inputs_only parameter is incompatible with html output request"
1119 0 : )));
1120 0 : }
1121 :
1122 : /// The type resides in the pageserver not to expose `ModelInputs`.
1123 0 : #[derive(serde::Serialize)]
1124 : struct TenantHistorySize {
1125 : id: TenantId,
1126 : /// Size is a mixture of WAL and logical size, so the unit is bytes.
1127 : ///
1128 : /// Will be none if `?inputs_only=true` was given.
1129 : size: Option<u64>,
1130 : /// Size of each segment used in the model.
1131 : /// Will be null if `?inputs_only=true` was given.
1132 : segment_sizes: Option<Vec<tenant_size_model::SegmentSizeResult>>,
1133 : inputs: crate::tenant::size::ModelInputs,
1134 : }
1135 :
1136 0 : json_response(
1137 0 : StatusCode::OK,
1138 0 : TenantHistorySize {
1139 0 : id: tenant_shard_id.tenant_id,
1140 0 : size: sizes.as_ref().map(|x| x.total_size),
1141 0 : segment_sizes: sizes.map(|x| x.segments),
1142 0 : inputs,
1143 0 : },
1144 0 : )
1145 0 : }
1146 :
1147 0 : async fn tenant_shard_split_handler(
1148 0 : mut request: Request<Body>,
1149 0 : _cancel: CancellationToken,
1150 0 : ) -> Result<Response<Body>, ApiError> {
1151 0 : let req: TenantShardSplitRequest = json_request(&mut request).await?;
1152 :
1153 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1154 0 : let state = get_state(&request);
1155 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
1156 :
1157 0 : let new_shards = state
1158 0 : .tenant_manager
1159 0 : .shard_split(tenant_shard_id, ShardCount::new(req.new_shard_count), &ctx)
1160 0 : .await
1161 0 : .map_err(ApiError::InternalServerError)?;
1162 :
1163 0 : json_response(StatusCode::OK, TenantShardSplitResponse { new_shards })
1164 0 : }
1165 :
1166 0 : async fn layer_map_info_handler(
1167 0 : request: Request<Body>,
1168 0 : _cancel: CancellationToken,
1169 0 : ) -> Result<Response<Body>, ApiError> {
1170 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1171 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1172 0 : let reset: LayerAccessStatsReset =
1173 0 : parse_query_param(&request, "reset")?.unwrap_or(LayerAccessStatsReset::NoReset);
1174 0 : let state = get_state(&request);
1175 0 :
1176 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1177 :
1178 0 : let timeline =
1179 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1180 0 : .await?;
1181 0 : let layer_map_info = timeline.layer_map_info(reset).await;
1182 :
1183 0 : json_response(StatusCode::OK, layer_map_info)
1184 0 : }
1185 :
1186 0 : async fn layer_download_handler(
1187 0 : request: Request<Body>,
1188 0 : _cancel: CancellationToken,
1189 0 : ) -> Result<Response<Body>, ApiError> {
1190 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1191 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1192 0 : let layer_file_name = get_request_param(&request, "layer_file_name")?;
1193 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1194 0 : let state = get_state(&request);
1195 :
1196 0 : let timeline =
1197 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1198 0 : .await?;
1199 0 : let downloaded = timeline
1200 0 : .download_layer(layer_file_name)
1201 0 : .await
1202 0 : .map_err(ApiError::InternalServerError)?;
1203 :
1204 0 : match downloaded {
1205 0 : Some(true) => json_response(StatusCode::OK, ()),
1206 0 : Some(false) => json_response(StatusCode::NOT_MODIFIED, ()),
1207 0 : None => json_response(
1208 0 : StatusCode::BAD_REQUEST,
1209 0 : format!("Layer {tenant_shard_id}/{timeline_id}/{layer_file_name} not found"),
1210 0 : ),
1211 : }
1212 0 : }
1213 :
1214 0 : async fn evict_timeline_layer_handler(
1215 0 : request: Request<Body>,
1216 0 : _cancel: CancellationToken,
1217 0 : ) -> Result<Response<Body>, ApiError> {
1218 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1219 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1220 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1221 0 : let layer_file_name = get_request_param(&request, "layer_file_name")?;
1222 0 : let state = get_state(&request);
1223 :
1224 0 : let timeline =
1225 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1226 0 : .await?;
1227 0 : let evicted = timeline
1228 0 : .evict_layer(layer_file_name)
1229 0 : .await
1230 0 : .map_err(ApiError::InternalServerError)?;
1231 :
1232 0 : match evicted {
1233 0 : Some(true) => json_response(StatusCode::OK, ()),
1234 0 : Some(false) => json_response(StatusCode::NOT_MODIFIED, ()),
1235 0 : None => json_response(
1236 0 : StatusCode::BAD_REQUEST,
1237 0 : format!("Layer {tenant_shard_id}/{timeline_id}/{layer_file_name} not found"),
1238 0 : ),
1239 : }
1240 0 : }
1241 :
1242 : /// Get tenant_size SVG graph along with the JSON data.
1243 0 : fn synthetic_size_html_response(
1244 0 : inputs: ModelInputs,
1245 0 : storage_model: StorageModel,
1246 0 : sizes: SizeResult,
1247 0 : ) -> Result<Response<Body>, ApiError> {
1248 0 : let mut timeline_ids: Vec<String> = Vec::new();
1249 0 : let mut timeline_map: HashMap<TimelineId, usize> = HashMap::new();
1250 0 : for (index, ti) in inputs.timeline_inputs.iter().enumerate() {
1251 0 : timeline_map.insert(ti.timeline_id, index);
1252 0 : timeline_ids.push(ti.timeline_id.to_string());
1253 0 : }
1254 0 : let seg_to_branch: Vec<usize> = inputs
1255 0 : .segments
1256 0 : .iter()
1257 0 : .map(|seg| *timeline_map.get(&seg.timeline_id).unwrap())
1258 0 : .collect();
1259 :
1260 0 : let svg =
1261 0 : tenant_size_model::svg::draw_svg(&storage_model, &timeline_ids, &seg_to_branch, &sizes)
1262 0 : .map_err(ApiError::InternalServerError)?;
1263 :
1264 0 : let mut response = String::new();
1265 0 :
1266 0 : use std::fmt::Write;
1267 0 : write!(response, "<html>\n<body>\n").unwrap();
1268 0 : write!(response, "<div>\n{svg}\n</div>").unwrap();
1269 0 : writeln!(response, "Project size: {}", sizes.total_size).unwrap();
1270 0 : writeln!(response, "<pre>").unwrap();
1271 0 : writeln!(
1272 0 : response,
1273 0 : "{}",
1274 0 : serde_json::to_string_pretty(&inputs).unwrap()
1275 0 : )
1276 0 : .unwrap();
1277 0 : writeln!(
1278 0 : response,
1279 0 : "{}",
1280 0 : serde_json::to_string_pretty(&sizes.segments).unwrap()
1281 0 : )
1282 0 : .unwrap();
1283 0 : writeln!(response, "</pre>").unwrap();
1284 0 : write!(response, "</body>\n</html>\n").unwrap();
1285 0 :
1286 0 : html_response(StatusCode::OK, response)
1287 0 : }
1288 :
1289 0 : pub fn html_response(status: StatusCode, data: String) -> Result<Response<Body>, ApiError> {
1290 0 : let response = Response::builder()
1291 0 : .status(status)
1292 0 : .header(header::CONTENT_TYPE, "text/html")
1293 0 : .body(Body::from(data.as_bytes().to_vec()))
1294 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
1295 0 : Ok(response)
1296 0 : }
1297 :
1298 : /// Helper for requests that may take a generation, which is mandatory
1299 : /// when control_plane_api is set, but otherwise defaults to Generation::none()
1300 0 : fn get_request_generation(state: &State, req_gen: Option<u32>) -> Result<Generation, ApiError> {
1301 0 : if state.conf.control_plane_api.is_some() {
1302 0 : req_gen
1303 0 : .map(Generation::new)
1304 0 : .ok_or(ApiError::BadRequest(anyhow!(
1305 0 : "generation attribute missing"
1306 0 : )))
1307 : } else {
1308 : // Legacy mode: all tenants operate with no generation
1309 0 : Ok(Generation::none())
1310 : }
1311 0 : }
1312 :
1313 0 : async fn tenant_create_handler(
1314 0 : mut request: Request<Body>,
1315 0 : _cancel: CancellationToken,
1316 0 : ) -> Result<Response<Body>, ApiError> {
1317 0 : let request_data: TenantCreateRequest = json_request(&mut request).await?;
1318 0 : let target_tenant_id = request_data.new_tenant_id;
1319 0 : check_permission(&request, None)?;
1320 :
1321 0 : let _timer = STORAGE_TIME_GLOBAL
1322 0 : .get_metric_with_label_values(&[StorageTimeOperation::CreateTenant.into()])
1323 0 : .expect("bug")
1324 0 : .start_timer();
1325 :
1326 0 : let tenant_conf =
1327 0 : TenantConfOpt::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
1328 :
1329 0 : let state = get_state(&request);
1330 :
1331 0 : let generation = get_request_generation(state, request_data.generation)?;
1332 :
1333 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
1334 0 :
1335 0 : let location_conf =
1336 0 : LocationConf::attached_single(tenant_conf, generation, &request_data.shard_parameters);
1337 :
1338 0 : let new_tenant = state
1339 0 : .tenant_manager
1340 0 : .upsert_location(
1341 0 : target_tenant_id,
1342 0 : location_conf,
1343 0 : None,
1344 0 : SpawnMode::Create,
1345 0 : &ctx,
1346 0 : )
1347 0 : .await?;
1348 :
1349 0 : let Some(new_tenant) = new_tenant else {
1350 : // This should never happen: indicates a bug in upsert_location
1351 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!(
1352 0 : "Upsert succeeded but didn't return tenant!"
1353 0 : )));
1354 : };
1355 : // We created the tenant. Existing API semantics are that the tenant
1356 : // is Active when this function returns.
1357 0 : new_tenant
1358 0 : .wait_to_become_active(ACTIVE_TENANT_TIMEOUT)
1359 0 : .await?;
1360 :
1361 0 : json_response(
1362 0 : StatusCode::CREATED,
1363 0 : TenantCreateResponse(new_tenant.tenant_shard_id().tenant_id),
1364 0 : )
1365 0 : }
1366 :
1367 0 : async fn get_tenant_config_handler(
1368 0 : request: Request<Body>,
1369 0 : _cancel: CancellationToken,
1370 0 : ) -> Result<Response<Body>, ApiError> {
1371 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1372 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1373 :
1374 0 : let tenant = mgr::get_tenant(tenant_shard_id, false)?;
1375 :
1376 0 : let response = HashMap::from([
1377 : (
1378 : "tenant_specific_overrides",
1379 0 : serde_json::to_value(tenant.tenant_specific_overrides())
1380 0 : .context("serializing tenant specific overrides")
1381 0 : .map_err(ApiError::InternalServerError)?,
1382 : ),
1383 : (
1384 0 : "effective_config",
1385 0 : serde_json::to_value(tenant.effective_config())
1386 0 : .context("serializing effective config")
1387 0 : .map_err(ApiError::InternalServerError)?,
1388 : ),
1389 : ]);
1390 :
1391 0 : json_response(StatusCode::OK, response)
1392 0 : }
1393 :
1394 0 : async fn update_tenant_config_handler(
1395 0 : mut request: Request<Body>,
1396 0 : _cancel: CancellationToken,
1397 0 : ) -> Result<Response<Body>, ApiError> {
1398 0 : let request_data: TenantConfigRequest = json_request(&mut request).await?;
1399 0 : let tenant_id = request_data.tenant_id;
1400 0 : check_permission(&request, Some(tenant_id))?;
1401 :
1402 0 : let tenant_conf =
1403 0 : TenantConfOpt::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
1404 :
1405 0 : let state = get_state(&request);
1406 0 : mgr::set_new_tenant_config(state.conf, tenant_conf, tenant_id)
1407 0 : .instrument(info_span!("tenant_config", %tenant_id))
1408 0 : .await?;
1409 :
1410 0 : json_response(StatusCode::OK, ())
1411 0 : }
1412 :
1413 0 : async fn put_tenant_location_config_handler(
1414 0 : mut request: Request<Body>,
1415 0 : _cancel: CancellationToken,
1416 0 : ) -> Result<Response<Body>, ApiError> {
1417 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1418 :
1419 0 : let request_data: TenantLocationConfigRequest = json_request(&mut request).await?;
1420 0 : let flush = parse_query_param(&request, "flush_ms")?.map(Duration::from_millis);
1421 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1422 :
1423 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
1424 0 : let state = get_state(&request);
1425 0 : let conf = state.conf;
1426 0 :
1427 0 : // The `Detached` state is special, it doesn't upsert a tenant, it removes
1428 0 : // its local disk content and drops it from memory.
1429 0 : if let LocationConfigMode::Detached = request_data.config.mode {
1430 0 : if let Err(e) =
1431 0 : mgr::detach_tenant(conf, tenant_shard_id, true, &state.deletion_queue_client)
1432 : .instrument(info_span!("tenant_detach",
1433 : tenant_id = %tenant_shard_id.tenant_id,
1434 0 : shard_id = %tenant_shard_id.shard_slug()
1435 : ))
1436 0 : .await
1437 : {
1438 0 : match e {
1439 0 : TenantStateError::SlotError(TenantSlotError::NotFound(_)) => {
1440 0 : // This API is idempotent: a NotFound on a detach is fine.
1441 0 : }
1442 0 : _ => return Err(e.into()),
1443 : }
1444 0 : }
1445 0 : return json_response(StatusCode::OK, ());
1446 0 : }
1447 :
1448 0 : let location_conf =
1449 0 : LocationConf::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
1450 :
1451 0 : let attached = state
1452 0 : .tenant_manager
1453 0 : .upsert_location(
1454 0 : tenant_shard_id,
1455 0 : location_conf,
1456 0 : flush,
1457 0 : tenant::SpawnMode::Normal,
1458 0 : &ctx,
1459 0 : )
1460 0 : .await?
1461 0 : .is_some();
1462 :
1463 0 : if let Some(_flush_ms) = flush {
1464 0 : match state
1465 0 : .secondary_controller
1466 0 : .upload_tenant(tenant_shard_id)
1467 0 : .await
1468 : {
1469 : Ok(()) => {
1470 0 : tracing::info!("Uploaded heatmap during flush");
1471 : }
1472 0 : Err(e) => {
1473 0 : tracing::warn!("Failed to flush heatmap: {e}");
1474 : }
1475 : }
1476 : } else {
1477 0 : tracing::info!("No flush requested when configuring");
1478 : }
1479 :
1480 : // This API returns a vector of pageservers where the tenant is attached: this is
1481 : // primarily for use in the sharding service. For compatibilty, we also return this
1482 : // when called directly on a pageserver, but the payload is always zero or one shards.
1483 0 : let mut response = TenantLocationConfigResponse { shards: Vec::new() };
1484 0 : if attached {
1485 0 : response.shards.push(TenantShardLocation {
1486 0 : shard_id: tenant_shard_id,
1487 0 : node_id: state.conf.id,
1488 0 : })
1489 0 : }
1490 :
1491 0 : json_response(StatusCode::OK, response)
1492 0 : }
1493 :
1494 0 : async fn list_location_config_handler(
1495 0 : request: Request<Body>,
1496 0 : _cancel: CancellationToken,
1497 0 : ) -> Result<Response<Body>, ApiError> {
1498 0 : let state = get_state(&request);
1499 0 : let slots = state.tenant_manager.list();
1500 0 : let result = LocationConfigListResponse {
1501 0 : tenant_shards: slots
1502 0 : .into_iter()
1503 0 : .map(|(tenant_shard_id, slot)| {
1504 0 : let v = match slot {
1505 0 : TenantSlot::Attached(t) => Some(t.get_location_conf()),
1506 0 : TenantSlot::Secondary(s) => Some(s.get_location_conf()),
1507 0 : TenantSlot::InProgress(_) => None,
1508 : };
1509 0 : (tenant_shard_id, v)
1510 0 : })
1511 0 : .collect(),
1512 0 : };
1513 0 : json_response(StatusCode::OK, result)
1514 0 : }
1515 :
1516 : // Do a time travel recovery on the given tenant/tenant shard. Tenant needs to be detached
1517 : // (from all pageservers) as it invalidates consistency assumptions.
1518 0 : async fn tenant_time_travel_remote_storage_handler(
1519 0 : request: Request<Body>,
1520 0 : cancel: CancellationToken,
1521 0 : ) -> Result<Response<Body>, ApiError> {
1522 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1523 :
1524 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1525 :
1526 0 : let timestamp_raw = must_get_query_param(&request, "travel_to")?;
1527 0 : let timestamp = humantime::parse_rfc3339(×tamp_raw)
1528 0 : .with_context(|| format!("Invalid time for travel_to: {timestamp_raw:?}"))
1529 0 : .map_err(ApiError::BadRequest)?;
1530 :
1531 0 : let done_if_after_raw = must_get_query_param(&request, "done_if_after")?;
1532 0 : let done_if_after = humantime::parse_rfc3339(&done_if_after_raw)
1533 0 : .with_context(|| format!("Invalid time for done_if_after: {done_if_after_raw:?}"))
1534 0 : .map_err(ApiError::BadRequest)?;
1535 :
1536 : // This is just a sanity check to fend off naive wrong usages of the API:
1537 : // the tenant needs to be detached *everywhere*
1538 0 : let state = get_state(&request);
1539 0 : let we_manage_tenant = state.tenant_manager.manages_tenant_shard(tenant_shard_id);
1540 0 : if we_manage_tenant {
1541 0 : return Err(ApiError::BadRequest(anyhow!(
1542 0 : "Tenant {tenant_shard_id} is already attached at this pageserver"
1543 0 : )));
1544 0 : }
1545 :
1546 0 : let Some(storage) = state.remote_storage.as_ref() else {
1547 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!(
1548 0 : "remote storage not configured, cannot run time travel"
1549 0 : )));
1550 : };
1551 :
1552 0 : if timestamp > done_if_after {
1553 0 : return Err(ApiError::BadRequest(anyhow!(
1554 0 : "The done_if_after timestamp comes before the timestamp to recover to"
1555 0 : )));
1556 0 : }
1557 :
1558 0 : tracing::info!("Issuing time travel request internally. timestamp={timestamp_raw}, done_if_after={done_if_after_raw}");
1559 :
1560 0 : remote_timeline_client::upload::time_travel_recover_tenant(
1561 0 : storage,
1562 0 : &tenant_shard_id,
1563 0 : timestamp,
1564 0 : done_if_after,
1565 0 : &cancel,
1566 0 : )
1567 0 : .await
1568 0 : .map_err(|e| match e {
1569 0 : TimeTravelError::BadInput(e) => {
1570 0 : warn!("bad input error: {e}");
1571 0 : ApiError::BadRequest(anyhow!("bad input error"))
1572 : }
1573 : TimeTravelError::Unimplemented => {
1574 0 : ApiError::BadRequest(anyhow!("unimplemented for the configured remote storage"))
1575 : }
1576 0 : TimeTravelError::Cancelled => ApiError::InternalServerError(anyhow!("cancelled")),
1577 : TimeTravelError::TooManyVersions => {
1578 0 : ApiError::InternalServerError(anyhow!("too many versions in remote storage"))
1579 : }
1580 0 : TimeTravelError::Other(e) => {
1581 0 : warn!("internal error: {e}");
1582 0 : ApiError::InternalServerError(anyhow!("internal error"))
1583 : }
1584 0 : })?;
1585 :
1586 0 : json_response(StatusCode::OK, ())
1587 0 : }
1588 :
1589 : /// Testing helper to transition a tenant to [`crate::tenant::TenantState::Broken`].
1590 0 : async fn handle_tenant_break(
1591 0 : r: Request<Body>,
1592 0 : _cancel: CancellationToken,
1593 0 : ) -> Result<Response<Body>, ApiError> {
1594 0 : let tenant_shard_id: TenantShardId = parse_request_param(&r, "tenant_shard_id")?;
1595 :
1596 0 : let tenant = crate::tenant::mgr::get_tenant(tenant_shard_id, true)
1597 0 : .map_err(|_| ApiError::Conflict(String::from("no active tenant found")))?;
1598 :
1599 0 : tenant.set_broken("broken from test".to_owned()).await;
1600 :
1601 0 : json_response(StatusCode::OK, ())
1602 0 : }
1603 :
1604 : // Run GC immediately on given timeline.
1605 0 : async fn timeline_gc_handler(
1606 0 : mut request: Request<Body>,
1607 0 : cancel: CancellationToken,
1608 0 : ) -> Result<Response<Body>, ApiError> {
1609 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1610 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1611 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1612 :
1613 0 : let gc_req: TimelineGcRequest = json_request(&mut request).await?;
1614 :
1615 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1616 0 : let wait_task_done =
1617 0 : mgr::immediate_gc(tenant_shard_id, timeline_id, gc_req, cancel, &ctx).await?;
1618 0 : let gc_result = wait_task_done
1619 0 : .await
1620 0 : .context("wait for gc task")
1621 0 : .map_err(ApiError::InternalServerError)?
1622 0 : .map_err(ApiError::InternalServerError)?;
1623 :
1624 0 : json_response(StatusCode::OK, gc_result)
1625 0 : }
1626 :
1627 : // Run compaction immediately on given timeline.
1628 0 : async fn timeline_compact_handler(
1629 0 : request: Request<Body>,
1630 0 : cancel: CancellationToken,
1631 0 : ) -> Result<Response<Body>, ApiError> {
1632 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1633 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1634 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1635 :
1636 0 : let state = get_state(&request);
1637 0 :
1638 0 : let mut flags = EnumSet::empty();
1639 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_repartition")? {
1640 0 : flags |= CompactFlags::ForceRepartition;
1641 0 : }
1642 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_image_layer_creation")? {
1643 0 : flags |= CompactFlags::ForceImageLayerCreation;
1644 0 : }
1645 :
1646 0 : async {
1647 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1648 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
1649 0 : timeline
1650 0 : .compact(&cancel, flags, &ctx)
1651 0 : .await
1652 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
1653 0 : json_response(StatusCode::OK, ())
1654 0 : }
1655 0 : .instrument(info_span!("manual_compaction", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
1656 0 : .await
1657 0 : }
1658 :
1659 : // Run checkpoint immediately on given timeline.
1660 0 : async fn timeline_checkpoint_handler(
1661 0 : request: Request<Body>,
1662 0 : cancel: CancellationToken,
1663 0 : ) -> Result<Response<Body>, ApiError> {
1664 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1665 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1666 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1667 :
1668 0 : let state = get_state(&request);
1669 0 :
1670 0 : let mut flags = EnumSet::empty();
1671 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_repartition")? {
1672 0 : flags |= CompactFlags::ForceRepartition;
1673 0 : }
1674 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_image_layer_creation")? {
1675 0 : flags |= CompactFlags::ForceImageLayerCreation;
1676 0 : }
1677 :
1678 0 : async {
1679 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1680 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
1681 0 : timeline
1682 0 : .freeze_and_flush()
1683 0 : .await
1684 0 : .map_err(ApiError::InternalServerError)?;
1685 0 : timeline
1686 0 : .compact(&cancel, flags, &ctx)
1687 0 : .await
1688 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
1689 :
1690 0 : json_response(StatusCode::OK, ())
1691 0 : }
1692 0 : .instrument(info_span!("manual_checkpoint", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
1693 0 : .await
1694 0 : }
1695 :
1696 0 : async fn timeline_download_remote_layers_handler_post(
1697 0 : mut request: Request<Body>,
1698 0 : _cancel: CancellationToken,
1699 0 : ) -> Result<Response<Body>, ApiError> {
1700 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1701 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1702 0 : let body: DownloadRemoteLayersTaskSpawnRequest = json_request(&mut request).await?;
1703 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1704 :
1705 0 : let state = get_state(&request);
1706 :
1707 0 : let timeline =
1708 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1709 0 : .await?;
1710 0 : match timeline.spawn_download_all_remote_layers(body).await {
1711 0 : Ok(st) => json_response(StatusCode::ACCEPTED, st),
1712 0 : Err(st) => json_response(StatusCode::CONFLICT, st),
1713 : }
1714 0 : }
1715 :
1716 0 : async fn timeline_download_remote_layers_handler_get(
1717 0 : request: Request<Body>,
1718 0 : _cancel: CancellationToken,
1719 0 : ) -> Result<Response<Body>, ApiError> {
1720 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1721 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1722 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1723 0 : let state = get_state(&request);
1724 :
1725 0 : let timeline =
1726 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1727 0 : .await?;
1728 0 : let info = timeline
1729 0 : .get_download_all_remote_layers_task_info()
1730 0 : .context("task never started since last pageserver process start")
1731 0 : .map_err(|e| ApiError::NotFound(e.into()))?;
1732 0 : json_response(StatusCode::OK, info)
1733 0 : }
1734 :
1735 0 : async fn deletion_queue_flush(
1736 0 : r: Request<Body>,
1737 0 : cancel: CancellationToken,
1738 0 : ) -> Result<Response<Body>, ApiError> {
1739 0 : let state = get_state(&r);
1740 0 :
1741 0 : if state.remote_storage.is_none() {
1742 : // Nothing to do if remote storage is disabled.
1743 0 : return json_response(StatusCode::OK, ());
1744 0 : }
1745 :
1746 0 : let execute = parse_query_param(&r, "execute")?.unwrap_or(false);
1747 0 :
1748 0 : let flush = async {
1749 0 : if execute {
1750 0 : state.deletion_queue_client.flush_execute().await
1751 : } else {
1752 0 : state.deletion_queue_client.flush().await
1753 : }
1754 0 : }
1755 : // DeletionQueueError's only case is shutting down.
1756 0 : .map_err(|_| ApiError::ShuttingDown);
1757 0 :
1758 0 : tokio::select! {
1759 0 : res = flush => {
1760 0 : res.map(|()| json_response(StatusCode::OK, ()))?
1761 : }
1762 : _ = cancel.cancelled() => {
1763 : Err(ApiError::ShuttingDown)
1764 : }
1765 : }
1766 0 : }
1767 :
1768 : /// Try if `GetPage@Lsn` is successful, useful for manual debugging.
1769 0 : async fn getpage_at_lsn_handler(
1770 0 : request: Request<Body>,
1771 0 : _cancel: CancellationToken,
1772 0 : ) -> Result<Response<Body>, ApiError> {
1773 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1774 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1775 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1776 0 : let state = get_state(&request);
1777 :
1778 : struct Key(crate::repository::Key);
1779 :
1780 : impl std::str::FromStr for Key {
1781 : type Err = anyhow::Error;
1782 :
1783 0 : fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
1784 0 : crate::repository::Key::from_hex(s).map(Key)
1785 0 : }
1786 : }
1787 :
1788 0 : let key: Key = parse_query_param(&request, "key")?
1789 0 : .ok_or_else(|| ApiError::BadRequest(anyhow!("missing 'key' query parameter")))?;
1790 0 : let lsn: Lsn = parse_query_param(&request, "lsn")?
1791 0 : .ok_or_else(|| ApiError::BadRequest(anyhow!("missing 'lsn' query parameter")))?;
1792 :
1793 0 : async {
1794 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1795 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
1796 :
1797 0 : let page = timeline.get(key.0, lsn, &ctx).await?;
1798 :
1799 0 : Result::<_, ApiError>::Ok(
1800 0 : Response::builder()
1801 0 : .status(StatusCode::OK)
1802 0 : .header(header::CONTENT_TYPE, "application/octet-stream")
1803 0 : .body(hyper::Body::from(page))
1804 0 : .unwrap(),
1805 0 : )
1806 0 : }
1807 0 : .instrument(info_span!("timeline_get", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
1808 0 : .await
1809 0 : }
1810 :
1811 0 : async fn timeline_collect_keyspace(
1812 0 : request: Request<Body>,
1813 0 : _cancel: CancellationToken,
1814 0 : ) -> Result<Response<Body>, ApiError> {
1815 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1816 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1817 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1818 0 : let state = get_state(&request);
1819 :
1820 0 : let at_lsn: Option<Lsn> = parse_query_param(&request, "at_lsn")?;
1821 :
1822 0 : async {
1823 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1824 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
1825 0 : let at_lsn = at_lsn.unwrap_or_else(|| timeline.get_last_record_lsn());
1826 0 : let keys = timeline
1827 0 : .collect_keyspace(at_lsn, &ctx)
1828 0 : .await
1829 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
1830 :
1831 0 : let res = pageserver_api::models::partitioning::Partitioning { keys, at_lsn };
1832 0 :
1833 0 : json_response(StatusCode::OK, res)
1834 0 : }
1835 0 : .instrument(info_span!("timeline_collect_keyspace", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
1836 0 : .await
1837 0 : }
1838 :
1839 0 : async fn active_timeline_of_active_tenant(
1840 0 : tenant_manager: &TenantManager,
1841 0 : tenant_shard_id: TenantShardId,
1842 0 : timeline_id: TimelineId,
1843 0 : ) -> Result<Arc<Timeline>, ApiError> {
1844 0 : let tenant = tenant_manager.get_attached_tenant_shard(tenant_shard_id, false)?;
1845 :
1846 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
1847 :
1848 0 : tenant
1849 0 : .get_timeline(timeline_id, true)
1850 0 : .map_err(|e| ApiError::NotFound(e.into()))
1851 0 : }
1852 :
1853 0 : async fn always_panic_handler(
1854 0 : req: Request<Body>,
1855 0 : _cancel: CancellationToken,
1856 0 : ) -> Result<Response<Body>, ApiError> {
1857 0 : // Deliberately cause a panic to exercise the panic hook registered via std::panic::set_hook().
1858 0 : // For pageserver, the relevant panic hook is `tracing_panic_hook` , and the `sentry` crate's wrapper around it.
1859 0 : // Use catch_unwind to ensure that tokio nor hyper are distracted by our panic.
1860 0 : let query = req.uri().query();
1861 0 : let _ = std::panic::catch_unwind(|| {
1862 0 : panic!("unconditional panic for testing panic hook integration; request query: {query:?}")
1863 0 : });
1864 0 : json_response(StatusCode::NO_CONTENT, ())
1865 0 : }
1866 :
1867 0 : async fn disk_usage_eviction_run(
1868 0 : mut r: Request<Body>,
1869 0 : cancel: CancellationToken,
1870 0 : ) -> Result<Response<Body>, ApiError> {
1871 0 : check_permission(&r, None)?;
1872 :
1873 0 : #[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
1874 : struct Config {
1875 : /// How many bytes to evict before reporting that pressure is relieved.
1876 : evict_bytes: u64,
1877 :
1878 : #[serde(default)]
1879 : eviction_order: crate::disk_usage_eviction_task::EvictionOrder,
1880 : }
1881 :
1882 0 : #[derive(Debug, Clone, Copy, serde::Serialize)]
1883 : struct Usage {
1884 : // remains unchanged after instantiation of the struct
1885 : evict_bytes: u64,
1886 : // updated by `add_available_bytes`
1887 : freed_bytes: u64,
1888 : }
1889 :
1890 : impl crate::disk_usage_eviction_task::Usage for Usage {
1891 0 : fn has_pressure(&self) -> bool {
1892 0 : self.evict_bytes > self.freed_bytes
1893 0 : }
1894 :
1895 0 : fn add_available_bytes(&mut self, bytes: u64) {
1896 0 : self.freed_bytes += bytes;
1897 0 : }
1898 : }
1899 :
1900 0 : let config = json_request::<Config>(&mut r).await?;
1901 :
1902 0 : let usage = Usage {
1903 0 : evict_bytes: config.evict_bytes,
1904 0 : freed_bytes: 0,
1905 0 : };
1906 0 :
1907 0 : let state = get_state(&r);
1908 :
1909 0 : let Some(storage) = state.remote_storage.as_ref() else {
1910 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!(
1911 0 : "remote storage not configured, cannot run eviction iteration"
1912 0 : )));
1913 : };
1914 :
1915 0 : let eviction_state = state.disk_usage_eviction_state.clone();
1916 :
1917 0 : let res = crate::disk_usage_eviction_task::disk_usage_eviction_task_iteration_impl(
1918 0 : &eviction_state,
1919 0 : storage,
1920 0 : usage,
1921 0 : &state.tenant_manager,
1922 0 : config.eviction_order,
1923 0 : &cancel,
1924 0 : )
1925 0 : .await;
1926 :
1927 0 : info!(?res, "disk_usage_eviction_task_iteration_impl finished");
1928 :
1929 0 : let res = res.map_err(ApiError::InternalServerError)?;
1930 :
1931 0 : json_response(StatusCode::OK, res)
1932 0 : }
1933 :
1934 0 : async fn secondary_upload_handler(
1935 0 : request: Request<Body>,
1936 0 : _cancel: CancellationToken,
1937 0 : ) -> Result<Response<Body>, ApiError> {
1938 0 : let state = get_state(&request);
1939 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1940 0 : state
1941 0 : .secondary_controller
1942 0 : .upload_tenant(tenant_shard_id)
1943 0 : .await
1944 0 : .map_err(ApiError::InternalServerError)?;
1945 :
1946 0 : json_response(StatusCode::OK, ())
1947 0 : }
1948 :
1949 0 : async fn secondary_download_handler(
1950 0 : request: Request<Body>,
1951 0 : _cancel: CancellationToken,
1952 0 : ) -> Result<Response<Body>, ApiError> {
1953 0 : let state = get_state(&request);
1954 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1955 0 : state
1956 0 : .secondary_controller
1957 0 : .download_tenant(tenant_shard_id)
1958 0 : .await
1959 0 : .map_err(ApiError::InternalServerError)?;
1960 :
1961 0 : json_response(StatusCode::OK, ())
1962 0 : }
1963 :
1964 0 : async fn handler_404(_: Request<Body>) -> Result<Response<Body>, ApiError> {
1965 0 : json_response(
1966 0 : StatusCode::NOT_FOUND,
1967 0 : HttpErrorBody::from_msg("page not found".to_owned()),
1968 0 : )
1969 0 : }
1970 :
1971 0 : async fn post_tracing_event_handler(
1972 0 : mut r: Request<Body>,
1973 0 : _cancel: CancellationToken,
1974 0 : ) -> Result<Response<Body>, ApiError> {
1975 0 : #[derive(Debug, serde::Deserialize)]
1976 : #[serde(rename_all = "lowercase")]
1977 : enum Level {
1978 : Error,
1979 : Warn,
1980 : Info,
1981 : Debug,
1982 : Trace,
1983 : }
1984 0 : #[derive(Debug, serde::Deserialize)]
1985 : struct Request {
1986 : level: Level,
1987 : message: String,
1988 : }
1989 0 : let body: Request = json_request(&mut r)
1990 0 : .await
1991 0 : .map_err(|_| ApiError::BadRequest(anyhow::anyhow!("invalid JSON body")))?;
1992 :
1993 0 : match body.level {
1994 0 : Level::Error => tracing::error!(?body.message),
1995 0 : Level::Warn => tracing::warn!(?body.message),
1996 0 : Level::Info => tracing::info!(?body.message),
1997 0 : Level::Debug => tracing::debug!(?body.message),
1998 0 : Level::Trace => tracing::trace!(?body.message),
1999 : }
2000 :
2001 0 : json_response(StatusCode::OK, ())
2002 0 : }
2003 :
2004 0 : async fn put_io_engine_handler(
2005 0 : mut r: Request<Body>,
2006 0 : _cancel: CancellationToken,
2007 0 : ) -> Result<Response<Body>, ApiError> {
2008 0 : check_permission(&r, None)?;
2009 0 : let kind: crate::virtual_file::IoEngineKind = json_request(&mut r).await?;
2010 0 : crate::virtual_file::io_engine::set(kind);
2011 0 : json_response(StatusCode::OK, ())
2012 0 : }
2013 :
2014 : /// Polled by control plane.
2015 : ///
2016 : /// See [`crate::utilization`].
2017 0 : async fn get_utilization(
2018 0 : r: Request<Body>,
2019 0 : _cancel: CancellationToken,
2020 0 : ) -> Result<Response<Body>, ApiError> {
2021 0 : // this probably could be completely public, but lets make that change later.
2022 0 : check_permission(&r, None)?;
2023 :
2024 0 : let state = get_state(&r);
2025 0 : let mut g = state.latest_utilization.lock().await;
2026 :
2027 0 : let regenerate_every = Duration::from_secs(1);
2028 0 : let still_valid = g
2029 0 : .as_ref()
2030 0 : .is_some_and(|(captured_at, _)| captured_at.elapsed() < regenerate_every);
2031 0 :
2032 0 : // avoid needless statvfs calls even though those should be non-blocking fast.
2033 0 : // regenerate at most 1Hz to allow polling at any rate.
2034 0 : if !still_valid {
2035 0 : let path = state.conf.tenants_path();
2036 0 : let doc = crate::utilization::regenerate(path.as_std_path())
2037 0 : .map_err(ApiError::InternalServerError)?;
2038 :
2039 0 : let mut buf = Vec::new();
2040 0 : serde_json::to_writer(&mut buf, &doc)
2041 0 : .context("serialize")
2042 0 : .map_err(ApiError::InternalServerError)?;
2043 :
2044 0 : let body = bytes::Bytes::from(buf);
2045 0 :
2046 0 : *g = Some((std::time::Instant::now(), body));
2047 0 : }
2048 :
2049 : // hyper 0.14 doesn't yet have Response::clone so this is a bit of extra legwork
2050 0 : let cached = g.as_ref().expect("just set").1.clone();
2051 0 :
2052 0 : Response::builder()
2053 0 : .header(hyper::http::header::CONTENT_TYPE, "application/json")
2054 0 : // thought of using http date header, but that is second precision which does not give any
2055 0 : // debugging aid
2056 0 : .status(StatusCode::OK)
2057 0 : .body(hyper::Body::from(cached))
2058 0 : .context("build response")
2059 0 : .map_err(ApiError::InternalServerError)
2060 0 : }
2061 :
2062 : /// Common functionality of all the HTTP API handlers.
2063 : ///
2064 : /// - Adds a tracing span to each request (by `request_span`)
2065 : /// - Logs the request depending on the request method (by `request_span`)
2066 : /// - Logs the response if it was not successful (by `request_span`
2067 : /// - Shields the handler function from async cancellations. Hyper can drop the handler
2068 : /// Future if the connection to the client is lost, but most of the pageserver code is
2069 : /// not async cancellation safe. This converts the dropped future into a graceful cancellation
2070 : /// request with a CancellationToken.
2071 0 : async fn api_handler<R, H>(request: Request<Body>, handler: H) -> Result<Response<Body>, ApiError>
2072 0 : where
2073 0 : R: std::future::Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
2074 0 : H: FnOnce(Request<Body>, CancellationToken) -> R + Send + Sync + 'static,
2075 0 : {
2076 0 : // Spawn a new task to handle the request, to protect the handler from unexpected
2077 0 : // async cancellations. Most pageserver functions are not async cancellation safe.
2078 0 : // We arm a drop-guard, so that if Hyper drops the Future, we signal the task
2079 0 : // with the cancellation token.
2080 0 : let token = CancellationToken::new();
2081 0 : let cancel_guard = token.clone().drop_guard();
2082 0 : let result = request_span(request, move |r| async {
2083 0 : let handle = tokio::spawn(
2084 0 : async {
2085 0 : let token_cloned = token.clone();
2086 0 : let result = handler(r, token).await;
2087 0 : if token_cloned.is_cancelled() {
2088 : // dropguard has executed: we will never turn this result into response.
2089 : //
2090 : // at least temporarily do {:?} logging; these failures are rare enough but
2091 : // could hide difficult errors.
2092 0 : match &result {
2093 0 : Ok(response) => {
2094 0 : let status = response.status();
2095 0 : info!(%status, "Cancelled request finished successfully")
2096 : }
2097 0 : Err(e) => error!("Cancelled request finished with an error: {e:?}"),
2098 : }
2099 0 : }
2100 : // only logging for cancelled panicked request handlers is the tracing_panic_hook,
2101 : // which should suffice.
2102 : //
2103 : // there is still a chance to lose the result due to race between
2104 : // returning from here and the actual connection closing happening
2105 : // before outer task gets to execute. leaving that up for #5815.
2106 0 : result
2107 0 : }
2108 0 : .in_current_span(),
2109 0 : );
2110 0 :
2111 0 : match handle.await {
2112 : // TODO: never actually return Err from here, always Ok(...) so that we can log
2113 : // spanned errors. Call api_error_handler instead and return appropriate Body.
2114 0 : Ok(result) => result,
2115 0 : Err(e) => {
2116 0 : // The handler task panicked. We have a global panic handler that logs the
2117 0 : // panic with its backtrace, so no need to log that here. Only log a brief
2118 0 : // message to make it clear that we returned the error to the client.
2119 0 : error!("HTTP request handler task panicked: {e:#}");
2120 :
2121 : // Don't return an Error here, because then fallback error handler that was
2122 : // installed in make_router() will print the error. Instead, construct the
2123 : // HTTP error response and return that.
2124 0 : Ok(
2125 0 : ApiError::InternalServerError(anyhow!("HTTP request handler task panicked"))
2126 0 : .into_response(),
2127 0 : )
2128 : }
2129 : }
2130 0 : })
2131 0 : .await;
2132 :
2133 0 : cancel_guard.disarm();
2134 0 :
2135 0 : result
2136 0 : }
2137 :
2138 : /// Like api_handler, but returns an error response if the server is built without
2139 : /// the 'testing' feature.
2140 0 : async fn testing_api_handler<R, H>(
2141 0 : desc: &str,
2142 0 : request: Request<Body>,
2143 0 : handler: H,
2144 0 : ) -> Result<Response<Body>, ApiError>
2145 0 : where
2146 0 : R: std::future::Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
2147 0 : H: FnOnce(Request<Body>, CancellationToken) -> R + Send + Sync + 'static,
2148 0 : {
2149 0 : if cfg!(feature = "testing") {
2150 0 : api_handler(request, handler).await
2151 : } else {
2152 0 : std::future::ready(Err(ApiError::BadRequest(anyhow!(
2153 0 : "Cannot {desc} because pageserver was compiled without testing APIs",
2154 0 : ))))
2155 0 : .await
2156 : }
2157 0 : }
2158 :
2159 0 : pub fn make_router(
2160 0 : state: Arc<State>,
2161 0 : launch_ts: &'static LaunchTimestamp,
2162 0 : auth: Option<Arc<SwappableJwtAuth>>,
2163 0 : ) -> anyhow::Result<RouterBuilder<hyper::Body, ApiError>> {
2164 0 : let spec = include_bytes!("openapi_spec.yml");
2165 0 : let mut router = attach_openapi_ui(endpoint::make_router(), spec, "/swagger.yml", "/v1/doc");
2166 0 : if auth.is_some() {
2167 0 : router = router.middleware(auth_middleware(|request| {
2168 0 : let state = get_state(request);
2169 0 : if state.allowlist_routes.contains(request.uri()) {
2170 0 : None
2171 : } else {
2172 0 : state.auth.as_deref()
2173 : }
2174 0 : }))
2175 0 : }
2176 :
2177 0 : router = router.middleware(
2178 0 : endpoint::add_response_header_middleware(
2179 0 : "PAGESERVER_LAUNCH_TIMESTAMP",
2180 0 : &launch_ts.to_string(),
2181 0 : )
2182 0 : .expect("construct launch timestamp header middleware"),
2183 0 : );
2184 0 :
2185 0 : Ok(router
2186 0 : .data(state)
2187 0 : .get("/v1/status", |r| api_handler(r, status_handler))
2188 0 : .put("/v1/failpoints", |r| {
2189 0 : testing_api_handler("manage failpoints", r, failpoints_handler)
2190 0 : })
2191 0 : .post("/v1/reload_auth_validation_keys", |r| {
2192 0 : api_handler(r, reload_auth_validation_keys_handler)
2193 0 : })
2194 0 : .get("/v1/tenant", |r| api_handler(r, tenant_list_handler))
2195 0 : .post("/v1/tenant", |r| api_handler(r, tenant_create_handler))
2196 0 : .get("/v1/tenant/:tenant_shard_id", |r| {
2197 0 : api_handler(r, tenant_status)
2198 0 : })
2199 0 : .delete("/v1/tenant/:tenant_shard_id", |r| {
2200 0 : api_handler(r, tenant_delete_handler)
2201 0 : })
2202 0 : .get("/v1/tenant/:tenant_shard_id/synthetic_size", |r| {
2203 0 : api_handler(r, tenant_size_handler)
2204 0 : })
2205 0 : .put("/v1/tenant/config", |r| {
2206 0 : api_handler(r, update_tenant_config_handler)
2207 0 : })
2208 0 : .put("/v1/tenant/:tenant_shard_id/shard_split", |r| {
2209 0 : api_handler(r, tenant_shard_split_handler)
2210 0 : })
2211 0 : .get("/v1/tenant/:tenant_shard_id/config", |r| {
2212 0 : api_handler(r, get_tenant_config_handler)
2213 0 : })
2214 0 : .put("/v1/tenant/:tenant_shard_id/location_config", |r| {
2215 0 : api_handler(r, put_tenant_location_config_handler)
2216 0 : })
2217 0 : .get("/v1/location_config", |r| {
2218 0 : api_handler(r, list_location_config_handler)
2219 0 : })
2220 0 : .put(
2221 0 : "/v1/tenant/:tenant_shard_id/time_travel_remote_storage",
2222 0 : |r| api_handler(r, tenant_time_travel_remote_storage_handler),
2223 0 : )
2224 0 : .get("/v1/tenant/:tenant_shard_id/timeline", |r| {
2225 0 : api_handler(r, timeline_list_handler)
2226 0 : })
2227 0 : .post("/v1/tenant/:tenant_shard_id/timeline", |r| {
2228 0 : api_handler(r, timeline_create_handler)
2229 0 : })
2230 0 : .post("/v1/tenant/:tenant_id/attach", |r| {
2231 0 : api_handler(r, tenant_attach_handler)
2232 0 : })
2233 0 : .post("/v1/tenant/:tenant_id/detach", |r| {
2234 0 : api_handler(r, tenant_detach_handler)
2235 0 : })
2236 0 : .post("/v1/tenant/:tenant_shard_id/reset", |r| {
2237 0 : api_handler(r, tenant_reset_handler)
2238 0 : })
2239 0 : .post("/v1/tenant/:tenant_id/load", |r| {
2240 0 : api_handler(r, tenant_load_handler)
2241 0 : })
2242 0 : .post("/v1/tenant/:tenant_id/ignore", |r| {
2243 0 : api_handler(r, tenant_ignore_handler)
2244 0 : })
2245 0 : .post(
2246 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/preserve_initdb_archive",
2247 0 : |r| api_handler(r, timeline_preserve_initdb_handler),
2248 0 : )
2249 0 : .get("/v1/tenant/:tenant_shard_id/timeline/:timeline_id", |r| {
2250 0 : api_handler(r, timeline_detail_handler)
2251 0 : })
2252 0 : .get(
2253 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/get_lsn_by_timestamp",
2254 0 : |r| api_handler(r, get_lsn_by_timestamp_handler),
2255 0 : )
2256 0 : .get(
2257 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/get_timestamp_of_lsn",
2258 0 : |r| api_handler(r, get_timestamp_of_lsn_handler),
2259 0 : )
2260 0 : .put(
2261 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/do_gc",
2262 0 : |r| api_handler(r, timeline_gc_handler),
2263 0 : )
2264 0 : .put(
2265 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/compact",
2266 0 : |r| testing_api_handler("run timeline compaction", r, timeline_compact_handler),
2267 0 : )
2268 0 : .put(
2269 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/checkpoint",
2270 0 : |r| testing_api_handler("run timeline checkpoint", r, timeline_checkpoint_handler),
2271 0 : )
2272 0 : .post(
2273 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/download_remote_layers",
2274 0 : |r| api_handler(r, timeline_download_remote_layers_handler_post),
2275 0 : )
2276 0 : .get(
2277 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/download_remote_layers",
2278 0 : |r| api_handler(r, timeline_download_remote_layers_handler_get),
2279 0 : )
2280 0 : .delete("/v1/tenant/:tenant_shard_id/timeline/:timeline_id", |r| {
2281 0 : api_handler(r, timeline_delete_handler)
2282 0 : })
2283 0 : .get(
2284 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/layer",
2285 0 : |r| api_handler(r, layer_map_info_handler),
2286 0 : )
2287 0 : .get(
2288 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/layer/:layer_file_name",
2289 0 : |r| api_handler(r, layer_download_handler),
2290 0 : )
2291 0 : .delete(
2292 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/layer/:layer_file_name",
2293 0 : |r| api_handler(r, evict_timeline_layer_handler),
2294 0 : )
2295 0 : .post("/v1/tenant/:tenant_shard_id/heatmap_upload", |r| {
2296 0 : api_handler(r, secondary_upload_handler)
2297 0 : })
2298 0 : .put("/v1/disk_usage_eviction/run", |r| {
2299 0 : api_handler(r, disk_usage_eviction_run)
2300 0 : })
2301 0 : .put("/v1/deletion_queue/flush", |r| {
2302 0 : api_handler(r, deletion_queue_flush)
2303 0 : })
2304 0 : .post("/v1/tenant/:tenant_shard_id/secondary/download", |r| {
2305 0 : api_handler(r, secondary_download_handler)
2306 0 : })
2307 0 : .put("/v1/tenant/:tenant_shard_id/break", |r| {
2308 0 : testing_api_handler("set tenant state to broken", r, handle_tenant_break)
2309 0 : })
2310 0 : .get("/v1/panic", |r| api_handler(r, always_panic_handler))
2311 0 : .post("/v1/tracing/event", |r| {
2312 0 : testing_api_handler("emit a tracing event", r, post_tracing_event_handler)
2313 0 : })
2314 0 : .get(
2315 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/getpage",
2316 0 : |r| testing_api_handler("getpage@lsn", r, getpage_at_lsn_handler),
2317 0 : )
2318 0 : .get(
2319 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/keyspace",
2320 0 : |r| api_handler(r, timeline_collect_keyspace),
2321 0 : )
2322 0 : .put("/v1/io_engine", |r| api_handler(r, put_io_engine_handler))
2323 0 : .get("/v1/utilization", |r| api_handler(r, get_utilization))
2324 0 : .any(handler_404))
2325 0 : }
|
