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