Line data Source code
1 : use crate::metrics::{
2 : HttpRequestLatencyLabelGroup, HttpRequestStatusLabelGroup, PageserverRequestLabelGroup,
3 : METRICS_REGISTRY,
4 : };
5 : use crate::reconciler::ReconcileError;
6 : use crate::service::{Service, STARTUP_RECONCILE_TIMEOUT};
7 : use anyhow::Context;
8 : use futures::Future;
9 : use hyper::header::CONTENT_TYPE;
10 : use hyper::{Body, Request, Response};
11 : use hyper::{StatusCode, Uri};
12 : use metrics::{BuildInfo, NeonMetrics};
13 : use pageserver_api::models::{
14 : TenantConfigRequest, TenantCreateRequest, TenantLocationConfigRequest, TenantShardSplitRequest,
15 : TenantTimeTravelRequest, TimelineCreateRequest,
16 : };
17 : use pageserver_api::shard::TenantShardId;
18 : use pageserver_client::mgmt_api;
19 : use std::sync::Arc;
20 : use std::time::{Duration, Instant};
21 : use tokio_util::sync::CancellationToken;
22 : use utils::auth::{Scope, SwappableJwtAuth};
23 : use utils::failpoint_support::failpoints_handler;
24 : use utils::http::endpoint::{auth_middleware, check_permission_with, request_span};
25 : use utils::http::request::{must_get_query_param, parse_query_param, parse_request_param};
26 : use utils::id::{TenantId, TimelineId};
27 :
28 : use utils::{
29 : http::{
30 : endpoint::{self},
31 : error::ApiError,
32 : json::{json_request, json_response},
33 : RequestExt, RouterBuilder,
34 : },
35 : id::NodeId,
36 : };
37 :
38 : use pageserver_api::controller_api::{
39 : NodeAvailability, NodeConfigureRequest, NodeRegisterRequest, TenantPolicyRequest,
40 : TenantShardMigrateRequest,
41 : };
42 : use pageserver_api::upcall_api::{ReAttachRequest, ValidateRequest};
43 :
44 : use control_plane::storage_controller::{AttachHookRequest, InspectRequest};
45 :
46 : use routerify::Middleware;
47 :
48 : /// State available to HTTP request handlers
49 : pub struct HttpState {
50 : service: Arc<crate::service::Service>,
51 : auth: Option<Arc<SwappableJwtAuth>>,
52 : neon_metrics: NeonMetrics,
53 : allowlist_routes: Vec<Uri>,
54 : }
55 :
56 : impl HttpState {
57 0 : pub fn new(
58 0 : service: Arc<crate::service::Service>,
59 0 : auth: Option<Arc<SwappableJwtAuth>>,
60 0 : build_info: BuildInfo,
61 0 : ) -> Self {
62 0 : let allowlist_routes = ["/status", "/ready", "/metrics"]
63 0 : .iter()
64 0 : .map(|v| v.parse().unwrap())
65 0 : .collect::<Vec<_>>();
66 0 : Self {
67 0 : service,
68 0 : auth,
69 0 : neon_metrics: NeonMetrics::new(build_info),
70 0 : allowlist_routes,
71 0 : }
72 0 : }
73 : }
74 :
75 : #[inline(always)]
76 0 : fn get_state(request: &Request<Body>) -> &HttpState {
77 0 : request
78 0 : .data::<Arc<HttpState>>()
79 0 : .expect("unknown state type")
80 0 : .as_ref()
81 0 : }
82 :
83 : /// Pageserver calls into this on startup, to learn which tenants it should attach
84 0 : async fn handle_re_attach(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
85 0 : check_permissions(&req, Scope::GenerationsApi)?;
86 :
87 0 : let reattach_req = json_request::<ReAttachRequest>(&mut req).await?;
88 0 : let state = get_state(&req);
89 0 : json_response(StatusCode::OK, state.service.re_attach(reattach_req).await?)
90 0 : }
91 :
92 : /// Pageserver calls into this before doing deletions, to confirm that it still
93 : /// holds the latest generation for the tenants with deletions enqueued
94 0 : async fn handle_validate(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
95 0 : check_permissions(&req, Scope::GenerationsApi)?;
96 :
97 0 : let validate_req = json_request::<ValidateRequest>(&mut req).await?;
98 0 : let state = get_state(&req);
99 0 : json_response(StatusCode::OK, state.service.validate(validate_req))
100 0 : }
101 :
102 : /// Call into this before attaching a tenant to a pageserver, to acquire a generation number
103 : /// (in the real control plane this is unnecessary, because the same program is managing
104 : /// generation numbers and doing attachments).
105 0 : async fn handle_attach_hook(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
106 0 : check_permissions(&req, Scope::Admin)?;
107 :
108 0 : let attach_req = json_request::<AttachHookRequest>(&mut req).await?;
109 0 : let state = get_state(&req);
110 0 :
111 0 : json_response(
112 0 : StatusCode::OK,
113 0 : state
114 0 : .service
115 0 : .attach_hook(attach_req)
116 0 : .await
117 0 : .map_err(ApiError::InternalServerError)?,
118 : )
119 0 : }
120 :
121 0 : async fn handle_inspect(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
122 0 : check_permissions(&req, Scope::Admin)?;
123 :
124 0 : let inspect_req = json_request::<InspectRequest>(&mut req).await?;
125 :
126 0 : let state = get_state(&req);
127 0 :
128 0 : json_response(StatusCode::OK, state.service.inspect(inspect_req))
129 0 : }
130 :
131 0 : async fn handle_tenant_create(
132 0 : service: Arc<Service>,
133 0 : mut req: Request<Body>,
134 0 : ) -> Result<Response<Body>, ApiError> {
135 0 : check_permissions(&req, Scope::PageServerApi)?;
136 :
137 0 : let create_req = json_request::<TenantCreateRequest>(&mut req).await?;
138 :
139 : json_response(
140 : StatusCode::CREATED,
141 0 : service.tenant_create(create_req).await?,
142 : )
143 0 : }
144 :
145 0 : async fn handle_tenant_location_config(
146 0 : service: Arc<Service>,
147 0 : mut req: Request<Body>,
148 0 : ) -> Result<Response<Body>, ApiError> {
149 0 : let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
150 0 : check_permissions(&req, Scope::PageServerApi)?;
151 :
152 0 : let config_req = json_request::<TenantLocationConfigRequest>(&mut req).await?;
153 : json_response(
154 : StatusCode::OK,
155 0 : service
156 0 : .tenant_location_config(tenant_shard_id, config_req)
157 0 : .await?,
158 : )
159 0 : }
160 :
161 0 : async fn handle_tenant_config_set(
162 0 : service: Arc<Service>,
163 0 : mut req: Request<Body>,
164 0 : ) -> Result<Response<Body>, ApiError> {
165 0 : check_permissions(&req, Scope::PageServerApi)?;
166 :
167 0 : let config_req = json_request::<TenantConfigRequest>(&mut req).await?;
168 :
169 0 : json_response(StatusCode::OK, service.tenant_config_set(config_req).await?)
170 0 : }
171 :
172 0 : async fn handle_tenant_config_get(
173 0 : service: Arc<Service>,
174 0 : req: Request<Body>,
175 0 : ) -> Result<Response<Body>, ApiError> {
176 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
177 0 : check_permissions(&req, Scope::PageServerApi)?;
178 :
179 0 : json_response(StatusCode::OK, service.tenant_config_get(tenant_id)?)
180 0 : }
181 :
182 0 : async fn handle_tenant_time_travel_remote_storage(
183 0 : service: Arc<Service>,
184 0 : mut req: Request<Body>,
185 0 : ) -> Result<Response<Body>, ApiError> {
186 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
187 0 : check_permissions(&req, Scope::PageServerApi)?;
188 :
189 0 : let time_travel_req = json_request::<TenantTimeTravelRequest>(&mut req).await?;
190 :
191 0 : let timestamp_raw = must_get_query_param(&req, "travel_to")?;
192 0 : let _timestamp = humantime::parse_rfc3339(×tamp_raw).map_err(|_e| {
193 0 : ApiError::BadRequest(anyhow::anyhow!(
194 0 : "Invalid time for travel_to: {timestamp_raw:?}"
195 0 : ))
196 0 : })?;
197 :
198 0 : let done_if_after_raw = must_get_query_param(&req, "done_if_after")?;
199 0 : let _done_if_after = humantime::parse_rfc3339(&done_if_after_raw).map_err(|_e| {
200 0 : ApiError::BadRequest(anyhow::anyhow!(
201 0 : "Invalid time for done_if_after: {done_if_after_raw:?}"
202 0 : ))
203 0 : })?;
204 :
205 0 : service
206 0 : .tenant_time_travel_remote_storage(
207 0 : &time_travel_req,
208 0 : tenant_id,
209 0 : timestamp_raw,
210 0 : done_if_after_raw,
211 0 : )
212 0 : .await?;
213 0 : json_response(StatusCode::OK, ())
214 0 : }
215 :
216 0 : fn map_reqwest_hyper_status(status: reqwest::StatusCode) -> Result<hyper::StatusCode, ApiError> {
217 0 : hyper::StatusCode::from_u16(status.as_u16())
218 0 : .context("invalid status code")
219 0 : .map_err(ApiError::InternalServerError)
220 0 : }
221 :
222 0 : async fn handle_tenant_secondary_download(
223 0 : service: Arc<Service>,
224 0 : req: Request<Body>,
225 0 : ) -> Result<Response<Body>, ApiError> {
226 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
227 0 : let wait = parse_query_param(&req, "wait_ms")?.map(Duration::from_millis);
228 :
229 0 : let (status, progress) = service.tenant_secondary_download(tenant_id, wait).await?;
230 0 : json_response(map_reqwest_hyper_status(status)?, progress)
231 0 : }
232 :
233 0 : async fn handle_tenant_delete(
234 0 : service: Arc<Service>,
235 0 : req: Request<Body>,
236 0 : ) -> Result<Response<Body>, ApiError> {
237 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
238 0 : check_permissions(&req, Scope::PageServerApi)?;
239 :
240 0 : let status_code = service
241 0 : .tenant_delete(tenant_id)
242 0 : .await
243 0 : .and_then(map_reqwest_hyper_status)?;
244 :
245 0 : if status_code == StatusCode::NOT_FOUND {
246 : // The pageserver uses 404 for successful deletion, but we use 200
247 0 : json_response(StatusCode::OK, ())
248 : } else {
249 0 : json_response(status_code, ())
250 : }
251 0 : }
252 :
253 0 : async fn handle_tenant_timeline_create(
254 0 : service: Arc<Service>,
255 0 : mut req: Request<Body>,
256 0 : ) -> Result<Response<Body>, ApiError> {
257 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
258 0 : check_permissions(&req, Scope::PageServerApi)?;
259 :
260 0 : let create_req = json_request::<TimelineCreateRequest>(&mut req).await?;
261 : json_response(
262 : StatusCode::CREATED,
263 0 : service
264 0 : .tenant_timeline_create(tenant_id, create_req)
265 0 : .await?,
266 : )
267 0 : }
268 :
269 0 : async fn handle_tenant_timeline_delete(
270 0 : service: Arc<Service>,
271 0 : req: Request<Body>,
272 0 : ) -> Result<Response<Body>, ApiError> {
273 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
274 0 : check_permissions(&req, Scope::PageServerApi)?;
275 :
276 0 : let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
277 :
278 : // For timeline deletions, which both implement an "initially return 202, then 404 once
279 : // we're done" semantic, we wrap with a retry loop to expose a simpler API upstream.
280 0 : async fn deletion_wrapper<R, F>(service: Arc<Service>, f: F) -> Result<Response<Body>, ApiError>
281 0 : where
282 0 : R: std::future::Future<Output = Result<StatusCode, ApiError>> + Send + 'static,
283 0 : F: Fn(Arc<Service>) -> R + Send + Sync + 'static,
284 0 : {
285 0 : let started_at = Instant::now();
286 0 : // To keep deletion reasonably snappy for small tenants, initially check after 1 second if deletion
287 0 : // completed.
288 0 : let mut retry_period = Duration::from_secs(1);
289 0 : // On subsequent retries, wait longer.
290 0 : let max_retry_period = Duration::from_secs(5);
291 0 : // Enable callers with a 30 second request timeout to reliably get a response
292 0 : let max_wait = Duration::from_secs(25);
293 :
294 : loop {
295 0 : let status = f(service.clone()).await?;
296 0 : match status {
297 : StatusCode::ACCEPTED => {
298 0 : tracing::info!("Deletion accepted, waiting to try again...");
299 0 : tokio::time::sleep(retry_period).await;
300 0 : retry_period = max_retry_period;
301 : }
302 : StatusCode::NOT_FOUND => {
303 0 : tracing::info!("Deletion complete");
304 0 : return json_response(StatusCode::OK, ());
305 : }
306 : _ => {
307 0 : tracing::warn!("Unexpected status {status}");
308 0 : return json_response(status, ());
309 : }
310 : }
311 :
312 0 : let now = Instant::now();
313 0 : if now + retry_period > started_at + max_wait {
314 0 : tracing::info!("Deletion timed out waiting for 404");
315 : // REQUEST_TIMEOUT would be more appropriate, but CONFLICT is already part of
316 : // the pageserver's swagger definition for this endpoint, and has the same desired
317 : // effect of causing the control plane to retry later.
318 0 : return json_response(StatusCode::CONFLICT, ());
319 0 : }
320 : }
321 0 : }
322 :
323 0 : deletion_wrapper(service, move |service| async move {
324 0 : service
325 0 : .tenant_timeline_delete(tenant_id, timeline_id)
326 0 : .await
327 0 : .and_then(map_reqwest_hyper_status)
328 0 : })
329 0 : .await
330 0 : }
331 :
332 0 : async fn handle_tenant_timeline_passthrough(
333 0 : service: Arc<Service>,
334 0 : req: Request<Body>,
335 0 : ) -> Result<Response<Body>, ApiError> {
336 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
337 0 : check_permissions(&req, Scope::PageServerApi)?;
338 :
339 0 : let Some(path) = req.uri().path_and_query() else {
340 : // This should never happen, our request router only calls us if there is a path
341 0 : return Err(ApiError::BadRequest(anyhow::anyhow!("Missing path")));
342 : };
343 :
344 0 : tracing::info!("Proxying request for tenant {} ({})", tenant_id, path);
345 :
346 : // Find the node that holds shard zero
347 0 : let (node, tenant_shard_id) = service.tenant_shard0_node(tenant_id)?;
348 :
349 : // Callers will always pass an unsharded tenant ID. Before proxying, we must
350 : // rewrite this to a shard-aware shard zero ID.
351 0 : let path = format!("{}", path);
352 0 : let tenant_str = tenant_id.to_string();
353 0 : let tenant_shard_str = format!("{}", tenant_shard_id);
354 0 : let path = path.replace(&tenant_str, &tenant_shard_str);
355 0 :
356 0 : let latency = &METRICS_REGISTRY
357 0 : .metrics_group
358 0 : .storage_controller_passthrough_request_latency;
359 0 :
360 0 : // This is a bit awkward. We remove the param from the request
361 0 : // and join the words by '_' to get a label for the request.
362 0 : let just_path = path.replace(&tenant_shard_str, "");
363 0 : let path_label = just_path
364 0 : .split('/')
365 0 : .filter(|token| !token.is_empty())
366 0 : .collect::<Vec<_>>()
367 0 : .join("_");
368 0 : let labels = PageserverRequestLabelGroup {
369 0 : pageserver_id: &node.get_id().to_string(),
370 0 : path: &path_label,
371 0 : method: crate::metrics::Method::Get,
372 0 : };
373 0 :
374 0 : let _timer = latency.start_timer(labels.clone());
375 0 :
376 0 : let client = mgmt_api::Client::new(node.base_url(), service.get_config().jwt_token.as_deref());
377 0 : let resp = client.get_raw(path).await.map_err(|_e|
378 : // FIXME: give APiError a proper Unavailable variant. We return 503 here because
379 : // if we can't successfully send a request to the pageserver, we aren't available.
380 0 : ApiError::ShuttingDown)?;
381 :
382 0 : if !resp.status().is_success() {
383 0 : let error_counter = &METRICS_REGISTRY
384 0 : .metrics_group
385 0 : .storage_controller_passthrough_request_error;
386 0 : error_counter.inc(labels);
387 0 : }
388 :
389 : // We have a reqest::Response, would like a http::Response
390 0 : let mut builder = hyper::Response::builder().status(map_reqwest_hyper_status(resp.status())?);
391 0 : for (k, v) in resp.headers() {
392 0 : builder = builder.header(k.as_str(), v.as_bytes());
393 0 : }
394 :
395 0 : let response = builder
396 0 : .body(Body::wrap_stream(resp.bytes_stream()))
397 0 : .map_err(|e| ApiError::InternalServerError(e.into()))?;
398 :
399 0 : Ok(response)
400 0 : }
401 :
402 0 : async fn handle_tenant_locate(
403 0 : service: Arc<Service>,
404 0 : req: Request<Body>,
405 0 : ) -> Result<Response<Body>, ApiError> {
406 0 : check_permissions(&req, Scope::Admin)?;
407 :
408 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
409 0 : json_response(StatusCode::OK, service.tenant_locate(tenant_id)?)
410 0 : }
411 :
412 0 : async fn handle_tenant_describe(
413 0 : service: Arc<Service>,
414 0 : req: Request<Body>,
415 0 : ) -> Result<Response<Body>, ApiError> {
416 0 : check_permissions(&req, Scope::Admin)?;
417 :
418 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
419 0 : json_response(StatusCode::OK, service.tenant_describe(tenant_id)?)
420 0 : }
421 :
422 0 : async fn handle_tenant_list(
423 0 : service: Arc<Service>,
424 0 : req: Request<Body>,
425 0 : ) -> Result<Response<Body>, ApiError> {
426 0 : check_permissions(&req, Scope::Admin)?;
427 :
428 0 : json_response(StatusCode::OK, service.tenant_list())
429 0 : }
430 :
431 0 : async fn handle_node_register(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
432 0 : check_permissions(&req, Scope::Admin)?;
433 :
434 0 : let register_req = json_request::<NodeRegisterRequest>(&mut req).await?;
435 0 : let state = get_state(&req);
436 0 : state.service.node_register(register_req).await?;
437 0 : json_response(StatusCode::OK, ())
438 0 : }
439 :
440 0 : async fn handle_node_list(req: Request<Body>) -> Result<Response<Body>, ApiError> {
441 0 : check_permissions(&req, Scope::Admin)?;
442 :
443 0 : let state = get_state(&req);
444 0 : let nodes = state.service.node_list().await?;
445 0 : let api_nodes = nodes.into_iter().map(|n| n.describe()).collect::<Vec<_>>();
446 0 :
447 0 : json_response(StatusCode::OK, api_nodes)
448 0 : }
449 :
450 0 : async fn handle_node_drop(req: Request<Body>) -> Result<Response<Body>, ApiError> {
451 0 : check_permissions(&req, Scope::Admin)?;
452 :
453 0 : let state = get_state(&req);
454 0 : let node_id: NodeId = parse_request_param(&req, "node_id")?;
455 0 : json_response(StatusCode::OK, state.service.node_drop(node_id).await?)
456 0 : }
457 :
458 0 : async fn handle_node_configure(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
459 0 : check_permissions(&req, Scope::Admin)?;
460 :
461 0 : let node_id: NodeId = parse_request_param(&req, "node_id")?;
462 0 : let config_req = json_request::<NodeConfigureRequest>(&mut req).await?;
463 0 : if node_id != config_req.node_id {
464 0 : return Err(ApiError::BadRequest(anyhow::anyhow!(
465 0 : "Path and body node_id differ"
466 0 : )));
467 0 : }
468 0 : let state = get_state(&req);
469 0 :
470 0 : json_response(
471 0 : StatusCode::OK,
472 0 : state
473 0 : .service
474 0 : .node_configure(
475 0 : config_req.node_id,
476 0 : config_req.availability.map(NodeAvailability::from),
477 0 : config_req.scheduling,
478 0 : )
479 0 : .await?,
480 : )
481 0 : }
482 :
483 0 : async fn handle_node_status(req: Request<Body>) -> Result<Response<Body>, ApiError> {
484 0 : check_permissions(&req, Scope::Admin)?;
485 :
486 0 : let state = get_state(&req);
487 0 : let node_id: NodeId = parse_request_param(&req, "node_id")?;
488 :
489 0 : let node_status = state.service.get_node(node_id).await?;
490 :
491 0 : json_response(StatusCode::OK, node_status)
492 0 : }
493 :
494 0 : async fn handle_node_drain(req: Request<Body>) -> Result<Response<Body>, ApiError> {
495 0 : check_permissions(&req, Scope::Admin)?;
496 :
497 0 : let state = get_state(&req);
498 0 : let node_id: NodeId = parse_request_param(&req, "node_id")?;
499 :
500 0 : state.service.start_node_drain(node_id).await?;
501 :
502 0 : json_response(StatusCode::ACCEPTED, ())
503 0 : }
504 :
505 0 : async fn handle_cancel_node_drain(req: Request<Body>) -> Result<Response<Body>, ApiError> {
506 0 : check_permissions(&req, Scope::Admin)?;
507 :
508 0 : let state = get_state(&req);
509 0 : let node_id: NodeId = parse_request_param(&req, "node_id")?;
510 :
511 0 : state.service.cancel_node_drain(node_id).await?;
512 :
513 0 : json_response(StatusCode::ACCEPTED, ())
514 0 : }
515 :
516 0 : async fn handle_node_fill(req: Request<Body>) -> Result<Response<Body>, ApiError> {
517 0 : check_permissions(&req, Scope::Admin)?;
518 :
519 0 : let state = get_state(&req);
520 0 : let node_id: NodeId = parse_request_param(&req, "node_id")?;
521 :
522 0 : state.service.start_node_fill(node_id).await?;
523 :
524 0 : json_response(StatusCode::ACCEPTED, ())
525 0 : }
526 :
527 0 : async fn handle_cancel_node_fill(req: Request<Body>) -> Result<Response<Body>, ApiError> {
528 0 : check_permissions(&req, Scope::Admin)?;
529 :
530 0 : let state = get_state(&req);
531 0 : let node_id: NodeId = parse_request_param(&req, "node_id")?;
532 :
533 0 : state.service.cancel_node_fill(node_id).await?;
534 :
535 0 : json_response(StatusCode::ACCEPTED, ())
536 0 : }
537 :
538 0 : async fn handle_tenant_shard_split(
539 0 : service: Arc<Service>,
540 0 : mut req: Request<Body>,
541 0 : ) -> Result<Response<Body>, ApiError> {
542 0 : check_permissions(&req, Scope::Admin)?;
543 :
544 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
545 0 : let split_req = json_request::<TenantShardSplitRequest>(&mut req).await?;
546 :
547 : json_response(
548 : StatusCode::OK,
549 0 : service.tenant_shard_split(tenant_id, split_req).await?,
550 : )
551 0 : }
552 :
553 0 : async fn handle_tenant_shard_migrate(
554 0 : service: Arc<Service>,
555 0 : mut req: Request<Body>,
556 0 : ) -> Result<Response<Body>, ApiError> {
557 0 : check_permissions(&req, Scope::Admin)?;
558 :
559 0 : let tenant_shard_id: TenantShardId = parse_request_param(&req, "tenant_shard_id")?;
560 0 : let migrate_req = json_request::<TenantShardMigrateRequest>(&mut req).await?;
561 : json_response(
562 : StatusCode::OK,
563 0 : service
564 0 : .tenant_shard_migrate(tenant_shard_id, migrate_req)
565 0 : .await?,
566 : )
567 0 : }
568 :
569 0 : async fn handle_tenant_update_policy(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
570 0 : check_permissions(&req, Scope::Admin)?;
571 :
572 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
573 0 : let update_req = json_request::<TenantPolicyRequest>(&mut req).await?;
574 0 : let state = get_state(&req);
575 0 :
576 0 : json_response(
577 0 : StatusCode::OK,
578 0 : state
579 0 : .service
580 0 : .tenant_update_policy(tenant_id, update_req)
581 0 : .await?,
582 : )
583 0 : }
584 :
585 0 : async fn handle_tenant_drop(req: Request<Body>) -> Result<Response<Body>, ApiError> {
586 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
587 0 : check_permissions(&req, Scope::PageServerApi)?;
588 :
589 0 : let state = get_state(&req);
590 0 :
591 0 : json_response(StatusCode::OK, state.service.tenant_drop(tenant_id).await?)
592 0 : }
593 :
594 0 : async fn handle_tenant_import(req: Request<Body>) -> Result<Response<Body>, ApiError> {
595 0 : let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
596 0 : check_permissions(&req, Scope::PageServerApi)?;
597 :
598 0 : let state = get_state(&req);
599 0 :
600 0 : json_response(
601 0 : StatusCode::OK,
602 0 : state.service.tenant_import(tenant_id).await?,
603 : )
604 0 : }
605 :
606 0 : async fn handle_tenants_dump(req: Request<Body>) -> Result<Response<Body>, ApiError> {
607 0 : check_permissions(&req, Scope::Admin)?;
608 :
609 0 : let state = get_state(&req);
610 0 : state.service.tenants_dump()
611 0 : }
612 :
613 0 : async fn handle_scheduler_dump(req: Request<Body>) -> Result<Response<Body>, ApiError> {
614 0 : check_permissions(&req, Scope::Admin)?;
615 :
616 0 : let state = get_state(&req);
617 0 : state.service.scheduler_dump()
618 0 : }
619 :
620 0 : async fn handle_consistency_check(req: Request<Body>) -> Result<Response<Body>, ApiError> {
621 0 : check_permissions(&req, Scope::Admin)?;
622 :
623 0 : let state = get_state(&req);
624 0 :
625 0 : json_response(StatusCode::OK, state.service.consistency_check().await?)
626 0 : }
627 :
628 0 : async fn handle_reconcile_all(req: Request<Body>) -> Result<Response<Body>, ApiError> {
629 0 : check_permissions(&req, Scope::Admin)?;
630 :
631 0 : let state = get_state(&req);
632 0 :
633 0 : json_response(StatusCode::OK, state.service.reconcile_all_now().await?)
634 0 : }
635 :
636 : /// Status endpoint is just used for checking that our HTTP listener is up
637 0 : async fn handle_status(_req: Request<Body>) -> Result<Response<Body>, ApiError> {
638 0 : json_response(StatusCode::OK, ())
639 0 : }
640 :
641 : /// Readiness endpoint indicates when we're done doing startup I/O (e.g. reconciling
642 : /// with remote pageserver nodes). This is intended for use as a kubernetes readiness probe.
643 0 : async fn handle_ready(req: Request<Body>) -> Result<Response<Body>, ApiError> {
644 0 : let state = get_state(&req);
645 0 : if state.service.startup_complete.is_ready() {
646 0 : json_response(StatusCode::OK, ())
647 : } else {
648 0 : json_response(StatusCode::SERVICE_UNAVAILABLE, ())
649 : }
650 0 : }
651 :
652 : impl From<ReconcileError> for ApiError {
653 0 : fn from(value: ReconcileError) -> Self {
654 0 : ApiError::Conflict(format!("Reconciliation error: {}", value))
655 0 : }
656 : }
657 :
658 : /// Common wrapper for request handlers that call into Service and will operate on tenants: they must only
659 : /// be allowed to run if Service has finished its initial reconciliation.
660 0 : async fn tenant_service_handler<R, H>(
661 0 : request: Request<Body>,
662 0 : handler: H,
663 0 : request_name: RequestName,
664 0 : ) -> R::Output
665 0 : where
666 0 : R: std::future::Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
667 0 : H: FnOnce(Arc<Service>, Request<Body>) -> R + Send + Sync + 'static,
668 0 : {
669 0 : let state = get_state(&request);
670 0 : let service = state.service.clone();
671 0 :
672 0 : let startup_complete = service.startup_complete.clone();
673 0 : if tokio::time::timeout(STARTUP_RECONCILE_TIMEOUT, startup_complete.wait())
674 0 : .await
675 0 : .is_err()
676 : {
677 : // This shouldn't happen: it is the responsibilty of [`Service::startup_reconcile`] to use appropriate
678 : // timeouts around its remote calls, to bound its runtime.
679 0 : return Err(ApiError::Timeout(
680 0 : "Timed out waiting for service readiness".into(),
681 0 : ));
682 0 : }
683 0 :
684 0 : named_request_span(
685 0 : request,
686 0 : |request| async move { handler(service, request).await },
687 0 : request_name,
688 0 : )
689 0 : .await
690 0 : }
691 :
692 : /// Check if the required scope is held in the request's token, or if the request has
693 : /// a token with 'admin' scope then always permit it.
694 0 : fn check_permissions(request: &Request<Body>, required_scope: Scope) -> Result<(), ApiError> {
695 0 : check_permission_with(request, |claims| {
696 0 : match crate::auth::check_permission(claims, required_scope) {
697 0 : Err(e) => match crate::auth::check_permission(claims, Scope::Admin) {
698 0 : Ok(()) => Ok(()),
699 0 : Err(_) => Err(e),
700 : },
701 0 : Ok(()) => Ok(()),
702 : }
703 0 : })
704 0 : }
705 :
706 : #[derive(Clone, Debug)]
707 : struct RequestMeta {
708 : method: hyper::http::Method,
709 : at: Instant,
710 : }
711 :
712 0 : fn prologue_metrics_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>(
713 0 : ) -> Middleware<B, ApiError> {
714 0 : Middleware::pre(move |req| async move {
715 0 : let meta = RequestMeta {
716 0 : method: req.method().clone(),
717 0 : at: Instant::now(),
718 0 : };
719 0 :
720 0 : req.set_context(meta);
721 0 :
722 0 : Ok(req)
723 0 : })
724 0 : }
725 :
726 0 : fn epilogue_metrics_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>(
727 0 : ) -> Middleware<B, ApiError> {
728 0 : Middleware::post_with_info(move |resp, req_info| async move {
729 0 : let request_name = match req_info.context::<RequestName>() {
730 0 : Some(name) => name,
731 0 : None => {
732 0 : return Ok(resp);
733 0 : }
734 0 : };
735 0 :
736 0 : if let Some(meta) = req_info.context::<RequestMeta>() {
737 0 : let status = &crate::metrics::METRICS_REGISTRY
738 0 : .metrics_group
739 0 : .storage_controller_http_request_status;
740 0 : let latency = &crate::metrics::METRICS_REGISTRY
741 0 : .metrics_group
742 0 : .storage_controller_http_request_latency;
743 0 :
744 0 : status.inc(HttpRequestStatusLabelGroup {
745 0 : path: request_name.0,
746 0 : method: meta.method.clone().into(),
747 0 : status: crate::metrics::StatusCode(resp.status()),
748 0 : });
749 0 :
750 0 : latency.observe(
751 0 : HttpRequestLatencyLabelGroup {
752 0 : path: request_name.0,
753 0 : method: meta.method.into(),
754 0 : },
755 0 : meta.at.elapsed().as_secs_f64(),
756 0 : );
757 0 : }
758 0 : Ok(resp)
759 0 : })
760 0 : }
761 :
762 0 : pub async fn measured_metrics_handler(req: Request<Body>) -> Result<Response<Body>, ApiError> {
763 0 : pub const TEXT_FORMAT: &str = "text/plain; version=0.0.4";
764 0 :
765 0 : let state = get_state(&req);
766 0 : let payload = crate::metrics::METRICS_REGISTRY.encode(&state.neon_metrics);
767 0 : let response = Response::builder()
768 0 : .status(200)
769 0 : .header(CONTENT_TYPE, TEXT_FORMAT)
770 0 : .body(payload.into())
771 0 : .unwrap();
772 0 :
773 0 : Ok(response)
774 0 : }
775 :
776 : #[derive(Clone)]
777 : struct RequestName(&'static str);
778 :
779 0 : async fn named_request_span<R, H>(
780 0 : request: Request<Body>,
781 0 : handler: H,
782 0 : name: RequestName,
783 0 : ) -> R::Output
784 0 : where
785 0 : R: Future<Output = Result<Response<Body>, ApiError>> + Send + 'static,
786 0 : H: FnOnce(Request<Body>) -> R + Send + Sync + 'static,
787 0 : {
788 0 : request.set_context(name);
789 0 : request_span(request, handler).await
790 0 : }
791 :
792 0 : pub fn make_router(
793 0 : service: Arc<Service>,
794 0 : auth: Option<Arc<SwappableJwtAuth>>,
795 0 : build_info: BuildInfo,
796 0 : ) -> RouterBuilder<hyper::Body, ApiError> {
797 0 : let mut router = endpoint::make_router()
798 0 : .middleware(prologue_metrics_middleware())
799 0 : .middleware(epilogue_metrics_middleware());
800 0 : if auth.is_some() {
801 0 : router = router.middleware(auth_middleware(|request| {
802 0 : let state = get_state(request);
803 0 : if state.allowlist_routes.contains(request.uri()) {
804 0 : None
805 : } else {
806 0 : state.auth.as_deref()
807 : }
808 0 : }));
809 0 : }
810 :
811 0 : router
812 0 : .data(Arc::new(HttpState::new(service, auth, build_info)))
813 0 : .get("/metrics", |r| {
814 0 : named_request_span(r, measured_metrics_handler, RequestName("metrics"))
815 0 : })
816 0 : // Non-prefixed generic endpoints (status, metrics)
817 0 : .get("/status", |r| {
818 0 : named_request_span(r, handle_status, RequestName("status"))
819 0 : })
820 0 : .get("/ready", |r| {
821 0 : named_request_span(r, handle_ready, RequestName("ready"))
822 0 : })
823 0 : // Upcalls for the pageserver: point the pageserver's `control_plane_api` config to this prefix
824 0 : .post("/upcall/v1/re-attach", |r| {
825 0 : named_request_span(r, handle_re_attach, RequestName("upcall_v1_reattach"))
826 0 : })
827 0 : .post("/upcall/v1/validate", |r| {
828 0 : named_request_span(r, handle_validate, RequestName("upcall_v1_validate"))
829 0 : })
830 0 : // Test/dev/debug endpoints
831 0 : .post("/debug/v1/attach-hook", |r| {
832 0 : named_request_span(r, handle_attach_hook, RequestName("debug_v1_attach_hook"))
833 0 : })
834 0 : .post("/debug/v1/inspect", |r| {
835 0 : named_request_span(r, handle_inspect, RequestName("debug_v1_inspect"))
836 0 : })
837 0 : .post("/debug/v1/tenant/:tenant_id/drop", |r| {
838 0 : named_request_span(r, handle_tenant_drop, RequestName("debug_v1_tenant_drop"))
839 0 : })
840 0 : .post("/debug/v1/node/:node_id/drop", |r| {
841 0 : named_request_span(r, handle_node_drop, RequestName("debug_v1_node_drop"))
842 0 : })
843 0 : .post("/debug/v1/tenant/:tenant_id/import", |r| {
844 0 : named_request_span(
845 0 : r,
846 0 : handle_tenant_import,
847 0 : RequestName("debug_v1_tenant_import"),
848 0 : )
849 0 : })
850 0 : .get("/debug/v1/tenant", |r| {
851 0 : named_request_span(r, handle_tenants_dump, RequestName("debug_v1_tenant"))
852 0 : })
853 0 : .get("/debug/v1/tenant/:tenant_id/locate", |r| {
854 0 : tenant_service_handler(
855 0 : r,
856 0 : handle_tenant_locate,
857 0 : RequestName("debug_v1_tenant_locate"),
858 0 : )
859 0 : })
860 0 : .get("/debug/v1/scheduler", |r| {
861 0 : named_request_span(r, handle_scheduler_dump, RequestName("debug_v1_scheduler"))
862 0 : })
863 0 : .post("/debug/v1/consistency_check", |r| {
864 0 : named_request_span(
865 0 : r,
866 0 : handle_consistency_check,
867 0 : RequestName("debug_v1_consistency_check"),
868 0 : )
869 0 : })
870 0 : .post("/debug/v1/reconcile_all", |r| {
871 0 : request_span(r, handle_reconcile_all)
872 0 : })
873 0 : .put("/debug/v1/failpoints", |r| {
874 0 : request_span(r, |r| failpoints_handler(r, CancellationToken::new()))
875 0 : })
876 0 : // Node operations
877 0 : .post("/control/v1/node", |r| {
878 0 : named_request_span(r, handle_node_register, RequestName("control_v1_node"))
879 0 : })
880 0 : .get("/control/v1/node", |r| {
881 0 : named_request_span(r, handle_node_list, RequestName("control_v1_node"))
882 0 : })
883 0 : .put("/control/v1/node/:node_id/config", |r| {
884 0 : named_request_span(
885 0 : r,
886 0 : handle_node_configure,
887 0 : RequestName("control_v1_node_config"),
888 0 : )
889 0 : })
890 0 : .get("/control/v1/node/:node_id", |r| {
891 0 : named_request_span(r, handle_node_status, RequestName("control_v1_node_status"))
892 0 : })
893 0 : .put("/control/v1/node/:node_id/drain", |r| {
894 0 : named_request_span(r, handle_node_drain, RequestName("control_v1_node_drain"))
895 0 : })
896 0 : .delete("/control/v1/node/:node_id/drain", |r| {
897 0 : named_request_span(
898 0 : r,
899 0 : handle_cancel_node_drain,
900 0 : RequestName("control_v1_cancel_node_drain"),
901 0 : )
902 0 : })
903 0 : .put("/control/v1/node/:node_id/fill", |r| {
904 0 : named_request_span(r, handle_node_fill, RequestName("control_v1_node_fill"))
905 0 : })
906 0 : .delete("/control/v1/node/:node_id/fill", |r| {
907 0 : named_request_span(
908 0 : r,
909 0 : handle_cancel_node_fill,
910 0 : RequestName("control_v1_cancel_node_fill"),
911 0 : )
912 0 : })
913 0 : // TODO(vlad): endpoint for cancelling drain and fill
914 0 : // Tenant Shard operations
915 0 : .put("/control/v1/tenant/:tenant_shard_id/migrate", |r| {
916 0 : tenant_service_handler(
917 0 : r,
918 0 : handle_tenant_shard_migrate,
919 0 : RequestName("control_v1_tenant_migrate"),
920 0 : )
921 0 : })
922 0 : .put("/control/v1/tenant/:tenant_id/shard_split", |r| {
923 0 : tenant_service_handler(
924 0 : r,
925 0 : handle_tenant_shard_split,
926 0 : RequestName("control_v1_tenant_shard_split"),
927 0 : )
928 0 : })
929 0 : .get("/control/v1/tenant/:tenant_id", |r| {
930 0 : tenant_service_handler(
931 0 : r,
932 0 : handle_tenant_describe,
933 0 : RequestName("control_v1_tenant_describe"),
934 0 : )
935 0 : })
936 0 : .get("/control/v1/tenant", |r| {
937 0 : tenant_service_handler(r, handle_tenant_list, RequestName("control_v1_tenant_list"))
938 0 : })
939 0 : .put("/control/v1/tenant/:tenant_id/policy", |r| {
940 0 : named_request_span(
941 0 : r,
942 0 : handle_tenant_update_policy,
943 0 : RequestName("control_v1_tenant_policy"),
944 0 : )
945 0 : })
946 0 : // Tenant operations
947 0 : // The ^/v1/ endpoints act as a "Virtual Pageserver", enabling shard-naive clients to call into
948 0 : // this service to manage tenants that actually consist of many tenant shards, as if they are a single entity.
949 0 : .post("/v1/tenant", |r| {
950 0 : tenant_service_handler(r, handle_tenant_create, RequestName("v1_tenant"))
951 0 : })
952 0 : .delete("/v1/tenant/:tenant_id", |r| {
953 0 : tenant_service_handler(r, handle_tenant_delete, RequestName("v1_tenant"))
954 0 : })
955 0 : .put("/v1/tenant/config", |r| {
956 0 : tenant_service_handler(r, handle_tenant_config_set, RequestName("v1_tenant_config"))
957 0 : })
958 0 : .get("/v1/tenant/:tenant_id/config", |r| {
959 0 : tenant_service_handler(r, handle_tenant_config_get, RequestName("v1_tenant_config"))
960 0 : })
961 0 : .put("/v1/tenant/:tenant_shard_id/location_config", |r| {
962 0 : tenant_service_handler(
963 0 : r,
964 0 : handle_tenant_location_config,
965 0 : RequestName("v1_tenant_location_config"),
966 0 : )
967 0 : })
968 0 : .put("/v1/tenant/:tenant_id/time_travel_remote_storage", |r| {
969 0 : tenant_service_handler(
970 0 : r,
971 0 : handle_tenant_time_travel_remote_storage,
972 0 : RequestName("v1_tenant_time_travel_remote_storage"),
973 0 : )
974 0 : })
975 0 : .post("/v1/tenant/:tenant_id/secondary/download", |r| {
976 0 : tenant_service_handler(
977 0 : r,
978 0 : handle_tenant_secondary_download,
979 0 : RequestName("v1_tenant_secondary_download"),
980 0 : )
981 0 : })
982 0 : // Timeline operations
983 0 : .delete("/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
984 0 : tenant_service_handler(
985 0 : r,
986 0 : handle_tenant_timeline_delete,
987 0 : RequestName("v1_tenant_timeline"),
988 0 : )
989 0 : })
990 0 : .post("/v1/tenant/:tenant_id/timeline", |r| {
991 0 : tenant_service_handler(
992 0 : r,
993 0 : handle_tenant_timeline_create,
994 0 : RequestName("v1_tenant_timeline"),
995 0 : )
996 0 : })
997 0 : // Tenant detail GET passthrough to shard zero:
998 0 : .get("/v1/tenant/:tenant_id", |r| {
999 0 : tenant_service_handler(
1000 0 : r,
1001 0 : handle_tenant_timeline_passthrough,
1002 0 : RequestName("v1_tenant_passthrough"),
1003 0 : )
1004 0 : })
1005 0 : // The `*` in the URL is a wildcard: any tenant/timeline GET APIs on the pageserver
1006 0 : // are implicitly exposed here. This must be last in the list to avoid
1007 0 : // taking precedence over other GET methods we might implement by hand.
1008 0 : .get("/v1/tenant/:tenant_id/*", |r| {
1009 0 : tenant_service_handler(
1010 0 : r,
1011 0 : handle_tenant_timeline_passthrough,
1012 0 : RequestName("v1_tenant_passthrough"),
1013 0 : )
1014 0 : })
1015 0 : }
|
