LCOV - code coverage report
Current view: top level - safekeeper/src/http - routes.rs (source / functions) Coverage Total Hit
Test: 5445d246133daeceb0507e6cc0797ab7c1c70cb8.info Lines: 2.6 % 582 15
Test Date: 2025-03-12 18:05:02 Functions: 1.2 % 85 1

            Line data    Source code
       1              : use std::collections::HashMap;
       2              : use std::fmt;
       3              : use std::io::Write as _;
       4              : use std::str::FromStr;
       5              : use std::sync::Arc;
       6              : 
       7              : use http_utils::endpoint::{
       8              :     self, ChannelWriter, auth_middleware, check_permission_with, profile_cpu_handler,
       9              :     profile_heap_handler, prometheus_metrics_handler, request_span,
      10              : };
      11              : use http_utils::error::ApiError;
      12              : use http_utils::failpoints::failpoints_handler;
      13              : use http_utils::json::{json_request, json_response};
      14              : use http_utils::request::{ensure_no_body, parse_query_param, parse_request_param};
      15              : use http_utils::{RequestExt, RouterBuilder};
      16              : use hyper::{Body, Request, Response, StatusCode};
      17              : use postgres_ffi::WAL_SEGMENT_SIZE;
      18              : use safekeeper_api::models::{
      19              :     AcceptorStateStatus, PullTimelineRequest, SafekeeperStatus, SkTimelineInfo, TermSwitchApiEntry,
      20              :     TimelineCopyRequest, TimelineCreateRequest, TimelineDeleteResult, TimelineStatus,
      21              :     TimelineTermBumpRequest,
      22              : };
      23              : use safekeeper_api::{ServerInfo, membership, models};
      24              : use storage_broker::proto::{SafekeeperTimelineInfo, TenantTimelineId as ProtoTenantTimelineId};
      25              : use tokio::sync::mpsc;
      26              : use tokio::task;
      27              : use tokio_stream::wrappers::ReceiverStream;
      28              : use tokio_util::sync::CancellationToken;
      29              : use tracing::{Instrument, info_span};
      30              : use utils::auth::SwappableJwtAuth;
      31              : use utils::id::{TenantId, TenantTimelineId, TimelineId};
      32              : use utils::lsn::Lsn;
      33              : 
      34              : use crate::debug_dump::TimelineDigestRequest;
      35              : use crate::safekeeper::TermLsn;
      36              : use crate::timelines_global_map::DeleteOrExclude;
      37              : use crate::{
      38              :     GlobalTimelines, SafeKeeperConf, copy_timeline, debug_dump, patch_control_file, pull_timeline,
      39              : };
      40              : 
      41              : /// Healthcheck handler.
      42            0 : async fn status_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
      43            0 :     check_permission(&request, None)?;
      44            0 :     let conf = get_conf(&request);
      45            0 :     let status = SafekeeperStatus { id: conf.my_id };
      46            0 :     json_response(StatusCode::OK, status)
      47            0 : }
      48              : 
      49            0 : fn get_conf(request: &Request<Body>) -> &SafeKeeperConf {
      50            0 :     request
      51            0 :         .data::<Arc<SafeKeeperConf>>()
      52            0 :         .expect("unknown state type")
      53            0 :         .as_ref()
      54            0 : }
      55              : 
      56            0 : fn get_global_timelines(request: &Request<Body>) -> Arc<GlobalTimelines> {
      57            0 :     request
      58            0 :         .data::<Arc<GlobalTimelines>>()
      59            0 :         .expect("unknown state type")
      60            0 :         .clone()
      61            0 : }
      62              : 
      63            0 : fn check_permission(request: &Request<Body>, tenant_id: Option<TenantId>) -> Result<(), ApiError> {
      64            0 :     check_permission_with(request, |claims| {
      65            0 :         crate::auth::check_permission(claims, tenant_id)
      66            0 :     })
      67            0 : }
      68              : 
      69              : /// Deactivates all timelines for the tenant and removes its data directory.
      70              : /// See `timeline_delete_handler`.
      71            0 : async fn tenant_delete_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
      72            0 :     let tenant_id = parse_request_param(&request, "tenant_id")?;
      73            0 :     let only_local = parse_query_param(&request, "only_local")?.unwrap_or(false);
      74            0 :     check_permission(&request, Some(tenant_id))?;
      75            0 :     ensure_no_body(&mut request).await?;
      76            0 :     let global_timelines = get_global_timelines(&request);
      77            0 :     let action = if only_local {
      78            0 :         DeleteOrExclude::DeleteLocal
      79              :     } else {
      80            0 :         DeleteOrExclude::Delete
      81              :     };
      82            0 :     let delete_info = global_timelines
      83            0 :         .delete_all_for_tenant(&tenant_id, action)
      84            0 :         .await
      85            0 :         .map_err(ApiError::InternalServerError)?;
      86            0 :     json_response(
      87            0 :         StatusCode::OK,
      88            0 :         delete_info
      89            0 :             .iter()
      90            0 :             .map(|(ttid, resp)| (format!("{}", ttid.timeline_id), *resp))
      91            0 :             .collect::<HashMap<String, TimelineDeleteResult>>(),
      92            0 :     )
      93            0 : }
      94              : 
      95            0 : async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
      96            0 :     let request_data: TimelineCreateRequest = json_request(&mut request).await?;
      97              : 
      98            0 :     let ttid = TenantTimelineId {
      99            0 :         tenant_id: request_data.tenant_id,
     100            0 :         timeline_id: request_data.timeline_id,
     101            0 :     };
     102            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     103              : 
     104            0 :     let server_info = ServerInfo {
     105            0 :         pg_version: request_data.pg_version,
     106            0 :         system_id: request_data.system_id.unwrap_or(0),
     107            0 :         wal_seg_size: request_data.wal_seg_size.unwrap_or(WAL_SEGMENT_SIZE as u32),
     108            0 :     };
     109            0 :     let global_timelines = get_global_timelines(&request);
     110            0 :     global_timelines
     111            0 :         .create(
     112            0 :             ttid,
     113            0 :             request_data.mconf,
     114            0 :             server_info,
     115            0 :             request_data.start_lsn,
     116            0 :             request_data.commit_lsn.unwrap_or(request_data.start_lsn),
     117            0 :         )
     118            0 :         .await
     119            0 :         .map_err(ApiError::InternalServerError)?;
     120              : 
     121            0 :     json_response(StatusCode::OK, ())
     122            0 : }
     123              : 
     124            0 : async fn utilization_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     125            0 :     check_permission(&request, None)?;
     126            0 :     let global_timelines = get_global_timelines(&request);
     127            0 :     let utilization = global_timelines.get_timeline_counts();
     128            0 :     json_response(StatusCode::OK, utilization)
     129            0 : }
     130              : 
     131              : /// List all (not deleted) timelines.
     132              : /// Note: it is possible to do the same with debug_dump.
     133            0 : async fn timeline_list_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     134            0 :     check_permission(&request, None)?;
     135            0 :     let global_timelines = get_global_timelines(&request);
     136            0 :     let res: Vec<TenantTimelineId> = global_timelines
     137            0 :         .get_all()
     138            0 :         .iter()
     139            0 :         .map(|tli| tli.ttid)
     140            0 :         .collect();
     141            0 :     json_response(StatusCode::OK, res)
     142            0 : }
     143              : 
     144              : impl From<TermSwitchApiEntry> for TermLsn {
     145            0 :     fn from(api_val: TermSwitchApiEntry) -> Self {
     146            0 :         TermLsn {
     147            0 :             term: api_val.term,
     148            0 :             lsn: api_val.lsn,
     149            0 :         }
     150            0 :     }
     151              : }
     152              : 
     153              : /// Report info about timeline.
     154            0 : async fn timeline_status_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     155            0 :     let ttid = TenantTimelineId::new(
     156            0 :         parse_request_param(&request, "tenant_id")?,
     157            0 :         parse_request_param(&request, "timeline_id")?,
     158              :     );
     159            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     160              : 
     161            0 :     let global_timelines = get_global_timelines(&request);
     162            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     163            0 :     let (inmem, state) = tli.get_state().await;
     164            0 :     let flush_lsn = tli.get_flush_lsn().await;
     165              : 
     166            0 :     let last_log_term = state.acceptor_state.get_last_log_term(flush_lsn);
     167            0 :     let term_history = state
     168            0 :         .acceptor_state
     169            0 :         .term_history
     170            0 :         .0
     171            0 :         .into_iter()
     172            0 :         .map(|ts| TermSwitchApiEntry {
     173            0 :             term: ts.term,
     174            0 :             lsn: ts.lsn,
     175            0 :         })
     176            0 :         .collect();
     177            0 :     let acc_state = AcceptorStateStatus {
     178            0 :         term: state.acceptor_state.term,
     179            0 :         epoch: last_log_term,
     180            0 :         term_history,
     181            0 :     };
     182            0 : 
     183            0 :     let conf = get_conf(&request);
     184              :     // Note: we report in memory values which can be lost.
     185            0 :     let status = TimelineStatus {
     186            0 :         tenant_id: ttid.tenant_id,
     187            0 :         timeline_id: ttid.timeline_id,
     188            0 :         mconf: state.mconf,
     189            0 :         acceptor_state: acc_state,
     190            0 :         pg_info: state.server,
     191            0 :         flush_lsn,
     192            0 :         timeline_start_lsn: state.timeline_start_lsn,
     193            0 :         local_start_lsn: state.local_start_lsn,
     194            0 :         commit_lsn: inmem.commit_lsn,
     195            0 :         backup_lsn: inmem.backup_lsn,
     196            0 :         peer_horizon_lsn: inmem.peer_horizon_lsn,
     197            0 :         remote_consistent_lsn: inmem.remote_consistent_lsn,
     198            0 :         peers: tli.get_peers(conf).await,
     199            0 :         walsenders: tli.get_walsenders().get_all_public(),
     200            0 :         walreceivers: tli.get_walreceivers().get_all(),
     201            0 :     };
     202            0 :     json_response(StatusCode::OK, status)
     203            0 : }
     204              : 
     205              : /// Deactivates the timeline and removes its data directory.
     206            0 : async fn timeline_delete_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     207            0 :     let ttid = TenantTimelineId::new(
     208            0 :         parse_request_param(&request, "tenant_id")?,
     209            0 :         parse_request_param(&request, "timeline_id")?,
     210              :     );
     211            0 :     let only_local = parse_query_param(&request, "only_local")?.unwrap_or(false);
     212            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     213            0 :     ensure_no_body(&mut request).await?;
     214            0 :     let global_timelines = get_global_timelines(&request);
     215            0 :     let action = if only_local {
     216            0 :         DeleteOrExclude::DeleteLocal
     217              :     } else {
     218            0 :         DeleteOrExclude::Delete
     219              :     };
     220            0 :     let resp = global_timelines
     221            0 :         .delete_or_exclude(&ttid, action)
     222            0 :         .await
     223            0 :         .map_err(ApiError::from)?;
     224            0 :     json_response(StatusCode::OK, resp)
     225            0 : }
     226              : 
     227              : /// Pull timeline from peer safekeeper instances.
     228            0 : async fn timeline_pull_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     229            0 :     check_permission(&request, None)?;
     230              : 
     231            0 :     let data: PullTimelineRequest = json_request(&mut request).await?;
     232            0 :     let conf = get_conf(&request);
     233            0 :     let global_timelines = get_global_timelines(&request);
     234              : 
     235            0 :     let resp = pull_timeline::handle_request(data, conf.sk_auth_token.clone(), global_timelines)
     236            0 :         .await
     237            0 :         .map_err(ApiError::InternalServerError)?;
     238            0 :     json_response(StatusCode::OK, resp)
     239            0 : }
     240              : 
     241              : /// Stream tar archive with all timeline data.
     242            0 : async fn timeline_snapshot_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     243            0 :     let destination = parse_request_param(&request, "destination_id")?;
     244            0 :     let ttid = TenantTimelineId::new(
     245            0 :         parse_request_param(&request, "tenant_id")?,
     246            0 :         parse_request_param(&request, "timeline_id")?,
     247              :     );
     248            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     249              : 
     250            0 :     let global_timelines = get_global_timelines(&request);
     251            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     252              : 
     253              :     // To stream the body use wrap_stream which wants Stream of Result<Bytes>,
     254              :     // so create the chan and write to it in another task.
     255            0 :     let (tx, rx) = mpsc::channel(1);
     256            0 : 
     257            0 :     let conf = get_conf(&request);
     258            0 :     task::spawn(pull_timeline::stream_snapshot(
     259            0 :         tli,
     260            0 :         conf.my_id,
     261            0 :         destination,
     262            0 :         tx,
     263            0 :     ));
     264            0 : 
     265            0 :     let rx_stream = ReceiverStream::new(rx);
     266            0 :     let body = Body::wrap_stream(rx_stream);
     267            0 : 
     268            0 :     let response = Response::builder()
     269            0 :         .status(200)
     270            0 :         .header(hyper::header::CONTENT_TYPE, "application/octet-stream")
     271            0 :         .body(body)
     272            0 :         .unwrap();
     273            0 : 
     274            0 :     Ok(response)
     275            0 : }
     276              : 
     277              : /// Error type for delete_or_exclude: either generation conflict or something
     278              : /// internal.
     279              : #[derive(thiserror::Error, Debug)]
     280              : pub enum DeleteOrExcludeError {
     281              :     #[error("refused to switch into excluding mconf {requested}, current: {current}")]
     282              :     Conflict {
     283              :         requested: membership::Configuration,
     284              :         current: membership::Configuration,
     285              :     },
     286              :     #[error(transparent)]
     287              :     Other(#[from] anyhow::Error),
     288              : }
     289              : 
     290              : /// Convert DeleteOrExcludeError to ApiError.
     291              : impl From<DeleteOrExcludeError> for ApiError {
     292            0 :     fn from(de: DeleteOrExcludeError) -> ApiError {
     293            0 :         match de {
     294              :             DeleteOrExcludeError::Conflict {
     295              :                 requested: _,
     296              :                 current: _,
     297            0 :             } => ApiError::Conflict(de.to_string()),
     298            0 :             DeleteOrExcludeError::Other(e) => ApiError::InternalServerError(e),
     299              :         }
     300            0 :     }
     301              : }
     302              : 
     303              : /// Remove timeline locally after this node has been excluded from the
     304              : /// membership configuration. The body is the same as in the membership endpoint
     305              : /// -- conf where node is excluded -- and in principle single ep could be used
     306              : /// for both actions, but since this is a data deletion op let's keep them
     307              : /// separate.
     308            0 : async fn timeline_exclude_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     309            0 :     let ttid = TenantTimelineId::new(
     310            0 :         parse_request_param(&request, "tenant_id")?,
     311            0 :         parse_request_param(&request, "timeline_id")?,
     312              :     );
     313            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     314              : 
     315            0 :     let global_timelines = get_global_timelines(&request);
     316            0 :     let data: models::TimelineMembershipSwitchRequest = json_request(&mut request).await?;
     317            0 :     let my_id = get_conf(&request).my_id;
     318            0 :     // If request doesn't exclude us, membership switch endpoint should be used
     319            0 :     // instead.
     320            0 :     if data.mconf.contains(my_id) {
     321            0 :         return Err(ApiError::Forbidden(format!(
     322            0 :             "refused to switch into {}, node {} is member of it",
     323            0 :             data.mconf, my_id
     324            0 :         )));
     325            0 :     }
     326            0 :     let action = DeleteOrExclude::Exclude(data.mconf);
     327              : 
     328            0 :     let resp = global_timelines
     329            0 :         .delete_or_exclude(&ttid, action)
     330            0 :         .await
     331            0 :         .map_err(ApiError::from)?;
     332            0 :     json_response(StatusCode::OK, resp)
     333            0 : }
     334              : 
     335              : /// Consider switching timeline membership configuration to the provided one.
     336            0 : async fn timeline_membership_handler(
     337            0 :     mut request: Request<Body>,
     338            0 : ) -> Result<Response<Body>, ApiError> {
     339            0 :     let ttid = TenantTimelineId::new(
     340            0 :         parse_request_param(&request, "tenant_id")?,
     341            0 :         parse_request_param(&request, "timeline_id")?,
     342              :     );
     343            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     344              : 
     345            0 :     let global_timelines = get_global_timelines(&request);
     346            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     347              : 
     348            0 :     let data: models::TimelineMembershipSwitchRequest = json_request(&mut request).await?;
     349            0 :     let my_id = get_conf(&request).my_id;
     350            0 :     // If request excludes us, exclude endpoint should be used instead.
     351            0 :     if !data.mconf.contains(my_id) {
     352            0 :         return Err(ApiError::Forbidden(format!(
     353            0 :             "refused to switch into {}, node {} is not a member of it",
     354            0 :             data.mconf, my_id
     355            0 :         )));
     356            0 :     }
     357            0 :     let req_gen = data.mconf.generation;
     358            0 :     let response = tli
     359            0 :         .membership_switch(data.mconf)
     360            0 :         .await
     361            0 :         .map_err(ApiError::InternalServerError)?;
     362              : 
     363              :     // Return 409 if request was ignored.
     364            0 :     if req_gen == response.current_conf.generation {
     365            0 :         json_response(StatusCode::OK, response)
     366              :     } else {
     367            0 :         Err(ApiError::Conflict(format!(
     368            0 :             "request to switch into {} ignored, current generation {}",
     369            0 :             req_gen, response.current_conf.generation
     370            0 :         )))
     371              :     }
     372            0 : }
     373              : 
     374            0 : async fn timeline_copy_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     375            0 :     check_permission(&request, None)?;
     376              : 
     377            0 :     let request_data: TimelineCopyRequest = json_request(&mut request).await?;
     378            0 :     let source_ttid = TenantTimelineId::new(
     379            0 :         parse_request_param(&request, "tenant_id")?,
     380            0 :         parse_request_param(&request, "source_timeline_id")?,
     381              :     );
     382              : 
     383            0 :     let global_timelines = get_global_timelines(&request);
     384            0 : 
     385            0 :     copy_timeline::handle_request(copy_timeline::Request{
     386            0 :         source_ttid,
     387            0 :         until_lsn: request_data.until_lsn,
     388            0 :         destination_ttid: TenantTimelineId::new(source_ttid.tenant_id, request_data.target_timeline_id),
     389            0 :     }, global_timelines)
     390            0 :         .instrument(info_span!("copy_timeline", from=%source_ttid, to=%request_data.target_timeline_id, until_lsn=%request_data.until_lsn))
     391            0 :         .await
     392            0 :         .map_err(ApiError::InternalServerError)?;
     393              : 
     394            0 :     json_response(StatusCode::OK, ())
     395            0 : }
     396              : 
     397            0 : async fn patch_control_file_handler(
     398            0 :     mut request: Request<Body>,
     399            0 : ) -> Result<Response<Body>, ApiError> {
     400            0 :     check_permission(&request, None)?;
     401              : 
     402            0 :     let ttid = TenantTimelineId::new(
     403            0 :         parse_request_param(&request, "tenant_id")?,
     404            0 :         parse_request_param(&request, "timeline_id")?,
     405              :     );
     406              : 
     407            0 :     let global_timelines = get_global_timelines(&request);
     408            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     409              : 
     410            0 :     let patch_request: patch_control_file::Request = json_request(&mut request).await?;
     411            0 :     let response = patch_control_file::handle_request(tli, patch_request)
     412            0 :         .await
     413            0 :         .map_err(ApiError::InternalServerError)?;
     414              : 
     415            0 :     json_response(StatusCode::OK, response)
     416            0 : }
     417              : 
     418              : /// Force persist control file.
     419            0 : async fn timeline_checkpoint_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     420            0 :     check_permission(&request, None)?;
     421              : 
     422            0 :     let ttid = TenantTimelineId::new(
     423            0 :         parse_request_param(&request, "tenant_id")?,
     424            0 :         parse_request_param(&request, "timeline_id")?,
     425              :     );
     426              : 
     427            0 :     let global_timelines = get_global_timelines(&request);
     428            0 :     let tli = global_timelines.get(ttid)?;
     429            0 :     tli.write_shared_state()
     430            0 :         .await
     431              :         .sk
     432            0 :         .state_mut()
     433            0 :         .flush()
     434            0 :         .await
     435            0 :         .map_err(ApiError::InternalServerError)?;
     436            0 :     json_response(StatusCode::OK, ())
     437            0 : }
     438              : 
     439            0 : async fn timeline_digest_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     440            0 :     let ttid = TenantTimelineId::new(
     441            0 :         parse_request_param(&request, "tenant_id")?,
     442            0 :         parse_request_param(&request, "timeline_id")?,
     443              :     );
     444            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     445              : 
     446            0 :     let global_timelines = get_global_timelines(&request);
     447            0 :     let from_lsn: Option<Lsn> = parse_query_param(&request, "from_lsn")?;
     448            0 :     let until_lsn: Option<Lsn> = parse_query_param(&request, "until_lsn")?;
     449              : 
     450            0 :     let request = TimelineDigestRequest {
     451            0 :         from_lsn: from_lsn.ok_or(ApiError::BadRequest(anyhow::anyhow!(
     452            0 :             "from_lsn is required"
     453            0 :         )))?,
     454            0 :         until_lsn: until_lsn.ok_or(ApiError::BadRequest(anyhow::anyhow!(
     455            0 :             "until_lsn is required"
     456            0 :         )))?,
     457              :     };
     458              : 
     459            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     460            0 :     let tli = tli
     461            0 :         .wal_residence_guard()
     462            0 :         .await
     463            0 :         .map_err(ApiError::InternalServerError)?;
     464              : 
     465            0 :     let response = debug_dump::calculate_digest(&tli, request)
     466            0 :         .await
     467            0 :         .map_err(ApiError::InternalServerError)?;
     468            0 :     json_response(StatusCode::OK, response)
     469            0 : }
     470              : 
     471              : /// Unevict timeline and remove uploaded partial segment(s) from the remote storage.
     472              : /// Successfull response returns list of segments existed before the deletion.
     473              : /// Aimed for one-off usage not normally needed.
     474            0 : async fn timeline_backup_partial_reset(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     475            0 :     let ttid = TenantTimelineId::new(
     476            0 :         parse_request_param(&request, "tenant_id")?,
     477            0 :         parse_request_param(&request, "timeline_id")?,
     478              :     );
     479            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     480              : 
     481            0 :     let global_timelines = get_global_timelines(&request);
     482            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     483              : 
     484            0 :     let response = tli
     485            0 :         .backup_partial_reset()
     486            0 :         .await
     487            0 :         .map_err(ApiError::InternalServerError)?;
     488            0 :     json_response(StatusCode::OK, response)
     489            0 : }
     490              : 
     491              : /// Make term at least as high as one in request. If one in request is None,
     492              : /// increment current one.
     493            0 : async fn timeline_term_bump_handler(
     494            0 :     mut request: Request<Body>,
     495            0 : ) -> Result<Response<Body>, ApiError> {
     496            0 :     let ttid = TenantTimelineId::new(
     497            0 :         parse_request_param(&request, "tenant_id")?,
     498            0 :         parse_request_param(&request, "timeline_id")?,
     499              :     );
     500            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     501              : 
     502            0 :     let request_data: TimelineTermBumpRequest = json_request(&mut request).await?;
     503              : 
     504            0 :     let global_timelines = get_global_timelines(&request);
     505            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     506            0 :     let response = tli
     507            0 :         .term_bump(request_data.term)
     508            0 :         .await
     509            0 :         .map_err(ApiError::InternalServerError)?;
     510              : 
     511            0 :     json_response(StatusCode::OK, response)
     512            0 : }
     513              : 
     514              : /// Used only in tests to hand craft required data.
     515            0 : async fn record_safekeeper_info(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     516            0 :     let ttid = TenantTimelineId::new(
     517            0 :         parse_request_param(&request, "tenant_id")?,
     518            0 :         parse_request_param(&request, "timeline_id")?,
     519              :     );
     520            0 :     check_permission(&request, Some(ttid.tenant_id))?;
     521            0 :     let sk_info: SkTimelineInfo = json_request(&mut request).await?;
     522            0 :     let proto_sk_info = SafekeeperTimelineInfo {
     523            0 :         safekeeper_id: 0,
     524            0 :         tenant_timeline_id: Some(ProtoTenantTimelineId {
     525            0 :             tenant_id: ttid.tenant_id.as_ref().to_owned(),
     526            0 :             timeline_id: ttid.timeline_id.as_ref().to_owned(),
     527            0 :         }),
     528            0 :         term: sk_info.term.unwrap_or(0),
     529            0 :         last_log_term: sk_info.last_log_term.unwrap_or(0),
     530            0 :         flush_lsn: sk_info.flush_lsn.0,
     531            0 :         commit_lsn: sk_info.commit_lsn.0,
     532            0 :         remote_consistent_lsn: sk_info.remote_consistent_lsn.0,
     533            0 :         peer_horizon_lsn: sk_info.peer_horizon_lsn.0,
     534            0 :         safekeeper_connstr: sk_info.safekeeper_connstr.unwrap_or_else(|| "".to_owned()),
     535            0 :         http_connstr: sk_info.http_connstr.unwrap_or_else(|| "".to_owned()),
     536            0 :         backup_lsn: sk_info.backup_lsn.0,
     537            0 :         local_start_lsn: sk_info.local_start_lsn.0,
     538            0 :         availability_zone: None,
     539            0 :         standby_horizon: sk_info.standby_horizon.0,
     540            0 :     };
     541            0 : 
     542            0 :     let global_timelines = get_global_timelines(&request);
     543            0 :     let tli = global_timelines.get(ttid).map_err(ApiError::from)?;
     544            0 :     tli.record_safekeeper_info(proto_sk_info)
     545            0 :         .await
     546            0 :         .map_err(ApiError::InternalServerError)?;
     547              : 
     548            0 :     json_response(StatusCode::OK, ())
     549            0 : }
     550              : 
     551            0 : fn parse_kv_str<E: fmt::Display, T: FromStr<Err = E>>(k: &str, v: &str) -> Result<T, ApiError> {
     552            0 :     v.parse()
     553            0 :         .map_err(|e| ApiError::BadRequest(anyhow::anyhow!("cannot parse {k}: {e}")))
     554            0 : }
     555              : 
     556              : /// Dump debug info about all available safekeeper state.
     557            0 : async fn dump_debug_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     558            0 :     check_permission(&request, None)?;
     559            0 :     ensure_no_body(&mut request).await?;
     560              : 
     561            0 :     let mut dump_all: Option<bool> = None;
     562            0 :     let mut dump_control_file: Option<bool> = None;
     563            0 :     let mut dump_memory: Option<bool> = None;
     564            0 :     let mut dump_disk_content: Option<bool> = None;
     565            0 :     let mut dump_term_history: Option<bool> = None;
     566            0 :     let mut dump_wal_last_modified: Option<bool> = None;
     567            0 :     let mut tenant_id: Option<TenantId> = None;
     568            0 :     let mut timeline_id: Option<TimelineId> = None;
     569            0 : 
     570            0 :     let query = request.uri().query().unwrap_or("");
     571            0 :     let mut values = url::form_urlencoded::parse(query.as_bytes());
     572              : 
     573            0 :     for (k, v) in &mut values {
     574            0 :         match k.as_ref() {
     575            0 :             "dump_all" => dump_all = Some(parse_kv_str(&k, &v)?),
     576            0 :             "dump_control_file" => dump_control_file = Some(parse_kv_str(&k, &v)?),
     577            0 :             "dump_memory" => dump_memory = Some(parse_kv_str(&k, &v)?),
     578            0 :             "dump_disk_content" => dump_disk_content = Some(parse_kv_str(&k, &v)?),
     579            0 :             "dump_term_history" => dump_term_history = Some(parse_kv_str(&k, &v)?),
     580            0 :             "dump_wal_last_modified" => dump_wal_last_modified = Some(parse_kv_str(&k, &v)?),
     581            0 :             "tenant_id" => tenant_id = Some(parse_kv_str(&k, &v)?),
     582            0 :             "timeline_id" => timeline_id = Some(parse_kv_str(&k, &v)?),
     583            0 :             _ => Err(ApiError::BadRequest(anyhow::anyhow!(
     584            0 :                 "Unknown query parameter: {}",
     585            0 :                 k
     586            0 :             )))?,
     587              :         }
     588              :     }
     589              : 
     590            0 :     let dump_all = dump_all.unwrap_or(false);
     591            0 :     let dump_control_file = dump_control_file.unwrap_or(dump_all);
     592            0 :     let dump_memory = dump_memory.unwrap_or(dump_all);
     593            0 :     let dump_disk_content = dump_disk_content.unwrap_or(dump_all);
     594            0 :     let dump_term_history = dump_term_history.unwrap_or(true);
     595            0 :     let dump_wal_last_modified = dump_wal_last_modified.unwrap_or(dump_all);
     596            0 : 
     597            0 :     let global_timelines = get_global_timelines(&request);
     598            0 : 
     599            0 :     let args = debug_dump::Args {
     600            0 :         dump_all,
     601            0 :         dump_control_file,
     602            0 :         dump_memory,
     603            0 :         dump_disk_content,
     604            0 :         dump_term_history,
     605            0 :         dump_wal_last_modified,
     606            0 :         tenant_id,
     607            0 :         timeline_id,
     608            0 :     };
     609              : 
     610            0 :     let resp = debug_dump::build(args, global_timelines)
     611            0 :         .await
     612            0 :         .map_err(ApiError::InternalServerError)?;
     613              : 
     614            0 :     let started_at = std::time::Instant::now();
     615            0 : 
     616            0 :     let (tx, rx) = mpsc::channel(1);
     617            0 : 
     618            0 :     let body = Body::wrap_stream(ReceiverStream::new(rx));
     619            0 : 
     620            0 :     let mut writer = ChannelWriter::new(128 * 1024, tx);
     621            0 : 
     622            0 :     let response = Response::builder()
     623            0 :         .status(200)
     624            0 :         .header(hyper::header::CONTENT_TYPE, "application/octet-stream")
     625            0 :         .body(body)
     626            0 :         .unwrap();
     627              : 
     628            0 :     let span = info_span!("blocking");
     629            0 :     tokio::task::spawn_blocking(move || {
     630            0 :         let _span = span.entered();
     631            0 : 
     632            0 :         let res = serde_json::to_writer(&mut writer, &resp)
     633            0 :             .map_err(std::io::Error::from)
     634            0 :             .and_then(|_| writer.flush());
     635            0 : 
     636            0 :         match res {
     637              :             Ok(()) => {
     638            0 :                 tracing::info!(
     639            0 :                     bytes = writer.flushed_bytes(),
     640            0 :                     elapsed_ms = started_at.elapsed().as_millis(),
     641            0 :                     "responded /v1/debug_dump"
     642              :                 );
     643              :             }
     644            0 :             Err(e) => {
     645            0 :                 tracing::warn!("failed to write out /v1/debug_dump response: {e:#}");
     646              :                 // semantics of this error are quite... unclear. we want to error the stream out to
     647              :                 // abort the response to somehow notify the client that we failed.
     648              :                 //
     649              :                 // though, most likely the reason for failure is that the receiver is already gone.
     650            0 :                 drop(
     651            0 :                     writer
     652            0 :                         .tx
     653            0 :                         .blocking_send(Err(std::io::ErrorKind::BrokenPipe.into())),
     654            0 :                 );
     655              :             }
     656              :         }
     657            0 :     });
     658            0 : 
     659            0 :     Ok(response)
     660            0 : }
     661              : 
     662              : /// Safekeeper http router.
     663            0 : pub fn make_router(
     664            0 :     conf: Arc<SafeKeeperConf>,
     665            0 :     global_timelines: Arc<GlobalTimelines>,
     666            0 : ) -> RouterBuilder<hyper::Body, ApiError> {
     667            0 :     let mut router = endpoint::make_router();
     668            0 :     if conf.http_auth.is_some() {
     669            0 :         router = router.middleware(auth_middleware(|request| {
     670              :             const ALLOWLIST_ROUTES: &[&str] =
     671              :                 &["/v1/status", "/metrics", "/profile/cpu", "/profile/heap"];
     672            0 :             if ALLOWLIST_ROUTES.contains(&request.uri().path()) {
     673            0 :                 None
     674              :             } else {
     675              :                 // Option<Arc<SwappableJwtAuth>> is always provided as data below, hence unwrap().
     676            0 :                 request
     677            0 :                     .data::<Option<Arc<SwappableJwtAuth>>>()
     678            0 :                     .unwrap()
     679            0 :                     .as_deref()
     680              :             }
     681            0 :         }))
     682            0 :     }
     683              : 
     684              :     // NB: on any changes do not forget to update the OpenAPI spec
     685              :     // located nearby (/safekeeper/src/http/openapi_spec.yaml).
     686            0 :     let auth = conf.http_auth.clone();
     687            0 :     router
     688            0 :         .data(conf)
     689            0 :         .data(global_timelines)
     690            0 :         .data(auth)
     691            0 :         .get("/metrics", |r| request_span(r, prometheus_metrics_handler))
     692            0 :         .get("/profile/cpu", |r| request_span(r, profile_cpu_handler))
     693            0 :         .get("/profile/heap", |r| request_span(r, profile_heap_handler))
     694            0 :         .get("/v1/status", |r| request_span(r, status_handler))
     695            0 :         .put("/v1/failpoints", |r| {
     696            0 :             request_span(r, move |r| async {
     697            0 :                 check_permission(&r, None)?;
     698            0 :                 let cancel = CancellationToken::new();
     699            0 :                 failpoints_handler(r, cancel).await
     700            0 :             })
     701            0 :         })
     702            0 :         .get("/v1/utilization", |r| request_span(r, utilization_handler))
     703            0 :         .delete("/v1/tenant/:tenant_id", |r| {
     704            0 :             request_span(r, tenant_delete_handler)
     705            0 :         })
     706            0 :         // Will be used in the future instead of implicit timeline creation
     707            0 :         .post("/v1/tenant/timeline", |r| {
     708            0 :             request_span(r, timeline_create_handler)
     709            0 :         })
     710            0 :         .get("/v1/tenant/timeline", |r| {
     711            0 :             request_span(r, timeline_list_handler)
     712            0 :         })
     713            0 :         .get("/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
     714            0 :             request_span(r, timeline_status_handler)
     715            0 :         })
     716            0 :         .delete("/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
     717            0 :             request_span(r, timeline_delete_handler)
     718            0 :         })
     719            0 :         .post("/v1/pull_timeline", |r| {
     720            0 :             request_span(r, timeline_pull_handler)
     721            0 :         })
     722            0 :         .put("/v1/tenant/:tenant_id/timeline/:timeline_id/exclude", |r| {
     723            0 :             request_span(r, timeline_exclude_handler)
     724            0 :         })
     725            0 :         .get(
     726            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/snapshot/:destination_id",
     727            0 :             |r| request_span(r, timeline_snapshot_handler),
     728            0 :         )
     729            0 :         .put(
     730            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/membership",
     731            0 :             |r| request_span(r, timeline_membership_handler),
     732            0 :         )
     733            0 :         .post(
     734            0 :             "/v1/tenant/:tenant_id/timeline/:source_timeline_id/copy",
     735            0 :             |r| request_span(r, timeline_copy_handler),
     736            0 :         )
     737            0 :         .patch(
     738            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/control_file",
     739            0 :             |r| request_span(r, patch_control_file_handler),
     740            0 :         )
     741            0 :         .post(
     742            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/checkpoint",
     743            0 :             |r| request_span(r, timeline_checkpoint_handler),
     744            0 :         )
     745            0 :         .get("/v1/tenant/:tenant_id/timeline/:timeline_id/digest", |r| {
     746            0 :             request_span(r, timeline_digest_handler)
     747            0 :         })
     748            0 :         .post(
     749            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/backup_partial_reset",
     750            0 :             |r| request_span(r, timeline_backup_partial_reset),
     751            0 :         )
     752            0 :         .post(
     753            0 :             "/v1/tenant/:tenant_id/timeline/:timeline_id/term_bump",
     754            0 :             |r| request_span(r, timeline_term_bump_handler),
     755            0 :         )
     756            0 :         .post("/v1/record_safekeeper_info/:tenant_id/:timeline_id", |r| {
     757            0 :             request_span(r, record_safekeeper_info)
     758            0 :         })
     759            0 :         .get("/v1/debug_dump", |r| request_span(r, dump_debug_handler))
     760            0 : }
     761              : 
     762              : #[cfg(test)]
     763              : mod tests {
     764              :     use super::*;
     765              : 
     766              :     #[test]
     767            1 :     fn test_term_switch_entry_api_serialize() {
     768            1 :         let state = AcceptorStateStatus {
     769            1 :             term: 1,
     770            1 :             epoch: 1,
     771            1 :             term_history: vec![TermSwitchApiEntry {
     772            1 :                 term: 1,
     773            1 :                 lsn: Lsn(0x16FFDDDD),
     774            1 :             }],
     775            1 :         };
     776            1 :         let json = serde_json::to_string(&state).unwrap();
     777            1 :         assert_eq!(
     778            1 :             json,
     779            1 :             "{\"term\":1,\"epoch\":1,\"term_history\":[{\"term\":1,\"lsn\":\"0/16FFDDDD\"}]}"
     780            1 :         );
     781            1 :     }
     782              : }
        

Generated by: LCOV version 2.1-beta