LCOV - code coverage report
Current view: top level - libs/pageserver_api/src - controller_api.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 29.3 % 116 34
Test Date: 2025-02-20 13:11:02 Functions: 2.2 % 314 7

            Line data    Source code
       1              : use std::collections::{HashMap, HashSet};
       2              : use std::fmt::Display;
       3              : use std::str::FromStr;
       4              : use std::time::{Duration, Instant};
       5              : 
       6              : /// Request/response types for the storage controller
       7              : /// API (`/control/v1` prefix).  Implemented by the server
       8              : /// in [`storage_controller::http`]
       9              : use serde::{Deserialize, Serialize};
      10              : use utils::id::{NodeId, TenantId};
      11              : 
      12              : use crate::models::PageserverUtilization;
      13              : use crate::{
      14              :     models::{ShardParameters, TenantConfig},
      15              :     shard::{ShardStripeSize, TenantShardId},
      16              : };
      17              : 
      18            2 : #[derive(Serialize, Deserialize, Debug)]
      19              : #[serde(deny_unknown_fields)]
      20              : pub struct TenantCreateRequest {
      21              :     pub new_tenant_id: TenantShardId,
      22              :     #[serde(default)]
      23              :     #[serde(skip_serializing_if = "Option::is_none")]
      24              :     pub generation: Option<u32>,
      25              : 
      26              :     // If omitted, create a single shard with TenantShardId::unsharded()
      27              :     #[serde(default)]
      28              :     #[serde(skip_serializing_if = "ShardParameters::is_unsharded")]
      29              :     pub shard_parameters: ShardParameters,
      30              : 
      31              :     #[serde(default)]
      32              :     #[serde(skip_serializing_if = "Option::is_none")]
      33              :     pub placement_policy: Option<PlacementPolicy>,
      34              : 
      35              :     #[serde(flatten)]
      36              :     pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
      37              : }
      38              : 
      39            0 : #[derive(Serialize, Deserialize)]
      40              : pub struct TenantCreateResponseShard {
      41              :     pub shard_id: TenantShardId,
      42              :     pub node_id: NodeId,
      43              :     pub generation: u32,
      44              : }
      45              : 
      46            0 : #[derive(Serialize, Deserialize)]
      47              : pub struct TenantCreateResponse {
      48              :     pub shards: Vec<TenantCreateResponseShard>,
      49              : }
      50              : 
      51            0 : #[derive(Serialize, Deserialize, Debug, Clone)]
      52              : pub struct NodeRegisterRequest {
      53              :     pub node_id: NodeId,
      54              : 
      55              :     pub listen_pg_addr: String,
      56              :     pub listen_pg_port: u16,
      57              : 
      58              :     pub listen_http_addr: String,
      59              :     pub listen_http_port: u16,
      60              : 
      61              :     pub availability_zone_id: AvailabilityZone,
      62              : }
      63              : 
      64            0 : #[derive(Serialize, Deserialize)]
      65              : pub struct NodeConfigureRequest {
      66              :     pub node_id: NodeId,
      67              : 
      68              :     pub availability: Option<NodeAvailabilityWrapper>,
      69              :     pub scheduling: Option<NodeSchedulingPolicy>,
      70              : }
      71              : 
      72            0 : #[derive(Serialize, Deserialize)]
      73              : pub struct TenantPolicyRequest {
      74              :     pub placement: Option<PlacementPolicy>,
      75              :     pub scheduling: Option<ShardSchedulingPolicy>,
      76              : }
      77              : 
      78            0 : #[derive(Clone, Serialize, Deserialize, PartialEq, Eq, Hash, Debug, PartialOrd, Ord)]
      79              : pub struct AvailabilityZone(pub String);
      80              : 
      81              : impl Display for AvailabilityZone {
      82          300 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
      83          300 :         write!(f, "{}", self.0)
      84          300 :     }
      85              : }
      86              : 
      87            0 : #[derive(Serialize, Deserialize)]
      88              : pub struct ShardsPreferredAzsRequest {
      89              :     #[serde(flatten)]
      90              :     pub preferred_az_ids: HashMap<TenantShardId, Option<AvailabilityZone>>,
      91              : }
      92              : 
      93            0 : #[derive(Serialize, Deserialize)]
      94              : pub struct ShardsPreferredAzsResponse {
      95              :     pub updated: Vec<TenantShardId>,
      96              : }
      97              : 
      98            0 : #[derive(Serialize, Deserialize, Debug)]
      99              : pub struct TenantLocateResponseShard {
     100              :     pub shard_id: TenantShardId,
     101              :     pub node_id: NodeId,
     102              : 
     103              :     pub listen_pg_addr: String,
     104              :     pub listen_pg_port: u16,
     105              : 
     106              :     pub listen_http_addr: String,
     107              :     pub listen_http_port: u16,
     108              : }
     109              : 
     110            0 : #[derive(Serialize, Deserialize)]
     111              : pub struct TenantLocateResponse {
     112              :     pub shards: Vec<TenantLocateResponseShard>,
     113              :     pub shard_params: ShardParameters,
     114              : }
     115              : 
     116            0 : #[derive(Serialize, Deserialize, Debug)]
     117              : pub struct TenantDescribeResponse {
     118              :     pub tenant_id: TenantId,
     119              :     pub shards: Vec<TenantDescribeResponseShard>,
     120              :     pub stripe_size: ShardStripeSize,
     121              :     pub policy: PlacementPolicy,
     122              :     pub config: TenantConfig,
     123              : }
     124              : 
     125            0 : #[derive(Serialize, Deserialize, Debug)]
     126              : pub struct NodeShardResponse {
     127              :     pub node_id: NodeId,
     128              :     pub shards: Vec<NodeShard>,
     129              : }
     130              : 
     131            0 : #[derive(Serialize, Deserialize, Debug)]
     132              : pub struct NodeShard {
     133              :     pub tenant_shard_id: TenantShardId,
     134              :     /// Whether the shard is observed secondary on a specific node. True = yes, False = no, None = not on this node.
     135              :     pub is_observed_secondary: Option<bool>,
     136              :     /// Whether the shard is intended to be a secondary on a specific node. True = yes, False = no, None = not on this node.
     137              :     pub is_intended_secondary: Option<bool>,
     138              : }
     139              : 
     140            0 : #[derive(Serialize, Deserialize)]
     141              : pub struct NodeDescribeResponse {
     142              :     pub id: NodeId,
     143              : 
     144              :     pub availability: NodeAvailabilityWrapper,
     145              :     pub scheduling: NodeSchedulingPolicy,
     146              : 
     147              :     pub availability_zone_id: String,
     148              : 
     149              :     pub listen_http_addr: String,
     150              :     pub listen_http_port: u16,
     151              : 
     152              :     pub listen_pg_addr: String,
     153              :     pub listen_pg_port: u16,
     154              : }
     155              : 
     156            0 : #[derive(Serialize, Deserialize, Debug)]
     157              : pub struct TenantDescribeResponseShard {
     158              :     pub tenant_shard_id: TenantShardId,
     159              : 
     160              :     pub node_attached: Option<NodeId>,
     161              :     pub node_secondary: Vec<NodeId>,
     162              : 
     163              :     pub last_error: String,
     164              : 
     165              :     /// A task is currently running to reconcile this tenant's intent state with the state on pageservers
     166              :     pub is_reconciling: bool,
     167              :     /// This shard failed in sending a compute notification to the cloud control plane, and a retry is pending.
     168              :     pub is_pending_compute_notification: bool,
     169              :     /// A shard split is currently underway
     170              :     pub is_splitting: bool,
     171              : 
     172              :     pub scheduling_policy: ShardSchedulingPolicy,
     173              : 
     174              :     pub preferred_az_id: Option<String>,
     175              : }
     176              : 
     177              : /// Migration request for a given tenant shard to a given node.
     178              : ///
     179              : /// Explicitly migrating a particular shard is a low level operation
     180              : /// TODO: higher level "Reschedule tenant" operation where the request
     181              : /// specifies some constraints, e.g. asking it to get off particular node(s)
     182            0 : #[derive(Serialize, Deserialize, Debug)]
     183              : pub struct TenantShardMigrateRequest {
     184              :     pub node_id: NodeId,
     185              :     #[serde(default)]
     186              :     pub migration_config: Option<MigrationConfig>,
     187              : }
     188              : 
     189            0 : #[derive(Serialize, Deserialize, Debug)]
     190              : pub struct MigrationConfig {
     191              :     #[serde(default)]
     192              :     #[serde(with = "humantime_serde")]
     193              :     pub secondary_warmup_timeout: Option<Duration>,
     194              :     #[serde(default)]
     195              :     #[serde(with = "humantime_serde")]
     196              :     pub secondary_download_request_timeout: Option<Duration>,
     197              : }
     198              : 
     199              : #[derive(Serialize, Clone, Debug)]
     200              : #[serde(into = "NodeAvailabilityWrapper")]
     201              : pub enum NodeAvailability {
     202              :     // Normal, happy state
     203              :     Active(PageserverUtilization),
     204              :     // Node is warming up, but we expect it to become available soon. Covers
     205              :     // the time span between the re-attach response being composed on the storage controller
     206              :     // and the first successful heartbeat after the processing of the re-attach response
     207              :     // finishes on the pageserver.
     208              :     WarmingUp(Instant),
     209              :     // Offline: Tenants shouldn't try to attach here, but they may assume that their
     210              :     // secondary locations on this node still exist.  Newly added nodes are in this
     211              :     // state until we successfully contact them.
     212              :     Offline,
     213              : }
     214              : 
     215              : impl PartialEq for NodeAvailability {
     216            0 :     fn eq(&self, other: &Self) -> bool {
     217              :         use NodeAvailability::*;
     218            0 :         matches!(
     219            0 :             (self, other),
     220              :             (Active(_), Active(_)) | (Offline, Offline) | (WarmingUp(_), WarmingUp(_))
     221              :         )
     222            0 :     }
     223              : }
     224              : 
     225              : impl Eq for NodeAvailability {}
     226              : 
     227              : // This wrapper provides serde functionality and it should only be used to
     228              : // communicate with external callers which don't know or care about the
     229              : // utilisation score of the pageserver it is targeting.
     230            0 : #[derive(Serialize, Deserialize, Clone, Copy, Debug)]
     231              : pub enum NodeAvailabilityWrapper {
     232              :     Active,
     233              :     WarmingUp,
     234              :     Offline,
     235              : }
     236              : 
     237              : impl From<NodeAvailabilityWrapper> for NodeAvailability {
     238            0 :     fn from(val: NodeAvailabilityWrapper) -> Self {
     239            0 :         match val {
     240              :             // Assume the worst utilisation score to begin with. It will later be updated by
     241              :             // the heartbeats.
     242              :             NodeAvailabilityWrapper::Active => {
     243            0 :                 NodeAvailability::Active(PageserverUtilization::full())
     244              :             }
     245            0 :             NodeAvailabilityWrapper::WarmingUp => NodeAvailability::WarmingUp(Instant::now()),
     246            0 :             NodeAvailabilityWrapper::Offline => NodeAvailability::Offline,
     247              :         }
     248            0 :     }
     249              : }
     250              : 
     251              : impl From<NodeAvailability> for NodeAvailabilityWrapper {
     252            0 :     fn from(val: NodeAvailability) -> Self {
     253            0 :         match val {
     254            0 :             NodeAvailability::Active(_) => NodeAvailabilityWrapper::Active,
     255            0 :             NodeAvailability::WarmingUp(_) => NodeAvailabilityWrapper::WarmingUp,
     256            0 :             NodeAvailability::Offline => NodeAvailabilityWrapper::Offline,
     257              :         }
     258            0 :     }
     259              : }
     260              : 
     261              : /// Scheduling policy enables us to selectively disable some automatic actions that the
     262              : /// controller performs on a tenant shard. This is only set to a non-default value by
     263              : /// human intervention, and it is reset to the default value (Active) when the tenant's
     264              : /// placement policy is modified away from Attached.
     265              : ///
     266              : /// The typical use of a non-Active scheduling policy is one of:
     267              : /// - Pinnning a shard to a node (i.e. migrating it there & setting a non-Active scheduling policy)
     268              : /// - Working around a bug (e.g. if something is flapping and we need to stop it until the bug is fixed)
     269              : ///
     270              : /// If you're not sure which policy to use to pin a shard to its current location, you probably
     271              : /// want Pause.
     272            0 : #[derive(Serialize, Deserialize, Clone, Copy, Eq, PartialEq, Debug)]
     273              : pub enum ShardSchedulingPolicy {
     274              :     // Normal mode: the tenant's scheduled locations may be updated at will, including
     275              :     // for non-essential optimization.
     276              :     Active,
     277              : 
     278              :     // Disable optimizations, but permit scheduling when necessary to fulfil the PlacementPolicy.
     279              :     // For example, this still permits a node's attachment location to change to a secondary in
     280              :     // response to a node failure, or to assign a new secondary if a node was removed.
     281              :     Essential,
     282              : 
     283              :     // No scheduling: leave the shard running wherever it currently is.  Even if the shard is
     284              :     // unavailable, it will not be rescheduled to another node.
     285              :     Pause,
     286              : 
     287              :     // No reconciling: we will make no location_conf API calls to pageservers at all.  If the
     288              :     // shard is unavailable, it stays that way.  If a node fails, this shard doesn't get failed over.
     289              :     Stop,
     290              : }
     291              : 
     292              : impl Default for ShardSchedulingPolicy {
     293        12841 :     fn default() -> Self {
     294        12841 :         Self::Active
     295        12841 :     }
     296              : }
     297              : 
     298            0 : #[derive(Serialize, Deserialize, Clone, Copy, Eq, PartialEq, Debug)]
     299              : pub enum NodeSchedulingPolicy {
     300              :     Active,
     301              :     Filling,
     302              :     Pause,
     303              :     PauseForRestart,
     304              :     Draining,
     305              : }
     306              : 
     307              : impl FromStr for NodeSchedulingPolicy {
     308              :     type Err = anyhow::Error;
     309              : 
     310            0 :     fn from_str(s: &str) -> Result<Self, Self::Err> {
     311            0 :         match s {
     312            0 :             "active" => Ok(Self::Active),
     313            0 :             "filling" => Ok(Self::Filling),
     314            0 :             "pause" => Ok(Self::Pause),
     315            0 :             "pause_for_restart" => Ok(Self::PauseForRestart),
     316            0 :             "draining" => Ok(Self::Draining),
     317            0 :             _ => Err(anyhow::anyhow!("Unknown scheduling state '{s}'")),
     318              :         }
     319            0 :     }
     320              : }
     321              : 
     322              : impl From<NodeSchedulingPolicy> for String {
     323            0 :     fn from(value: NodeSchedulingPolicy) -> String {
     324              :         use NodeSchedulingPolicy::*;
     325            0 :         match value {
     326            0 :             Active => "active",
     327            0 :             Filling => "filling",
     328            0 :             Pause => "pause",
     329            0 :             PauseForRestart => "pause_for_restart",
     330            0 :             Draining => "draining",
     331              :         }
     332            0 :         .to_string()
     333            0 :     }
     334              : }
     335              : 
     336            0 : #[derive(Serialize, Deserialize, Clone, Copy, Eq, PartialEq, Debug)]
     337              : pub enum SkSchedulingPolicy {
     338              :     Active,
     339              :     Pause,
     340              :     Decomissioned,
     341              : }
     342              : 
     343              : impl FromStr for SkSchedulingPolicy {
     344              :     type Err = anyhow::Error;
     345              : 
     346            0 :     fn from_str(s: &str) -> Result<Self, Self::Err> {
     347            0 :         Ok(match s {
     348            0 :             "active" => Self::Active,
     349            0 :             "pause" => Self::Pause,
     350            0 :             "decomissioned" => Self::Decomissioned,
     351              :             _ => {
     352            0 :                 return Err(anyhow::anyhow!(
     353            0 :                     "Unknown scheduling policy '{s}', try active,pause,decomissioned"
     354            0 :                 ))
     355              :             }
     356              :         })
     357            0 :     }
     358              : }
     359              : 
     360              : impl From<SkSchedulingPolicy> for String {
     361            0 :     fn from(value: SkSchedulingPolicy) -> String {
     362              :         use SkSchedulingPolicy::*;
     363            0 :         match value {
     364            0 :             Active => "active",
     365            0 :             Pause => "pause",
     366            0 :             Decomissioned => "decomissioned",
     367              :         }
     368            0 :         .to_string()
     369            0 :     }
     370              : }
     371              : 
     372              : /// Controls how tenant shards are mapped to locations on pageservers, e.g. whether
     373              : /// to create secondary locations.
     374            2 : #[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Eq)]
     375              : pub enum PlacementPolicy {
     376              :     /// Normal live state: one attached pageserver and zero or more secondaries.
     377              :     Attached(usize),
     378              :     /// Create one secondary mode locations. This is useful when onboarding
     379              :     /// a tenant, or for an idle tenant that we might want to bring online quickly.
     380              :     Secondary,
     381              : 
     382              :     /// Do not attach to any pageservers.  This is appropriate for tenants that
     383              :     /// have been idle for a long time, where we do not mind some delay in making
     384              :     /// them available in future.
     385              :     Detached,
     386              : }
     387              : 
     388              : impl PlacementPolicy {
     389           53 :     pub fn want_secondaries(&self) -> usize {
     390           53 :         match self {
     391           50 :             PlacementPolicy::Attached(secondary_count) => *secondary_count,
     392            3 :             PlacementPolicy::Secondary => 1,
     393            0 :             PlacementPolicy::Detached => 0,
     394              :         }
     395           53 :     }
     396              : }
     397              : 
     398            0 : #[derive(Serialize, Deserialize, Debug)]
     399              : pub struct TenantShardMigrateResponse {}
     400              : 
     401              : /// Metadata health record posted from scrubber.
     402            0 : #[derive(Serialize, Deserialize, Debug)]
     403              : pub struct MetadataHealthRecord {
     404              :     pub tenant_shard_id: TenantShardId,
     405              :     pub healthy: bool,
     406              :     pub last_scrubbed_at: chrono::DateTime<chrono::Utc>,
     407              : }
     408              : 
     409            0 : #[derive(Serialize, Deserialize, Debug)]
     410              : pub struct MetadataHealthUpdateRequest {
     411              :     pub healthy_tenant_shards: HashSet<TenantShardId>,
     412              :     pub unhealthy_tenant_shards: HashSet<TenantShardId>,
     413              : }
     414              : 
     415            0 : #[derive(Serialize, Deserialize, Debug)]
     416              : pub struct MetadataHealthUpdateResponse {}
     417              : 
     418            0 : #[derive(Serialize, Deserialize, Debug)]
     419              : pub struct MetadataHealthListUnhealthyResponse {
     420              :     pub unhealthy_tenant_shards: Vec<TenantShardId>,
     421              : }
     422              : 
     423            0 : #[derive(Serialize, Deserialize, Debug)]
     424              : pub struct MetadataHealthListOutdatedRequest {
     425              :     #[serde(with = "humantime_serde")]
     426              :     pub not_scrubbed_for: Duration,
     427              : }
     428              : 
     429            0 : #[derive(Serialize, Deserialize, Debug)]
     430              : pub struct MetadataHealthListOutdatedResponse {
     431              :     pub health_records: Vec<MetadataHealthRecord>,
     432              : }
     433              : 
     434              : /// Publicly exposed safekeeper description
     435            0 : #[derive(Serialize, Deserialize, Clone)]
     436              : pub struct SafekeeperDescribeResponse {
     437              :     pub id: NodeId,
     438              :     pub region_id: String,
     439              :     /// 1 is special, it means just created (not currently posted to storcon).
     440              :     /// Zero or negative is not really expected.
     441              :     /// Otherwise the number from `release-$(number_of_commits_on_branch)` tag.
     442              :     pub version: i64,
     443              :     pub host: String,
     444              :     pub port: i32,
     445              :     pub http_port: i32,
     446              :     pub availability_zone_id: String,
     447              :     pub scheduling_policy: SkSchedulingPolicy,
     448              : }
     449              : 
     450            0 : #[derive(Serialize, Deserialize, Clone)]
     451              : pub struct SafekeeperSchedulingPolicyRequest {
     452              :     pub scheduling_policy: SkSchedulingPolicy,
     453              : }
     454              : 
     455              : #[cfg(test)]
     456              : mod test {
     457              :     use super::*;
     458              :     use serde_json;
     459              : 
     460              :     /// Check stability of PlacementPolicy's serialization
     461              :     #[test]
     462            1 :     fn placement_policy_encoding() -> anyhow::Result<()> {
     463            1 :         let v = PlacementPolicy::Attached(1);
     464            1 :         let encoded = serde_json::to_string(&v)?;
     465            1 :         assert_eq!(encoded, "{\"Attached\":1}");
     466            1 :         assert_eq!(serde_json::from_str::<PlacementPolicy>(&encoded)?, v);
     467              : 
     468            1 :         let v = PlacementPolicy::Detached;
     469            1 :         let encoded = serde_json::to_string(&v)?;
     470            1 :         assert_eq!(encoded, "\"Detached\"");
     471            1 :         assert_eq!(serde_json::from_str::<PlacementPolicy>(&encoded)?, v);
     472            1 :         Ok(())
     473            1 :     }
     474              : 
     475              :     #[test]
     476            1 :     fn test_reject_unknown_field() {
     477            1 :         let id = TenantId::generate();
     478            1 :         let create_request = serde_json::json!({
     479            1 :             "new_tenant_id": id.to_string(),
     480            1 :             "unknown_field": "unknown_value".to_string(),
     481            1 :         });
     482            1 :         let err = serde_json::from_value::<TenantCreateRequest>(create_request).unwrap_err();
     483            1 :         assert!(
     484            1 :             err.to_string().contains("unknown field `unknown_field`"),
     485            0 :             "expect unknown field `unknown_field` error, got: {}",
     486              :             err
     487              :         );
     488            1 :     }
     489              : }
        

Generated by: LCOV version 2.1-beta