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 : self, 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 : #[error("request timed out: {0}")]
42 : Timeout(String),
43 : }
44 :
45 : pub type Result<T> = std::result::Result<T, Error>;
46 :
47 : pub trait ResponseErrorMessageExt: Sized {
48 : fn error_from_body(self) -> impl std::future::Future<Output = Result<Self>> + Send;
49 : }
50 :
51 : /// If status is not ok, try to extract error message from the body.
52 : impl ResponseErrorMessageExt for reqwest::Response {
53 0 : async fn error_from_body(self) -> Result<Self> {
54 0 : let status = self.status();
55 0 : if !(status.is_client_error() || status.is_server_error()) {
56 0 : return Ok(self);
57 0 : }
58 0 :
59 0 : let url = self.url().to_owned();
60 0 : Err(match self.json::<HttpErrorBody>().await {
61 0 : Ok(HttpErrorBody { msg }) => Error::ApiError(status, msg),
62 : Err(_) => {
63 0 : Error::ReceiveErrorBody(format!("http error ({}) at {}.", status.as_u16(), url))
64 : }
65 : })
66 0 : }
67 : }
68 :
69 : impl Client {
70 0 : pub fn new(
71 0 : client: reqwest::Client,
72 0 : mgmt_api_endpoint: String,
73 0 : jwt: Option<SecretString>,
74 0 : ) -> Self {
75 0 : Self {
76 0 : mgmt_api_endpoint,
77 0 : authorization_header: jwt
78 0 : .map(|jwt| SecretString::from(format!("Bearer {}", jwt.get_contents()))),
79 0 : client,
80 0 : }
81 0 : }
82 :
83 0 : pub async fn create_timeline(&self, req: &TimelineCreateRequest) -> Result<reqwest::Response> {
84 0 : let uri = format!("{}/v1/tenant/timeline", self.mgmt_api_endpoint);
85 0 : let resp = self.post(&uri, req).await?;
86 0 : Ok(resp)
87 0 : }
88 :
89 0 : pub async fn pull_timeline(&self, req: &PullTimelineRequest) -> Result<PullTimelineResponse> {
90 0 : let uri = format!("{}/v1/pull_timeline", self.mgmt_api_endpoint);
91 0 : let resp = self.post(&uri, req).await?;
92 0 : resp.json().await.map_err(Error::ReceiveBody)
93 0 : }
94 :
95 0 : pub async fn exclude_timeline(
96 0 : &self,
97 0 : tenant_id: TenantId,
98 0 : timeline_id: TimelineId,
99 0 : req: &models::TimelineMembershipSwitchRequest,
100 0 : ) -> Result<models::TimelineDeleteResult> {
101 0 : let uri = format!(
102 0 : "{}/v1/tenant/{}/timeline/{}/exclude",
103 0 : self.mgmt_api_endpoint, tenant_id, timeline_id
104 0 : );
105 0 : let resp = self.put(&uri, req).await?;
106 0 : resp.json().await.map_err(Error::ReceiveBody)
107 0 : }
108 :
109 0 : pub async fn delete_timeline(
110 0 : &self,
111 0 : tenant_id: TenantId,
112 0 : timeline_id: TimelineId,
113 0 : ) -> Result<models::TimelineDeleteResult> {
114 0 : let uri = format!(
115 0 : "{}/v1/tenant/{}/timeline/{}",
116 0 : self.mgmt_api_endpoint, tenant_id, timeline_id
117 0 : );
118 0 : let resp = self
119 0 : .request_maybe_body(Method::DELETE, &uri, None::<()>)
120 0 : .await?;
121 0 : resp.json().await.map_err(Error::ReceiveBody)
122 0 : }
123 :
124 0 : pub async fn delete_tenant(&self, tenant_id: TenantId) -> Result<models::TenantDeleteResult> {
125 0 : let uri = format!("{}/v1/tenant/{}", self.mgmt_api_endpoint, tenant_id);
126 0 : let resp = self
127 0 : .request_maybe_body(Method::DELETE, &uri, None::<()>)
128 0 : .await?;
129 0 : resp.json().await.map_err(Error::ReceiveBody)
130 0 : }
131 :
132 0 : pub async fn bump_timeline_term(
133 0 : &self,
134 0 : tenant_id: TenantId,
135 0 : timeline_id: TimelineId,
136 0 : req: &models::TimelineTermBumpRequest,
137 0 : ) -> Result<models::TimelineTermBumpResponse> {
138 0 : let uri = format!(
139 0 : "{}/v1/tenant/{}/timeline/{}/term_bump",
140 0 : self.mgmt_api_endpoint, tenant_id, timeline_id
141 0 : );
142 0 : let resp = self.post(&uri, req).await?;
143 0 : resp.json().await.map_err(Error::ReceiveBody)
144 0 : }
145 :
146 0 : pub async fn timeline_status(
147 0 : &self,
148 0 : tenant_id: TenantId,
149 0 : timeline_id: TimelineId,
150 0 : ) -> Result<TimelineStatus> {
151 0 : let uri = format!(
152 0 : "{}/v1/tenant/{}/timeline/{}",
153 0 : self.mgmt_api_endpoint, tenant_id, timeline_id
154 0 : );
155 0 : let resp = self.get(&uri).await?;
156 0 : resp.json().await.map_err(Error::ReceiveBody)
157 0 : }
158 :
159 0 : pub async fn snapshot(
160 0 : &self,
161 0 : tenant_id: TenantId,
162 0 : timeline_id: TimelineId,
163 0 : stream_to: NodeId,
164 0 : ) -> Result<reqwest::Response> {
165 0 : let uri = format!(
166 0 : "{}/v1/tenant/{}/timeline/{}/snapshot/{}",
167 0 : self.mgmt_api_endpoint, tenant_id, timeline_id, stream_to.0
168 0 : );
169 0 : self.get(&uri).await
170 0 : }
171 :
172 0 : pub async fn utilization(&self) -> Result<SafekeeperUtilization> {
173 0 : let uri = format!("{}/v1/utilization", self.mgmt_api_endpoint);
174 0 : let resp = self.get(&uri).await?;
175 0 : resp.json().await.map_err(Error::ReceiveBody)
176 0 : }
177 :
178 0 : async fn post<B: serde::Serialize, U: IntoUrl>(
179 0 : &self,
180 0 : uri: U,
181 0 : body: B,
182 0 : ) -> Result<reqwest::Response> {
183 0 : self.request(Method::POST, uri, body).await
184 0 : }
185 :
186 0 : async fn put<B: serde::Serialize, U: IntoUrl>(
187 0 : &self,
188 0 : uri: U,
189 0 : body: B,
190 0 : ) -> Result<reqwest::Response> {
191 0 : self.request(Method::PUT, uri, body).await
192 0 : }
193 :
194 0 : async fn get<U: IntoUrl>(&self, uri: U) -> Result<reqwest::Response> {
195 0 : self.request(Method::GET, uri, ()).await
196 0 : }
197 :
198 : /// Send the request and check that the status code is good.
199 0 : async fn request<B: serde::Serialize, U: reqwest::IntoUrl>(
200 0 : &self,
201 0 : method: Method,
202 0 : uri: U,
203 0 : body: B,
204 0 : ) -> Result<reqwest::Response> {
205 0 : self.request_maybe_body(method, uri, Some(body)).await
206 0 : }
207 :
208 : /// Send the request and check that the status code is good, with an optional body.
209 0 : async fn request_maybe_body<B: serde::Serialize, U: reqwest::IntoUrl>(
210 0 : &self,
211 0 : method: Method,
212 0 : uri: U,
213 0 : body: Option<B>,
214 0 : ) -> Result<reqwest::Response> {
215 0 : let res = self.request_noerror(method, uri, body).await?;
216 0 : let response = res.error_from_body().await?;
217 0 : Ok(response)
218 0 : }
219 :
220 : /// Just send the request.
221 0 : async fn request_noerror<B: serde::Serialize, U: reqwest::IntoUrl>(
222 0 : &self,
223 0 : method: Method,
224 0 : uri: U,
225 0 : body: Option<B>,
226 0 : ) -> Result<reqwest::Response> {
227 0 : let mut req = self.client.request(method, uri);
228 0 : if let Some(value) = &self.authorization_header {
229 0 : req = req.header(reqwest::header::AUTHORIZATION, value.get_contents())
230 0 : }
231 0 : if let Some(body) = body {
232 0 : req = req.json(&body);
233 0 : }
234 0 : req.send().await.map_err(Error::ReceiveBody)
235 0 : }
236 : }
|
