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::future::join_all;
14 : use futures::StreamExt;
15 : use futures::TryFutureExt;
16 : use http_utils::endpoint::{
17 : profile_cpu_handler, profile_heap_handler, prometheus_metrics_handler, request_span,
18 : };
19 : use http_utils::failpoints::failpoints_handler;
20 : use http_utils::request::must_parse_query_param;
21 : use http_utils::request::{get_request_param, must_get_query_param, parse_query_param};
22 : use humantime::format_rfc3339;
23 : use hyper::header;
24 : use hyper::StatusCode;
25 : use hyper::{Body, Request, Response, Uri};
26 : use metrics::launch_timestamp::LaunchTimestamp;
27 : use pageserver_api::models::virtual_file::IoMode;
28 : use pageserver_api::models::DownloadRemoteLayersTaskSpawnRequest;
29 : use pageserver_api::models::IngestAuxFilesRequest;
30 : use pageserver_api::models::ListAuxFilesRequest;
31 : use pageserver_api::models::LocationConfig;
32 : use pageserver_api::models::LocationConfigListResponse;
33 : use pageserver_api::models::LocationConfigMode;
34 : use pageserver_api::models::LsnLease;
35 : use pageserver_api::models::LsnLeaseRequest;
36 : use pageserver_api::models::OffloadedTimelineInfo;
37 : use pageserver_api::models::PageTraceEvent;
38 : use pageserver_api::models::ShardParameters;
39 : use pageserver_api::models::TenantConfigPatchRequest;
40 : use pageserver_api::models::TenantDetails;
41 : use pageserver_api::models::TenantLocationConfigRequest;
42 : use pageserver_api::models::TenantLocationConfigResponse;
43 : use pageserver_api::models::TenantScanRemoteStorageResponse;
44 : use pageserver_api::models::TenantScanRemoteStorageShard;
45 : use pageserver_api::models::TenantShardLocation;
46 : use pageserver_api::models::TenantShardSplitRequest;
47 : use pageserver_api::models::TenantShardSplitResponse;
48 : use pageserver_api::models::TenantSorting;
49 : use pageserver_api::models::TenantState;
50 : use pageserver_api::models::TenantWaitLsnRequest;
51 : use pageserver_api::models::TimelineArchivalConfigRequest;
52 : use pageserver_api::models::TimelineCreateRequestMode;
53 : use pageserver_api::models::TimelineCreateRequestModeImportPgdata;
54 : use pageserver_api::models::TimelinesInfoAndOffloaded;
55 : use pageserver_api::models::TopTenantShardItem;
56 : use pageserver_api::models::TopTenantShardsRequest;
57 : use pageserver_api::models::TopTenantShardsResponse;
58 : use pageserver_api::shard::ShardCount;
59 : use pageserver_api::shard::TenantShardId;
60 : use remote_storage::DownloadError;
61 : use remote_storage::GenericRemoteStorage;
62 : use remote_storage::TimeTravelError;
63 : use scopeguard::defer;
64 : use tenant_size_model::{svg::SvgBranchKind, SizeResult, StorageModel};
65 : use tokio::time::Instant;
66 : use tokio_util::io::StreamReader;
67 : use tokio_util::sync::CancellationToken;
68 : use tracing::*;
69 :
70 : use crate::config::PageServerConf;
71 : use crate::context::{DownloadBehavior, RequestContext};
72 : use crate::deletion_queue::DeletionQueueClient;
73 : use crate::pgdatadir_mapping::LsnForTimestamp;
74 : use crate::task_mgr::TaskKind;
75 : use crate::tenant::config::{LocationConf, TenantConfOpt};
76 : use crate::tenant::mgr::GetActiveTenantError;
77 : use crate::tenant::mgr::{
78 : GetTenantError, TenantManager, TenantMapError, TenantMapInsertError, TenantSlotError,
79 : TenantSlotUpsertError, TenantStateError,
80 : };
81 : use crate::tenant::mgr::{TenantSlot, UpsertLocationError};
82 : use crate::tenant::remote_timeline_client;
83 : use crate::tenant::remote_timeline_client::download_index_part;
84 : use crate::tenant::remote_timeline_client::list_remote_tenant_shards;
85 : use crate::tenant::remote_timeline_client::list_remote_timelines;
86 : use crate::tenant::secondary::SecondaryController;
87 : use crate::tenant::size::ModelInputs;
88 : use crate::tenant::storage_layer::IoConcurrency;
89 : use crate::tenant::storage_layer::LayerAccessStatsReset;
90 : use crate::tenant::storage_layer::LayerName;
91 : use crate::tenant::timeline::import_pgdata;
92 : use crate::tenant::timeline::offload::offload_timeline;
93 : use crate::tenant::timeline::offload::OffloadError;
94 : use crate::tenant::timeline::CompactFlags;
95 : use crate::tenant::timeline::CompactOptions;
96 : use crate::tenant::timeline::CompactRequest;
97 : use crate::tenant::timeline::CompactionError;
98 : use crate::tenant::timeline::Timeline;
99 : use crate::tenant::timeline::WaitLsnTimeout;
100 : use crate::tenant::timeline::WaitLsnWaiter;
101 : use crate::tenant::GetTimelineError;
102 : use crate::tenant::OffloadedTimeline;
103 : use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError};
104 : use crate::DEFAULT_PG_VERSION;
105 : use crate::{disk_usage_eviction_task, tenant};
106 : use http_utils::{
107 : endpoint::{self, attach_openapi_ui, auth_middleware, check_permission_with},
108 : error::{ApiError, HttpErrorBody},
109 : json::{json_request, json_request_maybe, json_response},
110 : request::parse_request_param,
111 : RequestExt, RouterBuilder,
112 : };
113 : use pageserver_api::models::{
114 : StatusResponse, TenantConfigRequest, TenantInfo, TimelineCreateRequest, TimelineGcRequest,
115 : TimelineInfo,
116 : };
117 : use utils::{
118 : auth::SwappableJwtAuth,
119 : generation::Generation,
120 : id::{TenantId, TimelineId},
121 : lsn::Lsn,
122 : };
123 :
124 : // For APIs that require an Active tenant, how long should we block waiting for that state?
125 : // This is not functionally necessary (clients will retry), but avoids generating a lot of
126 : // failed API calls while tenants are activating.
127 : #[cfg(not(feature = "testing"))]
128 : pub(crate) const ACTIVE_TENANT_TIMEOUT: Duration = Duration::from_millis(5000);
129 :
130 : // Tests run on slow/oversubscribed nodes, and may need to wait much longer for tenants to
131 : // finish attaching, if calls to remote storage are slow.
132 : #[cfg(feature = "testing")]
133 : pub(crate) const ACTIVE_TENANT_TIMEOUT: Duration = Duration::from_millis(30000);
134 :
135 : pub struct State {
136 : conf: &'static PageServerConf,
137 : tenant_manager: Arc<TenantManager>,
138 : auth: Option<Arc<SwappableJwtAuth>>,
139 : allowlist_routes: &'static [&'static str],
140 : remote_storage: GenericRemoteStorage,
141 : broker_client: storage_broker::BrokerClientChannel,
142 : disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
143 : deletion_queue_client: DeletionQueueClient,
144 : secondary_controller: SecondaryController,
145 : latest_utilization: tokio::sync::Mutex<Option<(std::time::Instant, bytes::Bytes)>>,
146 : }
147 :
148 : impl State {
149 : #[allow(clippy::too_many_arguments)]
150 0 : pub fn new(
151 0 : conf: &'static PageServerConf,
152 0 : tenant_manager: Arc<TenantManager>,
153 0 : auth: Option<Arc<SwappableJwtAuth>>,
154 0 : remote_storage: GenericRemoteStorage,
155 0 : broker_client: storage_broker::BrokerClientChannel,
156 0 : disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
157 0 : deletion_queue_client: DeletionQueueClient,
158 0 : secondary_controller: SecondaryController,
159 0 : ) -> anyhow::Result<Self> {
160 0 : let allowlist_routes = &[
161 0 : "/v1/status",
162 0 : "/v1/doc",
163 0 : "/swagger.yml",
164 0 : "/metrics",
165 0 : "/profile/cpu",
166 0 : "/profile/heap",
167 0 : ];
168 0 : Ok(Self {
169 0 : conf,
170 0 : tenant_manager,
171 0 : auth,
172 0 : allowlist_routes,
173 0 : remote_storage,
174 0 : broker_client,
175 0 : disk_usage_eviction_state,
176 0 : deletion_queue_client,
177 0 : secondary_controller,
178 0 : latest_utilization: Default::default(),
179 0 : })
180 0 : }
181 : }
182 :
183 : #[inline(always)]
184 0 : fn get_state(request: &Request<Body>) -> &State {
185 0 : request
186 0 : .data::<Arc<State>>()
187 0 : .expect("unknown state type")
188 0 : .as_ref()
189 0 : }
190 :
191 : #[inline(always)]
192 0 : fn get_config(request: &Request<Body>) -> &'static PageServerConf {
193 0 : get_state(request).conf
194 0 : }
195 :
196 : /// Check that the requester is authorized to operate on given tenant
197 0 : fn check_permission(request: &Request<Body>, tenant_id: Option<TenantId>) -> Result<(), ApiError> {
198 0 : check_permission_with(request, |claims| {
199 0 : crate::auth::check_permission(claims, tenant_id)
200 0 : })
201 0 : }
202 :
203 : impl From<PageReconstructError> for ApiError {
204 0 : fn from(pre: PageReconstructError) -> ApiError {
205 0 : match pre {
206 0 : PageReconstructError::Other(other) => ApiError::InternalServerError(other),
207 0 : PageReconstructError::MissingKey(e) => ApiError::InternalServerError(e.into()),
208 0 : PageReconstructError::Cancelled => ApiError::Cancelled,
209 0 : PageReconstructError::AncestorLsnTimeout(e) => ApiError::Timeout(format!("{e}").into()),
210 0 : PageReconstructError::WalRedo(pre) => ApiError::InternalServerError(pre),
211 : }
212 0 : }
213 : }
214 :
215 : impl From<TenantMapInsertError> for ApiError {
216 0 : fn from(tmie: TenantMapInsertError) -> ApiError {
217 0 : match tmie {
218 0 : TenantMapInsertError::SlotError(e) => e.into(),
219 0 : TenantMapInsertError::SlotUpsertError(e) => e.into(),
220 0 : TenantMapInsertError::Other(e) => ApiError::InternalServerError(e),
221 : }
222 0 : }
223 : }
224 :
225 : impl From<TenantSlotError> for ApiError {
226 0 : fn from(e: TenantSlotError) -> ApiError {
227 : use TenantSlotError::*;
228 0 : match e {
229 0 : NotFound(tenant_id) => {
230 0 : ApiError::NotFound(anyhow::anyhow!("NotFound: tenant {tenant_id}").into())
231 : }
232 : InProgress => {
233 0 : ApiError::ResourceUnavailable("Tenant is being modified concurrently".into())
234 : }
235 0 : MapState(e) => e.into(),
236 : }
237 0 : }
238 : }
239 :
240 : impl From<TenantSlotUpsertError> for ApiError {
241 0 : fn from(e: TenantSlotUpsertError) -> ApiError {
242 : use TenantSlotUpsertError::*;
243 0 : match e {
244 0 : InternalError(e) => ApiError::InternalServerError(anyhow::anyhow!("{e}")),
245 0 : MapState(e) => e.into(),
246 0 : ShuttingDown(_) => ApiError::ShuttingDown,
247 : }
248 0 : }
249 : }
250 :
251 : impl From<UpsertLocationError> for ApiError {
252 0 : fn from(e: UpsertLocationError) -> ApiError {
253 : use UpsertLocationError::*;
254 0 : match e {
255 0 : BadRequest(e) => ApiError::BadRequest(e),
256 0 : Unavailable(_) => ApiError::ShuttingDown,
257 0 : e @ InProgress => ApiError::Conflict(format!("{e}")),
258 0 : Flush(e) | InternalError(e) => ApiError::InternalServerError(e),
259 : }
260 0 : }
261 : }
262 :
263 : impl From<TenantMapError> for ApiError {
264 0 : fn from(e: TenantMapError) -> ApiError {
265 : use TenantMapError::*;
266 0 : match e {
267 : StillInitializing | ShuttingDown => {
268 0 : ApiError::ResourceUnavailable(format!("{e}").into())
269 0 : }
270 0 : }
271 0 : }
272 : }
273 :
274 : impl From<TenantStateError> for ApiError {
275 0 : fn from(tse: TenantStateError) -> ApiError {
276 0 : match tse {
277 : TenantStateError::IsStopping(_) => {
278 0 : ApiError::ResourceUnavailable("Tenant is stopping".into())
279 : }
280 0 : TenantStateError::SlotError(e) => e.into(),
281 0 : TenantStateError::SlotUpsertError(e) => e.into(),
282 0 : TenantStateError::Other(e) => ApiError::InternalServerError(anyhow!(e)),
283 : }
284 0 : }
285 : }
286 :
287 : impl From<GetTenantError> for ApiError {
288 0 : fn from(tse: GetTenantError) -> ApiError {
289 0 : match tse {
290 0 : GetTenantError::NotFound(tid) => ApiError::NotFound(anyhow!("tenant {tid}").into()),
291 0 : GetTenantError::ShardNotFound(tid) => {
292 0 : ApiError::NotFound(anyhow!("tenant {tid}").into())
293 : }
294 : GetTenantError::NotActive(_) => {
295 : // Why is this not `ApiError::NotFound`?
296 : // Because we must be careful to never return 404 for a tenant if it does
297 : // in fact exist locally. If we did, the caller could draw the conclusion
298 : // that it can attach the tenant to another PS and we'd be in split-brain.
299 0 : ApiError::ResourceUnavailable("Tenant not yet active".into())
300 : }
301 0 : GetTenantError::MapState(e) => ApiError::ResourceUnavailable(format!("{e}").into()),
302 : }
303 0 : }
304 : }
305 :
306 : impl From<GetTimelineError> for ApiError {
307 0 : fn from(gte: GetTimelineError) -> Self {
308 0 : // Rationale: tenant is activated only after eligble timelines activate
309 0 : ApiError::NotFound(gte.into())
310 0 : }
311 : }
312 :
313 : impl From<GetActiveTenantError> for ApiError {
314 0 : fn from(e: GetActiveTenantError) -> ApiError {
315 0 : match e {
316 0 : GetActiveTenantError::Broken(reason) => {
317 0 : ApiError::InternalServerError(anyhow!("tenant is broken: {}", reason))
318 : }
319 : GetActiveTenantError::WillNotBecomeActive(TenantState::Stopping { .. }) => {
320 0 : ApiError::ShuttingDown
321 : }
322 0 : GetActiveTenantError::WillNotBecomeActive(_) => ApiError::Conflict(format!("{}", e)),
323 0 : GetActiveTenantError::Cancelled => ApiError::ShuttingDown,
324 0 : GetActiveTenantError::NotFound(gte) => gte.into(),
325 : GetActiveTenantError::WaitForActiveTimeout { .. } => {
326 0 : ApiError::ResourceUnavailable(format!("{}", e).into())
327 : }
328 : GetActiveTenantError::SwitchedTenant => {
329 : // in our HTTP handlers, this error doesn't happen
330 : // TODO: separate error types
331 0 : ApiError::ResourceUnavailable("switched tenant".into())
332 : }
333 : }
334 0 : }
335 : }
336 :
337 : impl From<crate::tenant::DeleteTimelineError> for ApiError {
338 0 : fn from(value: crate::tenant::DeleteTimelineError) -> Self {
339 : use crate::tenant::DeleteTimelineError::*;
340 0 : match value {
341 0 : NotFound => ApiError::NotFound(anyhow::anyhow!("timeline not found").into()),
342 0 : HasChildren(children) => ApiError::PreconditionFailed(
343 0 : format!("Cannot delete timeline which has child timelines: {children:?}")
344 0 : .into_boxed_str(),
345 0 : ),
346 0 : a @ AlreadyInProgress(_) => ApiError::Conflict(a.to_string()),
347 0 : Cancelled => ApiError::ResourceUnavailable("shutting down".into()),
348 0 : Other(e) => ApiError::InternalServerError(e),
349 : }
350 0 : }
351 : }
352 :
353 : impl From<crate::tenant::TimelineArchivalError> for ApiError {
354 0 : fn from(value: crate::tenant::TimelineArchivalError) -> Self {
355 : use crate::tenant::TimelineArchivalError::*;
356 0 : match value {
357 0 : NotFound => ApiError::NotFound(anyhow::anyhow!("timeline not found").into()),
358 0 : Timeout => ApiError::Timeout("hit pageserver internal timeout".into()),
359 0 : Cancelled => ApiError::ShuttingDown,
360 0 : e @ HasArchivedParent(_) => {
361 0 : ApiError::PreconditionFailed(e.to_string().into_boxed_str())
362 : }
363 0 : HasUnarchivedChildren(children) => ApiError::PreconditionFailed(
364 0 : format!(
365 0 : "Cannot archive timeline which has non-archived child timelines: {children:?}"
366 0 : )
367 0 : .into_boxed_str(),
368 0 : ),
369 0 : a @ AlreadyInProgress => ApiError::Conflict(a.to_string()),
370 0 : Other(e) => ApiError::InternalServerError(e),
371 : }
372 0 : }
373 : }
374 :
375 : impl From<crate::tenant::mgr::DeleteTimelineError> for ApiError {
376 0 : fn from(value: crate::tenant::mgr::DeleteTimelineError) -> Self {
377 : use crate::tenant::mgr::DeleteTimelineError::*;
378 0 : match value {
379 : // Report Precondition failed so client can distinguish between
380 : // "tenant is missing" case from "timeline is missing"
381 0 : Tenant(GetTenantError::NotFound(..)) => ApiError::PreconditionFailed(
382 0 : "Requested tenant is missing".to_owned().into_boxed_str(),
383 0 : ),
384 0 : Tenant(t) => ApiError::from(t),
385 0 : Timeline(t) => ApiError::from(t),
386 : }
387 0 : }
388 : }
389 :
390 : impl From<crate::tenant::mgr::DeleteTenantError> for ApiError {
391 0 : fn from(value: crate::tenant::mgr::DeleteTenantError) -> Self {
392 : use crate::tenant::mgr::DeleteTenantError::*;
393 0 : match value {
394 0 : SlotError(e) => e.into(),
395 0 : Other(o) => ApiError::InternalServerError(o),
396 0 : Cancelled => ApiError::ShuttingDown,
397 : }
398 0 : }
399 : }
400 :
401 : impl From<crate::tenant::secondary::SecondaryTenantError> for ApiError {
402 0 : fn from(ste: crate::tenant::secondary::SecondaryTenantError) -> ApiError {
403 : use crate::tenant::secondary::SecondaryTenantError;
404 0 : match ste {
405 0 : SecondaryTenantError::GetTenant(gte) => gte.into(),
406 0 : SecondaryTenantError::ShuttingDown => ApiError::ShuttingDown,
407 : }
408 0 : }
409 : }
410 :
411 : // Helper function to construct a TimelineInfo struct for a timeline
412 0 : async fn build_timeline_info(
413 0 : timeline: &Arc<Timeline>,
414 0 : include_non_incremental_logical_size: bool,
415 0 : force_await_initial_logical_size: bool,
416 0 : ctx: &RequestContext,
417 0 : ) -> anyhow::Result<TimelineInfo> {
418 0 : crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
419 0 :
420 0 : if force_await_initial_logical_size {
421 0 : timeline.clone().await_initial_logical_size().await
422 0 : }
423 :
424 0 : let mut info = build_timeline_info_common(
425 0 : timeline,
426 0 : ctx,
427 0 : tenant::timeline::GetLogicalSizePriority::Background,
428 0 : )
429 0 : .await?;
430 0 : if include_non_incremental_logical_size {
431 : // XXX we should be using spawn_ondemand_logical_size_calculation here.
432 : // Otherwise, if someone deletes the timeline / detaches the tenant while
433 : // we're executing this function, we will outlive the timeline on-disk state.
434 : info.current_logical_size_non_incremental = Some(
435 0 : timeline
436 0 : .get_current_logical_size_non_incremental(info.last_record_lsn, ctx)
437 0 : .await?,
438 : );
439 0 : }
440 0 : Ok(info)
441 0 : }
442 :
443 0 : async fn build_timeline_info_common(
444 0 : timeline: &Arc<Timeline>,
445 0 : ctx: &RequestContext,
446 0 : logical_size_task_priority: tenant::timeline::GetLogicalSizePriority,
447 0 : ) -> anyhow::Result<TimelineInfo> {
448 0 : crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
449 0 : let initdb_lsn = timeline.initdb_lsn;
450 0 : let last_record_lsn = timeline.get_last_record_lsn();
451 0 : let (wal_source_connstr, last_received_msg_lsn, last_received_msg_ts) = {
452 0 : let guard = timeline.last_received_wal.lock().unwrap();
453 0 : if let Some(info) = guard.as_ref() {
454 0 : (
455 0 : Some(format!("{}", info.wal_source_connconf)), // Password is hidden, but it's for statistics only.
456 0 : Some(info.last_received_msg_lsn),
457 0 : Some(info.last_received_msg_ts),
458 0 : )
459 : } else {
460 0 : (None, None, None)
461 : }
462 : };
463 :
464 0 : let ancestor_timeline_id = timeline.get_ancestor_timeline_id();
465 0 : let ancestor_lsn = match timeline.get_ancestor_lsn() {
466 0 : Lsn(0) => None,
467 0 : lsn @ Lsn(_) => Some(lsn),
468 : };
469 0 : let current_logical_size = timeline.get_current_logical_size(logical_size_task_priority, ctx);
470 0 : let current_physical_size = Some(timeline.layer_size_sum().await);
471 0 : let state = timeline.current_state();
472 0 : // Report is_archived = false if the timeline is still loading
473 0 : let is_archived = timeline.is_archived().unwrap_or(false);
474 0 : let remote_consistent_lsn_projected = timeline
475 0 : .get_remote_consistent_lsn_projected()
476 0 : .unwrap_or(Lsn(0));
477 0 : let remote_consistent_lsn_visible = timeline
478 0 : .get_remote_consistent_lsn_visible()
479 0 : .unwrap_or(Lsn(0));
480 0 :
481 0 : let walreceiver_status = timeline.walreceiver_status();
482 0 :
483 0 : let (pitr_history_size, within_ancestor_pitr) = timeline.get_pitr_history_stats();
484 0 :
485 0 : let min_readable_lsn = std::cmp::max(
486 0 : timeline.get_gc_cutoff_lsn(),
487 0 : *timeline.get_applied_gc_cutoff_lsn(),
488 0 : );
489 :
490 0 : let info = TimelineInfo {
491 0 : tenant_id: timeline.tenant_shard_id,
492 0 : timeline_id: timeline.timeline_id,
493 0 : ancestor_timeline_id,
494 0 : ancestor_lsn,
495 0 : disk_consistent_lsn: timeline.get_disk_consistent_lsn(),
496 0 : remote_consistent_lsn: remote_consistent_lsn_projected,
497 0 : remote_consistent_lsn_visible,
498 0 : initdb_lsn,
499 0 : last_record_lsn,
500 0 : prev_record_lsn: Some(timeline.get_prev_record_lsn()),
501 0 : // Externally, expose the lowest LSN that can be used to create a branch as the "GC cutoff", although internally
502 0 : // we distinguish between the "planned" GC cutoff (PITR point) and the "latest" GC cutoff (where we
503 0 : // actually trimmed data to), which can pass each other when PITR is changed.
504 0 : latest_gc_cutoff_lsn: min_readable_lsn,
505 0 : min_readable_lsn,
506 0 : applied_gc_cutoff_lsn: *timeline.get_applied_gc_cutoff_lsn(),
507 0 : current_logical_size: current_logical_size.size_dont_care_about_accuracy(),
508 0 : current_logical_size_is_accurate: match current_logical_size.accuracy() {
509 0 : tenant::timeline::logical_size::Accuracy::Approximate => false,
510 0 : tenant::timeline::logical_size::Accuracy::Exact => true,
511 : },
512 0 : directory_entries_counts: timeline.get_directory_metrics().to_vec(),
513 0 : current_physical_size,
514 0 : current_logical_size_non_incremental: None,
515 0 : pitr_history_size,
516 0 : within_ancestor_pitr,
517 0 : timeline_dir_layer_file_size_sum: None,
518 0 : wal_source_connstr,
519 0 : last_received_msg_lsn,
520 0 : last_received_msg_ts,
521 0 : pg_version: timeline.pg_version,
522 0 :
523 0 : state,
524 0 : is_archived: Some(is_archived),
525 0 :
526 0 : walreceiver_status,
527 0 : };
528 0 : Ok(info)
529 0 : }
530 :
531 0 : fn build_timeline_offloaded_info(offloaded: &Arc<OffloadedTimeline>) -> OffloadedTimelineInfo {
532 0 : let &OffloadedTimeline {
533 0 : tenant_shard_id,
534 0 : timeline_id,
535 0 : ancestor_retain_lsn,
536 0 : ancestor_timeline_id,
537 0 : archived_at,
538 0 : ..
539 0 : } = offloaded.as_ref();
540 0 : OffloadedTimelineInfo {
541 0 : tenant_id: tenant_shard_id,
542 0 : timeline_id,
543 0 : ancestor_retain_lsn,
544 0 : ancestor_timeline_id,
545 0 : archived_at: archived_at.and_utc(),
546 0 : }
547 0 : }
548 :
549 : // healthcheck handler
550 0 : async fn status_handler(
551 0 : request: Request<Body>,
552 0 : _cancel: CancellationToken,
553 0 : ) -> Result<Response<Body>, ApiError> {
554 0 : check_permission(&request, None)?;
555 0 : let config = get_config(&request);
556 0 : json_response(StatusCode::OK, StatusResponse { id: config.id })
557 0 : }
558 :
559 0 : async fn reload_auth_validation_keys_handler(
560 0 : request: Request<Body>,
561 0 : _cancel: CancellationToken,
562 0 : ) -> Result<Response<Body>, ApiError> {
563 0 : check_permission(&request, None)?;
564 0 : let config = get_config(&request);
565 0 : let state = get_state(&request);
566 0 : let Some(shared_auth) = &state.auth else {
567 0 : return json_response(StatusCode::BAD_REQUEST, ());
568 : };
569 : // unwrap is ok because check is performed when creating config, so path is set and exists
570 0 : let key_path = config.auth_validation_public_key_path.as_ref().unwrap();
571 0 : info!("Reloading public key(s) for verifying JWT tokens from {key_path:?}");
572 :
573 0 : match utils::auth::JwtAuth::from_key_path(key_path) {
574 0 : Ok(new_auth) => {
575 0 : shared_auth.swap(new_auth);
576 0 : json_response(StatusCode::OK, ())
577 : }
578 0 : Err(e) => {
579 0 : let err_msg = "Error reloading public keys";
580 0 : warn!("Error reloading public keys from {key_path:?}: {e:}");
581 0 : json_response(
582 0 : StatusCode::INTERNAL_SERVER_ERROR,
583 0 : HttpErrorBody::from_msg(err_msg.to_string()),
584 0 : )
585 : }
586 : }
587 0 : }
588 :
589 0 : async fn timeline_create_handler(
590 0 : mut request: Request<Body>,
591 0 : _cancel: CancellationToken,
592 0 : ) -> Result<Response<Body>, ApiError> {
593 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
594 0 : let request_data: TimelineCreateRequest = json_request(&mut request).await?;
595 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
596 :
597 0 : let new_timeline_id = request_data.new_timeline_id;
598 : // fill in the default pg_version if not provided & convert request into domain model
599 0 : let params: tenant::CreateTimelineParams = match request_data.mode {
600 : TimelineCreateRequestMode::Bootstrap {
601 0 : existing_initdb_timeline_id,
602 0 : pg_version,
603 0 : } => tenant::CreateTimelineParams::Bootstrap(tenant::CreateTimelineParamsBootstrap {
604 0 : new_timeline_id,
605 0 : existing_initdb_timeline_id,
606 0 : pg_version: pg_version.unwrap_or(DEFAULT_PG_VERSION),
607 0 : }),
608 : TimelineCreateRequestMode::Branch {
609 0 : ancestor_timeline_id,
610 0 : ancestor_start_lsn,
611 0 : pg_version: _,
612 0 : } => tenant::CreateTimelineParams::Branch(tenant::CreateTimelineParamsBranch {
613 0 : new_timeline_id,
614 0 : ancestor_timeline_id,
615 0 : ancestor_start_lsn,
616 0 : }),
617 : TimelineCreateRequestMode::ImportPgdata {
618 : import_pgdata:
619 : TimelineCreateRequestModeImportPgdata {
620 0 : location,
621 0 : idempotency_key,
622 0 : },
623 0 : } => tenant::CreateTimelineParams::ImportPgdata(tenant::CreateTimelineParamsImportPgdata {
624 0 : idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new(
625 0 : idempotency_key.0,
626 0 : ),
627 0 : new_timeline_id,
628 : location: {
629 0 : use import_pgdata::index_part_format::Location;
630 0 : use pageserver_api::models::ImportPgdataLocation;
631 0 : match location {
632 : #[cfg(feature = "testing")]
633 0 : ImportPgdataLocation::LocalFs { path } => Location::LocalFs { path },
634 : ImportPgdataLocation::AwsS3 {
635 0 : region,
636 0 : bucket,
637 0 : key,
638 0 : } => Location::AwsS3 {
639 0 : region,
640 0 : bucket,
641 0 : key,
642 0 : },
643 : }
644 : },
645 : }),
646 : };
647 :
648 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
649 0 :
650 0 : let state = get_state(&request);
651 :
652 0 : async {
653 0 : let tenant = state
654 0 : .tenant_manager
655 0 : .get_attached_tenant_shard(tenant_shard_id)?;
656 :
657 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
658 :
659 : // earlier versions of the code had pg_version and ancestor_lsn in the span
660 : // => continue to provide that information, but, through a log message that doesn't require us to destructure
661 0 : tracing::info!(?params, "creating timeline");
662 :
663 0 : match tenant
664 0 : .create_timeline(params, state.broker_client.clone(), &ctx)
665 0 : .await
666 : {
667 0 : Ok(new_timeline) => {
668 : // Created. Construct a TimelineInfo for it.
669 0 : let timeline_info = build_timeline_info_common(
670 0 : &new_timeline,
671 0 : &ctx,
672 0 : tenant::timeline::GetLogicalSizePriority::User,
673 0 : )
674 0 : .await
675 0 : .map_err(ApiError::InternalServerError)?;
676 0 : json_response(StatusCode::CREATED, timeline_info)
677 : }
678 0 : Err(_) if tenant.cancel.is_cancelled() => {
679 0 : // In case we get some ugly error type during shutdown, cast it into a clean 503.
680 0 : json_response(
681 0 : StatusCode::SERVICE_UNAVAILABLE,
682 0 : HttpErrorBody::from_msg("Tenant shutting down".to_string()),
683 0 : )
684 : }
685 0 : Err(e @ tenant::CreateTimelineError::Conflict) => {
686 0 : json_response(StatusCode::CONFLICT, HttpErrorBody::from_msg(e.to_string()))
687 : }
688 0 : Err(e @ tenant::CreateTimelineError::AlreadyCreating) => json_response(
689 0 : StatusCode::TOO_MANY_REQUESTS,
690 0 : HttpErrorBody::from_msg(e.to_string()),
691 0 : ),
692 0 : Err(tenant::CreateTimelineError::AncestorLsn(err)) => json_response(
693 0 : StatusCode::NOT_ACCEPTABLE,
694 0 : HttpErrorBody::from_msg(format!("{err:#}")),
695 0 : ),
696 0 : Err(e @ tenant::CreateTimelineError::AncestorNotActive) => json_response(
697 0 : StatusCode::SERVICE_UNAVAILABLE,
698 0 : HttpErrorBody::from_msg(e.to_string()),
699 0 : ),
700 0 : Err(e @ tenant::CreateTimelineError::AncestorArchived) => json_response(
701 0 : StatusCode::NOT_ACCEPTABLE,
702 0 : HttpErrorBody::from_msg(e.to_string()),
703 0 : ),
704 0 : Err(tenant::CreateTimelineError::ShuttingDown) => json_response(
705 0 : StatusCode::SERVICE_UNAVAILABLE,
706 0 : HttpErrorBody::from_msg("tenant shutting down".to_string()),
707 0 : ),
708 0 : Err(tenant::CreateTimelineError::Other(err)) => Err(ApiError::InternalServerError(err)),
709 : }
710 0 : }
711 0 : .instrument(info_span!("timeline_create",
712 : tenant_id = %tenant_shard_id.tenant_id,
713 0 : shard_id = %tenant_shard_id.shard_slug(),
714 : timeline_id = %new_timeline_id,
715 : ))
716 0 : .await
717 0 : }
718 :
719 0 : async fn timeline_list_handler(
720 0 : request: Request<Body>,
721 0 : _cancel: CancellationToken,
722 0 : ) -> Result<Response<Body>, ApiError> {
723 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
724 0 : let include_non_incremental_logical_size: Option<bool> =
725 0 : parse_query_param(&request, "include-non-incremental-logical-size")?;
726 0 : let force_await_initial_logical_size: Option<bool> =
727 0 : parse_query_param(&request, "force-await-initial-logical-size")?;
728 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
729 :
730 0 : let state = get_state(&request);
731 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
732 :
733 0 : let response_data = async {
734 0 : let tenant = state
735 0 : .tenant_manager
736 0 : .get_attached_tenant_shard(tenant_shard_id)?;
737 :
738 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
739 :
740 0 : let timelines = tenant.list_timelines();
741 0 :
742 0 : let mut response_data = Vec::with_capacity(timelines.len());
743 0 : for timeline in timelines {
744 0 : let timeline_info = build_timeline_info(
745 0 : &timeline,
746 0 : include_non_incremental_logical_size.unwrap_or(false),
747 0 : force_await_initial_logical_size.unwrap_or(false),
748 0 : &ctx,
749 0 : )
750 0 : .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
751 0 : .await
752 0 : .context("Failed to build timeline info")
753 0 : .map_err(ApiError::InternalServerError)?;
754 :
755 0 : response_data.push(timeline_info);
756 : }
757 0 : Ok::<Vec<TimelineInfo>, ApiError>(response_data)
758 0 : }
759 0 : .instrument(info_span!("timeline_list",
760 : tenant_id = %tenant_shard_id.tenant_id,
761 0 : shard_id = %tenant_shard_id.shard_slug()))
762 0 : .await?;
763 :
764 0 : json_response(StatusCode::OK, response_data)
765 0 : }
766 :
767 0 : async fn timeline_and_offloaded_list_handler(
768 0 : request: Request<Body>,
769 0 : _cancel: CancellationToken,
770 0 : ) -> Result<Response<Body>, ApiError> {
771 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
772 0 : let include_non_incremental_logical_size: Option<bool> =
773 0 : parse_query_param(&request, "include-non-incremental-logical-size")?;
774 0 : let force_await_initial_logical_size: Option<bool> =
775 0 : parse_query_param(&request, "force-await-initial-logical-size")?;
776 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
777 :
778 0 : let state = get_state(&request);
779 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
780 :
781 0 : let response_data = async {
782 0 : let tenant = state
783 0 : .tenant_manager
784 0 : .get_attached_tenant_shard(tenant_shard_id)?;
785 :
786 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
787 :
788 0 : let (timelines, offloadeds) = tenant.list_timelines_and_offloaded();
789 0 :
790 0 : let mut timeline_infos = Vec::with_capacity(timelines.len());
791 0 : for timeline in timelines {
792 0 : let timeline_info = build_timeline_info(
793 0 : &timeline,
794 0 : include_non_incremental_logical_size.unwrap_or(false),
795 0 : force_await_initial_logical_size.unwrap_or(false),
796 0 : &ctx,
797 0 : )
798 0 : .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
799 0 : .await
800 0 : .context("Failed to build timeline info")
801 0 : .map_err(ApiError::InternalServerError)?;
802 :
803 0 : timeline_infos.push(timeline_info);
804 : }
805 0 : let offloaded_infos = offloadeds
806 0 : .into_iter()
807 0 : .map(|offloaded| build_timeline_offloaded_info(&offloaded))
808 0 : .collect::<Vec<_>>();
809 0 : let res = TimelinesInfoAndOffloaded {
810 0 : timelines: timeline_infos,
811 0 : offloaded: offloaded_infos,
812 0 : };
813 0 : Ok::<TimelinesInfoAndOffloaded, ApiError>(res)
814 0 : }
815 0 : .instrument(info_span!("timeline_and_offloaded_list",
816 : tenant_id = %tenant_shard_id.tenant_id,
817 0 : shard_id = %tenant_shard_id.shard_slug()))
818 0 : .await?;
819 :
820 0 : json_response(StatusCode::OK, response_data)
821 0 : }
822 :
823 0 : async fn timeline_preserve_initdb_handler(
824 0 : request: Request<Body>,
825 0 : _cancel: CancellationToken,
826 0 : ) -> Result<Response<Body>, ApiError> {
827 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
828 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
829 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
830 0 : let state = get_state(&request);
831 :
832 : // Part of the process for disaster recovery from safekeeper-stored WAL:
833 : // If we don't recover into a new timeline but want to keep the timeline ID,
834 : // then the initdb archive is deleted. This endpoint copies it to a different
835 : // location where timeline recreation cand find it.
836 :
837 0 : async {
838 0 : let tenant = state
839 0 : .tenant_manager
840 0 : .get_attached_tenant_shard(tenant_shard_id)?;
841 :
842 0 : let timeline = tenant.get_timeline(timeline_id, false)?;
843 :
844 0 : timeline
845 0 : .preserve_initdb_archive()
846 0 : .await
847 0 : .context("preserving initdb archive")
848 0 : .map_err(ApiError::InternalServerError)?;
849 :
850 0 : Ok::<_, ApiError>(())
851 0 : }
852 0 : .instrument(info_span!("timeline_preserve_initdb_archive",
853 : tenant_id = %tenant_shard_id.tenant_id,
854 0 : shard_id = %tenant_shard_id.shard_slug(),
855 : %timeline_id))
856 0 : .await?;
857 :
858 0 : json_response(StatusCode::OK, ())
859 0 : }
860 :
861 0 : async fn timeline_archival_config_handler(
862 0 : mut request: Request<Body>,
863 0 : _cancel: CancellationToken,
864 0 : ) -> Result<Response<Body>, ApiError> {
865 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
866 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
867 :
868 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
869 :
870 0 : let request_data: TimelineArchivalConfigRequest = json_request(&mut request).await?;
871 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
872 0 : let state = get_state(&request);
873 :
874 0 : async {
875 0 : let tenant = state
876 0 : .tenant_manager
877 0 : .get_attached_tenant_shard(tenant_shard_id)?;
878 :
879 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
880 :
881 0 : tenant
882 0 : .apply_timeline_archival_config(
883 0 : timeline_id,
884 0 : request_data.state,
885 0 : state.broker_client.clone(),
886 0 : ctx,
887 0 : )
888 0 : .await?;
889 0 : Ok::<_, ApiError>(())
890 0 : }
891 0 : .instrument(info_span!("timeline_archival_config",
892 : tenant_id = %tenant_shard_id.tenant_id,
893 0 : shard_id = %tenant_shard_id.shard_slug(),
894 : state = ?request_data.state,
895 : %timeline_id))
896 0 : .await?;
897 :
898 0 : json_response(StatusCode::OK, ())
899 0 : }
900 :
901 0 : async fn timeline_detail_handler(
902 0 : request: Request<Body>,
903 0 : _cancel: CancellationToken,
904 0 : ) -> Result<Response<Body>, ApiError> {
905 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
906 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
907 0 : let include_non_incremental_logical_size: Option<bool> =
908 0 : parse_query_param(&request, "include-non-incremental-logical-size")?;
909 0 : let force_await_initial_logical_size: Option<bool> =
910 0 : parse_query_param(&request, "force-await-initial-logical-size")?;
911 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
912 :
913 : // Logical size calculation needs downloading.
914 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
915 0 : let state = get_state(&request);
916 :
917 0 : let timeline_info = async {
918 0 : let tenant = state
919 0 : .tenant_manager
920 0 : .get_attached_tenant_shard(tenant_shard_id)?;
921 :
922 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
923 :
924 0 : let timeline = tenant.get_timeline(timeline_id, false)?;
925 :
926 0 : let timeline_info = build_timeline_info(
927 0 : &timeline,
928 0 : include_non_incremental_logical_size.unwrap_or(false),
929 0 : force_await_initial_logical_size.unwrap_or(false),
930 0 : &ctx,
931 0 : )
932 0 : .await
933 0 : .context("get local timeline info")
934 0 : .map_err(ApiError::InternalServerError)?;
935 :
936 0 : Ok::<_, ApiError>(timeline_info)
937 0 : }
938 0 : .instrument(info_span!("timeline_detail",
939 : tenant_id = %tenant_shard_id.tenant_id,
940 0 : shard_id = %tenant_shard_id.shard_slug(),
941 : %timeline_id))
942 0 : .await?;
943 :
944 0 : json_response(StatusCode::OK, timeline_info)
945 0 : }
946 :
947 0 : async fn get_lsn_by_timestamp_handler(
948 0 : request: Request<Body>,
949 0 : cancel: CancellationToken,
950 0 : ) -> Result<Response<Body>, ApiError> {
951 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
952 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
953 0 : let state = get_state(&request);
954 0 :
955 0 : if !tenant_shard_id.is_shard_zero() {
956 : // Requires SLRU contents, which are only stored on shard zero
957 0 : return Err(ApiError::BadRequest(anyhow!(
958 0 : "Size calculations are only available on shard zero"
959 0 : )));
960 0 : }
961 :
962 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
963 0 : let timestamp_raw = must_get_query_param(&request, "timestamp")?;
964 0 : let timestamp = humantime::parse_rfc3339(×tamp_raw)
965 0 : .with_context(|| format!("Invalid time: {:?}", timestamp_raw))
966 0 : .map_err(ApiError::BadRequest)?;
967 0 : let timestamp_pg = postgres_ffi::to_pg_timestamp(timestamp);
968 :
969 0 : let with_lease = parse_query_param(&request, "with_lease")?.unwrap_or(false);
970 0 :
971 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
972 :
973 0 : let timeline =
974 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
975 0 : .await?;
976 0 : let result = timeline
977 0 : .find_lsn_for_timestamp(timestamp_pg, &cancel, &ctx)
978 0 : .await?;
979 :
980 : #[derive(serde::Serialize, Debug)]
981 : struct Result {
982 : lsn: Lsn,
983 : kind: &'static str,
984 : #[serde(default)]
985 : #[serde(skip_serializing_if = "Option::is_none")]
986 : #[serde(flatten)]
987 : lease: Option<LsnLease>,
988 : }
989 0 : let (lsn, kind) = match result {
990 0 : LsnForTimestamp::Present(lsn) => (lsn, "present"),
991 0 : LsnForTimestamp::Future(lsn) => (lsn, "future"),
992 0 : LsnForTimestamp::Past(lsn) => (lsn, "past"),
993 0 : LsnForTimestamp::NoData(lsn) => (lsn, "nodata"),
994 : };
995 :
996 0 : let lease = if with_lease {
997 0 : timeline
998 0 : .init_lsn_lease(lsn, timeline.get_lsn_lease_length_for_ts(), &ctx)
999 0 : .inspect_err(|_| {
1000 0 : warn!("fail to grant a lease to {}", lsn);
1001 0 : })
1002 0 : .ok()
1003 : } else {
1004 0 : None
1005 : };
1006 :
1007 0 : let result = Result { lsn, kind, lease };
1008 0 : let valid_until = result
1009 0 : .lease
1010 0 : .as_ref()
1011 0 : .map(|l| humantime::format_rfc3339_millis(l.valid_until).to_string());
1012 0 : tracing::info!(
1013 : lsn=?result.lsn,
1014 : kind=%result.kind,
1015 : timestamp=%timestamp_raw,
1016 : valid_until=?valid_until,
1017 0 : "lsn_by_timestamp finished"
1018 : );
1019 0 : json_response(StatusCode::OK, result)
1020 0 : }
1021 :
1022 0 : async fn get_timestamp_of_lsn_handler(
1023 0 : request: Request<Body>,
1024 0 : _cancel: CancellationToken,
1025 0 : ) -> Result<Response<Body>, ApiError> {
1026 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1027 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1028 0 : let state = get_state(&request);
1029 0 :
1030 0 : if !tenant_shard_id.is_shard_zero() {
1031 : // Requires SLRU contents, which are only stored on shard zero
1032 0 : return Err(ApiError::BadRequest(anyhow!(
1033 0 : "Size calculations are only available on shard zero"
1034 0 : )));
1035 0 : }
1036 :
1037 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1038 :
1039 0 : let lsn_str = must_get_query_param(&request, "lsn")?;
1040 0 : let lsn = Lsn::from_str(&lsn_str)
1041 0 : .with_context(|| format!("Invalid LSN: {lsn_str:?}"))
1042 0 : .map_err(ApiError::BadRequest)?;
1043 :
1044 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1045 0 : let timeline =
1046 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1047 0 : .await?;
1048 0 : let result = timeline.get_timestamp_for_lsn(lsn, &ctx).await?;
1049 :
1050 0 : match result {
1051 0 : Some(time) => {
1052 0 : let time = format_rfc3339(
1053 0 : postgres_ffi::try_from_pg_timestamp(time).map_err(ApiError::InternalServerError)?,
1054 : )
1055 0 : .to_string();
1056 0 : json_response(StatusCode::OK, time)
1057 : }
1058 0 : None => Err(ApiError::NotFound(
1059 0 : anyhow::anyhow!("Timestamp for lsn {} not found", lsn).into(),
1060 0 : )),
1061 : }
1062 0 : }
1063 :
1064 0 : async fn timeline_delete_handler(
1065 0 : request: Request<Body>,
1066 0 : _cancel: CancellationToken,
1067 0 : ) -> Result<Response<Body>, ApiError> {
1068 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1069 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1070 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1071 :
1072 0 : let state = get_state(&request);
1073 :
1074 0 : let tenant = state
1075 0 : .tenant_manager
1076 0 : .get_attached_tenant_shard(tenant_shard_id)
1077 0 : .map_err(|e| {
1078 0 : match e {
1079 : // GetTenantError has a built-in conversion to ApiError, but in this context we don't
1080 : // want to treat missing tenants as 404, to avoid ambiguity with successful deletions.
1081 : GetTenantError::NotFound(_) | GetTenantError::ShardNotFound(_) => {
1082 0 : ApiError::PreconditionFailed(
1083 0 : "Requested tenant is missing".to_string().into_boxed_str(),
1084 0 : )
1085 : }
1086 0 : e => e.into(),
1087 : }
1088 0 : })?;
1089 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
1090 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))
1091 0 : .await?;
1092 :
1093 0 : json_response(StatusCode::ACCEPTED, ())
1094 0 : }
1095 :
1096 0 : async fn tenant_reset_handler(
1097 0 : request: Request<Body>,
1098 0 : _cancel: CancellationToken,
1099 0 : ) -> Result<Response<Body>, ApiError> {
1100 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1101 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1102 :
1103 0 : let drop_cache: Option<bool> = parse_query_param(&request, "drop_cache")?;
1104 :
1105 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
1106 0 : let state = get_state(&request);
1107 0 : state
1108 0 : .tenant_manager
1109 0 : .reset_tenant(tenant_shard_id, drop_cache.unwrap_or(false), &ctx)
1110 0 : .await
1111 0 : .map_err(ApiError::InternalServerError)?;
1112 :
1113 0 : json_response(StatusCode::OK, ())
1114 0 : }
1115 :
1116 0 : async fn tenant_list_handler(
1117 0 : request: Request<Body>,
1118 0 : _cancel: CancellationToken,
1119 0 : ) -> Result<Response<Body>, ApiError> {
1120 0 : check_permission(&request, None)?;
1121 0 : let state = get_state(&request);
1122 :
1123 0 : let response_data = state
1124 0 : .tenant_manager
1125 0 : .list_tenants()
1126 0 : .map_err(|_| {
1127 0 : ApiError::ResourceUnavailable("Tenant map is initializing or shutting down".into())
1128 0 : })?
1129 0 : .iter()
1130 0 : .map(|(id, state, gen)| TenantInfo {
1131 0 : id: *id,
1132 0 : state: state.clone(),
1133 0 : current_physical_size: None,
1134 0 : attachment_status: state.attachment_status(),
1135 0 : generation: (*gen)
1136 0 : .into()
1137 0 : .expect("Tenants are always attached with a generation"),
1138 0 : gc_blocking: None,
1139 0 : })
1140 0 : .collect::<Vec<TenantInfo>>();
1141 0 :
1142 0 : json_response(StatusCode::OK, response_data)
1143 0 : }
1144 :
1145 0 : async fn tenant_status(
1146 0 : request: Request<Body>,
1147 0 : _cancel: CancellationToken,
1148 0 : ) -> Result<Response<Body>, ApiError> {
1149 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1150 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1151 0 : let state = get_state(&request);
1152 0 :
1153 0 : // In tests, sometimes we want to query the state of a tenant without auto-activating it if it's currently waiting.
1154 0 : let activate = true;
1155 : #[cfg(feature = "testing")]
1156 0 : let activate = parse_query_param(&request, "activate")?.unwrap_or(activate);
1157 :
1158 0 : let tenant_info = async {
1159 0 : let tenant = state
1160 0 : .tenant_manager
1161 0 : .get_attached_tenant_shard(tenant_shard_id)?;
1162 :
1163 0 : if activate {
1164 : // This is advisory: we prefer to let the tenant activate on-demand when this function is
1165 : // called, but it is still valid to return 200 and describe the current state of the tenant
1166 : // if it doesn't make it into an active state.
1167 0 : tenant
1168 0 : .wait_to_become_active(ACTIVE_TENANT_TIMEOUT)
1169 0 : .await
1170 0 : .ok();
1171 0 : }
1172 :
1173 : // Calculate total physical size of all timelines
1174 0 : let mut current_physical_size = 0;
1175 0 : for timeline in tenant.list_timelines().iter() {
1176 0 : current_physical_size += timeline.layer_size_sum().await;
1177 : }
1178 :
1179 0 : let state = tenant.current_state();
1180 0 : Result::<_, ApiError>::Ok(TenantDetails {
1181 0 : tenant_info: TenantInfo {
1182 0 : id: tenant_shard_id,
1183 0 : state: state.clone(),
1184 0 : current_physical_size: Some(current_physical_size),
1185 0 : attachment_status: state.attachment_status(),
1186 0 : generation: tenant
1187 0 : .generation()
1188 0 : .into()
1189 0 : .expect("Tenants are always attached with a generation"),
1190 0 : gc_blocking: tenant.gc_block.summary().map(|x| format!("{x:?}")),
1191 0 : },
1192 0 : walredo: tenant.wal_redo_manager_status(),
1193 0 : timelines: tenant.list_timeline_ids(),
1194 0 : })
1195 0 : }
1196 0 : .instrument(info_span!("tenant_status_handler",
1197 : tenant_id = %tenant_shard_id.tenant_id,
1198 0 : shard_id = %tenant_shard_id.shard_slug()))
1199 0 : .await?;
1200 :
1201 0 : json_response(StatusCode::OK, tenant_info)
1202 0 : }
1203 :
1204 0 : async fn tenant_delete_handler(
1205 0 : request: Request<Body>,
1206 0 : _cancel: CancellationToken,
1207 0 : ) -> Result<Response<Body>, ApiError> {
1208 : // TODO openapi spec
1209 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1210 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1211 :
1212 0 : let state = get_state(&request);
1213 0 :
1214 0 : state
1215 0 : .tenant_manager
1216 0 : .delete_tenant(tenant_shard_id)
1217 0 : .instrument(info_span!("tenant_delete_handler",
1218 : tenant_id = %tenant_shard_id.tenant_id,
1219 0 : shard_id = %tenant_shard_id.shard_slug()
1220 : ))
1221 0 : .await?;
1222 :
1223 0 : json_response(StatusCode::OK, ())
1224 0 : }
1225 :
1226 : /// HTTP endpoint to query the current tenant_size of a tenant.
1227 : ///
1228 : /// This is not used by consumption metrics under [`crate::consumption_metrics`], but can be used
1229 : /// to debug any of the calculations. Requires `tenant_id` request parameter, supports
1230 : /// `inputs_only=true|false` (default false) which supports debugging failure to calculate model
1231 : /// values.
1232 : ///
1233 : /// 'retention_period' query parameter overrides the cutoff that is used to calculate the size
1234 : /// (only if it is shorter than the real cutoff).
1235 : ///
1236 : /// Note: we don't update the cached size and prometheus metric here.
1237 : /// The retention period might be different, and it's nice to have a method to just calculate it
1238 : /// without modifying anything anyway.
1239 0 : async fn tenant_size_handler(
1240 0 : request: Request<Body>,
1241 0 : cancel: CancellationToken,
1242 0 : ) -> Result<Response<Body>, ApiError> {
1243 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1244 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1245 0 : let inputs_only: Option<bool> = parse_query_param(&request, "inputs_only")?;
1246 0 : let retention_period: Option<u64> = parse_query_param(&request, "retention_period")?;
1247 0 : let headers = request.headers();
1248 0 : let state = get_state(&request);
1249 0 :
1250 0 : if !tenant_shard_id.is_shard_zero() {
1251 0 : return Err(ApiError::BadRequest(anyhow!(
1252 0 : "Size calculations are only available on shard zero"
1253 0 : )));
1254 0 : }
1255 0 :
1256 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1257 0 : let tenant = state
1258 0 : .tenant_manager
1259 0 : .get_attached_tenant_shard(tenant_shard_id)?;
1260 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
1261 :
1262 : // this can be long operation
1263 0 : let inputs = tenant
1264 0 : .gather_size_inputs(
1265 0 : retention_period,
1266 0 : LogicalSizeCalculationCause::TenantSizeHandler,
1267 0 : &cancel,
1268 0 : &ctx,
1269 0 : )
1270 0 : .await
1271 0 : .map_err(|e| match e {
1272 0 : crate::tenant::size::CalculateSyntheticSizeError::Cancelled => ApiError::ShuttingDown,
1273 0 : other => ApiError::InternalServerError(anyhow::anyhow!(other)),
1274 0 : })?;
1275 :
1276 0 : let mut sizes = None;
1277 0 : let accepts_html = headers
1278 0 : .get(header::ACCEPT)
1279 0 : .map(|v| v == "text/html")
1280 0 : .unwrap_or_default();
1281 0 : if !inputs_only.unwrap_or(false) {
1282 0 : let storage_model = inputs.calculate_model();
1283 0 : let size = storage_model.calculate();
1284 0 :
1285 0 : // If request header expects html, return html
1286 0 : if accepts_html {
1287 0 : return synthetic_size_html_response(inputs, storage_model, size);
1288 0 : }
1289 0 : sizes = Some(size);
1290 0 : } else if accepts_html {
1291 0 : return Err(ApiError::BadRequest(anyhow!(
1292 0 : "inputs_only parameter is incompatible with html output request"
1293 0 : )));
1294 0 : }
1295 :
1296 : /// The type resides in the pageserver not to expose `ModelInputs`.
1297 : #[derive(serde::Serialize)]
1298 : struct TenantHistorySize {
1299 : id: TenantId,
1300 : /// Size is a mixture of WAL and logical size, so the unit is bytes.
1301 : ///
1302 : /// Will be none if `?inputs_only=true` was given.
1303 : size: Option<u64>,
1304 : /// Size of each segment used in the model.
1305 : /// Will be null if `?inputs_only=true` was given.
1306 : segment_sizes: Option<Vec<tenant_size_model::SegmentSizeResult>>,
1307 : inputs: crate::tenant::size::ModelInputs,
1308 : }
1309 :
1310 0 : json_response(
1311 0 : StatusCode::OK,
1312 0 : TenantHistorySize {
1313 0 : id: tenant_shard_id.tenant_id,
1314 0 : size: sizes.as_ref().map(|x| x.total_size),
1315 0 : segment_sizes: sizes.map(|x| x.segments),
1316 0 : inputs,
1317 0 : },
1318 0 : )
1319 0 : }
1320 :
1321 0 : async fn tenant_shard_split_handler(
1322 0 : mut request: Request<Body>,
1323 0 : _cancel: CancellationToken,
1324 0 : ) -> Result<Response<Body>, ApiError> {
1325 0 : let req: TenantShardSplitRequest = json_request(&mut request).await?;
1326 :
1327 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1328 0 : let state = get_state(&request);
1329 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
1330 :
1331 0 : let tenant = state
1332 0 : .tenant_manager
1333 0 : .get_attached_tenant_shard(tenant_shard_id)?;
1334 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
1335 :
1336 0 : let new_shards = state
1337 0 : .tenant_manager
1338 0 : .shard_split(
1339 0 : tenant,
1340 0 : ShardCount::new(req.new_shard_count),
1341 0 : req.new_stripe_size,
1342 0 : &ctx,
1343 0 : )
1344 0 : .await
1345 0 : .map_err(ApiError::InternalServerError)?;
1346 :
1347 0 : json_response(StatusCode::OK, TenantShardSplitResponse { new_shards })
1348 0 : }
1349 :
1350 0 : async fn layer_map_info_handler(
1351 0 : request: Request<Body>,
1352 0 : _cancel: CancellationToken,
1353 0 : ) -> Result<Response<Body>, ApiError> {
1354 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1355 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1356 0 : let reset: LayerAccessStatsReset =
1357 0 : parse_query_param(&request, "reset")?.unwrap_or(LayerAccessStatsReset::NoReset);
1358 0 : let state = get_state(&request);
1359 0 :
1360 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1361 :
1362 0 : let timeline =
1363 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1364 0 : .await?;
1365 0 : let layer_map_info = timeline
1366 0 : .layer_map_info(reset)
1367 0 : .await
1368 0 : .map_err(|_shutdown| ApiError::ShuttingDown)?;
1369 :
1370 0 : json_response(StatusCode::OK, layer_map_info)
1371 0 : }
1372 :
1373 : #[instrument(skip_all, fields(tenant_id, shard_id, timeline_id, layer_name))]
1374 : async fn timeline_layer_scan_disposable_keys(
1375 : request: Request<Body>,
1376 : cancel: CancellationToken,
1377 : ) -> Result<Response<Body>, ApiError> {
1378 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1379 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1380 : let layer_name: LayerName = parse_request_param(&request, "layer_name")?;
1381 :
1382 : tracing::Span::current().record(
1383 : "tenant_id",
1384 : tracing::field::display(&tenant_shard_id.tenant_id),
1385 : );
1386 : tracing::Span::current().record(
1387 : "shard_id",
1388 : tracing::field::display(tenant_shard_id.shard_slug()),
1389 : );
1390 : tracing::Span::current().record("timeline_id", tracing::field::display(&timeline_id));
1391 : tracing::Span::current().record("layer_name", tracing::field::display(&layer_name));
1392 :
1393 : let state = get_state(&request);
1394 :
1395 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1396 :
1397 : // technically the timeline need not be active for this scan to complete
1398 : let timeline =
1399 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1400 : .await?;
1401 :
1402 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
1403 :
1404 : let guard = timeline.layers.read().await;
1405 : let Some(layer) = guard.try_get_from_key(&layer_name.clone().into()) else {
1406 : return Err(ApiError::NotFound(
1407 : anyhow::anyhow!("Layer {tenant_shard_id}/{timeline_id}/{layer_name} not found").into(),
1408 : ));
1409 : };
1410 :
1411 : let resident_layer = layer
1412 : .download_and_keep_resident()
1413 : .await
1414 0 : .map_err(|err| match err {
1415 : tenant::storage_layer::layer::DownloadError::TimelineShutdown
1416 : | tenant::storage_layer::layer::DownloadError::DownloadCancelled => {
1417 0 : ApiError::ShuttingDown
1418 : }
1419 : tenant::storage_layer::layer::DownloadError::ContextAndConfigReallyDeniesDownloads
1420 : | tenant::storage_layer::layer::DownloadError::DownloadRequired
1421 : | tenant::storage_layer::layer::DownloadError::NotFile(_)
1422 : | tenant::storage_layer::layer::DownloadError::DownloadFailed
1423 : | tenant::storage_layer::layer::DownloadError::PreStatFailed(_) => {
1424 0 : ApiError::InternalServerError(err.into())
1425 : }
1426 : #[cfg(test)]
1427 : tenant::storage_layer::layer::DownloadError::Failpoint(_) => {
1428 0 : ApiError::InternalServerError(err.into())
1429 : }
1430 0 : })?;
1431 :
1432 : let keys = resident_layer
1433 : .load_keys(&ctx)
1434 : .await
1435 : .map_err(ApiError::InternalServerError)?;
1436 :
1437 : let shard_identity = timeline.get_shard_identity();
1438 :
1439 : let mut disposable_count = 0;
1440 : let mut not_disposable_count = 0;
1441 : let cancel = cancel.clone();
1442 : for (i, key) in keys.into_iter().enumerate() {
1443 : if shard_identity.is_key_disposable(&key) {
1444 : disposable_count += 1;
1445 : tracing::debug!(key = %key, key.dbg=?key, "disposable key");
1446 : } else {
1447 : not_disposable_count += 1;
1448 : }
1449 : #[allow(clippy::collapsible_if)]
1450 : if i % 10000 == 0 {
1451 : if cancel.is_cancelled() || timeline.cancel.is_cancelled() || timeline.is_stopping() {
1452 : return Err(ApiError::ShuttingDown);
1453 : }
1454 : }
1455 : }
1456 :
1457 : json_response(
1458 : StatusCode::OK,
1459 : pageserver_api::models::ScanDisposableKeysResponse {
1460 : disposable_count,
1461 : not_disposable_count,
1462 : },
1463 : )
1464 : }
1465 :
1466 0 : async fn layer_download_handler(
1467 0 : request: Request<Body>,
1468 0 : _cancel: CancellationToken,
1469 0 : ) -> Result<Response<Body>, ApiError> {
1470 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1471 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1472 0 : let layer_file_name = get_request_param(&request, "layer_file_name")?;
1473 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1474 0 : let layer_name = LayerName::from_str(layer_file_name)
1475 0 : .map_err(|s| ApiError::BadRequest(anyhow::anyhow!(s)))?;
1476 0 : let state = get_state(&request);
1477 :
1478 0 : let timeline =
1479 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1480 0 : .await?;
1481 0 : let downloaded = timeline
1482 0 : .download_layer(&layer_name)
1483 0 : .await
1484 0 : .map_err(|e| match e {
1485 : tenant::storage_layer::layer::DownloadError::TimelineShutdown
1486 : | tenant::storage_layer::layer::DownloadError::DownloadCancelled => {
1487 0 : ApiError::ShuttingDown
1488 : }
1489 0 : other => ApiError::InternalServerError(other.into()),
1490 0 : })?;
1491 :
1492 0 : match downloaded {
1493 0 : Some(true) => json_response(StatusCode::OK, ()),
1494 0 : Some(false) => json_response(StatusCode::NOT_MODIFIED, ()),
1495 0 : None => json_response(
1496 0 : StatusCode::BAD_REQUEST,
1497 0 : format!("Layer {tenant_shard_id}/{timeline_id}/{layer_file_name} not found"),
1498 0 : ),
1499 : }
1500 0 : }
1501 :
1502 0 : async fn evict_timeline_layer_handler(
1503 0 : request: Request<Body>,
1504 0 : _cancel: CancellationToken,
1505 0 : ) -> Result<Response<Body>, ApiError> {
1506 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1507 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1508 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1509 0 : let layer_file_name = get_request_param(&request, "layer_file_name")?;
1510 0 : let state = get_state(&request);
1511 :
1512 0 : let layer_name = LayerName::from_str(layer_file_name)
1513 0 : .map_err(|s| ApiError::BadRequest(anyhow::anyhow!(s)))?;
1514 :
1515 0 : let timeline =
1516 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1517 0 : .await?;
1518 0 : let evicted = timeline
1519 0 : .evict_layer(&layer_name)
1520 0 : .await
1521 0 : .map_err(ApiError::InternalServerError)?;
1522 :
1523 0 : match evicted {
1524 0 : Some(true) => json_response(StatusCode::OK, ()),
1525 0 : Some(false) => json_response(StatusCode::NOT_MODIFIED, ()),
1526 0 : None => json_response(
1527 0 : StatusCode::BAD_REQUEST,
1528 0 : format!("Layer {tenant_shard_id}/{timeline_id}/{layer_file_name} not found"),
1529 0 : ),
1530 : }
1531 0 : }
1532 :
1533 0 : async fn timeline_gc_blocking_handler(
1534 0 : request: Request<Body>,
1535 0 : _cancel: CancellationToken,
1536 0 : ) -> Result<Response<Body>, ApiError> {
1537 0 : block_or_unblock_gc(request, true).await
1538 0 : }
1539 :
1540 0 : async fn timeline_gc_unblocking_handler(
1541 0 : request: Request<Body>,
1542 0 : _cancel: CancellationToken,
1543 0 : ) -> Result<Response<Body>, ApiError> {
1544 0 : block_or_unblock_gc(request, false).await
1545 0 : }
1546 :
1547 : /// Traces GetPage@LSN requests for a timeline, and emits metadata in an efficient binary encoding.
1548 : /// Use the `pagectl page-trace` command to decode and analyze the output.
1549 0 : async fn timeline_page_trace_handler(
1550 0 : request: Request<Body>,
1551 0 : cancel: CancellationToken,
1552 0 : ) -> Result<Response<Body>, ApiError> {
1553 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1554 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1555 0 : let state = get_state(&request);
1556 0 : check_permission(&request, None)?;
1557 :
1558 0 : let size_limit: usize = parse_query_param(&request, "size_limit_bytes")?.unwrap_or(1024 * 1024);
1559 0 : let time_limit_secs: u64 = parse_query_param(&request, "time_limit_secs")?.unwrap_or(5);
1560 :
1561 : // Convert size limit to event limit based on the serialized size of an event. The event size is
1562 : // fixed, as the default bincode serializer uses fixed-width integer encoding.
1563 0 : let event_size = bincode::serialize(&PageTraceEvent::default())
1564 0 : .map_err(|err| ApiError::InternalServerError(err.into()))?
1565 0 : .len();
1566 0 : let event_limit = size_limit / event_size;
1567 :
1568 0 : let timeline =
1569 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
1570 0 : .await?;
1571 :
1572 : // Install a page trace, unless one is already in progress. We just use a buffered channel,
1573 : // which may 2x the memory usage in the worst case, but it's still bounded.
1574 0 : let (trace_tx, mut trace_rx) = tokio::sync::mpsc::channel(event_limit);
1575 0 : let cur = timeline.page_trace.load();
1576 0 : let installed = cur.is_none()
1577 0 : && timeline
1578 0 : .page_trace
1579 0 : .compare_and_swap(cur, Some(Arc::new(trace_tx)))
1580 0 : .is_none();
1581 0 : if !installed {
1582 0 : return Err(ApiError::Conflict("page trace already active".to_string()));
1583 0 : }
1584 0 : defer!(timeline.page_trace.store(None)); // uninstall on return
1585 0 :
1586 0 : // Collect the trace and return it to the client. We could stream the response, but this is
1587 0 : // simple and fine.
1588 0 : let mut body = Vec::with_capacity(size_limit);
1589 0 : let deadline = Instant::now() + Duration::from_secs(time_limit_secs);
1590 :
1591 0 : while body.len() < size_limit {
1592 0 : tokio::select! {
1593 0 : event = trace_rx.recv() => {
1594 0 : let Some(event) = event else {
1595 0 : break; // shouldn't happen (sender doesn't close, unless timeline dropped)
1596 : };
1597 0 : bincode::serialize_into(&mut body, &event)
1598 0 : .map_err(|err| ApiError::InternalServerError(err.into()))?;
1599 : }
1600 0 : _ = tokio::time::sleep_until(deadline) => break, // time limit reached
1601 0 : _ = cancel.cancelled() => return Err(ApiError::Cancelled),
1602 : }
1603 : }
1604 :
1605 0 : Ok(Response::builder()
1606 0 : .status(StatusCode::OK)
1607 0 : .header(header::CONTENT_TYPE, "application/octet-stream")
1608 0 : .body(hyper::Body::from(body))
1609 0 : .unwrap())
1610 0 : }
1611 :
1612 : /// Adding a block is `POST ../block_gc`, removing a block is `POST ../unblock_gc`.
1613 : ///
1614 : /// Both are technically unsafe because they might fire off index uploads, thus they are POST.
1615 0 : async fn block_or_unblock_gc(
1616 0 : request: Request<Body>,
1617 0 : block: bool,
1618 0 : ) -> Result<Response<Body>, ApiError> {
1619 : use crate::tenant::{
1620 : remote_timeline_client::WaitCompletionError, upload_queue::NotInitialized,
1621 : };
1622 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1623 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1624 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
1625 0 : let state = get_state(&request);
1626 :
1627 0 : let tenant = state
1628 0 : .tenant_manager
1629 0 : .get_attached_tenant_shard(tenant_shard_id)?;
1630 :
1631 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
1632 :
1633 0 : let timeline = tenant.get_timeline(timeline_id, true)?;
1634 :
1635 0 : let fut = async {
1636 0 : if block {
1637 0 : timeline.block_gc(&tenant).await.map(|_| ())
1638 : } else {
1639 0 : timeline.unblock_gc(&tenant).await
1640 : }
1641 0 : };
1642 :
1643 0 : let span = tracing::info_span!(
1644 : "block_or_unblock_gc",
1645 : tenant_id = %tenant_shard_id.tenant_id,
1646 0 : shard_id = %tenant_shard_id.shard_slug(),
1647 : timeline_id = %timeline_id,
1648 : block = block,
1649 : );
1650 :
1651 0 : let res = fut.instrument(span).await;
1652 :
1653 0 : res.map_err(|e| {
1654 0 : if e.is::<NotInitialized>() || e.is::<WaitCompletionError>() {
1655 0 : ApiError::ShuttingDown
1656 : } else {
1657 0 : ApiError::InternalServerError(e)
1658 : }
1659 0 : })?;
1660 :
1661 0 : json_response(StatusCode::OK, ())
1662 0 : }
1663 :
1664 : /// Get tenant_size SVG graph along with the JSON data.
1665 0 : fn synthetic_size_html_response(
1666 0 : inputs: ModelInputs,
1667 0 : storage_model: StorageModel,
1668 0 : sizes: SizeResult,
1669 0 : ) -> Result<Response<Body>, ApiError> {
1670 0 : let mut timeline_ids: Vec<String> = Vec::new();
1671 0 : let mut timeline_map: HashMap<TimelineId, usize> = HashMap::new();
1672 0 : for (index, ti) in inputs.timeline_inputs.iter().enumerate() {
1673 0 : timeline_map.insert(ti.timeline_id, index);
1674 0 : timeline_ids.push(ti.timeline_id.to_string());
1675 0 : }
1676 0 : let seg_to_branch: Vec<(usize, SvgBranchKind)> = inputs
1677 0 : .segments
1678 0 : .iter()
1679 0 : .map(|seg| {
1680 0 : (
1681 0 : *timeline_map.get(&seg.timeline_id).unwrap(),
1682 0 : seg.kind.into(),
1683 0 : )
1684 0 : })
1685 0 : .collect();
1686 :
1687 0 : let svg =
1688 0 : tenant_size_model::svg::draw_svg(&storage_model, &timeline_ids, &seg_to_branch, &sizes)
1689 0 : .map_err(ApiError::InternalServerError)?;
1690 :
1691 0 : let mut response = String::new();
1692 :
1693 : use std::fmt::Write;
1694 0 : write!(response, "<html>\n<body>\n").unwrap();
1695 0 : write!(response, "<div>\n{svg}\n</div>").unwrap();
1696 0 : writeln!(response, "Project size: {}", sizes.total_size).unwrap();
1697 0 : writeln!(response, "<pre>").unwrap();
1698 0 : writeln!(
1699 0 : response,
1700 0 : "{}",
1701 0 : serde_json::to_string_pretty(&inputs).unwrap()
1702 0 : )
1703 0 : .unwrap();
1704 0 : writeln!(
1705 0 : response,
1706 0 : "{}",
1707 0 : serde_json::to_string_pretty(&sizes.segments).unwrap()
1708 0 : )
1709 0 : .unwrap();
1710 0 : writeln!(response, "</pre>").unwrap();
1711 0 : write!(response, "</body>\n</html>\n").unwrap();
1712 0 :
1713 0 : html_response(StatusCode::OK, response)
1714 0 : }
1715 :
1716 0 : pub fn html_response(status: StatusCode, data: String) -> Result<Response<Body>, ApiError> {
1717 0 : let response = Response::builder()
1718 0 : .status(status)
1719 0 : .header(header::CONTENT_TYPE, "text/html")
1720 0 : .body(Body::from(data.as_bytes().to_vec()))
1721 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
1722 0 : Ok(response)
1723 0 : }
1724 :
1725 0 : async fn get_tenant_config_handler(
1726 0 : request: Request<Body>,
1727 0 : _cancel: CancellationToken,
1728 0 : ) -> Result<Response<Body>, ApiError> {
1729 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1730 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1731 0 : let state = get_state(&request);
1732 :
1733 0 : let tenant = state
1734 0 : .tenant_manager
1735 0 : .get_attached_tenant_shard(tenant_shard_id)?;
1736 :
1737 0 : let response = HashMap::from([
1738 : (
1739 : "tenant_specific_overrides",
1740 0 : serde_json::to_value(tenant.tenant_specific_overrides())
1741 0 : .context("serializing tenant specific overrides")
1742 0 : .map_err(ApiError::InternalServerError)?,
1743 : ),
1744 : (
1745 0 : "effective_config",
1746 0 : serde_json::to_value(tenant.effective_config())
1747 0 : .context("serializing effective config")
1748 0 : .map_err(ApiError::InternalServerError)?,
1749 : ),
1750 : ]);
1751 :
1752 0 : json_response(StatusCode::OK, response)
1753 0 : }
1754 :
1755 0 : async fn update_tenant_config_handler(
1756 0 : mut request: Request<Body>,
1757 0 : _cancel: CancellationToken,
1758 0 : ) -> Result<Response<Body>, ApiError> {
1759 0 : let request_data: TenantConfigRequest = json_request(&mut request).await?;
1760 0 : let tenant_id = request_data.tenant_id;
1761 0 : check_permission(&request, Some(tenant_id))?;
1762 :
1763 0 : let new_tenant_conf =
1764 0 : TenantConfOpt::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
1765 :
1766 0 : let state = get_state(&request);
1767 0 :
1768 0 : let tenant_shard_id = TenantShardId::unsharded(tenant_id);
1769 :
1770 0 : let tenant = state
1771 0 : .tenant_manager
1772 0 : .get_attached_tenant_shard(tenant_shard_id)?;
1773 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
1774 :
1775 : // This is a legacy API that only operates on attached tenants: the preferred
1776 : // API to use is the location_config/ endpoint, which lets the caller provide
1777 : // the full LocationConf.
1778 0 : let location_conf = LocationConf::attached_single(
1779 0 : new_tenant_conf.clone(),
1780 0 : tenant.get_generation(),
1781 0 : &ShardParameters::default(),
1782 0 : );
1783 0 :
1784 0 : crate::tenant::Tenant::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
1785 0 : .await
1786 0 : .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
1787 :
1788 0 : let _ = tenant
1789 0 : .update_tenant_config(|_crnt| Ok(new_tenant_conf.clone()))
1790 0 : .expect("Closure returns Ok()");
1791 0 :
1792 0 : json_response(StatusCode::OK, ())
1793 0 : }
1794 :
1795 0 : async fn patch_tenant_config_handler(
1796 0 : mut request: Request<Body>,
1797 0 : _cancel: CancellationToken,
1798 0 : ) -> Result<Response<Body>, ApiError> {
1799 0 : let request_data: TenantConfigPatchRequest = json_request(&mut request).await?;
1800 0 : let tenant_id = request_data.tenant_id;
1801 0 : check_permission(&request, Some(tenant_id))?;
1802 :
1803 0 : let state = get_state(&request);
1804 0 :
1805 0 : let tenant_shard_id = TenantShardId::unsharded(tenant_id);
1806 :
1807 0 : let tenant = state
1808 0 : .tenant_manager
1809 0 : .get_attached_tenant_shard(tenant_shard_id)?;
1810 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
1811 :
1812 0 : let updated = tenant
1813 0 : .update_tenant_config(|crnt| crnt.apply_patch(request_data.config.clone()))
1814 0 : .map_err(ApiError::BadRequest)?;
1815 :
1816 : // This is a legacy API that only operates on attached tenants: the preferred
1817 : // API to use is the location_config/ endpoint, which lets the caller provide
1818 : // the full LocationConf.
1819 0 : let location_conf = LocationConf::attached_single(
1820 0 : updated,
1821 0 : tenant.get_generation(),
1822 0 : &ShardParameters::default(),
1823 0 : );
1824 0 :
1825 0 : crate::tenant::Tenant::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
1826 0 : .await
1827 0 : .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
1828 :
1829 0 : json_response(StatusCode::OK, ())
1830 0 : }
1831 :
1832 0 : async fn put_tenant_location_config_handler(
1833 0 : mut request: Request<Body>,
1834 0 : _cancel: CancellationToken,
1835 0 : ) -> Result<Response<Body>, ApiError> {
1836 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1837 :
1838 0 : let request_data: TenantLocationConfigRequest = json_request(&mut request).await?;
1839 0 : let flush = parse_query_param(&request, "flush_ms")?.map(Duration::from_millis);
1840 0 : let lazy = parse_query_param(&request, "lazy")?.unwrap_or(false);
1841 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1842 :
1843 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
1844 0 : let state = get_state(&request);
1845 0 : let conf = state.conf;
1846 0 :
1847 0 : // The `Detached` state is special, it doesn't upsert a tenant, it removes
1848 0 : // its local disk content and drops it from memory.
1849 0 : if let LocationConfigMode::Detached = request_data.config.mode {
1850 0 : if let Err(e) = state
1851 0 : .tenant_manager
1852 0 : .detach_tenant(conf, tenant_shard_id, &state.deletion_queue_client)
1853 0 : .instrument(info_span!("tenant_detach",
1854 : tenant_id = %tenant_shard_id.tenant_id,
1855 0 : shard_id = %tenant_shard_id.shard_slug()
1856 : ))
1857 0 : .await
1858 : {
1859 0 : match e {
1860 0 : TenantStateError::SlotError(TenantSlotError::NotFound(_)) => {
1861 0 : // This API is idempotent: a NotFound on a detach is fine.
1862 0 : }
1863 0 : _ => return Err(e.into()),
1864 : }
1865 0 : }
1866 0 : return json_response(StatusCode::OK, ());
1867 0 : }
1868 :
1869 0 : let location_conf =
1870 0 : LocationConf::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
1871 :
1872 : // lazy==true queues up for activation or jumps the queue like normal when a compute connects,
1873 : // similar to at startup ordering.
1874 0 : let spawn_mode = if lazy {
1875 0 : tenant::SpawnMode::Lazy
1876 : } else {
1877 0 : tenant::SpawnMode::Eager
1878 : };
1879 :
1880 0 : let tenant = state
1881 0 : .tenant_manager
1882 0 : .upsert_location(tenant_shard_id, location_conf, flush, spawn_mode, &ctx)
1883 0 : .await?;
1884 0 : let stripe_size = tenant.as_ref().map(|t| t.get_shard_stripe_size());
1885 0 : let attached = tenant.is_some();
1886 :
1887 0 : if let Some(_flush_ms) = flush {
1888 0 : match state
1889 0 : .secondary_controller
1890 0 : .upload_tenant(tenant_shard_id)
1891 0 : .await
1892 : {
1893 : Ok(()) => {
1894 0 : tracing::info!("Uploaded heatmap during flush");
1895 : }
1896 0 : Err(e) => {
1897 0 : tracing::warn!("Failed to flush heatmap: {e}");
1898 : }
1899 : }
1900 : } else {
1901 0 : tracing::info!("No flush requested when configuring");
1902 : }
1903 :
1904 : // This API returns a vector of pageservers where the tenant is attached: this is
1905 : // primarily for use in the sharding service. For compatibilty, we also return this
1906 : // when called directly on a pageserver, but the payload is always zero or one shards.
1907 0 : let mut response = TenantLocationConfigResponse {
1908 0 : shards: Vec::new(),
1909 0 : stripe_size: None,
1910 0 : };
1911 0 : if attached {
1912 0 : response.shards.push(TenantShardLocation {
1913 0 : shard_id: tenant_shard_id,
1914 0 : node_id: state.conf.id,
1915 0 : });
1916 0 : if tenant_shard_id.shard_count.count() > 1 {
1917 : // Stripe size should be set if we are attached
1918 0 : debug_assert!(stripe_size.is_some());
1919 0 : response.stripe_size = stripe_size;
1920 0 : }
1921 0 : }
1922 :
1923 0 : json_response(StatusCode::OK, response)
1924 0 : }
1925 :
1926 0 : async fn list_location_config_handler(
1927 0 : request: Request<Body>,
1928 0 : _cancel: CancellationToken,
1929 0 : ) -> Result<Response<Body>, ApiError> {
1930 0 : let state = get_state(&request);
1931 0 : let slots = state.tenant_manager.list();
1932 0 : let result = LocationConfigListResponse {
1933 0 : tenant_shards: slots
1934 0 : .into_iter()
1935 0 : .map(|(tenant_shard_id, slot)| {
1936 0 : let v = match slot {
1937 0 : TenantSlot::Attached(t) => Some(t.get_location_conf()),
1938 0 : TenantSlot::Secondary(s) => Some(s.get_location_conf()),
1939 0 : TenantSlot::InProgress(_) => None,
1940 : };
1941 0 : (tenant_shard_id, v)
1942 0 : })
1943 0 : .collect(),
1944 0 : };
1945 0 : json_response(StatusCode::OK, result)
1946 0 : }
1947 :
1948 0 : async fn get_location_config_handler(
1949 0 : request: Request<Body>,
1950 0 : _cancel: CancellationToken,
1951 0 : ) -> Result<Response<Body>, ApiError> {
1952 0 : let state = get_state(&request);
1953 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1954 0 : let slot = state.tenant_manager.get(tenant_shard_id);
1955 :
1956 0 : let Some(slot) = slot else {
1957 0 : return Err(ApiError::NotFound(
1958 0 : anyhow::anyhow!("Tenant shard not found").into(),
1959 0 : ));
1960 : };
1961 :
1962 0 : let result: Option<LocationConfig> = match slot {
1963 0 : TenantSlot::Attached(t) => Some(t.get_location_conf()),
1964 0 : TenantSlot::Secondary(s) => Some(s.get_location_conf()),
1965 0 : TenantSlot::InProgress(_) => None,
1966 : };
1967 :
1968 0 : json_response(StatusCode::OK, result)
1969 0 : }
1970 :
1971 : // Do a time travel recovery on the given tenant/tenant shard. Tenant needs to be detached
1972 : // (from all pageservers) as it invalidates consistency assumptions.
1973 0 : async fn tenant_time_travel_remote_storage_handler(
1974 0 : request: Request<Body>,
1975 0 : cancel: CancellationToken,
1976 0 : ) -> Result<Response<Body>, ApiError> {
1977 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
1978 :
1979 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
1980 :
1981 0 : let timestamp_raw = must_get_query_param(&request, "travel_to")?;
1982 0 : let timestamp = humantime::parse_rfc3339(×tamp_raw)
1983 0 : .with_context(|| format!("Invalid time for travel_to: {timestamp_raw:?}"))
1984 0 : .map_err(ApiError::BadRequest)?;
1985 :
1986 0 : let done_if_after_raw = must_get_query_param(&request, "done_if_after")?;
1987 0 : let done_if_after = humantime::parse_rfc3339(&done_if_after_raw)
1988 0 : .with_context(|| format!("Invalid time for done_if_after: {done_if_after_raw:?}"))
1989 0 : .map_err(ApiError::BadRequest)?;
1990 :
1991 : // This is just a sanity check to fend off naive wrong usages of the API:
1992 : // the tenant needs to be detached *everywhere*
1993 0 : let state = get_state(&request);
1994 0 : let we_manage_tenant = state.tenant_manager.manages_tenant_shard(tenant_shard_id);
1995 0 : if we_manage_tenant {
1996 0 : return Err(ApiError::BadRequest(anyhow!(
1997 0 : "Tenant {tenant_shard_id} is already attached at this pageserver"
1998 0 : )));
1999 0 : }
2000 0 :
2001 0 : if timestamp > done_if_after {
2002 0 : return Err(ApiError::BadRequest(anyhow!(
2003 0 : "The done_if_after timestamp comes before the timestamp to recover to"
2004 0 : )));
2005 0 : }
2006 0 :
2007 0 : tracing::info!("Issuing time travel request internally. timestamp={timestamp_raw}, done_if_after={done_if_after_raw}");
2008 :
2009 0 : remote_timeline_client::upload::time_travel_recover_tenant(
2010 0 : &state.remote_storage,
2011 0 : &tenant_shard_id,
2012 0 : timestamp,
2013 0 : done_if_after,
2014 0 : &cancel,
2015 0 : )
2016 0 : .await
2017 0 : .map_err(|e| match e {
2018 0 : TimeTravelError::BadInput(e) => {
2019 0 : warn!("bad input error: {e}");
2020 0 : ApiError::BadRequest(anyhow!("bad input error"))
2021 : }
2022 : TimeTravelError::Unimplemented => {
2023 0 : ApiError::BadRequest(anyhow!("unimplemented for the configured remote storage"))
2024 : }
2025 0 : TimeTravelError::Cancelled => ApiError::InternalServerError(anyhow!("cancelled")),
2026 : TimeTravelError::TooManyVersions => {
2027 0 : ApiError::InternalServerError(anyhow!("too many versions in remote storage"))
2028 : }
2029 0 : TimeTravelError::Other(e) => {
2030 0 : warn!("internal error: {e}");
2031 0 : ApiError::InternalServerError(anyhow!("internal error"))
2032 : }
2033 0 : })?;
2034 :
2035 0 : json_response(StatusCode::OK, ())
2036 0 : }
2037 :
2038 : /// Testing helper to transition a tenant to [`crate::tenant::TenantState::Broken`].
2039 0 : async fn handle_tenant_break(
2040 0 : r: Request<Body>,
2041 0 : _cancel: CancellationToken,
2042 0 : ) -> Result<Response<Body>, ApiError> {
2043 0 : let tenant_shard_id: TenantShardId = parse_request_param(&r, "tenant_shard_id")?;
2044 :
2045 0 : let state = get_state(&r);
2046 0 : state
2047 0 : .tenant_manager
2048 0 : .get_attached_tenant_shard(tenant_shard_id)?
2049 0 : .set_broken("broken from test".to_owned())
2050 0 : .await;
2051 :
2052 0 : json_response(StatusCode::OK, ())
2053 0 : }
2054 :
2055 : // Obtains an lsn lease on the given timeline.
2056 0 : async fn lsn_lease_handler(
2057 0 : mut request: Request<Body>,
2058 0 : _cancel: CancellationToken,
2059 0 : ) -> Result<Response<Body>, ApiError> {
2060 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2061 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2062 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2063 0 : let lsn = json_request::<LsnLeaseRequest>(&mut request).await?.lsn;
2064 :
2065 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
2066 0 :
2067 0 : let state = get_state(&request);
2068 :
2069 0 : let timeline =
2070 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
2071 0 : .await?;
2072 :
2073 0 : let result = async {
2074 0 : timeline
2075 0 : .init_lsn_lease(lsn, timeline.get_lsn_lease_length(), &ctx)
2076 0 : .map_err(|e| {
2077 0 : ApiError::InternalServerError(
2078 0 : e.context(format!("invalid lsn lease request at {lsn}")),
2079 0 : )
2080 0 : })
2081 0 : }
2082 0 : .instrument(info_span!("init_lsn_lease", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2083 0 : .await?;
2084 :
2085 0 : json_response(StatusCode::OK, result)
2086 0 : }
2087 :
2088 : // Run GC immediately on given timeline.
2089 0 : async fn timeline_gc_handler(
2090 0 : mut request: Request<Body>,
2091 0 : cancel: CancellationToken,
2092 0 : ) -> Result<Response<Body>, ApiError> {
2093 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2094 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2095 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2096 :
2097 0 : let gc_req: TimelineGcRequest = json_request(&mut request).await?;
2098 :
2099 0 : let state = get_state(&request);
2100 0 :
2101 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
2102 0 : let gc_result = state
2103 0 : .tenant_manager
2104 0 : .immediate_gc(tenant_shard_id, timeline_id, gc_req, cancel, &ctx)
2105 0 : .await?;
2106 :
2107 0 : json_response(StatusCode::OK, gc_result)
2108 0 : }
2109 :
2110 : // Cancel scheduled compaction tasks
2111 0 : async fn timeline_cancel_compact_handler(
2112 0 : request: Request<Body>,
2113 0 : _cancel: CancellationToken,
2114 0 : ) -> Result<Response<Body>, ApiError> {
2115 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2116 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2117 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2118 0 : let state = get_state(&request);
2119 0 : async {
2120 0 : let tenant = state
2121 0 : .tenant_manager
2122 0 : .get_attached_tenant_shard(tenant_shard_id)?;
2123 0 : tenant.cancel_scheduled_compaction(timeline_id);
2124 0 : json_response(StatusCode::OK, ())
2125 0 : }
2126 0 : .instrument(info_span!("timeline_cancel_compact", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2127 0 : .await
2128 0 : }
2129 :
2130 : // Get compact info of a timeline
2131 0 : async fn timeline_compact_info_handler(
2132 0 : request: Request<Body>,
2133 0 : _cancel: CancellationToken,
2134 0 : ) -> Result<Response<Body>, ApiError> {
2135 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2136 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2137 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2138 0 : let state = get_state(&request);
2139 0 : async {
2140 0 : let tenant = state
2141 0 : .tenant_manager
2142 0 : .get_attached_tenant_shard(tenant_shard_id)?;
2143 0 : let resp = tenant.get_scheduled_compaction_tasks(timeline_id);
2144 0 : json_response(StatusCode::OK, resp)
2145 0 : }
2146 0 : .instrument(info_span!("timeline_compact_info", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2147 0 : .await
2148 0 : }
2149 :
2150 : // Run compaction immediately on given timeline.
2151 0 : async fn timeline_compact_handler(
2152 0 : mut request: Request<Body>,
2153 0 : cancel: CancellationToken,
2154 0 : ) -> Result<Response<Body>, ApiError> {
2155 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2156 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2157 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2158 :
2159 0 : let compact_request = json_request_maybe::<Option<CompactRequest>>(&mut request).await?;
2160 :
2161 0 : let state = get_state(&request);
2162 0 :
2163 0 : let mut flags = EnumSet::empty();
2164 0 : flags |= CompactFlags::NoYield; // run compaction to completion
2165 0 :
2166 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_l0_compaction")? {
2167 0 : flags |= CompactFlags::ForceL0Compaction;
2168 0 : }
2169 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_repartition")? {
2170 0 : flags |= CompactFlags::ForceRepartition;
2171 0 : }
2172 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_image_layer_creation")? {
2173 0 : flags |= CompactFlags::ForceImageLayerCreation;
2174 0 : }
2175 0 : if Some(true) == parse_query_param::<_, bool>(&request, "enhanced_gc_bottom_most_compaction")? {
2176 0 : flags |= CompactFlags::EnhancedGcBottomMostCompaction;
2177 0 : }
2178 0 : if Some(true) == parse_query_param::<_, bool>(&request, "dry_run")? {
2179 0 : flags |= CompactFlags::DryRun;
2180 0 : }
2181 :
2182 0 : let wait_until_uploaded =
2183 0 : parse_query_param::<_, bool>(&request, "wait_until_uploaded")?.unwrap_or(false);
2184 :
2185 0 : let wait_until_scheduled_compaction_done =
2186 0 : parse_query_param::<_, bool>(&request, "wait_until_scheduled_compaction_done")?
2187 0 : .unwrap_or(false);
2188 0 :
2189 0 : let sub_compaction = compact_request
2190 0 : .as_ref()
2191 0 : .map(|r| r.sub_compaction)
2192 0 : .unwrap_or(false);
2193 0 : let sub_compaction_max_job_size_mb = compact_request
2194 0 : .as_ref()
2195 0 : .and_then(|r| r.sub_compaction_max_job_size_mb);
2196 0 :
2197 0 : let options = CompactOptions {
2198 0 : compact_key_range: compact_request
2199 0 : .as_ref()
2200 0 : .and_then(|r| r.compact_key_range.clone()),
2201 0 : compact_lsn_range: compact_request
2202 0 : .as_ref()
2203 0 : .and_then(|r| r.compact_lsn_range.clone()),
2204 0 : flags,
2205 0 : sub_compaction,
2206 0 : sub_compaction_max_job_size_mb,
2207 0 : };
2208 0 :
2209 0 : let scheduled = compact_request
2210 0 : .as_ref()
2211 0 : .map(|r| r.scheduled)
2212 0 : .unwrap_or(false);
2213 :
2214 0 : async {
2215 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
2216 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
2217 0 : if scheduled {
2218 0 : let tenant = state
2219 0 : .tenant_manager
2220 0 : .get_attached_tenant_shard(tenant_shard_id)?;
2221 0 : let rx = tenant.schedule_compaction(timeline_id, options).await.map_err(ApiError::InternalServerError)?;
2222 0 : if wait_until_scheduled_compaction_done {
2223 : // It is possible that this will take a long time, dropping the HTTP request will not cancel the compaction.
2224 0 : rx.await.ok();
2225 0 : }
2226 : } else {
2227 0 : timeline
2228 0 : .compact_with_options(&cancel, options, &ctx)
2229 0 : .await
2230 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
2231 0 : if wait_until_uploaded {
2232 0 : timeline.remote_client.wait_completion().await
2233 : // XXX map to correct ApiError for the cases where it's due to shutdown
2234 0 : .context("wait completion").map_err(ApiError::InternalServerError)?;
2235 0 : }
2236 : }
2237 0 : json_response(StatusCode::OK, ())
2238 0 : }
2239 0 : .instrument(info_span!("manual_compaction", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2240 0 : .await
2241 0 : }
2242 :
2243 : // Run offload immediately on given timeline.
2244 0 : async fn timeline_offload_handler(
2245 0 : 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 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2251 :
2252 0 : let state = get_state(&request);
2253 :
2254 0 : async {
2255 0 : let tenant = state
2256 0 : .tenant_manager
2257 0 : .get_attached_tenant_shard(tenant_shard_id)?;
2258 :
2259 0 : if tenant.get_offloaded_timeline(timeline_id).is_ok() {
2260 0 : return json_response(StatusCode::OK, ());
2261 0 : }
2262 0 : let timeline =
2263 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
2264 0 : .await?;
2265 :
2266 0 : if !tenant.timeline_has_no_attached_children(timeline_id) {
2267 0 : return Err(ApiError::PreconditionFailed(
2268 0 : "timeline has attached children".into(),
2269 0 : ));
2270 0 : }
2271 0 : if let (false, reason) = timeline.can_offload() {
2272 0 : return Err(ApiError::PreconditionFailed(
2273 0 : format!("Timeline::can_offload() check failed: {}", reason) .into(),
2274 0 : ));
2275 0 : }
2276 0 : offload_timeline(&tenant, &timeline)
2277 0 : .await
2278 0 : .map_err(|e| {
2279 0 : match e {
2280 0 : OffloadError::Cancelled => ApiError::ResourceUnavailable("Timeline shutting down".into()),
2281 0 : _ => ApiError::InternalServerError(anyhow!(e))
2282 : }
2283 0 : })?;
2284 :
2285 0 : json_response(StatusCode::OK, ())
2286 0 : }
2287 0 : .instrument(info_span!("manual_timeline_offload", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2288 0 : .await
2289 0 : }
2290 :
2291 : // Run checkpoint immediately on given timeline.
2292 0 : async fn timeline_checkpoint_handler(
2293 0 : request: Request<Body>,
2294 0 : cancel: CancellationToken,
2295 0 : ) -> Result<Response<Body>, ApiError> {
2296 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2297 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2298 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2299 :
2300 0 : let state = get_state(&request);
2301 0 :
2302 0 : let mut flags = EnumSet::empty();
2303 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_l0_compaction")? {
2304 0 : flags |= CompactFlags::ForceL0Compaction;
2305 0 : }
2306 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_repartition")? {
2307 0 : flags |= CompactFlags::ForceRepartition;
2308 0 : }
2309 0 : if Some(true) == parse_query_param::<_, bool>(&request, "force_image_layer_creation")? {
2310 0 : flags |= CompactFlags::ForceImageLayerCreation;
2311 0 : }
2312 :
2313 : // By default, checkpoints come with a compaction, but this may be optionally disabled by tests that just want to flush + upload.
2314 0 : let compact = parse_query_param::<_, bool>(&request, "compact")?.unwrap_or(true);
2315 :
2316 0 : let wait_until_flushed: bool =
2317 0 : parse_query_param(&request, "wait_until_flushed")?.unwrap_or(true);
2318 :
2319 0 : let wait_until_uploaded =
2320 0 : parse_query_param::<_, bool>(&request, "wait_until_uploaded")?.unwrap_or(false);
2321 :
2322 0 : async {
2323 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
2324 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
2325 0 : if wait_until_flushed {
2326 0 : timeline.freeze_and_flush().await
2327 : } else {
2328 0 : timeline.freeze().await.and(Ok(()))
2329 0 : }.map_err(|e| {
2330 0 : match e {
2331 0 : tenant::timeline::FlushLayerError::Cancelled => ApiError::ShuttingDown,
2332 0 : other => ApiError::InternalServerError(other.into()),
2333 :
2334 : }
2335 0 : })?;
2336 0 : if compact {
2337 0 : timeline
2338 0 : .compact(&cancel, flags, &ctx)
2339 0 : .await
2340 0 : .map_err(|e|
2341 0 : match e {
2342 0 : CompactionError::ShuttingDown => ApiError::ShuttingDown,
2343 0 : CompactionError::Offload(e) => ApiError::InternalServerError(anyhow::anyhow!(e)),
2344 0 : CompactionError::Other(e) => ApiError::InternalServerError(e)
2345 0 : }
2346 0 : )?;
2347 0 : }
2348 :
2349 0 : if wait_until_uploaded {
2350 0 : tracing::info!("Waiting for uploads to complete...");
2351 0 : timeline.remote_client.wait_completion().await
2352 : // XXX map to correct ApiError for the cases where it's due to shutdown
2353 0 : .context("wait completion").map_err(ApiError::InternalServerError)?;
2354 0 : tracing::info!("Uploads completed up to {}", timeline.get_remote_consistent_lsn_projected().unwrap_or(Lsn(0)));
2355 0 : }
2356 :
2357 0 : json_response(StatusCode::OK, ())
2358 0 : }
2359 0 : .instrument(info_span!("manual_checkpoint", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2360 0 : .await
2361 0 : }
2362 :
2363 0 : async fn timeline_download_remote_layers_handler_post(
2364 0 : mut request: Request<Body>,
2365 0 : _cancel: CancellationToken,
2366 0 : ) -> Result<Response<Body>, ApiError> {
2367 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2368 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2369 0 : let body: DownloadRemoteLayersTaskSpawnRequest = json_request(&mut request).await?;
2370 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2371 :
2372 0 : let state = get_state(&request);
2373 :
2374 0 : let timeline =
2375 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
2376 0 : .await?;
2377 0 : match timeline.spawn_download_all_remote_layers(body).await {
2378 0 : Ok(st) => json_response(StatusCode::ACCEPTED, st),
2379 0 : Err(st) => json_response(StatusCode::CONFLICT, st),
2380 : }
2381 0 : }
2382 :
2383 0 : async fn timeline_download_remote_layers_handler_get(
2384 0 : request: Request<Body>,
2385 0 : _cancel: CancellationToken,
2386 0 : ) -> Result<Response<Body>, ApiError> {
2387 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2388 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2389 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2390 0 : let state = get_state(&request);
2391 :
2392 0 : let timeline =
2393 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
2394 0 : .await?;
2395 0 : let info = timeline
2396 0 : .get_download_all_remote_layers_task_info()
2397 0 : .context("task never started since last pageserver process start")
2398 0 : .map_err(|e| ApiError::NotFound(e.into()))?;
2399 0 : json_response(StatusCode::OK, info)
2400 0 : }
2401 :
2402 0 : async fn timeline_detach_ancestor_handler(
2403 0 : request: Request<Body>,
2404 0 : _cancel: CancellationToken,
2405 0 : ) -> Result<Response<Body>, ApiError> {
2406 : use crate::tenant::timeline::detach_ancestor;
2407 : use pageserver_api::models::detach_ancestor::AncestorDetached;
2408 :
2409 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2410 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2411 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2412 :
2413 0 : let span = tracing::info_span!("detach_ancestor", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(), %timeline_id);
2414 :
2415 0 : async move {
2416 0 : let mut options = detach_ancestor::Options::default();
2417 :
2418 0 : let rewrite_concurrency =
2419 0 : parse_query_param::<_, std::num::NonZeroUsize>(&request, "rewrite_concurrency")?;
2420 0 : let copy_concurrency =
2421 0 : parse_query_param::<_, std::num::NonZeroUsize>(&request, "copy_concurrency")?;
2422 :
2423 0 : [
2424 0 : (&mut options.rewrite_concurrency, rewrite_concurrency),
2425 0 : (&mut options.copy_concurrency, copy_concurrency),
2426 0 : ]
2427 0 : .into_iter()
2428 0 : .filter_map(|(target, val)| val.map(|val| (target, val)))
2429 0 : .for_each(|(target, val)| *target = val);
2430 0 :
2431 0 : let state = get_state(&request);
2432 :
2433 0 : let tenant = state
2434 0 : .tenant_manager
2435 0 : .get_attached_tenant_shard(tenant_shard_id)?;
2436 :
2437 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
2438 :
2439 0 : let ctx = RequestContext::new(TaskKind::DetachAncestor, DownloadBehavior::Download);
2440 0 : let ctx = &ctx;
2441 :
2442 : // Flush the upload queues of all timelines before detaching ancestor. We do the same thing again
2443 : // during shutdown. This early upload ensures the pageserver does not need to upload too many
2444 : // things and creates downtime during timeline reloads.
2445 0 : for timeline in tenant.list_timelines() {
2446 0 : timeline
2447 0 : .remote_client
2448 0 : .wait_completion()
2449 0 : .await
2450 0 : .map_err(|e| {
2451 0 : ApiError::PreconditionFailed(format!("cannot drain upload queue: {e}").into())
2452 0 : })?;
2453 : }
2454 :
2455 0 : tracing::info!("all timeline upload queues are drained");
2456 :
2457 0 : let timeline = tenant.get_timeline(timeline_id, true)?;
2458 :
2459 0 : let progress = timeline
2460 0 : .prepare_to_detach_from_ancestor(&tenant, options, ctx)
2461 0 : .await?;
2462 :
2463 : // uncomment to allow early as possible Tenant::drop
2464 : // drop(tenant);
2465 :
2466 0 : let resp = match progress {
2467 0 : detach_ancestor::Progress::Prepared(attempt, prepared) => {
2468 : // it would be great to tag the guard on to the tenant activation future
2469 0 : let reparented_timelines = state
2470 0 : .tenant_manager
2471 0 : .complete_detaching_timeline_ancestor(
2472 0 : tenant_shard_id,
2473 0 : timeline_id,
2474 0 : prepared,
2475 0 : attempt,
2476 0 : ctx,
2477 0 : )
2478 0 : .await?;
2479 :
2480 0 : AncestorDetached {
2481 0 : reparented_timelines,
2482 0 : }
2483 : }
2484 0 : detach_ancestor::Progress::Done(resp) => resp,
2485 : };
2486 :
2487 0 : json_response(StatusCode::OK, resp)
2488 0 : }
2489 0 : .instrument(span)
2490 0 : .await
2491 0 : }
2492 :
2493 0 : async fn deletion_queue_flush(
2494 0 : r: Request<Body>,
2495 0 : cancel: CancellationToken,
2496 0 : ) -> Result<Response<Body>, ApiError> {
2497 0 : let state = get_state(&r);
2498 :
2499 0 : let execute = parse_query_param(&r, "execute")?.unwrap_or(false);
2500 0 :
2501 0 : let flush = async {
2502 0 : if execute {
2503 0 : state.deletion_queue_client.flush_execute().await
2504 : } else {
2505 0 : state.deletion_queue_client.flush().await
2506 : }
2507 0 : }
2508 : // DeletionQueueError's only case is shutting down.
2509 0 : .map_err(|_| ApiError::ShuttingDown);
2510 0 :
2511 0 : tokio::select! {
2512 0 : res = flush => {
2513 0 : res.map(|()| json_response(StatusCode::OK, ()))?
2514 : }
2515 0 : _ = cancel.cancelled() => {
2516 0 : Err(ApiError::ShuttingDown)
2517 : }
2518 : }
2519 0 : }
2520 :
2521 : /// Try if `GetPage@Lsn` is successful, useful for manual debugging.
2522 0 : async fn getpage_at_lsn_handler(
2523 0 : request: Request<Body>,
2524 0 : _cancel: CancellationToken,
2525 0 : ) -> Result<Response<Body>, ApiError> {
2526 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2527 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2528 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2529 0 : let state = get_state(&request);
2530 :
2531 : struct Key(pageserver_api::key::Key);
2532 :
2533 : impl std::str::FromStr for Key {
2534 : type Err = anyhow::Error;
2535 :
2536 0 : fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
2537 0 : pageserver_api::key::Key::from_hex(s).map(Key)
2538 0 : }
2539 : }
2540 :
2541 0 : let key: Key = parse_query_param(&request, "key")?
2542 0 : .ok_or_else(|| ApiError::BadRequest(anyhow!("missing 'key' query parameter")))?;
2543 0 : let lsn: Lsn = parse_query_param(&request, "lsn")?
2544 0 : .ok_or_else(|| ApiError::BadRequest(anyhow!("missing 'lsn' query parameter")))?;
2545 :
2546 0 : async {
2547 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
2548 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
2549 :
2550 0 : let page = timeline.get(key.0, lsn, &ctx).await?;
2551 :
2552 0 : Result::<_, ApiError>::Ok(
2553 0 : Response::builder()
2554 0 : .status(StatusCode::OK)
2555 0 : .header(header::CONTENT_TYPE, "application/octet-stream")
2556 0 : .body(hyper::Body::from(page))
2557 0 : .unwrap(),
2558 0 : )
2559 0 : }
2560 0 : .instrument(info_span!("timeline_get", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2561 0 : .await
2562 0 : }
2563 :
2564 0 : async fn timeline_collect_keyspace(
2565 0 : request: Request<Body>,
2566 0 : _cancel: CancellationToken,
2567 0 : ) -> Result<Response<Body>, ApiError> {
2568 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2569 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
2570 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
2571 0 : let state = get_state(&request);
2572 :
2573 0 : let at_lsn: Option<Lsn> = parse_query_param(&request, "at_lsn")?;
2574 :
2575 0 : async {
2576 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
2577 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
2578 0 : let at_lsn = at_lsn.unwrap_or_else(|| timeline.get_last_record_lsn());
2579 0 : let (dense_ks, sparse_ks) = timeline
2580 0 : .collect_keyspace(at_lsn, &ctx)
2581 0 : .await
2582 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
2583 :
2584 : // This API is currently used by pagebench. Pagebench will iterate all keys within the keyspace.
2585 : // Therefore, we split dense/sparse keys in this API.
2586 0 : let res = pageserver_api::models::partitioning::Partitioning { keys: dense_ks, sparse_keys: sparse_ks, at_lsn };
2587 0 :
2588 0 : json_response(StatusCode::OK, res)
2589 0 : }
2590 0 : .instrument(info_span!("timeline_collect_keyspace", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
2591 0 : .await
2592 0 : }
2593 :
2594 0 : async fn active_timeline_of_active_tenant(
2595 0 : tenant_manager: &TenantManager,
2596 0 : tenant_shard_id: TenantShardId,
2597 0 : timeline_id: TimelineId,
2598 0 : ) -> Result<Arc<Timeline>, ApiError> {
2599 0 : let tenant = tenant_manager.get_attached_tenant_shard(tenant_shard_id)?;
2600 :
2601 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
2602 :
2603 0 : Ok(tenant.get_timeline(timeline_id, true)?)
2604 0 : }
2605 :
2606 0 : async fn always_panic_handler(
2607 0 : req: Request<Body>,
2608 0 : _cancel: CancellationToken,
2609 0 : ) -> Result<Response<Body>, ApiError> {
2610 0 : // Deliberately cause a panic to exercise the panic hook registered via std::panic::set_hook().
2611 0 : // For pageserver, the relevant panic hook is `tracing_panic_hook` , and the `sentry` crate's wrapper around it.
2612 0 : // Use catch_unwind to ensure that tokio nor hyper are distracted by our panic.
2613 0 : let query = req.uri().query();
2614 0 : let _ = std::panic::catch_unwind(|| {
2615 0 : panic!("unconditional panic for testing panic hook integration; request query: {query:?}")
2616 0 : });
2617 0 : json_response(StatusCode::NO_CONTENT, ())
2618 0 : }
2619 :
2620 0 : async fn disk_usage_eviction_run(
2621 0 : mut r: Request<Body>,
2622 0 : cancel: CancellationToken,
2623 0 : ) -> Result<Response<Body>, ApiError> {
2624 0 : check_permission(&r, None)?;
2625 :
2626 0 : #[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
2627 : struct Config {
2628 : /// How many bytes to evict before reporting that pressure is relieved.
2629 : evict_bytes: u64,
2630 :
2631 : #[serde(default)]
2632 : eviction_order: pageserver_api::config::EvictionOrder,
2633 : }
2634 :
2635 : #[derive(Debug, Clone, Copy, serde::Serialize)]
2636 : struct Usage {
2637 : // remains unchanged after instantiation of the struct
2638 : evict_bytes: u64,
2639 : // updated by `add_available_bytes`
2640 : freed_bytes: u64,
2641 : }
2642 :
2643 : impl crate::disk_usage_eviction_task::Usage for Usage {
2644 0 : fn has_pressure(&self) -> bool {
2645 0 : self.evict_bytes > self.freed_bytes
2646 0 : }
2647 :
2648 0 : fn add_available_bytes(&mut self, bytes: u64) {
2649 0 : self.freed_bytes += bytes;
2650 0 : }
2651 : }
2652 :
2653 0 : let config = json_request::<Config>(&mut r).await?;
2654 :
2655 0 : let usage = Usage {
2656 0 : evict_bytes: config.evict_bytes,
2657 0 : freed_bytes: 0,
2658 0 : };
2659 0 :
2660 0 : let state = get_state(&r);
2661 0 : let eviction_state = state.disk_usage_eviction_state.clone();
2662 :
2663 0 : let res = crate::disk_usage_eviction_task::disk_usage_eviction_task_iteration_impl(
2664 0 : &eviction_state,
2665 0 : &state.remote_storage,
2666 0 : usage,
2667 0 : &state.tenant_manager,
2668 0 : config.eviction_order.into(),
2669 0 : &cancel,
2670 0 : )
2671 0 : .await;
2672 :
2673 0 : info!(?res, "disk_usage_eviction_task_iteration_impl finished");
2674 :
2675 0 : let res = res.map_err(ApiError::InternalServerError)?;
2676 :
2677 0 : json_response(StatusCode::OK, res)
2678 0 : }
2679 :
2680 0 : async fn secondary_upload_handler(
2681 0 : request: Request<Body>,
2682 0 : _cancel: CancellationToken,
2683 0 : ) -> Result<Response<Body>, ApiError> {
2684 0 : let state = get_state(&request);
2685 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2686 0 : state
2687 0 : .secondary_controller
2688 0 : .upload_tenant(tenant_shard_id)
2689 0 : .await?;
2690 :
2691 0 : json_response(StatusCode::OK, ())
2692 0 : }
2693 :
2694 0 : async fn tenant_scan_remote_handler(
2695 0 : request: Request<Body>,
2696 0 : cancel: CancellationToken,
2697 0 : ) -> Result<Response<Body>, ApiError> {
2698 0 : let state = get_state(&request);
2699 0 : let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
2700 :
2701 0 : let mut response = TenantScanRemoteStorageResponse::default();
2702 :
2703 0 : let (shards, _other_keys) =
2704 0 : list_remote_tenant_shards(&state.remote_storage, tenant_id, cancel.clone())
2705 0 : .await
2706 0 : .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
2707 :
2708 0 : for tenant_shard_id in shards {
2709 0 : let (timeline_ids, _other_keys) =
2710 0 : list_remote_timelines(&state.remote_storage, tenant_shard_id, cancel.clone())
2711 0 : .await
2712 0 : .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
2713 :
2714 0 : let mut generation = Generation::none();
2715 0 : for timeline_id in timeline_ids {
2716 0 : match download_index_part(
2717 0 : &state.remote_storage,
2718 0 : &tenant_shard_id,
2719 0 : &timeline_id,
2720 0 : Generation::MAX,
2721 0 : &cancel,
2722 0 : )
2723 0 : .instrument(info_span!("download_index_part",
2724 : tenant_id=%tenant_shard_id.tenant_id,
2725 0 : shard_id=%tenant_shard_id.shard_slug(),
2726 : %timeline_id))
2727 0 : .await
2728 : {
2729 0 : Ok((index_part, index_generation, _index_mtime)) => {
2730 0 : tracing::info!("Found timeline {tenant_shard_id}/{timeline_id} metadata (gen {index_generation:?}, {} layers, {} consistent LSN)",
2731 0 : index_part.layer_metadata.len(), index_part.metadata.disk_consistent_lsn());
2732 0 : generation = std::cmp::max(generation, index_generation);
2733 : }
2734 : Err(DownloadError::NotFound) => {
2735 : // This is normal for tenants that were created with multiple shards: they have an unsharded path
2736 : // containing the timeline's initdb tarball but no index. Otherwise it is a bit strange.
2737 0 : tracing::info!("Timeline path {tenant_shard_id}/{timeline_id} exists in remote storage but has no index, skipping");
2738 0 : continue;
2739 : }
2740 0 : Err(e) => {
2741 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!(e)));
2742 : }
2743 : };
2744 : }
2745 :
2746 0 : response.shards.push(TenantScanRemoteStorageShard {
2747 0 : tenant_shard_id,
2748 0 : generation: generation.into(),
2749 0 : });
2750 : }
2751 :
2752 0 : if response.shards.is_empty() {
2753 0 : return Err(ApiError::NotFound(
2754 0 : anyhow::anyhow!("No shards found for tenant ID {tenant_id}").into(),
2755 0 : ));
2756 0 : }
2757 0 :
2758 0 : json_response(StatusCode::OK, response)
2759 0 : }
2760 :
2761 0 : async fn secondary_download_handler(
2762 0 : request: Request<Body>,
2763 0 : _cancel: CancellationToken,
2764 0 : ) -> Result<Response<Body>, ApiError> {
2765 0 : let state = get_state(&request);
2766 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2767 0 : let wait = parse_query_param(&request, "wait_ms")?.map(Duration::from_millis);
2768 :
2769 : // We don't need this to issue the download request, but:
2770 : // - it enables us to cleanly return 404 if we get a request for an absent shard
2771 : // - we will use this to provide status feedback in the response
2772 0 : let Some(secondary_tenant) = state
2773 0 : .tenant_manager
2774 0 : .get_secondary_tenant_shard(tenant_shard_id)
2775 : else {
2776 0 : return Err(ApiError::NotFound(
2777 0 : anyhow::anyhow!("Shard {} not found", tenant_shard_id).into(),
2778 0 : ));
2779 : };
2780 :
2781 0 : let timeout = wait.unwrap_or(Duration::MAX);
2782 :
2783 0 : let result = tokio::time::timeout(
2784 0 : timeout,
2785 0 : state.secondary_controller.download_tenant(tenant_shard_id),
2786 0 : )
2787 0 : .await;
2788 :
2789 0 : let progress = secondary_tenant.progress.lock().unwrap().clone();
2790 :
2791 0 : let status = match result {
2792 : Ok(Ok(())) => {
2793 0 : if progress.layers_downloaded >= progress.layers_total {
2794 : // Download job ran to completion
2795 0 : StatusCode::OK
2796 : } else {
2797 : // Download dropped out without errors because it ran out of time budget
2798 0 : StatusCode::ACCEPTED
2799 : }
2800 : }
2801 : // Edge case: downloads aren't usually fallible: things like a missing heatmap are considered
2802 : // okay. We could get an error here in the unlikely edge case that the tenant
2803 : // was detached between our check above and executing the download job.
2804 0 : Ok(Err(e)) => return Err(e.into()),
2805 : // A timeout is not an error: we have started the download, we're just not done
2806 : // yet. The caller will get a response body indicating status.
2807 0 : Err(_) => StatusCode::ACCEPTED,
2808 : };
2809 :
2810 0 : json_response(status, progress)
2811 0 : }
2812 :
2813 0 : async fn wait_lsn_handler(
2814 0 : mut request: Request<Body>,
2815 0 : cancel: CancellationToken,
2816 0 : ) -> Result<Response<Body>, ApiError> {
2817 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2818 0 : let wait_lsn_request: TenantWaitLsnRequest = json_request(&mut request).await?;
2819 :
2820 0 : let state = get_state(&request);
2821 0 : let tenant = state
2822 0 : .tenant_manager
2823 0 : .get_attached_tenant_shard(tenant_shard_id)?;
2824 :
2825 0 : let mut wait_futures = Vec::default();
2826 0 : for timeline in tenant.list_timelines() {
2827 0 : let Some(lsn) = wait_lsn_request.timelines.get(&timeline.timeline_id) else {
2828 0 : continue;
2829 : };
2830 :
2831 0 : let fut = {
2832 0 : let timeline = timeline.clone();
2833 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
2834 0 : async move {
2835 0 : timeline
2836 0 : .wait_lsn(
2837 0 : *lsn,
2838 0 : WaitLsnWaiter::HttpEndpoint,
2839 0 : WaitLsnTimeout::Custom(wait_lsn_request.timeout),
2840 0 : &ctx,
2841 0 : )
2842 0 : .await
2843 0 : }
2844 0 : };
2845 0 : wait_futures.push(fut);
2846 0 : }
2847 :
2848 0 : if wait_futures.is_empty() {
2849 0 : return json_response(StatusCode::NOT_FOUND, ());
2850 0 : }
2851 :
2852 0 : let all_done = tokio::select! {
2853 0 : results = join_all(wait_futures) => {
2854 0 : results.iter().all(|res| res.is_ok())
2855 : },
2856 0 : _ = cancel.cancelled() => {
2857 0 : return Err(ApiError::Cancelled);
2858 : }
2859 : };
2860 :
2861 0 : let status = if all_done {
2862 0 : StatusCode::OK
2863 : } else {
2864 0 : StatusCode::ACCEPTED
2865 : };
2866 :
2867 0 : json_response(status, ())
2868 0 : }
2869 :
2870 0 : async fn secondary_status_handler(
2871 0 : request: Request<Body>,
2872 0 : _cancel: CancellationToken,
2873 0 : ) -> Result<Response<Body>, ApiError> {
2874 0 : let state = get_state(&request);
2875 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
2876 :
2877 0 : let Some(secondary_tenant) = state
2878 0 : .tenant_manager
2879 0 : .get_secondary_tenant_shard(tenant_shard_id)
2880 : else {
2881 0 : return Err(ApiError::NotFound(
2882 0 : anyhow::anyhow!("Shard {} not found", tenant_shard_id).into(),
2883 0 : ));
2884 : };
2885 :
2886 0 : let progress = secondary_tenant.progress.lock().unwrap().clone();
2887 0 :
2888 0 : json_response(StatusCode::OK, progress)
2889 0 : }
2890 :
2891 0 : async fn handler_404(_: Request<Body>) -> Result<Response<Body>, ApiError> {
2892 0 : json_response(
2893 0 : StatusCode::NOT_FOUND,
2894 0 : HttpErrorBody::from_msg("page not found".to_owned()),
2895 0 : )
2896 0 : }
2897 :
2898 0 : async fn post_tracing_event_handler(
2899 0 : mut r: Request<Body>,
2900 0 : _cancel: CancellationToken,
2901 0 : ) -> Result<Response<Body>, ApiError> {
2902 0 : #[derive(Debug, serde::Deserialize)]
2903 : #[serde(rename_all = "lowercase")]
2904 : enum Level {
2905 : Error,
2906 : Warn,
2907 : Info,
2908 : Debug,
2909 : Trace,
2910 : }
2911 0 : #[derive(Debug, serde::Deserialize)]
2912 : struct Request {
2913 : level: Level,
2914 : message: String,
2915 : }
2916 0 : let body: Request = json_request(&mut r)
2917 0 : .await
2918 0 : .map_err(|_| ApiError::BadRequest(anyhow::anyhow!("invalid JSON body")))?;
2919 :
2920 0 : match body.level {
2921 0 : Level::Error => tracing::error!(?body.message),
2922 0 : Level::Warn => tracing::warn!(?body.message),
2923 0 : Level::Info => tracing::info!(?body.message),
2924 0 : Level::Debug => tracing::debug!(?body.message),
2925 0 : Level::Trace => tracing::trace!(?body.message),
2926 : }
2927 :
2928 0 : json_response(StatusCode::OK, ())
2929 0 : }
2930 :
2931 0 : async fn put_io_engine_handler(
2932 0 : mut r: Request<Body>,
2933 0 : _cancel: CancellationToken,
2934 0 : ) -> Result<Response<Body>, ApiError> {
2935 0 : check_permission(&r, None)?;
2936 0 : let kind: crate::virtual_file::IoEngineKind = json_request(&mut r).await?;
2937 0 : crate::virtual_file::io_engine::set(kind);
2938 0 : json_response(StatusCode::OK, ())
2939 0 : }
2940 :
2941 0 : async fn put_io_mode_handler(
2942 0 : mut r: Request<Body>,
2943 0 : _cancel: CancellationToken,
2944 0 : ) -> Result<Response<Body>, ApiError> {
2945 0 : check_permission(&r, None)?;
2946 0 : let mode: IoMode = json_request(&mut r).await?;
2947 0 : crate::virtual_file::set_io_mode(mode);
2948 0 : json_response(StatusCode::OK, ())
2949 0 : }
2950 :
2951 : /// Polled by control plane.
2952 : ///
2953 : /// See [`crate::utilization`].
2954 0 : async fn get_utilization(
2955 0 : r: Request<Body>,
2956 0 : _cancel: CancellationToken,
2957 0 : ) -> Result<Response<Body>, ApiError> {
2958 0 : fail::fail_point!("get-utilization-http-handler", |_| {
2959 0 : Err(ApiError::ResourceUnavailable("failpoint".into()))
2960 0 : });
2961 :
2962 : // this probably could be completely public, but lets make that change later.
2963 0 : check_permission(&r, None)?;
2964 :
2965 0 : let state = get_state(&r);
2966 0 : let mut g = state.latest_utilization.lock().await;
2967 :
2968 0 : let regenerate_every = Duration::from_secs(1);
2969 0 : let still_valid = g
2970 0 : .as_ref()
2971 0 : .is_some_and(|(captured_at, _)| captured_at.elapsed() < regenerate_every);
2972 0 :
2973 0 : // avoid needless statvfs calls even though those should be non-blocking fast.
2974 0 : // regenerate at most 1Hz to allow polling at any rate.
2975 0 : if !still_valid {
2976 0 : let path = state.conf.tenants_path();
2977 0 : let doc =
2978 0 : crate::utilization::regenerate(state.conf, path.as_std_path(), &state.tenant_manager)
2979 0 : .map_err(ApiError::InternalServerError)?;
2980 :
2981 0 : let mut buf = Vec::new();
2982 0 : serde_json::to_writer(&mut buf, &doc)
2983 0 : .context("serialize")
2984 0 : .map_err(ApiError::InternalServerError)?;
2985 :
2986 0 : let body = bytes::Bytes::from(buf);
2987 0 :
2988 0 : *g = Some((std::time::Instant::now(), body));
2989 0 : }
2990 :
2991 : // hyper 0.14 doesn't yet have Response::clone so this is a bit of extra legwork
2992 0 : let cached = g.as_ref().expect("just set").1.clone();
2993 0 :
2994 0 : Response::builder()
2995 0 : .header(hyper::http::header::CONTENT_TYPE, "application/json")
2996 0 : // thought of using http date header, but that is second precision which does not give any
2997 0 : // debugging aid
2998 0 : .status(StatusCode::OK)
2999 0 : .body(hyper::Body::from(cached))
3000 0 : .context("build response")
3001 0 : .map_err(ApiError::InternalServerError)
3002 0 : }
3003 :
3004 0 : async fn list_aux_files(
3005 0 : mut request: Request<Body>,
3006 0 : _cancel: CancellationToken,
3007 0 : ) -> Result<Response<Body>, ApiError> {
3008 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
3009 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
3010 0 : let body: ListAuxFilesRequest = json_request(&mut request).await?;
3011 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
3012 :
3013 0 : let state = get_state(&request);
3014 :
3015 0 : let timeline =
3016 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
3017 0 : .await?;
3018 :
3019 0 : let io_concurrency = IoConcurrency::spawn_from_conf(
3020 0 : state.conf,
3021 0 : timeline.gate.enter().map_err(|_| ApiError::Cancelled)?,
3022 : );
3023 :
3024 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
3025 0 : let files = timeline
3026 0 : .list_aux_files(body.lsn, &ctx, io_concurrency)
3027 0 : .await?;
3028 0 : json_response(StatusCode::OK, files)
3029 0 : }
3030 :
3031 0 : async fn perf_info(
3032 0 : request: Request<Body>,
3033 0 : _cancel: CancellationToken,
3034 0 : ) -> Result<Response<Body>, ApiError> {
3035 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
3036 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
3037 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
3038 :
3039 0 : let state = get_state(&request);
3040 :
3041 0 : let timeline =
3042 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
3043 0 : .await?;
3044 :
3045 0 : let result = timeline.perf_info().await;
3046 :
3047 0 : json_response(StatusCode::OK, result)
3048 0 : }
3049 :
3050 0 : async fn ingest_aux_files(
3051 0 : mut request: Request<Body>,
3052 0 : _cancel: CancellationToken,
3053 0 : ) -> Result<Response<Body>, ApiError> {
3054 0 : let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
3055 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
3056 0 : let body: IngestAuxFilesRequest = json_request(&mut request).await?;
3057 0 : check_permission(&request, Some(tenant_shard_id.tenant_id))?;
3058 :
3059 0 : let state = get_state(&request);
3060 :
3061 0 : let timeline =
3062 0 : active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
3063 0 : .await?;
3064 :
3065 0 : let mut modification = timeline.begin_modification(
3066 0 : Lsn(timeline.get_last_record_lsn().0 + 8), /* advance LSN by 8 */
3067 0 : );
3068 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
3069 0 : for (fname, content) in body.aux_files {
3070 0 : modification
3071 0 : .put_file(&fname, content.as_bytes(), &ctx)
3072 0 : .await
3073 0 : .map_err(ApiError::InternalServerError)?;
3074 : }
3075 0 : modification
3076 0 : .commit(&ctx)
3077 0 : .await
3078 0 : .map_err(ApiError::InternalServerError)?;
3079 :
3080 0 : json_response(StatusCode::OK, ())
3081 0 : }
3082 :
3083 : /// Report on the largest tenants on this pageserver, for the storage controller to identify
3084 : /// candidates for splitting
3085 0 : async fn post_top_tenants(
3086 0 : mut r: Request<Body>,
3087 0 : _cancel: CancellationToken,
3088 0 : ) -> Result<Response<Body>, ApiError> {
3089 0 : check_permission(&r, None)?;
3090 0 : let request: TopTenantShardsRequest = json_request(&mut r).await?;
3091 0 : let state = get_state(&r);
3092 :
3093 0 : fn get_size_metric(sizes: &TopTenantShardItem, order_by: &TenantSorting) -> u64 {
3094 0 : match order_by {
3095 0 : TenantSorting::ResidentSize => sizes.resident_size,
3096 0 : TenantSorting::MaxLogicalSize => sizes.max_logical_size,
3097 : }
3098 0 : }
3099 :
3100 : #[derive(Eq, PartialEq)]
3101 : struct HeapItem {
3102 : metric: u64,
3103 : sizes: TopTenantShardItem,
3104 : }
3105 :
3106 : impl PartialOrd for HeapItem {
3107 0 : fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
3108 0 : Some(self.cmp(other))
3109 0 : }
3110 : }
3111 :
3112 : /// Heap items have reverse ordering on their metric: this enables using BinaryHeap, which
3113 : /// supports popping the greatest item but not the smallest.
3114 : impl Ord for HeapItem {
3115 0 : fn cmp(&self, other: &Self) -> std::cmp::Ordering {
3116 0 : Reverse(self.metric).cmp(&Reverse(other.metric))
3117 0 : }
3118 : }
3119 :
3120 0 : let mut top_n: BinaryHeap<HeapItem> = BinaryHeap::with_capacity(request.limit);
3121 :
3122 : // FIXME: this is a lot of clones to take this tenant list
3123 0 : for (tenant_shard_id, tenant_slot) in state.tenant_manager.list() {
3124 0 : if let Some(shards_lt) = request.where_shards_lt {
3125 : // Ignore tenants which already have >= this many shards
3126 0 : if tenant_shard_id.shard_count >= shards_lt {
3127 0 : continue;
3128 0 : }
3129 0 : }
3130 :
3131 0 : let sizes = match tenant_slot {
3132 0 : TenantSlot::Attached(tenant) => tenant.get_sizes(),
3133 : TenantSlot::Secondary(_) | TenantSlot::InProgress(_) => {
3134 0 : continue;
3135 : }
3136 : };
3137 0 : let metric = get_size_metric(&sizes, &request.order_by);
3138 :
3139 0 : if let Some(gt) = request.where_gt {
3140 : // Ignore tenants whose metric is <= the lower size threshold, to do less sorting work
3141 0 : if metric <= gt {
3142 0 : continue;
3143 0 : }
3144 0 : };
3145 :
3146 0 : match top_n.peek() {
3147 0 : None => {
3148 0 : // Top N list is empty: candidate becomes first member
3149 0 : top_n.push(HeapItem { metric, sizes });
3150 0 : }
3151 0 : Some(i) if i.metric > metric && top_n.len() < request.limit => {
3152 0 : // Lowest item in list is greater than our candidate, but we aren't at limit yet: push to end
3153 0 : top_n.push(HeapItem { metric, sizes });
3154 0 : }
3155 0 : Some(i) if i.metric > metric => {
3156 0 : // List is at limit and lowest value is greater than our candidate, drop it.
3157 0 : }
3158 0 : Some(_) => top_n.push(HeapItem { metric, sizes }),
3159 : }
3160 :
3161 0 : while top_n.len() > request.limit {
3162 0 : top_n.pop();
3163 0 : }
3164 : }
3165 :
3166 0 : json_response(
3167 0 : StatusCode::OK,
3168 0 : TopTenantShardsResponse {
3169 0 : shards: top_n.into_iter().map(|i| i.sizes).collect(),
3170 0 : },
3171 0 : )
3172 0 : }
3173 :
3174 0 : async fn put_tenant_timeline_import_basebackup(
3175 0 : request: Request<Body>,
3176 0 : _cancel: CancellationToken,
3177 0 : ) -> Result<Response<Body>, ApiError> {
3178 0 : let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
3179 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
3180 0 : let base_lsn: Lsn = must_parse_query_param(&request, "base_lsn")?;
3181 0 : let end_lsn: Lsn = must_parse_query_param(&request, "end_lsn")?;
3182 0 : let pg_version: u32 = must_parse_query_param(&request, "pg_version")?;
3183 :
3184 0 : check_permission(&request, Some(tenant_id))?;
3185 :
3186 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
3187 0 :
3188 0 : let tenant_shard_id = TenantShardId::unsharded(tenant_id);
3189 :
3190 0 : let span = info_span!("import_basebackup",
3191 0 : tenant_id=%tenant_id, timeline_id=%timeline_id, shard_id=%tenant_shard_id.shard_slug(),
3192 : base_lsn=%base_lsn, end_lsn=%end_lsn, pg_version=%pg_version);
3193 0 : async move {
3194 0 : let state = get_state(&request);
3195 0 : let tenant = state
3196 0 : .tenant_manager
3197 0 : .get_attached_tenant_shard(tenant_shard_id)?;
3198 :
3199 0 : let broker_client = state.broker_client.clone();
3200 0 :
3201 0 : let mut body = StreamReader::new(request.into_body().map(|res| {
3202 0 : res.map_err(|error| {
3203 0 : std::io::Error::new(std::io::ErrorKind::Other, anyhow::anyhow!(error))
3204 0 : })
3205 0 : }));
3206 0 :
3207 0 : tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
3208 :
3209 0 : let timeline = tenant
3210 0 : .create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)
3211 0 : .map_err(ApiError::InternalServerError)
3212 0 : .await?;
3213 :
3214 : // TODO mark timeline as not ready until it reaches end_lsn.
3215 : // We might have some wal to import as well, and we should prevent compute
3216 : // from connecting before that and writing conflicting wal.
3217 : //
3218 : // This is not relevant for pageserver->pageserver migrations, since there's
3219 : // no wal to import. But should be fixed if we want to import from postgres.
3220 :
3221 : // TODO leave clean state on error. For now you can use detach to clean
3222 : // up broken state from a failed import.
3223 :
3224 : // Import basebackup provided via CopyData
3225 0 : info!("importing basebackup");
3226 :
3227 0 : timeline
3228 0 : .import_basebackup_from_tar(tenant.clone(), &mut body, base_lsn, broker_client, &ctx)
3229 0 : .await
3230 0 : .map_err(ApiError::InternalServerError)?;
3231 :
3232 : // Read the end of the tar archive.
3233 0 : read_tar_eof(body)
3234 0 : .await
3235 0 : .map_err(ApiError::InternalServerError)?;
3236 :
3237 : // TODO check checksum
3238 : // Meanwhile you can verify client-side by taking fullbackup
3239 : // and checking that it matches in size with what was imported.
3240 : // It wouldn't work if base came from vanilla postgres though,
3241 : // since we discard some log files.
3242 :
3243 0 : info!("done");
3244 0 : json_response(StatusCode::OK, ())
3245 0 : }
3246 0 : .instrument(span)
3247 0 : .await
3248 0 : }
3249 :
3250 0 : async fn put_tenant_timeline_import_wal(
3251 0 : request: Request<Body>,
3252 0 : _cancel: CancellationToken,
3253 0 : ) -> Result<Response<Body>, ApiError> {
3254 0 : let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
3255 0 : let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
3256 0 : let start_lsn: Lsn = must_parse_query_param(&request, "start_lsn")?;
3257 0 : let end_lsn: Lsn = must_parse_query_param(&request, "end_lsn")?;
3258 :
3259 0 : check_permission(&request, Some(tenant_id))?;
3260 :
3261 0 : let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
3262 :
3263 0 : let span = info_span!("import_wal", tenant_id=%tenant_id, timeline_id=%timeline_id, start_lsn=%start_lsn, end_lsn=%end_lsn);
3264 0 : async move {
3265 0 : let state = get_state(&request);
3266 :
3267 0 : let timeline = active_timeline_of_active_tenant(&state.tenant_manager, TenantShardId::unsharded(tenant_id), timeline_id).await?;
3268 :
3269 0 : let mut body = StreamReader::new(request.into_body().map(|res| {
3270 0 : res.map_err(|error| {
3271 0 : std::io::Error::new(std::io::ErrorKind::Other, anyhow::anyhow!(error))
3272 0 : })
3273 0 : }));
3274 0 :
3275 0 : let last_record_lsn = timeline.get_last_record_lsn();
3276 0 : if last_record_lsn != start_lsn {
3277 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!("Cannot import WAL from Lsn {start_lsn} because timeline does not start from the same lsn: {last_record_lsn}")));
3278 0 : }
3279 0 :
3280 0 : // TODO leave clean state on error. For now you can use detach to clean
3281 0 : // up broken state from a failed import.
3282 0 :
3283 0 : // Import wal provided via CopyData
3284 0 : info!("importing wal");
3285 0 : crate::import_datadir::import_wal_from_tar(&timeline, &mut body, start_lsn, end_lsn, &ctx).await.map_err(ApiError::InternalServerError)?;
3286 0 : info!("wal import complete");
3287 :
3288 : // Read the end of the tar archive.
3289 0 : read_tar_eof(body).await.map_err(ApiError::InternalServerError)?;
3290 :
3291 : // TODO Does it make sense to overshoot?
3292 0 : if timeline.get_last_record_lsn() < end_lsn {
3293 0 : return Err(ApiError::InternalServerError(anyhow::anyhow!("Cannot import WAL from Lsn {start_lsn} because timeline does not start from the same lsn: {last_record_lsn}")));
3294 0 : }
3295 0 :
3296 0 : // Flush data to disk, then upload to s3. No need for a forced checkpoint.
3297 0 : // We only want to persist the data, and it doesn't matter if it's in the
3298 0 : // shape of deltas or images.
3299 0 : info!("flushing layers");
3300 0 : timeline.freeze_and_flush().await.map_err(|e| match e {
3301 0 : tenant::timeline::FlushLayerError::Cancelled => ApiError::ShuttingDown,
3302 0 : other => ApiError::InternalServerError(anyhow::anyhow!(other)),
3303 0 : })?;
3304 :
3305 0 : info!("done");
3306 :
3307 0 : json_response(StatusCode::OK, ())
3308 0 : }.instrument(span).await
3309 0 : }
3310 :
3311 : /// Read the end of a tar archive.
3312 : ///
3313 : /// A tar archive normally ends with two consecutive blocks of zeros, 512 bytes each.
3314 : /// `tokio_tar` already read the first such block. Read the second all-zeros block,
3315 : /// and check that there is no more data after the EOF marker.
3316 : ///
3317 : /// 'tar' command can also write extra blocks of zeros, up to a record
3318 : /// size, controlled by the --record-size argument. Ignore them too.
3319 0 : async fn read_tar_eof(mut reader: (impl tokio::io::AsyncRead + Unpin)) -> anyhow::Result<()> {
3320 : use tokio::io::AsyncReadExt;
3321 0 : let mut buf = [0u8; 512];
3322 0 :
3323 0 : // Read the all-zeros block, and verify it
3324 0 : let mut total_bytes = 0;
3325 0 : while total_bytes < 512 {
3326 0 : let nbytes = reader.read(&mut buf[total_bytes..]).await?;
3327 0 : total_bytes += nbytes;
3328 0 : if nbytes == 0 {
3329 0 : break;
3330 0 : }
3331 : }
3332 0 : if total_bytes < 512 {
3333 0 : anyhow::bail!("incomplete or invalid tar EOF marker");
3334 0 : }
3335 0 : if !buf.iter().all(|&x| x == 0) {
3336 0 : anyhow::bail!("invalid tar EOF marker");
3337 0 : }
3338 0 :
3339 0 : // Drain any extra zero-blocks after the EOF marker
3340 0 : let mut trailing_bytes = 0;
3341 0 : let mut seen_nonzero_bytes = false;
3342 : loop {
3343 0 : let nbytes = reader.read(&mut buf).await?;
3344 0 : trailing_bytes += nbytes;
3345 0 : if !buf.iter().all(|&x| x == 0) {
3346 0 : seen_nonzero_bytes = true;
3347 0 : }
3348 0 : if nbytes == 0 {
3349 0 : break;
3350 0 : }
3351 : }
3352 0 : if seen_nonzero_bytes {
3353 0 : anyhow::bail!("unexpected non-zero bytes after the tar archive");
3354 0 : }
3355 0 : if trailing_bytes % 512 != 0 {
3356 0 : anyhow::bail!("unexpected number of zeros ({trailing_bytes}), not divisible by tar block size (512 bytes), after the tar archive");
3357 0 : }
3358 0 : Ok(())
3359 0 : }
3360 :
3361 : /// Common functionality of all the HTTP API handlers.
3362 : ///
3363 : /// - Adds a tracing span to each request (by `request_span`)
3364 : /// - Logs the request depending on the request method (by `request_span`)
3365 : /// - Logs the response if it was not successful (by `request_span`
3366 : /// - Shields the handler function from async cancellations. Hyper can drop the handler
3367 : /// Future if the connection to the client is lost, but most of the pageserver code is
3368 : /// not async cancellation safe. This converts the dropped future into a graceful cancellation
3369 : /// request with a CancellationToken.
3370 0 : async fn api_handler<R, H>(request: Request<Body>, handler: H) -> Result<Response<Body>, ApiError>
3371 0 : where
3372 0 : R: std::future::Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
3373 0 : H: FnOnce(Request<Body>, CancellationToken) -> R + Send + Sync + 'static,
3374 0 : {
3375 0 : if request.uri() != &"/v1/failpoints".parse::<Uri>().unwrap() {
3376 0 : fail::fail_point!("api-503", |_| Err(ApiError::ResourceUnavailable(
3377 0 : "failpoint".into()
3378 0 : )));
3379 :
3380 0 : fail::fail_point!("api-500", |_| Err(ApiError::InternalServerError(
3381 0 : anyhow::anyhow!("failpoint")
3382 0 : )));
3383 0 : }
3384 :
3385 : // Spawn a new task to handle the request, to protect the handler from unexpected
3386 : // async cancellations. Most pageserver functions are not async cancellation safe.
3387 : // We arm a drop-guard, so that if Hyper drops the Future, we signal the task
3388 : // with the cancellation token.
3389 0 : let token = CancellationToken::new();
3390 0 : let cancel_guard = token.clone().drop_guard();
3391 0 : let result = request_span(request, move |r| async {
3392 0 : let handle = tokio::spawn(
3393 0 : async {
3394 0 : let token_cloned = token.clone();
3395 0 : let result = handler(r, token).await;
3396 0 : if token_cloned.is_cancelled() {
3397 : // dropguard has executed: we will never turn this result into response.
3398 : //
3399 : // at least temporarily do {:?} logging; these failures are rare enough but
3400 : // could hide difficult errors.
3401 0 : match &result {
3402 0 : Ok(response) => {
3403 0 : let status = response.status();
3404 0 : info!(%status, "Cancelled request finished successfully")
3405 : }
3406 0 : Err(e) => match e {
3407 : ApiError::ShuttingDown | ApiError::ResourceUnavailable(_) => {
3408 : // Don't log this at error severity: they are normal during lifecycle of tenants/process
3409 0 : info!("Cancelled request aborted for shutdown")
3410 : }
3411 : _ => {
3412 : // Log these in a highly visible way, because we have no client to send the response to, but
3413 : // would like to know that something went wrong.
3414 0 : error!("Cancelled request finished with an error: {e:?}")
3415 : }
3416 : },
3417 : }
3418 0 : }
3419 : // only logging for cancelled panicked request handlers is the tracing_panic_hook,
3420 : // which should suffice.
3421 : //
3422 : // there is still a chance to lose the result due to race between
3423 : // returning from here and the actual connection closing happening
3424 : // before outer task gets to execute. leaving that up for #5815.
3425 0 : result
3426 0 : }
3427 0 : .in_current_span(),
3428 0 : );
3429 0 :
3430 0 : match handle.await {
3431 : // TODO: never actually return Err from here, always Ok(...) so that we can log
3432 : // spanned errors. Call api_error_handler instead and return appropriate Body.
3433 0 : Ok(result) => result,
3434 0 : Err(e) => {
3435 0 : // The handler task panicked. We have a global panic handler that logs the
3436 0 : // panic with its backtrace, so no need to log that here. Only log a brief
3437 0 : // message to make it clear that we returned the error to the client.
3438 0 : error!("HTTP request handler task panicked: {e:#}");
3439 :
3440 : // Don't return an Error here, because then fallback error handler that was
3441 : // installed in make_router() will print the error. Instead, construct the
3442 : // HTTP error response and return that.
3443 0 : Ok(
3444 0 : ApiError::InternalServerError(anyhow!("HTTP request handler task panicked"))
3445 0 : .into_response(),
3446 0 : )
3447 : }
3448 : }
3449 0 : })
3450 0 : .await;
3451 :
3452 0 : cancel_guard.disarm();
3453 0 :
3454 0 : result
3455 0 : }
3456 :
3457 : /// Like api_handler, but returns an error response if the server is built without
3458 : /// the 'testing' feature.
3459 0 : async fn testing_api_handler<R, H>(
3460 0 : desc: &str,
3461 0 : request: Request<Body>,
3462 0 : handler: H,
3463 0 : ) -> Result<Response<Body>, ApiError>
3464 0 : where
3465 0 : R: std::future::Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
3466 0 : H: FnOnce(Request<Body>, CancellationToken) -> R + Send + Sync + 'static,
3467 0 : {
3468 0 : if cfg!(feature = "testing") {
3469 0 : api_handler(request, handler).await
3470 : } else {
3471 0 : std::future::ready(Err(ApiError::BadRequest(anyhow!(
3472 0 : "Cannot {desc} because pageserver was compiled without testing APIs",
3473 0 : ))))
3474 0 : .await
3475 : }
3476 0 : }
3477 :
3478 0 : pub fn make_router(
3479 0 : state: Arc<State>,
3480 0 : launch_ts: &'static LaunchTimestamp,
3481 0 : auth: Option<Arc<SwappableJwtAuth>>,
3482 0 : ) -> anyhow::Result<RouterBuilder<hyper::Body, ApiError>> {
3483 0 : let spec = include_bytes!("openapi_spec.yml");
3484 0 : let mut router = attach_openapi_ui(endpoint::make_router(), spec, "/swagger.yml", "/v1/doc");
3485 0 : if auth.is_some() {
3486 0 : router = router.middleware(auth_middleware(|request| {
3487 0 : let state = get_state(request);
3488 0 : if state.allowlist_routes.contains(&request.uri().path()) {
3489 0 : None
3490 : } else {
3491 0 : state.auth.as_deref()
3492 : }
3493 0 : }))
3494 0 : }
3495 :
3496 0 : router = router.middleware(
3497 0 : endpoint::add_response_header_middleware(
3498 0 : "PAGESERVER_LAUNCH_TIMESTAMP",
3499 0 : &launch_ts.to_string(),
3500 0 : )
3501 0 : .expect("construct launch timestamp header middleware"),
3502 0 : );
3503 0 :
3504 0 : Ok(router
3505 0 : .data(state)
3506 0 : .get("/metrics", |r| request_span(r, prometheus_metrics_handler))
3507 0 : .get("/profile/cpu", |r| request_span(r, profile_cpu_handler))
3508 0 : .get("/profile/heap", |r| request_span(r, profile_heap_handler))
3509 0 : .get("/v1/status", |r| api_handler(r, status_handler))
3510 0 : .put("/v1/failpoints", |r| {
3511 0 : testing_api_handler("manage failpoints", r, failpoints_handler)
3512 0 : })
3513 0 : .post("/v1/reload_auth_validation_keys", |r| {
3514 0 : api_handler(r, reload_auth_validation_keys_handler)
3515 0 : })
3516 0 : .get("/v1/tenant", |r| api_handler(r, tenant_list_handler))
3517 0 : .get("/v1/tenant/:tenant_shard_id", |r| {
3518 0 : api_handler(r, tenant_status)
3519 0 : })
3520 0 : .delete("/v1/tenant/:tenant_shard_id", |r| {
3521 0 : api_handler(r, tenant_delete_handler)
3522 0 : })
3523 0 : .get("/v1/tenant/:tenant_shard_id/synthetic_size", |r| {
3524 0 : api_handler(r, tenant_size_handler)
3525 0 : })
3526 0 : .patch("/v1/tenant/config", |r| {
3527 0 : api_handler(r, patch_tenant_config_handler)
3528 0 : })
3529 0 : .put("/v1/tenant/config", |r| {
3530 0 : api_handler(r, update_tenant_config_handler)
3531 0 : })
3532 0 : .put("/v1/tenant/:tenant_shard_id/shard_split", |r| {
3533 0 : api_handler(r, tenant_shard_split_handler)
3534 0 : })
3535 0 : .get("/v1/tenant/:tenant_shard_id/config", |r| {
3536 0 : api_handler(r, get_tenant_config_handler)
3537 0 : })
3538 0 : .put("/v1/tenant/:tenant_shard_id/location_config", |r| {
3539 0 : api_handler(r, put_tenant_location_config_handler)
3540 0 : })
3541 0 : .get("/v1/location_config", |r| {
3542 0 : api_handler(r, list_location_config_handler)
3543 0 : })
3544 0 : .get("/v1/location_config/:tenant_shard_id", |r| {
3545 0 : api_handler(r, get_location_config_handler)
3546 0 : })
3547 0 : .put(
3548 0 : "/v1/tenant/:tenant_shard_id/time_travel_remote_storage",
3549 0 : |r| api_handler(r, tenant_time_travel_remote_storage_handler),
3550 0 : )
3551 0 : .get("/v1/tenant/:tenant_shard_id/timeline", |r| {
3552 0 : api_handler(r, timeline_list_handler)
3553 0 : })
3554 0 : .get("/v1/tenant/:tenant_shard_id/timeline_and_offloaded", |r| {
3555 0 : api_handler(r, timeline_and_offloaded_list_handler)
3556 0 : })
3557 0 : .post("/v1/tenant/:tenant_shard_id/timeline", |r| {
3558 0 : api_handler(r, timeline_create_handler)
3559 0 : })
3560 0 : .post("/v1/tenant/:tenant_shard_id/reset", |r| {
3561 0 : api_handler(r, tenant_reset_handler)
3562 0 : })
3563 0 : .post(
3564 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/preserve_initdb_archive",
3565 0 : |r| api_handler(r, timeline_preserve_initdb_handler),
3566 0 : )
3567 0 : .put(
3568 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/archival_config",
3569 0 : |r| api_handler(r, timeline_archival_config_handler),
3570 0 : )
3571 0 : .get("/v1/tenant/:tenant_shard_id/timeline/:timeline_id", |r| {
3572 0 : api_handler(r, timeline_detail_handler)
3573 0 : })
3574 0 : .get(
3575 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/get_lsn_by_timestamp",
3576 0 : |r| api_handler(r, get_lsn_by_timestamp_handler),
3577 0 : )
3578 0 : .get(
3579 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/get_timestamp_of_lsn",
3580 0 : |r| api_handler(r, get_timestamp_of_lsn_handler),
3581 0 : )
3582 0 : .post(
3583 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/lsn_lease",
3584 0 : |r| api_handler(r, lsn_lease_handler),
3585 0 : )
3586 0 : .put(
3587 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/do_gc",
3588 0 : |r| api_handler(r, timeline_gc_handler),
3589 0 : )
3590 0 : .get(
3591 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/compact",
3592 0 : |r| api_handler(r, timeline_compact_info_handler),
3593 0 : )
3594 0 : .put(
3595 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/compact",
3596 0 : |r| api_handler(r, timeline_compact_handler),
3597 0 : )
3598 0 : .delete(
3599 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/compact",
3600 0 : |r| api_handler(r, timeline_cancel_compact_handler),
3601 0 : )
3602 0 : .put(
3603 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/offload",
3604 0 : |r| testing_api_handler("attempt timeline offload", r, timeline_offload_handler),
3605 0 : )
3606 0 : .put(
3607 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/checkpoint",
3608 0 : |r| testing_api_handler("run timeline checkpoint", r, timeline_checkpoint_handler),
3609 0 : )
3610 0 : .post(
3611 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/download_remote_layers",
3612 0 : |r| api_handler(r, timeline_download_remote_layers_handler_post),
3613 0 : )
3614 0 : .get(
3615 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/download_remote_layers",
3616 0 : |r| api_handler(r, timeline_download_remote_layers_handler_get),
3617 0 : )
3618 0 : .put(
3619 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/detach_ancestor",
3620 0 : |r| api_handler(r, timeline_detach_ancestor_handler),
3621 0 : )
3622 0 : .delete("/v1/tenant/:tenant_shard_id/timeline/:timeline_id", |r| {
3623 0 : api_handler(r, timeline_delete_handler)
3624 0 : })
3625 0 : .get(
3626 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/layer",
3627 0 : |r| api_handler(r, layer_map_info_handler),
3628 0 : )
3629 0 : .get(
3630 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/layer/:layer_file_name",
3631 0 : |r| api_handler(r, layer_download_handler),
3632 0 : )
3633 0 : .delete(
3634 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/layer/:layer_file_name",
3635 0 : |r| api_handler(r, evict_timeline_layer_handler),
3636 0 : )
3637 0 : .post(
3638 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/layer/:layer_name/scan_disposable_keys",
3639 0 : |r| testing_api_handler("timeline_layer_scan_disposable_keys", r, timeline_layer_scan_disposable_keys),
3640 0 : )
3641 0 : .post(
3642 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/block_gc",
3643 0 : |r| api_handler(r, timeline_gc_blocking_handler),
3644 0 : )
3645 0 : .post(
3646 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/unblock_gc",
3647 0 : |r| api_handler(r, timeline_gc_unblocking_handler),
3648 0 : )
3649 0 : .get(
3650 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/page_trace",
3651 0 : |r| api_handler(r, timeline_page_trace_handler),
3652 0 : )
3653 0 : .post("/v1/tenant/:tenant_shard_id/heatmap_upload", |r| {
3654 0 : api_handler(r, secondary_upload_handler)
3655 0 : })
3656 0 : .get("/v1/tenant/:tenant_id/scan_remote_storage", |r| {
3657 0 : api_handler(r, tenant_scan_remote_handler)
3658 0 : })
3659 0 : .put("/v1/disk_usage_eviction/run", |r| {
3660 0 : api_handler(r, disk_usage_eviction_run)
3661 0 : })
3662 0 : .put("/v1/deletion_queue/flush", |r| {
3663 0 : api_handler(r, deletion_queue_flush)
3664 0 : })
3665 0 : .get("/v1/tenant/:tenant_shard_id/secondary/status", |r| {
3666 0 : api_handler(r, secondary_status_handler)
3667 0 : })
3668 0 : .post("/v1/tenant/:tenant_shard_id/secondary/download", |r| {
3669 0 : api_handler(r, secondary_download_handler)
3670 0 : })
3671 0 : .post("/v1/tenant/:tenant_shard_id/wait_lsn", |r| {
3672 0 : api_handler(r, wait_lsn_handler)
3673 0 : })
3674 0 : .put("/v1/tenant/:tenant_shard_id/break", |r| {
3675 0 : testing_api_handler("set tenant state to broken", r, handle_tenant_break)
3676 0 : })
3677 0 : .get("/v1/panic", |r| api_handler(r, always_panic_handler))
3678 0 : .post("/v1/tracing/event", |r| {
3679 0 : testing_api_handler("emit a tracing event", r, post_tracing_event_handler)
3680 0 : })
3681 0 : .get(
3682 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/getpage",
3683 0 : |r| testing_api_handler("getpage@lsn", r, getpage_at_lsn_handler),
3684 0 : )
3685 0 : .get(
3686 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/keyspace",
3687 0 : |r| api_handler(r, timeline_collect_keyspace),
3688 0 : )
3689 0 : .put("/v1/io_engine", |r| api_handler(r, put_io_engine_handler))
3690 0 : .put("/v1/io_mode", |r| api_handler(r, put_io_mode_handler))
3691 0 : .get("/v1/utilization", |r| api_handler(r, get_utilization))
3692 0 : .post(
3693 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/ingest_aux_files",
3694 0 : |r| testing_api_handler("ingest_aux_files", r, ingest_aux_files),
3695 0 : )
3696 0 : .post(
3697 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/list_aux_files",
3698 0 : |r| testing_api_handler("list_aux_files", r, list_aux_files),
3699 0 : )
3700 0 : .post("/v1/top_tenants", |r| api_handler(r, post_top_tenants))
3701 0 : .post(
3702 0 : "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/perf_info",
3703 0 : |r| testing_api_handler("perf_info", r, perf_info),
3704 0 : )
3705 0 : .put(
3706 0 : "/v1/tenant/:tenant_id/timeline/:timeline_id/import_basebackup",
3707 0 : |r| api_handler(r, put_tenant_timeline_import_basebackup),
3708 0 : )
3709 0 : .put(
3710 0 : "/v1/tenant/:tenant_id/timeline/:timeline_id/import_wal",
3711 0 : |r| api_handler(r, put_tenant_timeline_import_wal),
3712 0 : )
3713 0 : .any(handler_404))
3714 0 : }
|
