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