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