LCOV - code coverage report
Current view: top level - safekeeper/client/src - mgmt_api.rs (source / functions) Coverage Total Hit
Test: 1b0a6a0c05cee5a7de360813c8034804e105ce1c.info Lines: 0.0 % 111 0
Test Date: 2025-03-12 00:01:28 Functions: 0.0 % 44 0

            Line data    Source code
       1              : //! Safekeeper http client.
       2              : //!
       3              : //! Partially copied from pageserver client; some parts might be better to be
       4              : //! united.
       5              : 
       6              : use std::error::Error as _;
       7              : 
       8              : use http_utils::error::HttpErrorBody;
       9              : use reqwest::{IntoUrl, Method, StatusCode};
      10              : use safekeeper_api::models::{
      11              :     PullTimelineRequest, PullTimelineResponse, SafekeeperUtilization, TimelineCreateRequest,
      12              :     TimelineStatus,
      13              : };
      14              : use utils::id::{NodeId, TenantId, TimelineId};
      15              : use utils::logging::SecretString;
      16              : 
      17              : #[derive(Debug, Clone)]
      18              : pub struct Client {
      19              :     mgmt_api_endpoint: String,
      20              :     authorization_header: Option<SecretString>,
      21              :     client: reqwest::Client,
      22              : }
      23              : 
      24              : #[derive(thiserror::Error, Debug)]
      25              : pub enum Error {
      26              :     /// Failed to receive body (reqwest error).
      27            0 :     #[error("receive body: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
      28              :     ReceiveBody(reqwest::Error),
      29              : 
      30              :     /// Status is not ok, but failed to parse body as `HttpErrorBody`.
      31              :     #[error("receive error body: {0}")]
      32              :     ReceiveErrorBody(String),
      33              : 
      34              :     /// Status is not ok; parsed error in body as `HttpErrorBody`.
      35              :     #[error("safekeeper API: {1}")]
      36              :     ApiError(StatusCode, String),
      37              : 
      38              :     #[error("Cancelled")]
      39              :     Cancelled,
      40              : 
      41              :     /// Failed to create client.
      42            0 :     #[error("create client: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
      43              :     CreateClient(reqwest::Error),
      44              : }
      45              : 
      46              : pub type Result<T> = std::result::Result<T, Error>;
      47              : 
      48              : pub trait ResponseErrorMessageExt: Sized {
      49              :     fn error_from_body(self) -> impl std::future::Future<Output = Result<Self>> + Send;
      50              : }
      51              : 
      52              : /// If status is not ok, try to extract error message from the body.
      53              : impl ResponseErrorMessageExt for reqwest::Response {
      54            0 :     async fn error_from_body(self) -> Result<Self> {
      55            0 :         let status = self.status();
      56            0 :         if !(status.is_client_error() || status.is_server_error()) {
      57            0 :             return Ok(self);
      58            0 :         }
      59            0 : 
      60            0 :         let url = self.url().to_owned();
      61            0 :         Err(match self.json::<HttpErrorBody>().await {
      62            0 :             Ok(HttpErrorBody { msg }) => Error::ApiError(status, msg),
      63              :             Err(_) => {
      64            0 :                 Error::ReceiveErrorBody(format!("http error ({}) at {}.", status.as_u16(), url))
      65              :             }
      66              :         })
      67            0 :     }
      68              : }
      69              : 
      70              : impl Client {
      71            0 :     pub fn new(
      72            0 :         client: reqwest::Client,
      73            0 :         mgmt_api_endpoint: String,
      74            0 :         jwt: Option<SecretString>,
      75            0 :     ) -> Self {
      76            0 :         Self {
      77            0 :             mgmt_api_endpoint,
      78            0 :             authorization_header: jwt
      79            0 :                 .map(|jwt| SecretString::from(format!("Bearer {}", jwt.get_contents()))),
      80            0 :             client,
      81            0 :         }
      82            0 :     }
      83              : 
      84            0 :     pub async fn create_timeline(&self, req: &TimelineCreateRequest) -> Result<TimelineStatus> {
      85            0 :         let uri = format!(
      86            0 :             "{}/v1/tenant/{}/timeline/{}",
      87            0 :             self.mgmt_api_endpoint, req.tenant_id, req.timeline_id
      88            0 :         );
      89            0 :         let resp = self.post(&uri, req).await?;
      90            0 :         resp.json().await.map_err(Error::ReceiveBody)
      91            0 :     }
      92              : 
      93            0 :     pub async fn pull_timeline(&self, req: &PullTimelineRequest) -> Result<PullTimelineResponse> {
      94            0 :         let uri = format!("{}/v1/pull_timeline", self.mgmt_api_endpoint);
      95            0 :         let resp = self.post(&uri, req).await?;
      96            0 :         resp.json().await.map_err(Error::ReceiveBody)
      97            0 :     }
      98              : 
      99            0 :     pub async fn delete_timeline(
     100            0 :         &self,
     101            0 :         tenant_id: TenantId,
     102            0 :         timeline_id: TimelineId,
     103            0 :     ) -> Result<TimelineStatus> {
     104            0 :         let uri = format!(
     105            0 :             "{}/v1/tenant/{}/timeline/{}",
     106            0 :             self.mgmt_api_endpoint, tenant_id, timeline_id
     107            0 :         );
     108            0 :         let resp = self.request(Method::DELETE, &uri, ()).await?;
     109            0 :         resp.json().await.map_err(Error::ReceiveBody)
     110            0 :     }
     111              : 
     112            0 :     pub async fn timeline_status(
     113            0 :         &self,
     114            0 :         tenant_id: TenantId,
     115            0 :         timeline_id: TimelineId,
     116            0 :     ) -> Result<TimelineStatus> {
     117            0 :         let uri = format!(
     118            0 :             "{}/v1/tenant/{}/timeline/{}",
     119            0 :             self.mgmt_api_endpoint, tenant_id, timeline_id
     120            0 :         );
     121            0 :         let resp = self.get(&uri).await?;
     122            0 :         resp.json().await.map_err(Error::ReceiveBody)
     123            0 :     }
     124              : 
     125            0 :     pub async fn snapshot(
     126            0 :         &self,
     127            0 :         tenant_id: TenantId,
     128            0 :         timeline_id: TimelineId,
     129            0 :         stream_to: NodeId,
     130            0 :     ) -> Result<reqwest::Response> {
     131            0 :         let uri = format!(
     132            0 :             "{}/v1/tenant/{}/timeline/{}/snapshot/{}",
     133            0 :             self.mgmt_api_endpoint, tenant_id, timeline_id, stream_to.0
     134            0 :         );
     135            0 :         self.get(&uri).await
     136            0 :     }
     137              : 
     138            0 :     pub async fn utilization(&self) -> Result<SafekeeperUtilization> {
     139            0 :         let uri = format!("{}/v1/utilization", self.mgmt_api_endpoint);
     140            0 :         let resp = self.get(&uri).await?;
     141            0 :         resp.json().await.map_err(Error::ReceiveBody)
     142            0 :     }
     143              : 
     144            0 :     async fn post<B: serde::Serialize, U: IntoUrl>(
     145            0 :         &self,
     146            0 :         uri: U,
     147            0 :         body: B,
     148            0 :     ) -> Result<reqwest::Response> {
     149            0 :         self.request(Method::POST, uri, body).await
     150            0 :     }
     151              : 
     152            0 :     async fn get<U: IntoUrl>(&self, uri: U) -> Result<reqwest::Response> {
     153            0 :         self.request(Method::GET, uri, ()).await
     154            0 :     }
     155              : 
     156              :     /// Send the request and check that the status code is good.
     157            0 :     async fn request<B: serde::Serialize, U: reqwest::IntoUrl>(
     158            0 :         &self,
     159            0 :         method: Method,
     160            0 :         uri: U,
     161            0 :         body: B,
     162            0 :     ) -> Result<reqwest::Response> {
     163            0 :         let res = self.request_noerror(method, uri, body).await?;
     164            0 :         let response = res.error_from_body().await?;
     165            0 :         Ok(response)
     166            0 :     }
     167              : 
     168              :     /// Just send the request.
     169            0 :     async fn request_noerror<B: serde::Serialize, U: reqwest::IntoUrl>(
     170            0 :         &self,
     171            0 :         method: Method,
     172            0 :         uri: U,
     173            0 :         body: B,
     174            0 :     ) -> Result<reqwest::Response> {
     175            0 :         let mut req = self.client.request(method, uri);
     176            0 :         if let Some(value) = &self.authorization_header {
     177            0 :             req = req.header(reqwest::header::AUTHORIZATION, value.get_contents())
     178            0 :         }
     179            0 :         req.json(&body).send().await.map_err(Error::ReceiveBody)
     180            0 :     }
     181              : }
        

Generated by: LCOV version 2.1-beta