LCOV - code coverage report
Current view: top level - storage_controller/src - reconciler.rs (source / functions) Coverage Total Hit
Test: 5445d246133daeceb0507e6cc0797ab7c1c70cb8.info Lines: 0.0 % 758 0
Test Date: 2025-03-12 18:05:02 Functions: 0.0 % 55 0

            Line data    Source code
       1              : use std::borrow::Cow;
       2              : use std::collections::HashMap;
       3              : use std::sync::Arc;
       4              : use std::time::{Duration, Instant};
       5              : 
       6              : use json_structural_diff::JsonDiff;
       7              : use pageserver_api::controller_api::{AvailabilityZone, MigrationConfig, PlacementPolicy};
       8              : use pageserver_api::models::{
       9              :     LocationConfig, LocationConfigMode, LocationConfigSecondary, TenantConfig, TenantWaitLsnRequest,
      10              : };
      11              : use pageserver_api::shard::{ShardIdentity, TenantShardId};
      12              : use pageserver_client::mgmt_api;
      13              : use reqwest::StatusCode;
      14              : use tokio_util::sync::CancellationToken;
      15              : use utils::backoff::exponential_backoff;
      16              : use utils::generation::Generation;
      17              : use utils::id::{NodeId, TimelineId};
      18              : use utils::lsn::Lsn;
      19              : use utils::pausable_failpoint;
      20              : use utils::sync::gate::GateGuard;
      21              : 
      22              : use crate::compute_hook::{ComputeHook, NotifyError};
      23              : use crate::node::Node;
      24              : use crate::pageserver_client::PageserverClient;
      25              : use crate::persistence::Persistence;
      26              : use crate::tenant_shard::{IntentState, ObservedState, ObservedStateDelta, ObservedStateLocation};
      27              : use crate::{compute_hook, service};
      28              : 
      29              : const DEFAULT_HEATMAP_PERIOD: Duration = Duration::from_secs(60);
      30              : 
      31              : /// Object with the lifetime of the background reconcile task that is created
      32              : /// for tenants which have a difference between their intent and observed states.
      33              : pub(super) struct Reconciler {
      34              :     /// See [`crate::tenant_shard::TenantShard`] for the meanings of these fields: they are a snapshot
      35              :     /// of a tenant's state from when we spawned a reconcile task.
      36              :     pub(super) tenant_shard_id: TenantShardId,
      37              :     pub(crate) shard: ShardIdentity,
      38              :     pub(crate) placement_policy: PlacementPolicy,
      39              :     pub(crate) generation: Option<Generation>,
      40              :     pub(crate) intent: TargetState,
      41              : 
      42              :     /// Nodes not referenced by [`Self::intent`], from which we should try
      43              :     /// to detach this tenant shard.
      44              :     pub(crate) detach: Vec<Node>,
      45              : 
      46              :     /// Configuration specific to this reconciler
      47              :     pub(crate) reconciler_config: ReconcilerConfig,
      48              : 
      49              :     pub(crate) config: TenantConfig,
      50              :     pub(crate) preferred_az: Option<AvailabilityZone>,
      51              : 
      52              :     /// Observed state from the point of view of the reconciler.
      53              :     /// This gets updated as the reconciliation makes progress.
      54              :     pub(crate) observed: ObservedState,
      55              : 
      56              :     /// Snapshot of the observed state at the point when the reconciler
      57              :     /// was spawned.
      58              :     pub(crate) original_observed: ObservedState,
      59              : 
      60              :     pub(crate) service_config: service::Config,
      61              : 
      62              :     /// A hook to notify the running postgres instances when we change the location
      63              :     /// of a tenant.  Use this via [`Self::compute_notify`] to update our failure flag
      64              :     /// and guarantee eventual retries.
      65              :     pub(crate) compute_hook: Arc<ComputeHook>,
      66              : 
      67              :     /// To avoid stalling if the cloud control plane is unavailable, we may proceed
      68              :     /// past failures in [`ComputeHook::notify`], but we _must_ remember that we failed
      69              :     /// so that we can set [`crate::tenant_shard::TenantShard::pending_compute_notification`] to ensure a later retry.
      70              :     pub(crate) compute_notify_failure: bool,
      71              : 
      72              :     /// Reconciler is responsible for keeping alive semaphore units that limit concurrency on how many
      73              :     /// we will spawn.
      74              :     pub(crate) _resource_units: ReconcileUnits,
      75              : 
      76              :     /// A means to abort background reconciliation: it is essential to
      77              :     /// call this when something changes in the original TenantShard that
      78              :     /// will make this reconciliation impossible or unnecessary, for
      79              :     /// example when a pageserver node goes offline, or the PlacementPolicy for
      80              :     /// the tenant is changed.
      81              :     pub(crate) cancel: CancellationToken,
      82              : 
      83              :     /// Reconcilers are registered with a Gate so that during a graceful shutdown we
      84              :     /// can wait for all the reconcilers to respond to their cancellation tokens.
      85              :     pub(crate) _gate_guard: GateGuard,
      86              : 
      87              :     /// Access to persistent storage for updating generation numbers
      88              :     pub(crate) persistence: Arc<Persistence>,
      89              : }
      90              : 
      91              : pub(crate) struct ReconcilerConfigBuilder {
      92              :     config: ReconcilerConfig,
      93              : }
      94              : 
      95              : impl ReconcilerConfigBuilder {
      96              :     /// Priority is special: you must pick one thoughtfully, do not just use 'normal' as the default
      97            0 :     pub(crate) fn new(priority: ReconcilerPriority) -> Self {
      98            0 :         Self {
      99            0 :             config: ReconcilerConfig::new(priority),
     100            0 :         }
     101            0 :     }
     102              : 
     103            0 :     pub(crate) fn secondary_warmup_timeout(self, value: Duration) -> Self {
     104            0 :         Self {
     105            0 :             config: ReconcilerConfig {
     106            0 :                 secondary_warmup_timeout: Some(value),
     107            0 :                 ..self.config
     108            0 :             },
     109            0 :         }
     110            0 :     }
     111              : 
     112            0 :     pub(crate) fn secondary_download_request_timeout(self, value: Duration) -> Self {
     113            0 :         Self {
     114            0 :             config: ReconcilerConfig {
     115            0 :                 secondary_download_request_timeout: Some(value),
     116            0 :                 ..self.config
     117            0 :             },
     118            0 :         }
     119            0 :     }
     120              : 
     121            0 :     pub(crate) fn tenant_creation_hint(self, hint: bool) -> Self {
     122            0 :         Self {
     123            0 :             config: ReconcilerConfig {
     124            0 :                 tenant_creation_hint: hint,
     125            0 :                 ..self.config
     126            0 :             },
     127            0 :         }
     128            0 :     }
     129              : 
     130            0 :     pub(crate) fn build(self) -> ReconcilerConfig {
     131            0 :         self.config
     132            0 :     }
     133              : }
     134              : 
     135              : // Higher priorities are used for user-facing tasks, so that a long backlog of housekeeping work (e.g. reconciling on startup, rescheduling
     136              : // things on node changes) does not starve user-facing tasks.
     137              : #[derive(Debug, Copy, Clone)]
     138              : pub(crate) enum ReconcilerPriority {
     139              :     Normal,
     140              :     High,
     141              : }
     142              : 
     143              : #[derive(Debug, Copy, Clone)]
     144              : pub(crate) struct ReconcilerConfig {
     145              :     pub(crate) priority: ReconcilerPriority,
     146              : 
     147              :     // During live migration give up on warming-up the secondary
     148              :     // after this timeout.
     149              :     secondary_warmup_timeout: Option<Duration>,
     150              : 
     151              :     // During live migrations this is the amount of time that
     152              :     // the pagserver will hold our poll.
     153              :     secondary_download_request_timeout: Option<Duration>,
     154              : 
     155              :     // A hint indicating whether this reconciliation is done on the
     156              :     // creation of a new tenant. This only informs logging behaviour.
     157              :     tenant_creation_hint: bool,
     158              : }
     159              : 
     160              : impl ReconcilerConfig {
     161              :     /// Configs are always constructed with an explicit priority, to force callers to think about whether
     162              :     /// the operation they're scheduling is high-priority or not. Normal priority is not a safe default, because
     163              :     /// scheduling something user-facing at normal priority can result in it getting starved out by background work.
     164            0 :     pub(crate) fn new(priority: ReconcilerPriority) -> Self {
     165            0 :         Self {
     166            0 :             priority,
     167            0 :             secondary_warmup_timeout: None,
     168            0 :             secondary_download_request_timeout: None,
     169            0 :             tenant_creation_hint: false,
     170            0 :         }
     171            0 :     }
     172              : 
     173            0 :     pub(crate) fn get_secondary_warmup_timeout(&self) -> Duration {
     174              :         const SECONDARY_WARMUP_TIMEOUT_DEFAULT: Duration = Duration::from_secs(300);
     175            0 :         self.secondary_warmup_timeout
     176            0 :             .unwrap_or(SECONDARY_WARMUP_TIMEOUT_DEFAULT)
     177            0 :     }
     178              : 
     179            0 :     pub(crate) fn get_secondary_download_request_timeout(&self) -> Duration {
     180              :         const SECONDARY_DOWNLOAD_REQUEST_TIMEOUT_DEFAULT: Duration = Duration::from_secs(20);
     181            0 :         self.secondary_download_request_timeout
     182            0 :             .unwrap_or(SECONDARY_DOWNLOAD_REQUEST_TIMEOUT_DEFAULT)
     183            0 :     }
     184              : 
     185            0 :     pub(crate) fn tenant_creation_hint(&self) -> bool {
     186            0 :         self.tenant_creation_hint
     187            0 :     }
     188              : }
     189              : 
     190              : impl From<&MigrationConfig> for ReconcilerConfig {
     191            0 :     fn from(value: &MigrationConfig) -> Self {
     192            0 :         // Run reconciler at high priority because MigrationConfig comes from human requests that should
     193            0 :         // be presumed urgent.
     194            0 :         let mut builder = ReconcilerConfigBuilder::new(ReconcilerPriority::High);
     195              : 
     196            0 :         if let Some(timeout) = value.secondary_warmup_timeout {
     197            0 :             builder = builder.secondary_warmup_timeout(timeout)
     198            0 :         }
     199              : 
     200            0 :         if let Some(timeout) = value.secondary_download_request_timeout {
     201            0 :             builder = builder.secondary_download_request_timeout(timeout)
     202            0 :         }
     203              : 
     204            0 :         builder.build()
     205            0 :     }
     206              : }
     207              : 
     208              : /// RAII resource units granted to a Reconciler, which it should keep alive until it finishes doing I/O
     209              : pub(crate) struct ReconcileUnits {
     210              :     _sem_units: tokio::sync::OwnedSemaphorePermit,
     211              : }
     212              : 
     213              : impl ReconcileUnits {
     214            0 :     pub(crate) fn new(sem_units: tokio::sync::OwnedSemaphorePermit) -> Self {
     215            0 :         Self {
     216            0 :             _sem_units: sem_units,
     217            0 :         }
     218            0 :     }
     219              : }
     220              : 
     221              : /// This is a snapshot of [`crate::tenant_shard::IntentState`], but it does not do any
     222              : /// reference counting for Scheduler.  The IntentState is what the scheduler works with,
     223              : /// and the TargetState is just the instruction for a particular Reconciler run.
     224              : #[derive(Debug)]
     225              : pub(crate) struct TargetState {
     226              :     pub(crate) attached: Option<Node>,
     227              :     pub(crate) secondary: Vec<Node>,
     228              : }
     229              : 
     230              : impl TargetState {
     231            0 :     pub(crate) fn from_intent(nodes: &HashMap<NodeId, Node>, intent: &IntentState) -> Self {
     232            0 :         Self {
     233            0 :             attached: intent.get_attached().map(|n| {
     234            0 :                 nodes
     235            0 :                     .get(&n)
     236            0 :                     .expect("Intent attached referenced non-existent node")
     237            0 :                     .clone()
     238            0 :             }),
     239            0 :             secondary: intent
     240            0 :                 .get_secondary()
     241            0 :                 .iter()
     242            0 :                 .map(|n| {
     243            0 :                     nodes
     244            0 :                         .get(n)
     245            0 :                         .expect("Intent secondary referenced non-existent node")
     246            0 :                         .clone()
     247            0 :                 })
     248            0 :                 .collect(),
     249            0 :         }
     250            0 :     }
     251              : }
     252              : 
     253              : #[derive(thiserror::Error, Debug)]
     254              : pub(crate) enum ReconcileError {
     255              :     #[error(transparent)]
     256              :     Remote(#[from] mgmt_api::Error),
     257              :     #[error(transparent)]
     258              :     Notify(#[from] NotifyError),
     259              :     #[error("Cancelled")]
     260              :     Cancel,
     261              :     #[error(transparent)]
     262              :     Other(#[from] anyhow::Error),
     263              : }
     264              : 
     265              : impl Reconciler {
     266            0 :     async fn location_config(
     267            0 :         &mut self,
     268            0 :         node: &Node,
     269            0 :         config: LocationConfig,
     270            0 :         flush_ms: Option<Duration>,
     271            0 :         lazy: bool,
     272            0 :     ) -> Result<(), ReconcileError> {
     273            0 :         if !node.is_available() && config.mode == LocationConfigMode::Detached {
     274              :             // [`crate::service::Service::node_activate_reconcile`] will update the observed state
     275              :             // when the node comes back online. At that point, the intent and observed states will
     276              :             // be mismatched and a background reconciliation will detach.
     277            0 :             tracing::info!(
     278            0 :                 "Node {node} is unavailable during detach: proceeding anyway, it will be detached via background reconciliation"
     279              :             );
     280            0 :             return Ok(());
     281            0 :         }
     282            0 : 
     283            0 :         self.observed
     284            0 :             .locations
     285            0 :             .insert(node.get_id(), ObservedStateLocation { conf: None });
     286            0 : 
     287            0 :         // TODO: amend locations that use long-polling: they will hit this timeout.
     288            0 :         let timeout = Duration::from_secs(25);
     289            0 : 
     290            0 :         tracing::info!("location_config({node}) calling: {:?}", config);
     291            0 :         let tenant_shard_id = self.tenant_shard_id;
     292            0 :         let config_ref = &config;
     293            0 :         match node
     294            0 :             .with_client_retries(
     295            0 :                 |client| async move {
     296            0 :                     let config = config_ref.clone();
     297            0 :                     client
     298            0 :                         .location_config(tenant_shard_id, config.clone(), flush_ms, lazy)
     299            0 :                         .await
     300            0 :                 },
     301            0 :                 &self.service_config.pageserver_jwt_token,
     302            0 :                 &self.service_config.ssl_ca_cert,
     303            0 :                 1,
     304            0 :                 3,
     305            0 :                 timeout,
     306            0 :                 &self.cancel,
     307            0 :             )
     308            0 :             .await
     309              :         {
     310            0 :             Some(Ok(_)) => {}
     311            0 :             Some(Err(e)) => return Err(e.into()),
     312            0 :             None => return Err(ReconcileError::Cancel),
     313              :         };
     314            0 :         tracing::info!("location_config({node}) complete: {:?}", config);
     315              : 
     316            0 :         match config.mode {
     317            0 :             LocationConfigMode::Detached => {
     318            0 :                 self.observed.locations.remove(&node.get_id());
     319            0 :             }
     320            0 :             _ => {
     321            0 :                 self.observed
     322            0 :                     .locations
     323            0 :                     .insert(node.get_id(), ObservedStateLocation { conf: Some(config) });
     324            0 :             }
     325              :         }
     326              : 
     327            0 :         Ok(())
     328            0 :     }
     329              : 
     330            0 :     fn get_node(&self, node_id: &NodeId) -> Option<&Node> {
     331            0 :         if let Some(node) = self.intent.attached.as_ref() {
     332            0 :             if node.get_id() == *node_id {
     333            0 :                 return Some(node);
     334            0 :             }
     335            0 :         }
     336              : 
     337            0 :         if let Some(node) = self
     338            0 :             .intent
     339            0 :             .secondary
     340            0 :             .iter()
     341            0 :             .find(|n| n.get_id() == *node_id)
     342              :         {
     343            0 :             return Some(node);
     344            0 :         }
     345              : 
     346            0 :         if let Some(node) = self.detach.iter().find(|n| n.get_id() == *node_id) {
     347            0 :             return Some(node);
     348            0 :         }
     349            0 : 
     350            0 :         None
     351            0 :     }
     352              : 
     353            0 :     async fn maybe_live_migrate(&mut self) -> Result<(), ReconcileError> {
     354            0 :         let destination = if let Some(node) = &self.intent.attached {
     355            0 :             match self.observed.locations.get(&node.get_id()) {
     356            0 :                 Some(conf) => {
     357              :                     // We will do a live migration only if the intended destination is not
     358              :                     // currently in an attached state.
     359            0 :                     match &conf.conf {
     360            0 :                         Some(conf) if conf.mode == LocationConfigMode::Secondary => {
     361            0 :                             // Fall through to do a live migration
     362            0 :                             node
     363              :                         }
     364              :                         None | Some(_) => {
     365              :                             // Attached or uncertain: don't do a live migration, proceed
     366              :                             // with a general-case reconciliation
     367            0 :                             tracing::info!("maybe_live_migrate: destination is None or attached");
     368            0 :                             return Ok(());
     369              :                         }
     370              :                     }
     371              :                 }
     372              :                 None => {
     373              :                     // Our destination is not attached: maybe live migrate if some other
     374              :                     // node is currently attached.  Fall through.
     375            0 :                     node
     376              :                 }
     377              :             }
     378              :         } else {
     379              :             // No intent to be attached
     380            0 :             tracing::info!("maybe_live_migrate: no attached intent");
     381            0 :             return Ok(());
     382              :         };
     383              : 
     384            0 :         let mut origin = None;
     385            0 :         for (node_id, state) in &self.observed.locations {
     386            0 :             if let Some(observed_conf) = &state.conf {
     387            0 :                 if observed_conf.mode == LocationConfigMode::AttachedSingle {
     388              :                     // We will only attempt live migration if the origin is not offline: this
     389              :                     // avoids trying to do it while reconciling after responding to an HA failover.
     390            0 :                     if let Some(node) = self.get_node(node_id) {
     391            0 :                         if node.is_available() {
     392            0 :                             origin = Some(node.clone());
     393            0 :                             break;
     394            0 :                         }
     395            0 :                     }
     396            0 :                 }
     397            0 :             }
     398              :         }
     399              : 
     400            0 :         let Some(origin) = origin else {
     401            0 :             tracing::info!("maybe_live_migrate: no origin found");
     402            0 :             return Ok(());
     403              :         };
     404              : 
     405              :         // We have an origin and a destination: proceed to do the live migration
     406            0 :         tracing::info!("Live migrating {}->{}", origin, destination);
     407            0 :         self.live_migrate(origin, destination.clone()).await?;
     408              : 
     409            0 :         Ok(())
     410            0 :     }
     411              : 
     412            0 :     async fn wait_lsn(
     413            0 :         &self,
     414            0 :         node: &Node,
     415            0 :         tenant_shard_id: TenantShardId,
     416            0 :         timelines: HashMap<TimelineId, Lsn>,
     417            0 :     ) -> Result<StatusCode, ReconcileError> {
     418              :         const TIMEOUT: Duration = Duration::from_secs(10);
     419              : 
     420            0 :         let client = PageserverClient::new(
     421            0 :             node.get_id(),
     422            0 :             node.base_url(),
     423            0 :             self.service_config.pageserver_jwt_token.as_deref(),
     424            0 :             self.service_config.ssl_ca_cert.clone(),
     425            0 :         )?;
     426              : 
     427            0 :         client
     428            0 :             .wait_lsn(
     429            0 :                 tenant_shard_id,
     430            0 :                 TenantWaitLsnRequest {
     431            0 :                     timelines,
     432            0 :                     timeout: TIMEOUT,
     433            0 :                 },
     434            0 :             )
     435            0 :             .await
     436            0 :             .map_err(|e| e.into())
     437            0 :     }
     438              : 
     439            0 :     async fn get_lsns(
     440            0 :         &self,
     441            0 :         tenant_shard_id: TenantShardId,
     442            0 :         node: &Node,
     443            0 :     ) -> anyhow::Result<HashMap<TimelineId, Lsn>> {
     444            0 :         let client = PageserverClient::new(
     445            0 :             node.get_id(),
     446            0 :             node.base_url(),
     447            0 :             self.service_config.pageserver_jwt_token.as_deref(),
     448            0 :             self.service_config.ssl_ca_cert.clone(),
     449            0 :         )?;
     450              : 
     451            0 :         let timelines = client.timeline_list(&tenant_shard_id).await?;
     452            0 :         Ok(timelines
     453            0 :             .into_iter()
     454            0 :             .map(|t| (t.timeline_id, t.last_record_lsn))
     455            0 :             .collect())
     456            0 :     }
     457              : 
     458            0 :     async fn secondary_download(
     459            0 :         &self,
     460            0 :         tenant_shard_id: TenantShardId,
     461            0 :         node: &Node,
     462            0 :     ) -> Result<(), ReconcileError> {
     463            0 :         // This is not the timeout for a request, but the total amount of time we're willing to wait
     464            0 :         // for a secondary location to get up to date before
     465            0 :         let total_download_timeout = self.reconciler_config.get_secondary_warmup_timeout();
     466            0 : 
     467            0 :         // This the long-polling interval for the secondary download requests we send to destination pageserver
     468            0 :         // during a migration.
     469            0 :         let request_download_timeout = self
     470            0 :             .reconciler_config
     471            0 :             .get_secondary_download_request_timeout();
     472            0 : 
     473            0 :         let started_at = Instant::now();
     474              : 
     475              :         loop {
     476            0 :             let (status, progress) = match node
     477            0 :                 .with_client_retries(
     478            0 :                     |client| async move {
     479            0 :                         client
     480            0 :                             .tenant_secondary_download(
     481            0 :                                 tenant_shard_id,
     482            0 :                                 Some(request_download_timeout),
     483            0 :                             )
     484            0 :                             .await
     485            0 :                     },
     486            0 :                     &self.service_config.pageserver_jwt_token,
     487            0 :                     &self.service_config.ssl_ca_cert,
     488            0 :                     1,
     489            0 :                     3,
     490            0 :                     request_download_timeout * 2,
     491            0 :                     &self.cancel,
     492            0 :                 )
     493            0 :                 .await
     494              :             {
     495            0 :                 None => Err(ReconcileError::Cancel),
     496            0 :                 Some(Ok(v)) => Ok(v),
     497            0 :                 Some(Err(e)) => {
     498            0 :                     // Give up, but proceed: it's unfortunate if we couldn't freshen the destination before
     499            0 :                     // attaching, but we should not let an issue with a secondary location stop us proceeding
     500            0 :                     // with a live migration.
     501            0 :                     tracing::warn!("Failed to prepare by downloading layers on node {node}: {e})");
     502            0 :                     return Ok(());
     503              :                 }
     504            0 :             }?;
     505              : 
     506            0 :             if status == StatusCode::OK {
     507            0 :                 tracing::info!(
     508            0 :                     "Downloads to {} complete: {}/{} layers, {}/{} bytes",
     509              :                     node,
     510              :                     progress.layers_downloaded,
     511              :                     progress.layers_total,
     512              :                     progress.bytes_downloaded,
     513              :                     progress.bytes_total
     514              :                 );
     515            0 :                 return Ok(());
     516            0 :             } else if status == StatusCode::ACCEPTED {
     517            0 :                 let total_runtime = started_at.elapsed();
     518            0 :                 if total_runtime > total_download_timeout {
     519            0 :                     tracing::warn!(
     520            0 :                         "Timed out after {}ms downloading layers to {node}.  Progress so far: {}/{} layers, {}/{} bytes",
     521            0 :                         total_runtime.as_millis(),
     522              :                         progress.layers_downloaded,
     523              :                         progress.layers_total,
     524              :                         progress.bytes_downloaded,
     525              :                         progress.bytes_total
     526              :                     );
     527              :                     // Give up, but proceed: an incompletely warmed destination doesn't prevent migration working,
     528              :                     // it just makes the I/O performance for users less good.
     529            0 :                     return Ok(());
     530            0 :                 }
     531            0 : 
     532            0 :                 // Log and proceed around the loop to retry.  We don't sleep between requests, because our HTTP call
     533            0 :                 // to the pageserver is a long-poll.
     534            0 :                 tracing::info!(
     535            0 :                     "Downloads to {} not yet complete: {}/{} layers, {}/{} bytes",
     536              :                     node,
     537              :                     progress.layers_downloaded,
     538              :                     progress.layers_total,
     539              :                     progress.bytes_downloaded,
     540              :                     progress.bytes_total
     541              :                 );
     542            0 :             }
     543              :         }
     544            0 :     }
     545              : 
     546              :     /// This function does _not_ mutate any state, so it is cancellation safe.
     547              :     ///
     548              :     /// This function does not respect [`Self::cancel`], callers should handle that.
     549            0 :     async fn await_lsn(
     550            0 :         &self,
     551            0 :         tenant_shard_id: TenantShardId,
     552            0 :         node: &Node,
     553            0 :         baseline: HashMap<TimelineId, Lsn>,
     554            0 :     ) -> anyhow::Result<()> {
     555              :         // Signal to the pageserver that it should ingest up to the baseline LSNs.
     556              :         loop {
     557            0 :             match self.wait_lsn(node, tenant_shard_id, baseline.clone()).await {
     558              :                 Ok(StatusCode::OK) => {
     559              :                     // Everything is caught up
     560            0 :                     return Ok(());
     561              :                 }
     562              :                 Ok(StatusCode::ACCEPTED) => {
     563              :                     // Some timelines are not caught up yet.
     564              :                     // They'll be polled below.
     565            0 :                     break;
     566              :                 }
     567              :                 Ok(StatusCode::NOT_FOUND) => {
     568              :                     // None of the timelines are present on the pageserver.
     569              :                     // This is correct if they've all been deleted, but
     570              :                     // let let the polling loop below cross check.
     571            0 :                     break;
     572              :                 }
     573            0 :                 Ok(status_code) => {
     574            0 :                     tracing::warn!(
     575            0 :                         "Unexpected status code ({status_code}) returned by wait_lsn endpoint"
     576              :                     );
     577            0 :                     break;
     578              :                 }
     579            0 :                 Err(e) => {
     580            0 :                     tracing::info!("🕑 Can't trigger LSN wait on {node} yet, waiting ({e})",);
     581            0 :                     tokio::time::sleep(Duration::from_millis(500)).await;
     582            0 :                     continue;
     583              :                 }
     584              :             }
     585              :         }
     586              : 
     587              :         // Poll the LSNs until they catch up
     588              :         loop {
     589            0 :             let latest = match self.get_lsns(tenant_shard_id, node).await {
     590            0 :                 Ok(l) => l,
     591            0 :                 Err(e) => {
     592            0 :                     tracing::info!("🕑 Can't get LSNs on node {node} yet, waiting ({e})",);
     593            0 :                     tokio::time::sleep(Duration::from_millis(500)).await;
     594            0 :                     continue;
     595              :                 }
     596              :             };
     597              : 
     598            0 :             let mut any_behind: bool = false;
     599            0 :             for (timeline_id, baseline_lsn) in &baseline {
     600            0 :                 match latest.get(timeline_id) {
     601            0 :                     Some(latest_lsn) => {
     602            0 :                         tracing::info!(timeline_id = %timeline_id, "🕑 LSN origin {baseline_lsn} vs destination {latest_lsn}");
     603            0 :                         if latest_lsn < baseline_lsn {
     604            0 :                             any_behind = true;
     605            0 :                         }
     606              :                     }
     607            0 :                     None => {
     608            0 :                         // Timeline was deleted in the meantime - ignore it
     609            0 :                     }
     610              :                 }
     611              :             }
     612              : 
     613            0 :             if !any_behind {
     614            0 :                 tracing::info!("✅ LSN caught up.  Proceeding...");
     615            0 :                 break;
     616              :             } else {
     617            0 :                 tokio::time::sleep(Duration::from_millis(500)).await;
     618              :             }
     619              :         }
     620              : 
     621            0 :         Ok(())
     622            0 :     }
     623              : 
     624            0 :     pub async fn live_migrate(
     625            0 :         &mut self,
     626            0 :         origin_ps: Node,
     627            0 :         dest_ps: Node,
     628            0 :     ) -> Result<(), ReconcileError> {
     629            0 :         // `maybe_live_migrate` is responsibble for sanity of inputs
     630            0 :         assert!(origin_ps.get_id() != dest_ps.get_id());
     631              : 
     632            0 :         fn build_location_config(
     633            0 :             shard: &ShardIdentity,
     634            0 :             config: &TenantConfig,
     635            0 :             mode: LocationConfigMode,
     636            0 :             generation: Option<Generation>,
     637            0 :             secondary_conf: Option<LocationConfigSecondary>,
     638            0 :         ) -> LocationConfig {
     639            0 :             LocationConfig {
     640            0 :                 mode,
     641            0 :                 generation: generation.map(|g| g.into().unwrap()),
     642            0 :                 secondary_conf,
     643            0 :                 tenant_conf: config.clone(),
     644            0 :                 shard_number: shard.number.0,
     645            0 :                 shard_count: shard.count.literal(),
     646            0 :                 shard_stripe_size: shard.stripe_size.0,
     647            0 :             }
     648            0 :         }
     649              : 
     650            0 :         tracing::info!("🔁 Switching origin node {origin_ps} to stale mode",);
     651              : 
     652              :         // FIXME: it is incorrect to use self.generation here, we should use the generation
     653              :         // from the ObservedState of the origin pageserver (it might be older than self.generation)
     654            0 :         let stale_conf = build_location_config(
     655            0 :             &self.shard,
     656            0 :             &self.config,
     657            0 :             LocationConfigMode::AttachedStale,
     658            0 :             self.generation,
     659            0 :             None,
     660            0 :         );
     661            0 :         self.location_config(&origin_ps, stale_conf, Some(Duration::from_secs(10)), false)
     662            0 :             .await?;
     663              : 
     664            0 :         let baseline_lsns = Some(self.get_lsns(self.tenant_shard_id, &origin_ps).await?);
     665              : 
     666              :         // If we are migrating to a destination that has a secondary location, warm it up first
     667            0 :         if let Some(destination_conf) = self.observed.locations.get(&dest_ps.get_id()) {
     668            0 :             if let Some(destination_conf) = &destination_conf.conf {
     669            0 :                 if destination_conf.mode == LocationConfigMode::Secondary {
     670            0 :                     tracing::info!("🔁 Downloading latest layers to destination node {dest_ps}",);
     671            0 :                     self.secondary_download(self.tenant_shard_id, &dest_ps)
     672            0 :                         .await?;
     673            0 :                 }
     674            0 :             }
     675            0 :         }
     676              : 
     677            0 :         pausable_failpoint!("reconciler-live-migrate-pre-generation-inc");
     678              : 
     679              :         // Increment generation before attaching to new pageserver
     680              :         self.generation = Some(
     681            0 :             self.persistence
     682            0 :                 .increment_generation(self.tenant_shard_id, dest_ps.get_id())
     683            0 :                 .await?,
     684              :         );
     685              : 
     686            0 :         let dest_conf = build_location_config(
     687            0 :             &self.shard,
     688            0 :             &self.config,
     689            0 :             LocationConfigMode::AttachedMulti,
     690            0 :             self.generation,
     691            0 :             None,
     692            0 :         );
     693            0 : 
     694            0 :         tracing::info!("🔁 Attaching to pageserver {dest_ps}");
     695            0 :         self.location_config(&dest_ps, dest_conf, None, false)
     696            0 :             .await?;
     697              : 
     698            0 :         pausable_failpoint!("reconciler-live-migrate-pre-await-lsn");
     699              : 
     700            0 :         if let Some(baseline) = baseline_lsns {
     701            0 :             tracing::info!("🕑 Waiting for LSN to catch up...");
     702            0 :             tokio::select! {
     703            0 :                 r = self.await_lsn(self.tenant_shard_id, &dest_ps, baseline) => {r?;}
     704            0 :                 _ = self.cancel.cancelled() => {return Err(ReconcileError::Cancel)}
     705              :             };
     706            0 :         }
     707              : 
     708            0 :         tracing::info!("🔁 Notifying compute to use pageserver {dest_ps}");
     709              : 
     710              :         // During a live migration it is unhelpful to proceed if we couldn't notify compute: if we detach
     711              :         // the origin without notifying compute, we will render the tenant unavailable.
     712            0 :         self.compute_notify_blocking(&origin_ps).await?;
     713            0 :         pausable_failpoint!("reconciler-live-migrate-post-notify");
     714              : 
     715              :         // Downgrade the origin to secondary.  If the tenant's policy is PlacementPolicy::Attached(0), then
     716              :         // this location will be deleted in the general case reconciliation that runs after this.
     717            0 :         let origin_secondary_conf = build_location_config(
     718            0 :             &self.shard,
     719            0 :             &self.config,
     720            0 :             LocationConfigMode::Secondary,
     721            0 :             None,
     722            0 :             Some(LocationConfigSecondary { warm: true }),
     723            0 :         );
     724            0 :         self.location_config(&origin_ps, origin_secondary_conf.clone(), None, false)
     725            0 :             .await?;
     726              :         // TODO: we should also be setting the ObservedState on earlier API calls, in case we fail
     727              :         // partway through.  In fact, all location conf API calls should be in a wrapper that sets
     728              :         // the observed state to None, then runs, then sets it to what we wrote.
     729            0 :         self.observed.locations.insert(
     730            0 :             origin_ps.get_id(),
     731            0 :             ObservedStateLocation {
     732            0 :                 conf: Some(origin_secondary_conf),
     733            0 :             },
     734            0 :         );
     735            0 : 
     736            0 :         pausable_failpoint!("reconciler-live-migrate-post-detach");
     737              : 
     738            0 :         tracing::info!("🔁 Switching to AttachedSingle mode on node {dest_ps}",);
     739            0 :         let dest_final_conf = build_location_config(
     740            0 :             &self.shard,
     741            0 :             &self.config,
     742            0 :             LocationConfigMode::AttachedSingle,
     743            0 :             self.generation,
     744            0 :             None,
     745            0 :         );
     746            0 :         self.location_config(&dest_ps, dest_final_conf.clone(), None, false)
     747            0 :             .await?;
     748            0 :         self.observed.locations.insert(
     749            0 :             dest_ps.get_id(),
     750            0 :             ObservedStateLocation {
     751            0 :                 conf: Some(dest_final_conf),
     752            0 :             },
     753            0 :         );
     754            0 : 
     755            0 :         tracing::info!("✅ Migration complete");
     756              : 
     757            0 :         Ok(())
     758            0 :     }
     759              : 
     760            0 :     async fn maybe_refresh_observed(&mut self) -> Result<(), ReconcileError> {
     761              :         // If the attached node has uncertain state, read it from the pageserver before proceeding: this
     762              :         // is important to avoid spurious generation increments.
     763              :         //
     764              :         // We don't need to do this for secondary/detach locations because it's harmless to just PUT their
     765              :         // location conf, whereas for attached locations it can interrupt clients if we spuriously destroy/recreate
     766              :         // the `Timeline` object in the pageserver.
     767              : 
     768            0 :         let Some(attached_node) = self.intent.attached.as_ref() else {
     769              :             // Nothing to do
     770            0 :             return Ok(());
     771              :         };
     772              : 
     773            0 :         if matches!(
     774            0 :             self.observed.locations.get(&attached_node.get_id()),
     775              :             Some(ObservedStateLocation { conf: None })
     776              :         ) {
     777            0 :             let tenant_shard_id = self.tenant_shard_id;
     778            0 :             let observed_conf = match attached_node
     779            0 :                 .with_client_retries(
     780            0 :                     |client| async move { client.get_location_config(tenant_shard_id).await },
     781            0 :                     &self.service_config.pageserver_jwt_token,
     782            0 :                     &self.service_config.ssl_ca_cert,
     783            0 :                     1,
     784            0 :                     1,
     785            0 :                     Duration::from_secs(5),
     786            0 :                     &self.cancel,
     787            0 :                 )
     788            0 :                 .await
     789              :             {
     790            0 :                 Some(Ok(observed)) => Some(observed),
     791            0 :                 Some(Err(mgmt_api::Error::ApiError(status, _msg)))
     792            0 :                     if status == StatusCode::NOT_FOUND =>
     793            0 :                 {
     794            0 :                     None
     795              :                 }
     796            0 :                 Some(Err(e)) => return Err(e.into()),
     797            0 :                 None => return Err(ReconcileError::Cancel),
     798              :             };
     799            0 :             tracing::info!("Scanned location configuration on {attached_node}: {observed_conf:?}");
     800            0 :             match observed_conf {
     801            0 :                 Some(conf) => {
     802            0 :                     // Pageserver returned a state: update it in observed.  This may still be an indeterminate (None) state,
     803            0 :                     // if internally the pageserver's TenantSlot was being mutated (e.g. some long running API call is still running)
     804            0 :                     self.observed
     805            0 :                         .locations
     806            0 :                         .insert(attached_node.get_id(), ObservedStateLocation { conf });
     807            0 :                 }
     808            0 :                 None => {
     809            0 :                     // Pageserver returned 404: we have confirmation that there is no state for this shard on that pageserver.
     810            0 :                     self.observed.locations.remove(&attached_node.get_id());
     811            0 :                 }
     812              :             }
     813            0 :         }
     814              : 
     815            0 :         Ok(())
     816            0 :     }
     817              : 
     818              :     /// Reconciling a tenant makes API calls to pageservers until the observed state
     819              :     /// matches the intended state.
     820              :     ///
     821              :     /// First we apply special case handling (e.g. for live migrations), and then a
     822              :     /// general case reconciliation where we walk through the intent by pageserver
     823              :     /// and call out to the pageserver to apply the desired state.
     824              :     ///
     825              :     /// An Ok(()) result indicates that we successfully attached the tenant, but _not_ that
     826              :     /// all locations for the tenant are in the expected state. When nodes that are to be detached
     827              :     /// or configured as secondary are unavailable, we may return Ok(()) but leave the shard in a
     828              :     /// state where it still requires later reconciliation.
     829            0 :     pub(crate) async fn reconcile(&mut self) -> Result<(), ReconcileError> {
     830            0 :         // Prepare: if we have uncertain `observed` state for our would-be attachement location, then refresh it
     831            0 :         self.maybe_refresh_observed().await?;
     832              : 
     833              :         // Special case: live migration
     834            0 :         self.maybe_live_migrate().await?;
     835              : 
     836              :         // If the attached pageserver is not attached, do so now.
     837            0 :         if let Some(node) = self.intent.attached.as_ref() {
     838              :             // If we are in an attached policy, then generation must have been set (null generations
     839              :             // are only present when a tenant is initially loaded with a secondary policy)
     840            0 :             debug_assert!(self.generation.is_some());
     841            0 :             let Some(generation) = self.generation else {
     842            0 :                 return Err(ReconcileError::Other(anyhow::anyhow!(
     843            0 :                     "Attempted to attach with NULL generation"
     844            0 :                 )));
     845              :             };
     846              : 
     847            0 :             let mut wanted_conf = attached_location_conf(
     848            0 :                 generation,
     849            0 :                 &self.shard,
     850            0 :                 &self.config,
     851            0 :                 &self.placement_policy,
     852            0 :             );
     853            0 :             match self.observed.locations.get(&node.get_id()) {
     854            0 :                 Some(conf) if conf.conf.as_ref() == Some(&wanted_conf) => {
     855            0 :                     // Nothing to do
     856            0 :                     tracing::info!(node_id=%node.get_id(), "Observed configuration already correct.")
     857              :                 }
     858            0 :                 observed => {
     859              :                     // In all cases other than a matching observed configuration, we will
     860              :                     // reconcile this location.  This includes locations with different configurations, as well
     861              :                     // as locations with unknown (None) observed state.
     862              : 
     863              :                     // Incrementing generation is the safe general case, but is inefficient for changes that only
     864              :                     // modify some details (e.g. the tenant's config).
     865            0 :                     let increment_generation = match observed {
     866            0 :                         None => true,
     867            0 :                         Some(ObservedStateLocation { conf: None }) => true,
     868              :                         Some(ObservedStateLocation {
     869            0 :                             conf: Some(observed),
     870            0 :                         }) => {
     871            0 :                             let generations_match = observed.generation == wanted_conf.generation;
     872            0 : 
     873            0 :                             // We may skip incrementing the generation if the location is already in the expected mode and
     874            0 :                             // generation.  In principle it would also be safe to skip from certain other modes (e.g. AttachedStale),
     875            0 :                             // but such states are handled inside `live_migrate`, and if we see that state here we're cleaning up
     876            0 :                             // after a restart/crash, so fall back to the universally safe path of incrementing generation.
     877            0 :                             !generations_match || (observed.mode != wanted_conf.mode)
     878              :                         }
     879              :                     };
     880              : 
     881            0 :                     if increment_generation {
     882            0 :                         pausable_failpoint!("reconciler-pre-increment-generation");
     883              : 
     884            0 :                         let generation = self
     885            0 :                             .persistence
     886            0 :                             .increment_generation(self.tenant_shard_id, node.get_id())
     887            0 :                             .await?;
     888            0 :                         self.generation = Some(generation);
     889            0 :                         wanted_conf.generation = generation.into();
     890            0 :                     }
     891              : 
     892            0 :                     let diff = match observed {
     893              :                         Some(ObservedStateLocation {
     894            0 :                             conf: Some(observed),
     895            0 :                         }) => {
     896            0 :                             let diff = JsonDiff::diff(
     897            0 :                                 &serde_json::to_value(observed.clone()).unwrap(),
     898            0 :                                 &serde_json::to_value(wanted_conf.clone()).unwrap(),
     899            0 :                                 false,
     900            0 :                             );
     901              : 
     902            0 :                             if let Some(json_diff) = diff.diff {
     903            0 :                                 serde_json::to_string(&json_diff).unwrap_or("diff err".to_string())
     904              :                             } else {
     905            0 :                                 "unknown".to_string()
     906              :                             }
     907              :                         }
     908            0 :                         _ => "full".to_string(),
     909              :                     };
     910              : 
     911            0 :                     tracing::info!(node_id=%node.get_id(), "Observed configuration requires update: {diff}");
     912              : 
     913              :                     // Because `node` comes from a ref to &self, clone it before calling into a &mut self
     914              :                     // function: this could be avoided by refactoring the state mutated by location_config into
     915              :                     // a separate type to Self.
     916            0 :                     let node = node.clone();
     917            0 : 
     918            0 :                     // Use lazy=true, because we may run many of Self concurrently, and do not want to
     919            0 :                     // overload the pageserver with logical size calculations.
     920            0 :                     self.location_config(&node, wanted_conf, None, true).await?;
     921            0 :                     self.compute_notify().await?;
     922              :                 }
     923              :             }
     924            0 :         }
     925              : 
     926              :         // Configure secondary locations: if these were previously attached this
     927              :         // implicitly downgrades them from attached to secondary.
     928            0 :         let mut changes = Vec::new();
     929            0 :         for node in &self.intent.secondary {
     930            0 :             let wanted_conf = secondary_location_conf(&self.shard, &self.config);
     931            0 :             match self.observed.locations.get(&node.get_id()) {
     932            0 :                 Some(conf) if conf.conf.as_ref() == Some(&wanted_conf) => {
     933            0 :                     // Nothing to do
     934            0 :                     tracing::info!(node_id=%node.get_id(), "Observed configuration already correct.")
     935              :                 }
     936              :                 _ => {
     937              :                     // Only try and configure secondary locations on nodes that are available.  This
     938              :                     // allows the reconciler to "succeed" while some secondaries are offline (e.g. after
     939              :                     // a node failure, where the failed node will have a secondary intent)
     940            0 :                     if node.is_available() {
     941            0 :                         tracing::info!(node_id=%node.get_id(), "Observed configuration requires update.");
     942            0 :                         changes.push((node.clone(), wanted_conf))
     943              :                     } else {
     944            0 :                         tracing::info!(node_id=%node.get_id(), "Skipping configuration as secondary, node is unavailable");
     945            0 :                         self.observed
     946            0 :                             .locations
     947            0 :                             .insert(node.get_id(), ObservedStateLocation { conf: None });
     948              :                     }
     949              :                 }
     950              :             }
     951              :         }
     952              : 
     953              :         // Detach any extraneous pageservers that are no longer referenced
     954              :         // by our intent.
     955            0 :         for node in &self.detach {
     956            0 :             changes.push((
     957            0 :                 node.clone(),
     958            0 :                 LocationConfig {
     959            0 :                     mode: LocationConfigMode::Detached,
     960            0 :                     generation: None,
     961            0 :                     secondary_conf: None,
     962            0 :                     shard_number: self.shard.number.0,
     963            0 :                     shard_count: self.shard.count.literal(),
     964            0 :                     shard_stripe_size: self.shard.stripe_size.0,
     965            0 :                     tenant_conf: self.config.clone(),
     966            0 :                 },
     967            0 :             ));
     968            0 :         }
     969              : 
     970            0 :         for (node, conf) in changes {
     971            0 :             if self.cancel.is_cancelled() {
     972            0 :                 return Err(ReconcileError::Cancel);
     973            0 :             }
     974            0 :             // We only try to configure secondary locations if the node is available.  This does
     975            0 :             // not stop us succeeding with the reconcile, because our core goal is to make the
     976            0 :             // shard _available_ (the attached location), and configuring secondary locations
     977            0 :             // can be done lazily when the node becomes available (via background reconciliation).
     978            0 :             if node.is_available() {
     979            0 :                 self.location_config(&node, conf, None, false).await?;
     980              :             } else {
     981              :                 // If the node is unavailable, we skip and consider the reconciliation successful: this
     982              :                 // is a common case where a pageserver is marked unavailable: we demote a location on
     983              :                 // that unavailable pageserver to secondary.
     984            0 :                 tracing::info!("Skipping configuring secondary location {node}, it is unavailable");
     985            0 :                 self.observed
     986            0 :                     .locations
     987            0 :                     .insert(node.get_id(), ObservedStateLocation { conf: None });
     988              :             }
     989              :         }
     990              : 
     991              :         // The condition below identifies a detach. We must have no attached intent and
     992              :         // must have been attached to something previously. Pass this information to
     993              :         // the [`ComputeHook`] such that it can update its tenant-wide state.
     994            0 :         if self.intent.attached.is_none() && !self.detach.is_empty() {
     995            0 :             // TODO: Consider notifying control plane about detaches. This would avoid situations
     996            0 :             // where the compute tries to start-up with a stale set of pageservers.
     997            0 :             self.compute_hook
     998            0 :                 .handle_detach(self.tenant_shard_id, self.shard.stripe_size);
     999            0 :         }
    1000              : 
    1001            0 :         pausable_failpoint!("reconciler-epilogue");
    1002              : 
    1003            0 :         Ok(())
    1004            0 :     }
    1005              : 
    1006            0 :     pub(crate) async fn compute_notify(&mut self) -> Result<(), NotifyError> {
    1007              :         // Whenever a particular Reconciler emits a notification, it is always notifying for the intended
    1008              :         // destination.
    1009            0 :         if let Some(node) = &self.intent.attached {
    1010            0 :             let result = self
    1011            0 :                 .compute_hook
    1012            0 :                 .notify(
    1013            0 :                     compute_hook::ShardUpdate {
    1014            0 :                         tenant_shard_id: self.tenant_shard_id,
    1015            0 :                         node_id: node.get_id(),
    1016            0 :                         stripe_size: self.shard.stripe_size,
    1017            0 :                         preferred_az: self.preferred_az.as_ref().map(Cow::Borrowed),
    1018            0 :                     },
    1019            0 :                     &self.cancel,
    1020            0 :                 )
    1021            0 :                 .await;
    1022            0 :             if let Err(e) = &result {
    1023              :                 // Set this flag so that in our ReconcileResult we will set the flag on the shard that it
    1024              :                 // needs to retry at some point.
    1025            0 :                 self.compute_notify_failure = true;
    1026              : 
    1027              :                 // It is up to the caller whether they want to drop out on this error, but they don't have to:
    1028              :                 // in general we should avoid letting unavailability of the cloud control plane stop us from
    1029              :                 // making progress.
    1030            0 :                 match e {
    1031            0 :                     // 404s from cplane during tenant creation are expected.
    1032            0 :                     // Cplane only persists the shards to the database after
    1033            0 :                     // creating the tenant and the timeline. If we notify before
    1034            0 :                     // that, we'll get a 404.
    1035            0 :                     //
    1036            0 :                     // This is fine because tenant creations happen via /location_config
    1037            0 :                     // and that returns the list of locations in the response. Hence, we
    1038            0 :                     // silence the error and return Ok(()) here. Reconciliation will still
    1039            0 :                     // be retried because we set [`Reconciler::compute_notify_failure`] above.
    1040            0 :                     NotifyError::Unexpected(hyper::StatusCode::NOT_FOUND)
    1041            0 :                         if self.reconciler_config.tenant_creation_hint() =>
    1042            0 :                     {
    1043            0 :                         return Ok(());
    1044              :                     }
    1045            0 :                     NotifyError::ShuttingDown => {}
    1046              :                     _ => {
    1047            0 :                         tracing::warn!(
    1048            0 :                             "Failed to notify compute of attached pageserver {node}: {e}"
    1049              :                         );
    1050              :                     }
    1051              :                 }
    1052            0 :             }
    1053            0 :             result
    1054              :         } else {
    1055            0 :             Ok(())
    1056              :         }
    1057            0 :     }
    1058              : 
    1059              :     /// Compare the observed state snapshot from when the reconcile was created
    1060              :     /// with the final observed state in order to generate observed state deltas.
    1061            0 :     pub(crate) fn observed_deltas(&self) -> Vec<ObservedStateDelta> {
    1062            0 :         let mut deltas = Vec::default();
    1063              : 
    1064            0 :         for (node_id, location) in &self.observed.locations {
    1065            0 :             let previous_location = self.original_observed.locations.get(node_id);
    1066            0 :             let do_upsert = match previous_location {
    1067              :                 // Location config changed for node
    1068            0 :                 Some(prev) if location.conf != prev.conf => true,
    1069              :                 // New location config for node
    1070            0 :                 None => true,
    1071              :                 // Location config has not changed for node
    1072            0 :                 _ => false,
    1073              :             };
    1074              : 
    1075            0 :             if do_upsert {
    1076            0 :                 deltas.push(ObservedStateDelta::Upsert(Box::new((
    1077            0 :                     *node_id,
    1078            0 :                     location.clone(),
    1079            0 :                 ))));
    1080            0 :             }
    1081              :         }
    1082              : 
    1083            0 :         for node_id in self.original_observed.locations.keys() {
    1084            0 :             if !self.observed.locations.contains_key(node_id) {
    1085            0 :                 deltas.push(ObservedStateDelta::Delete(*node_id));
    1086            0 :             }
    1087              :         }
    1088              : 
    1089            0 :         deltas
    1090            0 :     }
    1091              : 
    1092              :     /// Keep trying to notify the compute indefinitely, only dropping out if:
    1093              :     /// - the node `origin` becomes unavailable -> Ok(())
    1094              :     /// - the node `origin` no longer has our tenant shard attached -> Ok(())
    1095              :     /// - our cancellation token fires -> Err(ReconcileError::Cancelled)
    1096              :     ///
    1097              :     /// This is used during live migration, where we do not wish to detach
    1098              :     /// an origin location until the compute definitely knows about the new
    1099              :     /// location.
    1100              :     ///
    1101              :     /// In cases where the origin node becomes unavailable, we return success, indicating
    1102              :     /// to the caller that they should continue irrespective of whether the compute was notified,
    1103              :     /// because the origin node is unusable anyway.  Notification will be retried later via the
    1104              :     /// [`Self::compute_notify_failure`] flag.
    1105            0 :     async fn compute_notify_blocking(&mut self, origin: &Node) -> Result<(), ReconcileError> {
    1106            0 :         let mut notify_attempts = 0;
    1107            0 :         while let Err(e) = self.compute_notify().await {
    1108            0 :             match e {
    1109            0 :                 NotifyError::Fatal(_) => return Err(ReconcileError::Notify(e)),
    1110            0 :                 NotifyError::ShuttingDown => return Err(ReconcileError::Cancel),
    1111              :                 _ => {
    1112            0 :                     tracing::warn!(
    1113            0 :                         "Live migration blocked by compute notification error, retrying: {e}"
    1114              :                     );
    1115              :                 }
    1116              :             }
    1117              : 
    1118              :             // Did the origin pageserver become unavailable?
    1119            0 :             if !origin.is_available() {
    1120            0 :                 tracing::info!("Giving up on compute notification because {origin} is unavailable");
    1121            0 :                 break;
    1122            0 :             }
    1123            0 : 
    1124            0 :             // Does the origin pageserver still host the shard we are interested in?  We should only
    1125            0 :             // continue waiting for compute notification to be acked if the old location is still usable.
    1126            0 :             let tenant_shard_id = self.tenant_shard_id;
    1127            0 :             match origin
    1128            0 :                 .with_client_retries(
    1129            0 :                     |client| async move { client.get_location_config(tenant_shard_id).await },
    1130            0 :                     &self.service_config.pageserver_jwt_token,
    1131            0 :                     &self.service_config.ssl_ca_cert,
    1132            0 :                     1,
    1133            0 :                     3,
    1134            0 :                     Duration::from_secs(5),
    1135            0 :                     &self.cancel,
    1136            0 :                 )
    1137            0 :                 .await
    1138              :             {
    1139            0 :                 Some(Ok(Some(location_conf))) => {
    1140            0 :                     if matches!(
    1141            0 :                         location_conf.mode,
    1142              :                         LocationConfigMode::AttachedMulti
    1143              :                             | LocationConfigMode::AttachedSingle
    1144              :                             | LocationConfigMode::AttachedStale
    1145              :                     ) {
    1146            0 :                         tracing::debug!(
    1147            0 :                             "Still attached to {origin}, will wait & retry compute notification"
    1148              :                         );
    1149              :                     } else {
    1150            0 :                         tracing::info!(
    1151            0 :                             "Giving up on compute notification because {origin} is in state {:?}",
    1152              :                             location_conf.mode
    1153              :                         );
    1154            0 :                         return Ok(());
    1155              :                     }
    1156              :                     // Fall through
    1157              :                 }
    1158              :                 Some(Ok(None)) => {
    1159            0 :                     tracing::info!(
    1160            0 :                         "No longer attached to {origin}, giving up on compute notification"
    1161              :                     );
    1162            0 :                     return Ok(());
    1163              :                 }
    1164            0 :                 Some(Err(e)) => {
    1165            0 :                     match e {
    1166              :                         mgmt_api::Error::Cancelled => {
    1167            0 :                             tracing::info!(
    1168            0 :                                 "Giving up on compute notification because {origin} is unavailable"
    1169              :                             );
    1170            0 :                             return Ok(());
    1171              :                         }
    1172              :                         mgmt_api::Error::ApiError(StatusCode::NOT_FOUND, _) => {
    1173            0 :                             tracing::info!(
    1174            0 :                                 "No longer attached to {origin}, giving up on compute notification"
    1175              :                             );
    1176            0 :                             return Ok(());
    1177              :                         }
    1178            0 :                         e => {
    1179            0 :                             // Other API errors are unexpected here.
    1180            0 :                             tracing::warn!("Unexpected error checking location on {origin}: {e}");
    1181              : 
    1182              :                             // Fall through, we will retry compute notification.
    1183              :                         }
    1184              :                     }
    1185              :                 }
    1186            0 :                 None => return Err(ReconcileError::Cancel),
    1187              :             };
    1188              : 
    1189            0 :             exponential_backoff(
    1190            0 :                 notify_attempts,
    1191            0 :                 // Generous waits: control plane operations which might be blocking us usually complete on the order
    1192            0 :                 // of hundreds to thousands of milliseconds, so no point busy polling.
    1193            0 :                 1.0,
    1194            0 :                 10.0,
    1195            0 :                 &self.cancel,
    1196            0 :             )
    1197            0 :             .await;
    1198            0 :             notify_attempts += 1;
    1199              :         }
    1200              : 
    1201            0 :         Ok(())
    1202            0 :     }
    1203              : }
    1204              : 
    1205              : /// We tweak the externally-set TenantConfig while configuring
    1206              : /// locations, using our awareness of whether secondary locations
    1207              : /// are in use to automatically enable/disable heatmap uploads.
    1208            0 : fn ha_aware_config(config: &TenantConfig, has_secondaries: bool) -> TenantConfig {
    1209            0 :     let mut config = config.clone();
    1210            0 :     if has_secondaries {
    1211            0 :         if config.heatmap_period.is_none() {
    1212            0 :             config.heatmap_period = Some(DEFAULT_HEATMAP_PERIOD);
    1213            0 :         }
    1214            0 :     } else {
    1215            0 :         config.heatmap_period = None;
    1216            0 :     }
    1217            0 :     config
    1218            0 : }
    1219              : 
    1220            0 : pub(crate) fn attached_location_conf(
    1221            0 :     generation: Generation,
    1222            0 :     shard: &ShardIdentity,
    1223            0 :     config: &TenantConfig,
    1224            0 :     policy: &PlacementPolicy,
    1225            0 : ) -> LocationConfig {
    1226            0 :     let has_secondaries = match policy {
    1227              :         PlacementPolicy::Attached(0) | PlacementPolicy::Detached | PlacementPolicy::Secondary => {
    1228            0 :             false
    1229              :         }
    1230            0 :         PlacementPolicy::Attached(_) => true,
    1231              :     };
    1232              : 
    1233            0 :     LocationConfig {
    1234            0 :         mode: LocationConfigMode::AttachedSingle,
    1235            0 :         generation: generation.into(),
    1236            0 :         secondary_conf: None,
    1237            0 :         shard_number: shard.number.0,
    1238            0 :         shard_count: shard.count.literal(),
    1239            0 :         shard_stripe_size: shard.stripe_size.0,
    1240            0 :         tenant_conf: ha_aware_config(config, has_secondaries),
    1241            0 :     }
    1242            0 : }
    1243              : 
    1244            0 : pub(crate) fn secondary_location_conf(
    1245            0 :     shard: &ShardIdentity,
    1246            0 :     config: &TenantConfig,
    1247            0 : ) -> LocationConfig {
    1248            0 :     LocationConfig {
    1249            0 :         mode: LocationConfigMode::Secondary,
    1250            0 :         generation: None,
    1251            0 :         secondary_conf: Some(LocationConfigSecondary { warm: true }),
    1252            0 :         shard_number: shard.number.0,
    1253            0 :         shard_count: shard.count.literal(),
    1254            0 :         shard_stripe_size: shard.stripe_size.0,
    1255            0 :         tenant_conf: ha_aware_config(config, true),
    1256            0 :     }
    1257            0 : }
        

Generated by: LCOV version 2.1-beta