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

            Line data    Source code
       1              : use std::collections::HashMap;
       2              : 
       3              : use futures::Future;
       4              : use pageserver_api::config::NodeMetadata;
       5              : use pageserver_api::controller_api::{AvailabilityZone, NodeRegisterRequest};
       6              : use pageserver_api::shard::TenantShardId;
       7              : use pageserver_api::upcall_api::{
       8              :     ReAttachRequest, ReAttachResponse, ReAttachResponseTenant, ValidateRequest,
       9              :     ValidateRequestTenant, ValidateResponse,
      10              : };
      11              : use serde::Serialize;
      12              : use serde::de::DeserializeOwned;
      13              : use tokio_util::sync::CancellationToken;
      14              : use url::Url;
      15              : use utils::generation::Generation;
      16              : use utils::id::NodeId;
      17              : use utils::{backoff, failpoint_support};
      18              : 
      19              : use crate::config::PageServerConf;
      20              : use crate::virtual_file::on_fatal_io_error;
      21              : 
      22              : /// The Pageserver's client for using the storage controller upcall API: this is a small API
      23              : /// for dealing with generations (see docs/rfcs/025-generation-numbers.md).
      24              : ///
      25              : /// The server presenting this API may either be the storage controller or some other
      26              : /// service (such as the Neon control plane) providing a store of generation numbers.
      27              : pub struct ControllerUpcallClient {
      28              :     http_client: reqwest::Client,
      29              :     base_url: Url,
      30              :     node_id: NodeId,
      31              :     cancel: CancellationToken,
      32              : }
      33              : 
      34              : /// Represent operations which internally retry on all errors other than
      35              : /// cancellation token firing: the only way they can fail is ShuttingDown.
      36              : pub enum RetryForeverError {
      37              :     ShuttingDown,
      38              : }
      39              : 
      40              : pub trait ControlPlaneGenerationsApi {
      41              :     fn re_attach(
      42              :         &self,
      43              :         conf: &PageServerConf,
      44              :     ) -> impl Future<
      45              :         Output = Result<HashMap<TenantShardId, ReAttachResponseTenant>, RetryForeverError>,
      46              :     > + Send;
      47              :     fn validate(
      48              :         &self,
      49              :         tenants: Vec<(TenantShardId, Generation)>,
      50              :     ) -> impl Future<Output = Result<HashMap<TenantShardId, bool>, RetryForeverError>> + Send;
      51              : }
      52              : 
      53              : impl ControllerUpcallClient {
      54              :     /// A None return value indicates that the input `conf` object does not have control
      55              :     /// plane API enabled.
      56            0 :     pub fn new(conf: &'static PageServerConf, cancel: &CancellationToken) -> Option<Self> {
      57            0 :         let mut url = match conf.control_plane_api.as_ref() {
      58            0 :             Some(u) => u.clone(),
      59            0 :             None => return None,
      60              :         };
      61              : 
      62            0 :         if let Ok(mut segs) = url.path_segments_mut() {
      63            0 :             // This ensures that `url` ends with a slash if it doesn't already.
      64            0 :             // That way, we can subsequently use join() to safely attach extra path elements.
      65            0 :             segs.pop_if_empty().push("");
      66            0 :         }
      67              : 
      68            0 :         let mut client = reqwest::ClientBuilder::new();
      69              : 
      70            0 :         if let Some(jwt) = &conf.control_plane_api_token {
      71            0 :             let mut headers = reqwest::header::HeaderMap::new();
      72            0 :             headers.insert(
      73            0 :                 "Authorization",
      74            0 :                 format!("Bearer {}", jwt.get_contents()).parse().unwrap(),
      75            0 :             );
      76            0 :             client = client.default_headers(headers);
      77            0 :         }
      78              : 
      79            0 :         Some(Self {
      80            0 :             http_client: client.build().expect("Failed to construct HTTP client"),
      81            0 :             base_url: url,
      82            0 :             node_id: conf.id,
      83            0 :             cancel: cancel.clone(),
      84            0 :         })
      85            0 :     }
      86              : 
      87              :     #[tracing::instrument(skip_all)]
      88              :     async fn retry_http_forever<R, T>(
      89              :         &self,
      90              :         url: &url::Url,
      91              :         request: R,
      92              :     ) -> Result<T, RetryForeverError>
      93              :     where
      94              :         R: Serialize,
      95              :         T: DeserializeOwned,
      96              :     {
      97              :         let res = backoff::retry(
      98            0 :             || async {
      99            0 :                 let response = self
     100            0 :                     .http_client
     101            0 :                     .post(url.clone())
     102            0 :                     .json(&request)
     103            0 :                     .send()
     104            0 :                     .await?;
     105              : 
     106            0 :                 response.error_for_status_ref()?;
     107            0 :                 response.json::<T>().await
     108            0 :             },
     109            0 :             |_| false,
     110              :             3,
     111              :             u32::MAX,
     112              :             "storage controller upcall",
     113              :             &self.cancel,
     114              :         )
     115              :         .await
     116              :         .ok_or(RetryForeverError::ShuttingDown)?
     117              :         .expect("We retry forever, this should never be reached");
     118              : 
     119              :         Ok(res)
     120              :     }
     121              : 
     122            0 :     pub(crate) fn base_url(&self) -> &Url {
     123            0 :         &self.base_url
     124            0 :     }
     125              : }
     126              : 
     127              : impl ControlPlaneGenerationsApi for ControllerUpcallClient {
     128              :     /// Block until we get a successful response, or error out if we are shut down
     129              :     #[tracing::instrument(skip_all)] // so that warning logs from retry_http_forever have context
     130              :     async fn re_attach(
     131              :         &self,
     132              :         conf: &PageServerConf,
     133              :     ) -> Result<HashMap<TenantShardId, ReAttachResponseTenant>, RetryForeverError> {
     134              :         let url = self
     135              :             .base_url
     136              :             .join("re-attach")
     137              :             .expect("Failed to build re-attach path");
     138              : 
     139              :         // Include registration content in the re-attach request if a metadata file is readable
     140              :         let metadata_path = conf.metadata_path();
     141              :         let register = match tokio::fs::read_to_string(&metadata_path).await {
     142              :             Ok(metadata_str) => match serde_json::from_str::<NodeMetadata>(&metadata_str) {
     143              :                 Ok(m) => {
     144              :                     // Since we run one time at startup, be generous in our logging and
     145              :                     // dump all metadata.
     146              :                     tracing::info!(
     147              :                         "Loaded node metadata: postgres {}:{}, http {}:{}, other fields: {:?}",
     148              :                         m.postgres_host,
     149              :                         m.postgres_port,
     150              :                         m.http_host,
     151              :                         m.http_port,
     152              :                         m.other
     153              :                     );
     154              : 
     155              :                     let az_id = {
     156              :                         let az_id_from_metadata = m
     157              :                             .other
     158              :                             .get("availability_zone_id")
     159            0 :                             .and_then(|jv| jv.as_str().map(|str| str.to_owned()));
     160              : 
     161              :                         match az_id_from_metadata {
     162              :                             Some(az_id) => Some(AvailabilityZone(az_id)),
     163              :                             None => {
     164              :                                 tracing::warn!(
     165              :                                     "metadata.json does not contain an 'availability_zone_id' field"
     166              :                                 );
     167              :                                 conf.availability_zone.clone().map(AvailabilityZone)
     168              :                             }
     169              :                         }
     170              :                     };
     171              : 
     172              :                     if az_id.is_none() {
     173              :                         panic!(
     174              :                             "Availablity zone id could not be inferred from metadata.json or pageserver config"
     175              :                         );
     176              :                     }
     177              : 
     178              :                     Some(NodeRegisterRequest {
     179              :                         node_id: conf.id,
     180              :                         listen_pg_addr: m.postgres_host,
     181              :                         listen_pg_port: m.postgres_port,
     182              :                         listen_http_addr: m.http_host,
     183              :                         listen_http_port: m.http_port,
     184              :                         listen_https_port: m.https_port,
     185              :                         availability_zone_id: az_id.expect("Checked above"),
     186              :                     })
     187              :                 }
     188              :                 Err(e) => {
     189              :                     tracing::error!("Unreadable metadata in {metadata_path}: {e}");
     190              :                     None
     191              :                 }
     192              :             },
     193              :             Err(e) => {
     194              :                 if e.kind() == std::io::ErrorKind::NotFound {
     195              :                     // This is legal: we may have been deployed with some external script
     196              :                     // doing registration for us.
     197              :                     tracing::info!("Metadata file not found at {metadata_path}");
     198              :                 } else {
     199              :                     on_fatal_io_error(&e, &format!("Loading metadata at {metadata_path}"))
     200              :                 }
     201              :                 None
     202              :             }
     203              :         };
     204              : 
     205              :         let request = ReAttachRequest {
     206              :             node_id: self.node_id,
     207              :             register: register.clone(),
     208              :         };
     209              : 
     210              :         let response: ReAttachResponse = self.retry_http_forever(&url, request).await?;
     211              :         tracing::info!(
     212              :             "Received re-attach response with {} tenants (node {}, register: {:?})",
     213              :             response.tenants.len(),
     214              :             self.node_id,
     215              :             register,
     216              :         );
     217              : 
     218              :         failpoint_support::sleep_millis_async!("control-plane-client-re-attach");
     219              : 
     220              :         Ok(response
     221              :             .tenants
     222              :             .into_iter()
     223            0 :             .map(|rart| (rart.id, rart))
     224              :             .collect::<HashMap<_, _>>())
     225              :     }
     226              : 
     227              :     /// Block until we get a successful response, or error out if we are shut down
     228              :     #[tracing::instrument(skip_all)] // so that warning logs from retry_http_forever have context
     229              :     async fn validate(
     230              :         &self,
     231              :         tenants: Vec<(TenantShardId, Generation)>,
     232              :     ) -> Result<HashMap<TenantShardId, bool>, RetryForeverError> {
     233              :         let url = self
     234              :             .base_url
     235              :             .join("validate")
     236              :             .expect("Failed to build validate path");
     237              : 
     238              :         // When sending validate requests, break them up into chunks so that we
     239              :         // avoid possible edge cases of generating any HTTP requests that
     240              :         // require database I/O across many thousands of tenants.
     241              :         let mut result: HashMap<TenantShardId, bool> = HashMap::with_capacity(tenants.len());
     242              :         for tenant_chunk in (tenants).chunks(128) {
     243              :             let request = ValidateRequest {
     244              :                 tenants: tenant_chunk
     245              :                     .iter()
     246            0 :                     .map(|(id, generation)| ValidateRequestTenant {
     247            0 :                         id: *id,
     248            0 :                         r#gen: (*generation).into().expect(
     249            0 :                             "Generation should always be valid for a Tenant doing deletions",
     250            0 :                         ),
     251            0 :                     })
     252              :                     .collect(),
     253              :             };
     254              : 
     255              :             failpoint_support::sleep_millis_async!(
     256              :                 "control-plane-client-validate-sleep",
     257              :                 &self.cancel
     258              :             );
     259              :             if self.cancel.is_cancelled() {
     260              :                 return Err(RetryForeverError::ShuttingDown);
     261              :             }
     262              : 
     263              :             let response: ValidateResponse = self.retry_http_forever(&url, request).await?;
     264              :             for rt in response.tenants {
     265              :                 result.insert(rt.id, rt.valid);
     266              :             }
     267              :         }
     268              : 
     269              :         Ok(result.into_iter().collect())
     270              :     }
     271              : }
        

Generated by: LCOV version 2.1-beta