From 8cce27bedb11a780dd59095437541c26ff6e62dd Mon Sep 17 00:00:00 2001 From: Vlad Lazar Date: Mon, 14 Apr 2025 16:31:32 +0100 Subject: [PATCH 01/27] pageserver: add a randomized read path test (#11519) ## Problem Every time we make changes to the read path to fix a bug or add a feature, we end up adding another incomprehensible test. ## Summary of changes Add some generic infrastructure for generating a layer map from a type spec and use that for a read path test. The test is randomized but uses a fixed seed by default. A fuzzing mode is available for confidence building. See [Notion page](https://www.notion.so/neondatabase/Read-Path-Unit-Testing-Fuzzing-1d1f189e0047806c8e5cd37781b0a350?pvs=4) for a diagram of the layer map used. Just for fun I tried removing [this commit](https://github.com/neondatabase/neon/pull/11494/commits/9990199cb4f26a4ab2e64372bd1301f734ddcef5) from https://github.com/neondatabase/neon/pull/11494 and it caught the bug in the normal mode (no fuzzing required). --- pageserver/Cargo.toml | 2 + pageserver/src/tenant.rs | 528 ++++++++++++++++++++++++++++++ pageserver/src/tenant/timeline.rs | 2 +- 3 files changed, 531 insertions(+), 1 deletion(-) diff --git a/pageserver/Cargo.toml b/pageserver/Cargo.toml index 56d97bf8a9..74f3fce6e5 100644 --- a/pageserver/Cargo.toml +++ b/pageserver/Cargo.toml @@ -10,6 +10,8 @@ default = [] # which adds some runtime cost to run tests on outage conditions testing = ["fail/failpoints", "pageserver_api/testing", "wal_decoder/testing", "pageserver_client/testing"] +fuzz-read-path = ["testing"] + [dependencies] anyhow.workspace = true arc-swap.workspace = true diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index f14c7608fd..0ba70f45b2 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -5933,12 +5933,20 @@ mod tests { use models::CompactLsnRange; use pageserver_api::key::{AUX_KEY_PREFIX, Key, NON_INHERITED_RANGE, RELATION_SIZE_PREFIX}; use pageserver_api::keyspace::KeySpace; + #[cfg(feature = "testing")] + use pageserver_api::keyspace::KeySpaceRandomAccum; use pageserver_api::models::{CompactionAlgorithm, CompactionAlgorithmSettings}; #[cfg(feature = "testing")] use pageserver_api::record::NeonWalRecord; use pageserver_api::value::Value; use pageserver_compaction::helpers::overlaps_with; + #[cfg(feature = "testing")] + use rand::SeedableRng; + #[cfg(feature = "testing")] + use rand::rngs::StdRng; use rand::{Rng, thread_rng}; + #[cfg(feature = "testing")] + use std::ops::Range; use storage_layer::{IoConcurrency, PersistentLayerKey}; use tests::storage_layer::ValuesReconstructState; use tests::timeline::{GetVectoredError, ShutdownMode}; @@ -5960,6 +5968,318 @@ mod tests { static TEST_KEY: Lazy = Lazy::new(|| Key::from_slice(&hex!("010000000033333333444444445500000001"))); + #[cfg(feature = "testing")] + struct TestTimelineSpecification { + start_lsn: Lsn, + last_record_lsn: Lsn, + + in_memory_layers_shape: Vec<(Range, Range)>, + delta_layers_shape: Vec<(Range, Range)>, + image_layers_shape: Vec<(Range, Lsn)>, + + gap_chance: u8, + will_init_chance: u8, + } + + #[cfg(feature = "testing")] + struct Storage { + storage: HashMap<(Key, Lsn), Value>, + start_lsn: Lsn, + } + + #[cfg(feature = "testing")] + impl Storage { + fn get(&self, key: Key, lsn: Lsn) -> Bytes { + use bytes::BufMut; + + let mut crnt_lsn = lsn; + let mut got_base = false; + + let mut acc = Vec::new(); + + while crnt_lsn >= self.start_lsn { + if let Some(value) = self.storage.get(&(key, crnt_lsn)) { + acc.push(value.clone()); + + match value { + Value::WalRecord(NeonWalRecord::Test { will_init, .. }) => { + if *will_init { + got_base = true; + break; + } + } + Value::Image(_) => { + got_base = true; + break; + } + _ => unreachable!(), + } + } + + crnt_lsn = crnt_lsn.checked_sub(1u64).unwrap(); + } + + assert!( + got_base, + "Input data was incorrect. No base image for {key}@{lsn}" + ); + + tracing::debug!("Wal redo depth for {key}@{lsn} is {}", acc.len()); + + let mut blob = BytesMut::new(); + for value in acc.into_iter().rev() { + match value { + Value::WalRecord(NeonWalRecord::Test { append, .. }) => { + blob.extend_from_slice(append.as_bytes()); + } + Value::Image(img) => { + blob.put(img); + } + _ => unreachable!(), + } + } + + blob.into() + } + } + + #[cfg(feature = "testing")] + #[allow(clippy::too_many_arguments)] + async fn randomize_timeline( + tenant: &Arc, + new_timeline_id: TimelineId, + pg_version: u32, + spec: TestTimelineSpecification, + random: &mut rand::rngs::StdRng, + ctx: &RequestContext, + ) -> anyhow::Result<(Arc, Storage, Vec)> { + let mut storage: HashMap<(Key, Lsn), Value> = HashMap::default(); + let mut interesting_lsns = vec![spec.last_record_lsn]; + + for (key_range, lsn_range) in spec.in_memory_layers_shape.iter() { + let mut lsn = lsn_range.start; + while lsn < lsn_range.end { + let mut key = key_range.start; + while key < key_range.end { + let gap = random.gen_range(1..=100) <= spec.gap_chance; + let will_init = random.gen_range(1..=100) <= spec.will_init_chance; + + if gap { + continue; + } + + let record = if will_init { + Value::WalRecord(NeonWalRecord::wal_init(format!("[wil_init {key}@{lsn}]"))) + } else { + Value::WalRecord(NeonWalRecord::wal_append(format!("[delta {key}@{lsn}]"))) + }; + + storage.insert((key, lsn), record); + + key = key.next(); + } + lsn = Lsn(lsn.0 + 1); + } + + // Stash some interesting LSN for future use + for offset in [0, 5, 100].iter() { + if *offset == 0 { + interesting_lsns.push(lsn_range.start); + } else { + let below = lsn_range.start.checked_sub(*offset); + match below { + Some(v) if v >= spec.start_lsn => { + interesting_lsns.push(v); + } + _ => {} + } + + let above = Lsn(lsn_range.start.0 + offset); + interesting_lsns.push(above); + } + } + } + + for (key_range, lsn_range) in spec.delta_layers_shape.iter() { + let mut lsn = lsn_range.start; + while lsn < lsn_range.end { + let mut key = key_range.start; + while key < key_range.end { + let gap = random.gen_range(1..=100) <= spec.gap_chance; + let will_init = random.gen_range(1..=100) <= spec.will_init_chance; + + if gap { + continue; + } + + let record = if will_init { + Value::WalRecord(NeonWalRecord::wal_init(format!("[wil_init {key}@{lsn}]"))) + } else { + Value::WalRecord(NeonWalRecord::wal_append(format!("[delta {key}@{lsn}]"))) + }; + + storage.insert((key, lsn), record); + + key = key.next(); + } + lsn = Lsn(lsn.0 + 1); + } + + // Stash some interesting LSN for future use + for offset in [0, 5, 100].iter() { + if *offset == 0 { + interesting_lsns.push(lsn_range.start); + } else { + let below = lsn_range.start.checked_sub(*offset); + match below { + Some(v) if v >= spec.start_lsn => { + interesting_lsns.push(v); + } + _ => {} + } + + let above = Lsn(lsn_range.start.0 + offset); + interesting_lsns.push(above); + } + } + } + + for (key_range, lsn) in spec.image_layers_shape.iter() { + let mut key = key_range.start; + while key < key_range.end { + let blob = Bytes::from(format!("[image {key}@{lsn}]")); + let record = Value::Image(blob.clone()); + storage.insert((key, *lsn), record); + + key = key.next(); + } + + // Stash some interesting LSN for future use + for offset in [0, 5, 100].iter() { + if *offset == 0 { + interesting_lsns.push(*lsn); + } else { + let below = lsn.checked_sub(*offset); + match below { + Some(v) if v >= spec.start_lsn => { + interesting_lsns.push(v); + } + _ => {} + } + + let above = Lsn(lsn.0 + offset); + interesting_lsns.push(above); + } + } + } + + let in_memory_test_layers = { + let mut acc = Vec::new(); + + for (key_range, lsn_range) in spec.in_memory_layers_shape.iter() { + let mut data = Vec::new(); + + let mut lsn = lsn_range.start; + while lsn < lsn_range.end { + let mut key = key_range.start; + while key < key_range.end { + if let Some(record) = storage.get(&(key, lsn)) { + data.push((key, lsn, record.clone())); + } + + key = key.next(); + } + lsn = Lsn(lsn.0 + 1); + } + + acc.push(InMemoryLayerTestDesc { + data, + lsn_range: lsn_range.clone(), + is_open: false, + }) + } + + acc + }; + + let delta_test_layers = { + let mut acc = Vec::new(); + + for (key_range, lsn_range) in spec.delta_layers_shape.iter() { + let mut data = Vec::new(); + + let mut lsn = lsn_range.start; + while lsn < lsn_range.end { + let mut key = key_range.start; + while key < key_range.end { + if let Some(record) = storage.get(&(key, lsn)) { + data.push((key, lsn, record.clone())); + } + + key = key.next(); + } + lsn = Lsn(lsn.0 + 1); + } + + acc.push(DeltaLayerTestDesc { + data, + lsn_range: lsn_range.clone(), + key_range: key_range.clone(), + }) + } + + acc + }; + + let image_test_layers = { + let mut acc = Vec::new(); + + for (key_range, lsn) in spec.image_layers_shape.iter() { + let mut data = Vec::new(); + + let mut key = key_range.start; + while key < key_range.end { + if let Some(record) = storage.get(&(key, *lsn)) { + let blob = match record { + Value::Image(blob) => blob.clone(), + _ => unreachable!(), + }; + + data.push((key, blob)); + } + + key = key.next(); + } + + acc.push((*lsn, data)); + } + + acc + }; + + let tline = tenant + .create_test_timeline_with_layers( + new_timeline_id, + spec.start_lsn, + pg_version, + ctx, + in_memory_test_layers, + delta_test_layers, + image_test_layers, + spec.last_record_lsn, + ) + .await?; + + Ok(( + tline, + Storage { + storage, + start_lsn: spec.start_lsn, + }, + interesting_lsns, + )) + } + #[tokio::test] async fn test_basic() -> anyhow::Result<()> { let (tenant, ctx) = TenantHarness::create("test_basic").await?.load().await; @@ -10543,6 +10863,214 @@ mod tests { Ok(()) } + // A randomized read path test. Generates a layer map according to a deterministic + // specification. Fills the (key, LSN) space in random manner and then performs + // random scattered queries validating the results against in-memory storage. + // + // See this internal Notion page for a diagram of the layer map: + // https://www.notion.so/neondatabase/Read-Path-Unit-Testing-Fuzzing-1d1f189e0047806c8e5cd37781b0a350?pvs=4 + // + // A fuzzing mode is also supported. In this mode, the test will use a random + // seed instead of a hardcoded one. Use it in conjunction with `cargo stress` + // to run multiple instances in parallel: + // + // $ RUST_BACKTRACE=1 RUST_LOG=INFO \ + // cargo stress --package=pageserver --features=testing,fuzz-read-path --release -- test_read_path + #[cfg(feature = "testing")] + #[tokio::test] + async fn test_read_path() -> anyhow::Result<()> { + use rand::seq::SliceRandom; + + let seed = if cfg!(feature = "fuzz-read-path") { + let seed: u64 = thread_rng().r#gen(); + seed + } else { + // Use a hard-coded seed when not in fuzzing mode. + // Note that with the current approach results are not reproducible + // accross platforms and Rust releases. + const SEED: u64 = 0; + SEED + }; + + let mut random = StdRng::seed_from_u64(seed); + + let (queries, will_init_chance, gap_chance) = if cfg!(feature = "fuzz-read-path") { + const QUERIES: u64 = 5000; + let will_init_chance: u8 = random.gen_range(0..=10); + let gap_chance: u8 = random.gen_range(0..=50); + + (QUERIES, will_init_chance, gap_chance) + } else { + const QUERIES: u64 = 1000; + const WILL_INIT_CHANCE: u8 = 1; + const GAP_CHANCE: u8 = 5; + + (QUERIES, WILL_INIT_CHANCE, GAP_CHANCE) + }; + + let harness = TenantHarness::create("test_read_path").await?; + let (tenant, ctx) = harness.load().await; + + tracing::info!("Using random seed: {seed}"); + tracing::info!(%will_init_chance, %gap_chance, "Fill params"); + + // Define the layer map shape. Note that this part is not randomized. + + const KEY_DIMENSION_SIZE: u32 = 99; + let start_key = Key::from_hex("110000000033333333444444445500000000").unwrap(); + let end_key = start_key.add(KEY_DIMENSION_SIZE); + let total_key_range = start_key..end_key; + let total_key_range_size = end_key.to_i128() - start_key.to_i128(); + let total_start_lsn = Lsn(104); + let last_record_lsn = Lsn(504); + + assert!(total_key_range_size % 3 == 0); + + let in_memory_layers_shape = vec![ + (total_key_range.clone(), Lsn(304)..Lsn(400)), + (total_key_range.clone(), Lsn(400)..last_record_lsn), + ]; + + let delta_layers_shape = vec![ + ( + start_key..(start_key.add((total_key_range_size / 3) as u32)), + Lsn(200)..Lsn(304), + ), + ( + (start_key.add((total_key_range_size / 3) as u32)) + ..(start_key.add((total_key_range_size * 2 / 3) as u32)), + Lsn(200)..Lsn(304), + ), + ( + (start_key.add((total_key_range_size * 2 / 3) as u32)) + ..(start_key.add(total_key_range_size as u32)), + Lsn(200)..Lsn(304), + ), + ]; + + let image_layers_shape = vec![ + ( + start_key.add((total_key_range_size * 2 / 3 - 10) as u32) + ..start_key.add((total_key_range_size * 2 / 3 + 10) as u32), + Lsn(456), + ), + ( + start_key.add((total_key_range_size / 3 - 10) as u32) + ..start_key.add((total_key_range_size / 3 + 10) as u32), + Lsn(256), + ), + (total_key_range.clone(), total_start_lsn), + ]; + + let specification = TestTimelineSpecification { + start_lsn: total_start_lsn, + last_record_lsn, + in_memory_layers_shape, + delta_layers_shape, + image_layers_shape, + gap_chance, + will_init_chance, + }; + + // Create and randomly fill in the layers according to the specification + let (tline, storage, interesting_lsns) = randomize_timeline( + &tenant, + TIMELINE_ID, + DEFAULT_PG_VERSION, + specification, + &mut random, + &ctx, + ) + .await?; + + // Now generate queries based on the interesting lsns that we've collected. + // + // While there's still room in the query, pick and interesting LSN and a random + // key. Then roll the dice to see if the next key should also be included in + // the query. When the roll fails, break the "batch" and pick another point in the + // (key, LSN) space. + + const PICK_NEXT_CHANCE: u8 = 50; + for _ in 0..queries { + let query = { + let mut keyspaces_at_lsn: HashMap = HashMap::default(); + let mut used_keys: HashSet = HashSet::default(); + + while used_keys.len() < Timeline::MAX_GET_VECTORED_KEYS as usize { + let selected_lsn = interesting_lsns.choose(&mut random).expect("not empty"); + let mut selected_key = start_key.add(random.gen_range(0..KEY_DIMENSION_SIZE)); + + while used_keys.len() < Timeline::MAX_GET_VECTORED_KEYS as usize { + if used_keys.contains(&selected_key) + || selected_key >= start_key.add(KEY_DIMENSION_SIZE) + { + break; + } + + keyspaces_at_lsn + .entry(*selected_lsn) + .or_default() + .add_key(selected_key); + used_keys.insert(selected_key); + + let pick_next = random.gen_range(0..=100) <= PICK_NEXT_CHANCE; + if pick_next { + selected_key = selected_key.next(); + } else { + break; + } + } + } + + VersionedKeySpaceQuery::scattered( + keyspaces_at_lsn + .into_iter() + .map(|(lsn, acc)| (lsn, acc.to_keyspace())) + .collect(), + ) + }; + + // Run the query and validate the results + + let results = tline + .get_vectored(query.clone(), IoConcurrency::Sequential, &ctx) + .await; + + let blobs = match results { + Ok(ok) => ok, + Err(err) => { + panic!("seed={seed} Error returned for query {query}: {err}"); + } + }; + + for (key, key_res) in blobs.into_iter() { + match key_res { + Ok(blob) => { + let requested_at_lsn = query.map_key_to_lsn(&key); + let expected = storage.get(key, requested_at_lsn); + + if blob != expected { + tracing::error!( + "seed={seed} Mismatch for {key}@{requested_at_lsn} from query: {query}" + ); + } + + assert_eq!(blob, expected); + } + Err(err) => { + let requested_at_lsn = query.map_key_to_lsn(&key); + + panic!( + "seed={seed} Error returned for {key}@{requested_at_lsn} from query {query}: {err}" + ); + } + } + } + } + + Ok(()) + } + fn sort_layer_key(k1: &PersistentLayerKey, k2: &PersistentLayerKey) -> std::cmp::Ordering { ( k1.is_delta, diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 204bdb5eee..c27a4b62da 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -4026,7 +4026,7 @@ impl VersionedKeySpaceQuery { /// Returns LSN for a specific key. /// /// Invariant: requested key must be part of [`Self::total_keyspace`] - fn map_key_to_lsn(&self, key: &Key) -> Lsn { + pub(super) fn map_key_to_lsn(&self, key: &Key) -> Lsn { match self { Self::Uniform { lsn, .. } => *lsn, Self::Scattered { keyspaces_at_lsn } => { From 028a191040352db678d0bbe80c79d58b5e84cdd3 Mon Sep 17 00:00:00 2001 From: Tristan Partin Date: Mon, 14 Apr 2025 16:18:21 -0500 Subject: [PATCH 02/27] Continue with s/spec/config changes (#11574) Signed-off-by: Tristan Partin --- compute_tools/src/bin/compute_ctl.rs | 7 +++---- compute_tools/src/compute.rs | 14 +++++--------- 2 files changed, 8 insertions(+), 13 deletions(-) diff --git a/compute_tools/src/bin/compute_ctl.rs b/compute_tools/src/bin/compute_ctl.rs index ea8350e2f5..16fd51d79a 100644 --- a/compute_tools/src/bin/compute_ctl.rs +++ b/compute_tools/src/bin/compute_ctl.rs @@ -139,7 +139,7 @@ fn main() -> Result<()> { let scenario = failpoint_support::init(); - // For historical reasons, the main thread that processes the spec and launches postgres + // For historical reasons, the main thread that processes the config and launches postgres // is synchronous, but we always have this tokio runtime available and we "enter" it so // that you can use tokio::spawn() and tokio::runtime::Handle::current().block_on(...) // from all parts of compute_ctl. @@ -155,7 +155,7 @@ fn main() -> Result<()> { let connstr = Url::parse(&cli.connstr).context("cannot parse connstr as a URL")?; - let cli_spec = get_config(&cli)?; + let config = get_config(&cli)?; let compute_node = ComputeNode::new( ComputeNodeParams { @@ -176,8 +176,7 @@ fn main() -> Result<()> { #[cfg(target_os = "linux")] vm_monitor_addr: cli.vm_monitor_addr, }, - cli_spec.spec, - cli_spec.compute_ctl_config, + config, )?; let exit_code = compute_node.run()?; diff --git a/compute_tools/src/compute.rs b/compute_tools/src/compute.rs index 06d5bbb9c5..c7b4bdd240 100644 --- a/compute_tools/src/compute.rs +++ b/compute_tools/src/compute.rs @@ -11,7 +11,7 @@ use std::{env, fs}; use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use compute_api::privilege::Privilege; -use compute_api::responses::{ComputeCtlConfig, ComputeMetrics, ComputeStatus}; +use compute_api::responses::{ComputeConfig, ComputeCtlConfig, ComputeMetrics, ComputeStatus}; use compute_api::spec::{ ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PgIdent, }; @@ -303,11 +303,7 @@ struct StartVmMonitorResult { } impl ComputeNode { - pub fn new( - params: ComputeNodeParams, - cli_spec: Option, - compute_ctl_config: ComputeCtlConfig, - ) -> Result { + pub fn new(params: ComputeNodeParams, config: ComputeConfig) -> Result { let connstr = params.connstr.as_str(); let conn_conf = postgres::config::Config::from_str(connstr) .context("cannot build postgres config from connstr")?; @@ -315,8 +311,8 @@ impl ComputeNode { .context("cannot build tokio postgres config from connstr")?; let mut new_state = ComputeState::new(); - if let Some(cli_spec) = cli_spec { - let pspec = ParsedSpec::try_from(cli_spec).map_err(|msg| anyhow::anyhow!(msg))?; + if let Some(spec) = config.spec { + let pspec = ParsedSpec::try_from(spec).map_err(|msg| anyhow::anyhow!(msg))?; new_state.pspec = Some(pspec); } @@ -327,7 +323,7 @@ impl ComputeNode { state: Mutex::new(new_state), state_changed: Condvar::new(), ext_download_progress: RwLock::new(HashMap::new()), - compute_ctl_config, + compute_ctl_config: config.compute_ctl_config, }) } From cbd2fc2395a6be13c43a62248d607ee6cada6ee5 Mon Sep 17 00:00:00 2001 From: Tristan Partin Date: Mon, 14 Apr 2025 20:21:18 -0500 Subject: [PATCH 03/27] Clean up logs and error messages in compute_ctl authorize middleware (#11576) Signed-off-by: Tristan Partin --- compute_tools/src/http/middleware/authorize.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/compute_tools/src/http/middleware/authorize.rs b/compute_tools/src/http/middleware/authorize.rs index ee3a5cb953..f221752c38 100644 --- a/compute_tools/src/http/middleware/authorize.rs +++ b/compute_tools/src/http/middleware/authorize.rs @@ -11,7 +11,7 @@ use futures::future::BoxFuture; use http::{Request, Response, StatusCode}; use jsonwebtoken::{Algorithm, DecodingKey, TokenData, Validation, jwk::JwkSet}; use tower_http::auth::AsyncAuthorizeRequest; -use tracing::warn; +use tracing::{debug, warn}; use crate::http::{JsonResponse, extract::RequestId}; @@ -92,7 +92,7 @@ impl AsyncAuthorizeRequest for Authorize { if data.claims.compute_id != compute_id { return Err(JsonResponse::error( StatusCode::UNAUTHORIZED, - "invalid claims in authorization token", + "invalid compute ID in authorization token claims", )); } @@ -112,12 +112,14 @@ impl Authorize { token: &str, validation: &Validation, ) -> Result> { + debug!("verifying token {}", token); + for jwk in jwks.keys.iter() { let decoding_key = match DecodingKey::from_jwk(jwk) { Ok(key) => key, Err(e) => { warn!( - "Failed to construct decoding key from {}: {}", + "failed to construct decoding key from {}: {}", jwk.common.key_id.as_ref().unwrap(), e ); @@ -130,7 +132,7 @@ impl Authorize { Ok(data) => return Ok(data), Err(e) => { warn!( - "Failed to decode authorization token using {}: {}", + "failed to decode authorization token using {}: {}", jwk.common.key_id.as_ref().unwrap(), e ); @@ -140,6 +142,6 @@ impl Authorize { } } - Err(anyhow!("Failed to verify authorization token")) + Err(anyhow!("failed to verify authorization token")) } } From 8c77ccfc01bf5e634bf2adab9543a2f2e1b9420e Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Tue, 15 Apr 2025 09:25:09 +0200 Subject: [PATCH 04/27] pageserver: log total progress during shard ancestor compaction (#11565) ## Problem Shard ancestor compaction doesn't currently log any global progress information, only for the current batch. ## Summary of changes Log the number of layers checked for eligibility this iteration, and the total number of layers to check. This will indicate how far along the total shard ancestor compaction has gotten for this iteration. --- pageserver/src/tenant/layer_map.rs | 2 +- .../src/tenant/layer_map/historic_layer_coverage.rs | 2 +- pageserver/src/tenant/timeline/compaction.rs | 8 ++++++-- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/pageserver/src/tenant/layer_map.rs b/pageserver/src/tenant/layer_map.rs index 96cee922ff..23052ccee7 100644 --- a/pageserver/src/tenant/layer_map.rs +++ b/pageserver/src/tenant/layer_map.rs @@ -714,7 +714,7 @@ impl LayerMap { true } - pub fn iter_historic_layers(&self) -> impl '_ + Iterator> { + pub fn iter_historic_layers(&self) -> impl ExactSizeIterator> { self.historic.iter() } diff --git a/pageserver/src/tenant/layer_map/historic_layer_coverage.rs b/pageserver/src/tenant/layer_map/historic_layer_coverage.rs index b3dc8e56a3..5ccc75fff6 100644 --- a/pageserver/src/tenant/layer_map/historic_layer_coverage.rs +++ b/pageserver/src/tenant/layer_map/historic_layer_coverage.rs @@ -504,7 +504,7 @@ impl BufferedHistoricLayerCoverage { } /// Iterate all the layers - pub fn iter(&self) -> impl '_ + Iterator { + pub fn iter(&self) -> impl ExactSizeIterator { // NOTE we can actually perform this without rebuilding, // but it's not necessary for now. if !self.buffer.is_empty() { diff --git a/pageserver/src/tenant/timeline/compaction.rs b/pageserver/src/tenant/timeline/compaction.rs index 91cc8ca10c..2a450795fd 100644 --- a/pageserver/src/tenant/timeline/compaction.rs +++ b/pageserver/src/tenant/timeline/compaction.rs @@ -1273,7 +1273,10 @@ impl Timeline { let pitr_cutoff = self.gc_info.read().unwrap().cutoffs.time; let layers = self.layers.read().await; - for layer_desc in layers.layer_map()?.iter_historic_layers() { + let layers_iter = layers.layer_map()?.iter_historic_layers(); + let (layers_total, mut layers_checked) = (layers_iter.len(), 0); + for layer_desc in layers_iter { + layers_checked += 1; let layer = layers.get_from_desc(&layer_desc); if layer.metadata().shard.shard_count == self.shard_identity.count { // This layer does not belong to a historic ancestor, no need to re-image it. @@ -1371,7 +1374,8 @@ impl Timeline { } info!( - "starting shard ancestor compaction, rewriting {} layers and dropping {} layers \ + "starting shard ancestor compaction, rewriting {} layers and dropping {} layers, \ + checked {layers_checked}/{layers_total} layers \ (latest_gc_cutoff={} pitr_cutoff={})", layers_to_rewrite.len(), drop_layers.len(), From 9a6ace9bde2b46dd1b46ea2fe28f01c0b62a6745 Mon Sep 17 00:00:00 2001 From: Fedor Dikarev Date: Tue, 15 Apr 2025 10:21:44 +0200 Subject: [PATCH 05/27] introduce new runners: unit-perf and use them for benchmark jobs (#11409) ## Problem Benchmarks results are inconsistent on existing small-metal runners ## Summary of changes Introduce new `unit-perf` runners, and lets run benchmark on them. The new hardware has slower, but consistent, CPU frequency - if run with default governor schedutil. Thus we needed to adjust some testcases' timeouts and add some retry steps where hard-coded timeouts couldn't be increased without changing the system under test. - [wait_for_last_record_lsn](https://github.com/neondatabase/neon/blob/6592d69a6700a2bd2e9f60c22af138ea0dafbdd0/test_runner/fixtures/pageserver/utils.py#L193) 1000s -> 2000s - [test_branch_creation_many](https://github.com/neondatabase/neon/pull/11409/files#diff-2ebfe76f89004d563c7e53e3ca82462e1d85e92e6d5588e8e8f598bbe119e927) 1000s - [test_ingest_insert_bulk](https://github.com/neondatabase/neon/pull/11409/files#diff-e90e685be4a87053bc264a68740969e6a8872c8897b8b748d0e8c5f683a68d9f) - with back throttling disabled compute becomes unresponsive for more than 60 seconds (PG hard-coded client authentication connection timeout) - [test_sharded_ingest](https://github.com/neondatabase/neon/pull/11409/files#diff-e8d870165bd44acb9a6d8350f8640b301c1385a4108430b8d6d659b697e4a3f1) 600s -> 1200s Right now there are only 2 runners of that class, and if we decide to go with them, we have to check how much that type of runners we need, so jobs not stuck with waiting for that type of runners available. However we now decided to run those runners with governor performance instead of schedutil. This achieves almost same performance as previous runners but still achieves consistent results for same commit Related issue to activate performance governor on these runners https://github.com/neondatabase/runner/pull/138 ## Verification that it helps ### analyze runtimes on new runner for same commit Table of runtimes for the same commit on different runners in [run](https://github.com/neondatabase/neon/actions/runs/14417589789) | Run | Benchmarks (1) | Benchmarks (2) |Benchmarks (3) |Benchmarks (4) | Benchmarks (5) | |--------|--------|---------|---------|---------|---------| | 1 | 1950.37s | 6374.55s | 3646.15s | 4149.48s | 2330.22s | | 2 | - | 6369.27s | 3666.65s | 4162.42s | 2329.23s | | Delta % | - | 0,07 % | 0,5 % | 0,3 % | 0,04 % | | with governor performance | 1519.57s | 4131.62s | - | - | - | | second run gov. perf. | 1513.62s | 4134.67s | - | - | - | | Delta % | 0,3 % | 0,07 % | - | - | - | | speedup gov. performance | 22 % | 35 % | - | - | - | | current desktop class hetzner runners (main) | 1487.10s | 3699.67s | - | - | - | | slower than desktop class | 2 % | 12 % | - | - | - | In summary, the runtimes for the same commit on this hardware varies less than 1 %. --------- Co-authored-by: BodoBolero --- .github/actionlint.yml | 1 + .github/workflows/build_and_test.yml | 2 +- explained_queries.sql | 0 test_runner/fixtures/pageserver/utils.py | 2 +- test_runner/performance/test_branch_creation.py | 3 ++- .../performance/test_ingest_insert_bulk.py | 15 ++++++++++++++- test_runner/performance/test_sharded_ingest.py | 2 +- 7 files changed, 20 insertions(+), 5 deletions(-) create mode 100644 explained_queries.sql diff --git a/.github/actionlint.yml b/.github/actionlint.yml index edc456d611..1d1b50e458 100644 --- a/.github/actionlint.yml +++ b/.github/actionlint.yml @@ -6,6 +6,7 @@ self-hosted-runner: - small - small-metal - small-arm64 + - unit-perf - us-east-2 config-variables: - AWS_ECR_REGION diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 46c8cd6fc9..0e67a22bfc 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -284,7 +284,7 @@ jobs: statuses: write contents: write pull-requests: write - runs-on: [ self-hosted, small-metal ] + runs-on: [ self-hosted, unit-perf ] container: image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm credentials: diff --git a/explained_queries.sql b/explained_queries.sql new file mode 100644 index 0000000000..e69de29bb2 diff --git a/test_runner/fixtures/pageserver/utils.py b/test_runner/fixtures/pageserver/utils.py index bc5076758d..8f5234a2fa 100644 --- a/test_runner/fixtures/pageserver/utils.py +++ b/test_runner/fixtures/pageserver/utils.py @@ -199,7 +199,7 @@ def wait_for_last_record_lsn( """waits for pageserver to catch up to a certain lsn, returns the last observed lsn.""" current_lsn = Lsn(0) - for i in range(1000): + for i in range(2000): current_lsn = last_record_lsn(pageserver_http, tenant, timeline) if current_lsn >= lsn: return current_lsn diff --git a/test_runner/performance/test_branch_creation.py b/test_runner/performance/test_branch_creation.py index b2bd94fae7..a3ee30cda2 100644 --- a/test_runner/performance/test_branch_creation.py +++ b/test_runner/performance/test_branch_creation.py @@ -97,6 +97,7 @@ def test_branch_creation_heavy_write(neon_compare: NeonCompare, n_branches: int) _record_branch_creation_durations(neon_compare, branch_creation_durations) +@pytest.mark.timeout(1000) @pytest.mark.parametrize("n_branches", [500, 1024]) @pytest.mark.parametrize("shape", ["one_ancestor", "random"]) def test_branch_creation_many(neon_compare: NeonCompare, n_branches: int, shape: str): @@ -205,7 +206,7 @@ def wait_and_record_startup_metrics( assert len(matching) == len(expected_labels) return matching - samples = wait_until(metrics_are_filled) + samples = wait_until(metrics_are_filled, timeout=60) for sample in samples: phase = sample.labels["phase"] diff --git a/test_runner/performance/test_ingest_insert_bulk.py b/test_runner/performance/test_ingest_insert_bulk.py index 01836b82e9..ed0a6c70bd 100644 --- a/test_runner/performance/test_ingest_insert_bulk.py +++ b/test_runner/performance/test_ingest_insert_bulk.py @@ -52,6 +52,8 @@ def test_ingest_insert_bulk( # would compete with Pageserver for bandwidth. # neon_env_builder.enable_safekeeper_remote_storage(s3_storage()) + neon_env_builder.pageserver_config_override = "wait_lsn_timeout='600 s'" + neon_env_builder.disable_scrub_on_exit() # immediate shutdown may leave stray layers env = neon_env_builder.init_start() @@ -92,7 +94,18 @@ def test_ingest_insert_bulk( worker_rows = rows / CONCURRENCY pool.submit(insert_rows, endpoint, f"table{i}", worker_rows, value) - end_lsn = Lsn(endpoint.safe_psql("select pg_current_wal_lsn()")[0][0]) + for attempt in range(5): + try: + end_lsn = Lsn(endpoint.safe_psql("select pg_current_wal_lsn()")[0][0]) + break + except Exception as e: + # if we disable backpressure, postgres can become unresponsive for longer than a minute + # and new connection attempts time out in postgres after 1 minute + # so if this happens we retry new connection + log.error(f"Attempt {attempt + 1}/5: Failed to select current wal lsn: {e}") + if attempt == 4: + log.error("Exceeded maximum retry attempts for selecting current wal lsn") + raise # Wait for pageserver to ingest the WAL. client = env.pageserver.http_client() diff --git a/test_runner/performance/test_sharded_ingest.py b/test_runner/performance/test_sharded_ingest.py index 94fd54bade..293026d40a 100644 --- a/test_runner/performance/test_sharded_ingest.py +++ b/test_runner/performance/test_sharded_ingest.py @@ -13,7 +13,7 @@ from fixtures.neon_fixtures import ( ) -@pytest.mark.timeout(600) +@pytest.mark.timeout(1200) @pytest.mark.parametrize("shard_count", [1, 8, 32]) @pytest.mark.parametrize( "wal_receiver_protocol", From 63a106021ad7e40cdebaff1ac7dee85e52e8c092 Mon Sep 17 00:00:00 2001 From: Alexander Bayandin Date: Tue, 15 Apr 2025 10:29:36 +0100 Subject: [PATCH 06/27] CI(allure-report-generate): Install allure to /tmp (#11579) ## Problem The `/__w/neon/neon` directory is mounted from host to container and persists between runs. Sometimes the next workflow run fails to delete it: ``` Deleting the contents of '/__w/neon/neon' Error: File was unable to be removed Error: EACCES: permission denied, rmdir '/__w/neon/neon/allure-2.32.2/bin' ``` ## Summary of changes - Download and install allure to `/tmp` which exists in container only Ref https://github.com/neondatabase/cloud/issues/27186 --- .github/actions/allure-report-generate/action.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/actions/allure-report-generate/action.yml b/.github/actions/allure-report-generate/action.yml index b85ca7874d..c27311f24e 100644 --- a/.github/actions/allure-report-generate/action.yml +++ b/.github/actions/allure-report-generate/action.yml @@ -70,6 +70,7 @@ runs: - name: Install Allure shell: bash -euxo pipefail {0} + working-directory: /tmp run: | if ! which allure; then ALLURE_ZIP=allure-${ALLURE_VERSION}.zip From 5be94e28c4d4822847a22887dbca1b7bbda55c61 Mon Sep 17 00:00:00 2001 From: a-masterov <72613290+a-masterov@users.noreply.github.com> Date: Tue, 15 Apr 2025 13:00:25 +0200 Subject: [PATCH 07/27] Update the documentation of the cloud regress test (#11539) ## Problem The information in the README.md contained errors, and some information was missing. ## Summary of changes Found errors are fixed, and new information is added. --------- Co-authored-by: Alexander Bayandin --- test_runner/cloud_regress/README.md | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/test_runner/cloud_regress/README.md b/test_runner/cloud_regress/README.md index 9c460e2764..f341f3c818 100644 --- a/test_runner/cloud_regress/README.md +++ b/test_runner/cloud_regress/README.md @@ -3,19 +3,35 @@ * Create a Neon project on staging. * Grant the superuser privileges to the DB user. * (Optional) create a branch for testing -* Configure the endpoint by updating the control-plane database with the following settings: +* Add the following settings to the `pg_settings` section of the default endpoint configuration for the project using the admin interface: * `Timeone`: `America/Los_Angeles` * `DateStyle`: `Postgres,MDY` * `compute_query_id`: `off` +* Add the following section to the project configuration: +```json +"preload_libraries": { + "use_defaults": false, + "enabled_libraries": [] + } +``` * Checkout the actual `Neon` sources * Patch the sql and expected files for the specific PostgreSQL version, e.g. for v17: ```bash $ cd vendor/postgres-v17 $ patch -p1 <../../compute/patches/cloud_regress_pg17.patch ``` +* Set the environment variables (please modify according your configuration): +```bash +$ export DEFAULT_PG_VERSION=17 +$ export BUILD_TYPE=release +``` +* Build the Neon binaries see [README.md](../../README.md) * Set the environment variable `BENCHMARK_CONNSTR` to the connection URI of your project. -* Set the environment variable `PG_VERSION` to the version of your project. +* Update poetry, run +```bash +$ scripts/pysync +``` * Run ```bash -$ pytest -m remote_cluster -k cloud_regress +$ scripts/pytest -m remote_cluster -k cloud_regress ``` \ No newline at end of file From 19bea5fd0c048f7867cb41f5a1b705951bc1e08b Mon Sep 17 00:00:00 2001 From: Alexander Bayandin Date: Tue, 15 Apr 2025 12:23:41 +0100 Subject: [PATCH 08/27] CI: do not wait for tests to trigger deploy job (#11548) ## Problem There is too much delay between merging a PR into `main` and deploying the changes to staging ## Summary of changes - Trigger `deploy` job without waiting for `build-and-test-locally` job --- .github/workflows/build_and_test.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 0e67a22bfc..80c4511b36 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -1271,7 +1271,7 @@ jobs: exit 1 deploy: - needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, push-neon-image-prod, push-compute-image-prod, meta, build-and-test-locally, trigger-custom-extensions-build-and-wait ] + needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, push-neon-image-prod, push-compute-image-prod, meta, trigger-custom-extensions-build-and-wait ] # `!failure() && !cancelled()` is required because the workflow depends on the job that can be skipped: `push-neon-image-prod` and `push-compute-image-prod` if: ${{ contains(fromJSON('["push-main", "storage-release", "proxy-release", "compute-release"]'), needs.meta.outputs.run-kind) && !failure() && !cancelled() }} permissions: From a4ea7d6194c221e796f89dad563bded89e8bd94f Mon Sep 17 00:00:00 2001 From: "Alex Chi Z." <4198311+skyzh@users.noreply.github.com> Date: Tue, 15 Apr 2025 09:58:32 -0400 Subject: [PATCH 09/27] fix(pageserver): gc-compaction verification false failure (#11564) ## Problem https://github.com/neondatabase/neon/pull/11515 introduced a bug that some key history cannot be verified. If a key only exists above the horizon, the verification will fail for its first occurrence because the history does not exist at that point. As gc-compaction skips a key range whenever an error occurs, it might be doing some wasted work in staging/prod now. But I'm not planning a hotfix this week as the bug doesn't affect correctness/performance. ## Summary of changes Allow keys with only above horizon history in the verification. Signed-off-by: Alex Chi Z --- pageserver/src/tenant/timeline/compaction.rs | 31 +++++++++++++------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/pageserver/src/tenant/timeline/compaction.rs b/pageserver/src/tenant/timeline/compaction.rs index 2a450795fd..3d5f11aeb9 100644 --- a/pageserver/src/tenant/timeline/compaction.rs +++ b/pageserver/src/tenant/timeline/compaction.rs @@ -819,7 +819,15 @@ impl KeyHistoryRetention { base_img: &Option<(Lsn, &Bytes)>, history: &[(Lsn, &NeonWalRecord)], tline: &Arc, + skip_empty: bool, ) -> anyhow::Result<()> { + if base_img.is_none() && history.is_empty() { + if skip_empty { + return Ok(()); + } + anyhow::bail!("verification failed: key {} has no history at {}", key, lsn); + }; + let mut records = history .iter() .map(|(lsn, val)| (*lsn, (*val).clone())) @@ -860,17 +868,12 @@ impl KeyHistoryRetention { if *retain_lsn >= min_lsn { // Only verify after the key appears in the full history for the first time. - if base_img.is_none() && history.is_empty() { - anyhow::bail!( - "verificatoin failed: key {} has no history at {}", - key, - retain_lsn - ); - }; // We don't modify history: in theory, we could replace the history with a single // image as in `generate_key_retention` to make redos at later LSNs faster. But we // want to verify everything as if they are read from the real layer map. - collect_and_verify(key, *retain_lsn, &base_img, &history, tline).await?; + collect_and_verify(key, *retain_lsn, &base_img, &history, tline, false) + .await + .context("below horizon retain_lsn")?; } } @@ -878,13 +881,17 @@ impl KeyHistoryRetention { match val { Value::Image(img) => { // Above the GC horizon, we verify every time we see an image. - collect_and_verify(key, *lsn, &base_img, &history, tline).await?; + collect_and_verify(key, *lsn, &base_img, &history, tline, true) + .await + .context("above horizon full image")?; base_img = Some((*lsn, img)); history.clear(); } Value::WalRecord(rec) if val.will_init() => { // Above the GC horizon, we verify every time we see an init record. - collect_and_verify(key, *lsn, &base_img, &history, tline).await?; + collect_and_verify(key, *lsn, &base_img, &history, tline, true) + .await + .context("above horizon init record")?; base_img = None; history.clear(); history.push((*lsn, rec)); @@ -895,7 +902,9 @@ impl KeyHistoryRetention { } } // Ensure the latest record is readable. - collect_and_verify(key, max_lsn, &base_img, &history, tline).await?; + collect_and_verify(key, max_lsn, &base_img, &history, tline, false) + .await + .context("latest record")?; Ok(()) } } From e31455d93694545998aa3a173c866ae965b91190 Mon Sep 17 00:00:00 2001 From: a-masterov <72613290+a-masterov@users.noreply.github.com> Date: Tue, 15 Apr 2025 16:06:01 +0200 Subject: [PATCH 10/27] Add the tests for the extensions `pg_jsonschema` and `pg_session_jwt` (#11323) ## Problem `pg_jsonschema` and `pg_session_jwt` are not yet covered by tests ## Summary of changes Added the tests for these extensions. --- .../ext-src/pg_jsonschema-src/Makefile | 8 ++ .../expected/jsonschema_edge_cases.out | 87 +++++++++++++++++++ .../expected/jsonschema_valid_api.out | 65 ++++++++++++++ .../sql/jsonschema_edge_cases.sql | 66 ++++++++++++++ .../sql/jsonschema_valid_api.sql | 48 ++++++++++ .../ext-src/pg_session_jwt-src/Makefile | 9 ++ .../expected/basic_functions.out | 35 ++++++++ .../sql/basic_functions.sql | 19 ++++ 8 files changed, 337 insertions(+) create mode 100644 docker-compose/ext-src/pg_jsonschema-src/Makefile create mode 100644 docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_edge_cases.out create mode 100644 docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_valid_api.out create mode 100644 docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_edge_cases.sql create mode 100644 docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_valid_api.sql create mode 100644 docker-compose/ext-src/pg_session_jwt-src/Makefile create mode 100644 docker-compose/ext-src/pg_session_jwt-src/expected/basic_functions.out create mode 100644 docker-compose/ext-src/pg_session_jwt-src/sql/basic_functions.sql diff --git a/docker-compose/ext-src/pg_jsonschema-src/Makefile b/docker-compose/ext-src/pg_jsonschema-src/Makefile new file mode 100644 index 0000000000..d79364d8b5 --- /dev/null +++ b/docker-compose/ext-src/pg_jsonschema-src/Makefile @@ -0,0 +1,8 @@ +EXTENSION = pg_jsonschema +DATA = pg_jsonschema--1.0.sql +REGRESS = jsonschema_valid_api jsonschema_edge_cases +REGRESS_OPTS = --load-extension=pg_jsonschema + +PG_CONFIG ?= pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) diff --git a/docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_edge_cases.out b/docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_edge_cases.out new file mode 100644 index 0000000000..f4089bfb13 --- /dev/null +++ b/docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_edge_cases.out @@ -0,0 +1,87 @@ +-- Schema with enums, nulls, extra properties disallowed +SELECT jsonschema_is_valid('{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false +}'::json); + jsonschema_is_valid +--------------------- + t +(1 row) + +-- Valid enum and null email +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "active", "email": null}'::json +); + jsonschema_validation_errors +------------------------------ + {} +(1 row) + +-- Invalid enum value +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "disabled", "email": null}'::json +); + jsonschema_validation_errors +---------------------------------------------------------------------- + {"\"disabled\" is not one of [\"active\",\"inactive\",\"pending\"]"} +(1 row) + +-- Invalid email format (assuming format is validated) +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "active", "email": "not-an-email"}'::json +); + jsonschema_validation_errors +----------------------------------------- + {"\"not-an-email\" is not a \"email\""} +(1 row) + +-- Extra property not allowed +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "active", "extra": "should not be here"}'::json +); + jsonschema_validation_errors +-------------------------------------------------------------------- + {"Additional properties are not allowed ('extra' was unexpected)"} +(1 row) + diff --git a/docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_valid_api.out b/docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_valid_api.out new file mode 100644 index 0000000000..73f0a562e7 --- /dev/null +++ b/docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_valid_api.out @@ -0,0 +1,65 @@ +-- Define schema +SELECT jsonschema_is_valid('{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] +}'::json); + jsonschema_is_valid +--------------------- + t +(1 row) + +-- Valid instance +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] + }'::json, + '{"username": "alice", "age": 25}'::json +); + jsonschema_validation_errors +------------------------------ + {} +(1 row) + +-- Invalid instance: missing required "username" +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] + }'::json, + '{"age": 25}'::json +); + jsonschema_validation_errors +----------------------------------------- + {"\"username\" is a required property"} +(1 row) + +-- Invalid instance: wrong type for "age" +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] + }'::json, + '{"username": "bob", "age": "twenty"}'::json +); + jsonschema_validation_errors +------------------------------------------- + {"\"twenty\" is not of type \"integer\""} +(1 row) + diff --git a/docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_edge_cases.sql b/docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_edge_cases.sql new file mode 100644 index 0000000000..edad8cca16 --- /dev/null +++ b/docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_edge_cases.sql @@ -0,0 +1,66 @@ +-- Schema with enums, nulls, extra properties disallowed +SELECT jsonschema_is_valid('{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false +}'::json); + +-- Valid enum and null email +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "active", "email": null}'::json +); + +-- Invalid enum value +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "disabled", "email": null}'::json +); + +-- Invalid email format (assuming format is validated) +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "active", "email": "not-an-email"}'::json +); + +-- Extra property not allowed +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["active", "inactive", "pending"] }, + "email": { "type": ["string", "null"], "format": "email" } + }, + "required": ["status"], + "additionalProperties": false + }'::json, + '{"status": "active", "extra": "should not be here"}'::json +); diff --git a/docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_valid_api.sql b/docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_valid_api.sql new file mode 100644 index 0000000000..44539ed6ce --- /dev/null +++ b/docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_valid_api.sql @@ -0,0 +1,48 @@ +-- Define schema +SELECT jsonschema_is_valid('{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] +}'::json); + +-- Valid instance +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] + }'::json, + '{"username": "alice", "age": 25}'::json +); + +-- Invalid instance: missing required "username" +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] + }'::json, + '{"age": 25}'::json +); + +-- Invalid instance: wrong type for "age" +SELECT jsonschema_validation_errors( + '{ + "type": "object", + "properties": { + "username": { "type": "string" }, + "age": { "type": "integer" } + }, + "required": ["username"] + }'::json, + '{"username": "bob", "age": "twenty"}'::json +); diff --git a/docker-compose/ext-src/pg_session_jwt-src/Makefile b/docker-compose/ext-src/pg_session_jwt-src/Makefile new file mode 100644 index 0000000000..c61c9777ad --- /dev/null +++ b/docker-compose/ext-src/pg_session_jwt-src/Makefile @@ -0,0 +1,9 @@ +EXTENSION = pg_session_jwt + +REGRESS = basic_functions +REGRESS_OPTS = --load-extension=$(EXTENSION) +export PGOPTIONS = -c pg_session_jwt.jwk={"crv":"Ed25519","kty":"OKP","x":"R_Abz-63zJ00l-IraL5fQhwkhGVZCSooQFV5ntC3C7M"} + +PG_CONFIG ?= pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) \ No newline at end of file diff --git a/docker-compose/ext-src/pg_session_jwt-src/expected/basic_functions.out b/docker-compose/ext-src/pg_session_jwt-src/expected/basic_functions.out new file mode 100644 index 0000000000..ca54864ecd --- /dev/null +++ b/docker-compose/ext-src/pg_session_jwt-src/expected/basic_functions.out @@ -0,0 +1,35 @@ +-- Basic functionality tests for pg_session_jwt +-- Test auth.init() function +SELECT auth.init(); + init +------ + +(1 row) + +-- Test an invalid JWT +SELECT auth.jwt_session_init('INVALID-JWT'); +ERROR: invalid JWT encoding +-- Test creating a session with an expired JWT +SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjE3NDI1NjQ0MzIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MjQyNDIsInN1YiI6InVzZXIxMjMifQ.A6FwKuaSduHB9O7Gz37g0uoD_U9qVS0JNtT7YABGVgB7HUD1AMFc9DeyhNntWBqncg8k5brv-hrNTuUh5JYMAw'); +ERROR: Token used after it has expired +-- Test creating a session with a valid JWT +SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjQ4OTYxNjQyNTIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MzQzNDMsInN1YiI6InVzZXIxMjMifQ.2TXVgjb6JSUq6_adlvp-m_SdOxZSyGS30RS9TLB0xu2N83dMSs2NybwE1NMU8Fb0tcAZR_ET7M2rSxbTrphfCg'); + jwt_session_init +------------------ + +(1 row) + +-- Test auth.session() function +SELECT auth.session(); + session +------------------------------------------------------------------------- + {"exp": 4896164252, "iat": 1742564252, "jti": 434343, "sub": "user123"} +(1 row) + +-- Test auth.user_id() function +SELECT auth.user_id() AS user_id; + user_id +--------- + user123 +(1 row) + diff --git a/docker-compose/ext-src/pg_session_jwt-src/sql/basic_functions.sql b/docker-compose/ext-src/pg_session_jwt-src/sql/basic_functions.sql new file mode 100644 index 0000000000..6c1ab90c0c --- /dev/null +++ b/docker-compose/ext-src/pg_session_jwt-src/sql/basic_functions.sql @@ -0,0 +1,19 @@ +-- Basic functionality tests for pg_session_jwt + +-- Test auth.init() function +SELECT auth.init(); + +-- Test an invalid JWT +SELECT auth.jwt_session_init('INVALID-JWT'); + +-- Test creating a session with an expired JWT +SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjE3NDI1NjQ0MzIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MjQyNDIsInN1YiI6InVzZXIxMjMifQ.A6FwKuaSduHB9O7Gz37g0uoD_U9qVS0JNtT7YABGVgB7HUD1AMFc9DeyhNntWBqncg8k5brv-hrNTuUh5JYMAw'); + +-- Test creating a session with a valid JWT +SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjQ4OTYxNjQyNTIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MzQzNDMsInN1YiI6InVzZXIxMjMifQ.2TXVgjb6JSUq6_adlvp-m_SdOxZSyGS30RS9TLB0xu2N83dMSs2NybwE1NMU8Fb0tcAZR_ET7M2rSxbTrphfCg'); + +-- Test auth.session() function +SELECT auth.session(); + +-- Test auth.user_id() function +SELECT auth.user_id() AS user_id; \ No newline at end of file From bcef542d5b5411c2111a605dfa288461a2159b3a Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Tue, 15 Apr 2025 16:25:58 +0200 Subject: [PATCH 11/27] pageserver: don't rewrite invisible layers during ancestor compaction (#11580) ## Problem Shard ancestor compaction can be very expensive following shard splits of large tenants. We currently rewrite garbage layers after shard splits as well, which can be a significant amount of data. Touches https://github.com/neondatabase/cloud/issues/22532. ## Summary of changes Don't rewrite invisible layers after shard splits. --- pageserver/src/tenant/timeline/compaction.rs | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/pageserver/src/tenant/timeline/compaction.rs b/pageserver/src/tenant/timeline/compaction.rs index 3d5f11aeb9..5d5149e2d4 100644 --- a/pageserver/src/tenant/timeline/compaction.rs +++ b/pageserver/src/tenant/timeline/compaction.rs @@ -56,7 +56,8 @@ use crate::tenant::storage_layer::batch_split_writer::{ use crate::tenant::storage_layer::filter_iterator::FilterIterator; use crate::tenant::storage_layer::merge_iterator::MergeIterator; use crate::tenant::storage_layer::{ - AsLayerDesc, PersistentLayerDesc, PersistentLayerKey, ValueReconstructState, + AsLayerDesc, LayerVisibilityHint, PersistentLayerDesc, PersistentLayerKey, + ValueReconstructState, }; use crate::tenant::tasks::log_compaction_error; use crate::tenant::timeline::{ @@ -1348,12 +1349,19 @@ impl Timeline { continue; } + // We do not yet implement rewrite of delta layers. if layer_desc.is_delta() { - // We do not yet implement rewrite of delta layers debug!(%layer, "Skipping rewrite of delta layer"); continue; } + // We don't bother rewriting layers that aren't visible, since these won't be needed by + // reads and will likely be garbage collected soon. + if layer.visibility() != LayerVisibilityHint::Visible { + debug!(%layer, "Skipping rewrite of invisible layer"); + continue; + } + // Only rewrite layers if their generations differ. This guarantees: // - that local rewrite is safe, as local layer paths will differ between existing layer and rewritten one // - that the layer is persistent in remote storage, as we only see old-generation'd layer via loading from remote storage From 983d56502bb84d18288cc6498a258896e858a38c Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Tue, 15 Apr 2025 16:26:29 +0200 Subject: [PATCH 12/27] pageserver: reduce shard ancestor rewrite threshold to 30% (#11582) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem When doing power-of-two shard splits (i.e. 4 → 8 → 16), we end up rewriting all layers since half of the pages will be local due to striping. This causes a lot of resource usage when splitting large tenants. ## Summary of changes Drop the threshold of local/total pages to 30%, to reduce the amount of layer rewrites after splits. --- pageserver/src/tenant/timeline/compaction.rs | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/pageserver/src/tenant/timeline/compaction.rs b/pageserver/src/tenant/timeline/compaction.rs index 5d5149e2d4..76c153d60f 100644 --- a/pageserver/src/tenant/timeline/compaction.rs +++ b/pageserver/src/tenant/timeline/compaction.rs @@ -70,6 +70,13 @@ use crate::virtual_file::{MaybeFatalIo, VirtualFile}; /// Maximum number of deltas before generating an image layer in bottom-most compaction. const COMPACTION_DELTA_THRESHOLD: usize = 5; +/// Ratio of shard-local pages below which we trigger shard ancestor layer rewrites. 0.3 means that +/// <= 30% of layer pages must belong to the descendant shard to rewrite the layer. +/// +/// We choose a value < 0.5 to avoid rewriting all visible layers every time we do a power-of-two +/// shard split, which gets expensive for large tenants. +const ANCESTOR_COMPACTION_REWRITE_THRESHOLD: f64 = 0.3; + #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)] pub struct GcCompactionJobId(pub usize); @@ -1330,14 +1337,15 @@ impl Timeline { continue; } - // Don't bother re-writing a layer unless it will at least halve its size + // Only rewrite a layer if we can reclaim significant space. if layer_local_page_count != u32::MAX - && layer_local_page_count > layer_raw_page_count / 2 + && layer_local_page_count as f64 / layer_raw_page_count as f64 + <= ANCESTOR_COMPACTION_REWRITE_THRESHOLD { debug!(%layer, - "layer is already mostly local ({}/{}), not rewriting", - layer_local_page_count, - layer_raw_page_count + "layer has a large share of local pages \ + ({layer_local_page_count}/{layer_raw_page_count} > \ + {ANCESTOR_COMPACTION_REWRITE_THRESHOLD}), not rewriting", ); } From 0f7c2cc382af8bad3d796a485b125f9eacd3cd8b Mon Sep 17 00:00:00 2001 From: Alexander Bayandin Date: Tue, 15 Apr 2025 16:08:05 +0100 Subject: [PATCH 13/27] CI(release): add time to RC PR branch names (#11547) ## Problem We can't have more than one open release PR created on the same day (due to non-unique enough branch names). ## Summary of changes - Add time (hours and minutes) to RC PR branch names - Also make sure we use UTC for releases --- .github/workflows/_create-release-pr.yml | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/.github/workflows/_create-release-pr.yml b/.github/workflows/_create-release-pr.yml index bfbb45e30b..f96ed7d69b 100644 --- a/.github/workflows/_create-release-pr.yml +++ b/.github/workflows/_create-release-pr.yml @@ -53,10 +53,13 @@ jobs: || inputs.component-name == 'Compute' && 'release-compute' }} run: | - today=$(date +'%Y-%m-%d') - echo "title=${COMPONENT_NAME} release ${today}" | tee -a ${GITHUB_OUTPUT} - echo "rc-branch=rc/${RELEASE_BRANCH}/${today}" | tee -a ${GITHUB_OUTPUT} - echo "release-branch=${RELEASE_BRANCH}" | tee -a ${GITHUB_OUTPUT} + now_date=$(date -u +'%Y-%m-%d') + now_time=$(date -u +'%H-%M-%Z') + { + echo "title=${COMPONENT_NAME} release ${now_date}" + echo "rc-branch=rc/${RELEASE_BRANCH}/${now_date}_${now_time}" + echo "release-branch=${RELEASE_BRANCH}" + } | tee -a ${GITHUB_OUTPUT} - name: Configure git run: | From 931f8c43003275a4f90c8ca1b439e92455881830 Mon Sep 17 00:00:00 2001 From: "Alex Chi Z." <4198311+skyzh@users.noreply.github.com> Date: Tue, 15 Apr 2025 11:16:16 -0400 Subject: [PATCH 14/27] fix(pageserver): check if cancelled before waiting logical size (2/2) (#11575) ## Problem close https://github.com/neondatabase/neon/issues/11486, proceeding https://github.com/neondatabase/neon/pull/11531 ## Summary of changes This patch fixes the rest 50% of instability of `test_create_churn_during_restart`. During tenant warmup, we'll request logical size; however, if the startup gets cancelled, we won't be able to spawn the initial logical size calculation task that sets the `cancel_wait_for_background_loop_concurrency_limit_semaphore`. Therefore, we check `cancelled` before proceeding to get `cancel_wait_for_background_loop_concurrency_limit_semaphore`. There will still be a race if the timeline shutdown happens after L5710 and before L5711, but it should be enough to reduce the flakiness of the test. Signed-off-by: Alex Chi Z --- pageserver/src/tenant/timeline.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index c27a4b62da..613834dc88 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -5702,6 +5702,12 @@ impl Timeline { return; } + if self.cancel.is_cancelled() { + // We already requested stopping the tenant, so we cannot wait for the logical size + // calculation to complete given the task might have been already cancelled. + return; + } + if let Some(await_bg_cancel) = self .current_logical_size .cancel_wait_for_background_loop_concurrency_limit_semaphore From c5115518e93022e3231a312444201545af10f245 Mon Sep 17 00:00:00 2001 From: Fedor Dikarev Date: Tue, 15 Apr 2025 17:29:15 +0200 Subject: [PATCH 15/27] remove temp file from repo (#11586) ## Problem In https://github.com/neondatabase/neon/pull/11409 we added temp file to the repo. ## Summary of changes Remove temp file from the repo. --- explained_queries.sql | 0 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 explained_queries.sql diff --git a/explained_queries.sql b/explained_queries.sql deleted file mode 100644 index e69de29bb2..0000000000 From eadb05f78e70ad4e1eb474e6ae3194d840fe99c5 Mon Sep 17 00:00:00 2001 From: Tristan Partin Date: Tue, 15 Apr 2025 12:27:49 -0500 Subject: [PATCH 16/27] Teach neon_local to pass the Authorization header to compute_ctl (#11490) This allows us to remove hacks in the compute_ctl authorization middleware which allowed for bypasses of auth checks. Fixes: https://github.com/neondatabase/neon/issues/11316 Signed-off-by: Tristan Partin --- Cargo.lock | 14 +++- Cargo.toml | 2 + .../src/http/middleware/authorize.rs | 6 +- control_plane/Cargo.toml | 5 ++ control_plane/src/bin/neon_local.rs | 18 +++++ control_plane/src/endpoint.rs | 77 +++++++++++++++---- control_plane/src/local_env.rs | 33 +++++++- control_plane/src/storage_controller.rs | 17 ++-- libs/compute_api/src/responses.rs | 4 +- libs/http-utils/Cargo.toml | 1 + libs/http-utils/src/endpoint.rs | 3 +- libs/utils/Cargo.toml | 1 + libs/utils/src/auth.rs | 32 +++++--- pageserver/Cargo.toml | 1 + pageserver/src/page_service.rs | 3 +- safekeeper/Cargo.toml | 1 + safekeeper/src/handler.rs | 3 +- workspace_hack/Cargo.toml | 1 + 18 files changed, 178 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5d2cdcea27..5c9170b7de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1416,6 +1416,7 @@ name = "control_plane" version = "0.1.0" dependencies = [ "anyhow", + "base64 0.13.1", "camino", "clap", "comfy-table", @@ -1425,10 +1426,13 @@ dependencies = [ "humantime", "humantime-serde", "hyper 0.14.30", + "jsonwebtoken", "nix 0.27.1", "once_cell", "pageserver_api", "pageserver_client", + "pem", + "pkcs8 0.10.2", "postgres_backend", "postgres_connection", "regex", @@ -1437,6 +1441,7 @@ dependencies = [ "scopeguard", "serde", "serde_json", + "sha2", "storage_broker", "thiserror 1.0.69", "tokio", @@ -2817,6 +2822,7 @@ dependencies = [ "hyper 0.14.30", "itertools 0.10.5", "jemalloc_pprof", + "jsonwebtoken", "metrics", "once_cell", "pprof", @@ -4269,6 +4275,7 @@ dependencies = [ "hyper 0.14.30", "indoc", "itertools 0.10.5", + "jsonwebtoken", "md5", "metrics", "nix 0.27.1", @@ -5685,9 +5692,9 @@ dependencies = [ [[package]] name = "ring" -version = "0.17.13" +version = "0.17.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70ac5d832aa16abd7d1def883a8545280c20a60f523a370aa3a9617c2b8550ee" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" dependencies = [ "cc", "cfg-if", @@ -5988,6 +5995,7 @@ dependencies = [ "humantime", "hyper 0.14.30", "itertools 0.10.5", + "jsonwebtoken", "metrics", "once_cell", "pageserver_api", @@ -7872,6 +7880,7 @@ dependencies = [ "metrics", "nix 0.27.1", "once_cell", + "pem", "pin-project-lite", "postgres_connection", "pprof", @@ -8460,6 +8469,7 @@ dependencies = [ "once_cell", "p256 0.13.2", "parquet", + "pkcs8 0.10.2", "prettyplease", "proc-macro2", "prost 0.13.3", diff --git a/Cargo.toml b/Cargo.toml index d957fa9070..8fac3bb46c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -141,7 +141,9 @@ parking_lot = "0.12" parquet = { version = "53", default-features = false, features = ["zstd"] } parquet_derive = "53" pbkdf2 = { version = "0.12.1", features = ["simple", "std"] } +pem = "3.0.3" pin-project-lite = "0.2" +pkcs8 = "0.10.2" pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] } procfs = "0.16" prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency diff --git a/compute_tools/src/http/middleware/authorize.rs b/compute_tools/src/http/middleware/authorize.rs index f221752c38..f1137de0ab 100644 --- a/compute_tools/src/http/middleware/authorize.rs +++ b/compute_tools/src/http/middleware/authorize.rs @@ -54,8 +54,8 @@ impl AsyncAuthorizeRequest for Authorize { Box::pin(async move { let request_id = request.extract_parts::().await.unwrap(); - // TODO: Remove this stanza after teaching neon_local and the - // regression tests to use a JWT + JWKS. + // TODO(tristan957): Remove this stanza after teaching neon_local + // and the regression tests to use a JWT + JWKS. // // https://github.com/neondatabase/neon/issues/11316 if cfg!(feature = "testing") { @@ -112,6 +112,8 @@ impl Authorize { token: &str, validation: &Validation, ) -> Result> { + debug_assert!(!jwks.keys.is_empty()); + debug!("verifying token {}", token); for jwk in jwks.keys.iter() { diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 162c49ec7c..a0ea216d9c 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -6,13 +6,17 @@ license.workspace = true [dependencies] anyhow.workspace = true +base64.workspace = true camino.workspace = true clap.workspace = true comfy-table.workspace = true futures.workspace = true humantime.workspace = true +jsonwebtoken.workspace = true nix.workspace = true once_cell.workspace = true +pem.workspace = true +pkcs8.workspace = true humantime-serde.workspace = true hyper0.workspace = true regex.workspace = true @@ -20,6 +24,7 @@ reqwest = { workspace = true, features = ["blocking", "json"] } scopeguard.workspace = true serde.workspace = true serde_json.workspace = true +sha2.workspace = true thiserror.workspace = true toml.workspace = true toml_edit.workspace = true diff --git a/control_plane/src/bin/neon_local.rs b/control_plane/src/bin/neon_local.rs index db9715dc62..950b264163 100644 --- a/control_plane/src/bin/neon_local.rs +++ b/control_plane/src/bin/neon_local.rs @@ -552,6 +552,7 @@ enum EndpointCmd { Start(EndpointStartCmdArgs), Reconfigure(EndpointReconfigureCmdArgs), Stop(EndpointStopCmdArgs), + GenerateJwt(EndpointGenerateJwtCmdArgs), } #[derive(clap::Args)] @@ -699,6 +700,13 @@ struct EndpointStopCmdArgs { mode: String, } +#[derive(clap::Args)] +#[clap(about = "Generate a JWT for an endpoint")] +struct EndpointGenerateJwtCmdArgs { + #[clap(help = "Postgres endpoint id")] + endpoint_id: String, +} + #[derive(clap::Subcommand)] #[clap(about = "Manage neon_local branch name mappings")] enum MappingsCmd { @@ -1528,6 +1536,16 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?; endpoint.stop(&args.mode, args.destroy)?; } + EndpointCmd::GenerateJwt(args) => { + let endpoint_id = &args.endpoint_id; + let endpoint = cplane + .endpoints + .get(endpoint_id) + .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?; + let jwt = endpoint.generate_jwt()?; + + println!("{jwt}"); + } } Ok(()) diff --git a/control_plane/src/endpoint.rs b/control_plane/src/endpoint.rs index 2fa7a62f8f..0fe6975a6e 100644 --- a/control_plane/src/endpoint.rs +++ b/control_plane/src/endpoint.rs @@ -42,22 +42,29 @@ use std::path::PathBuf; use std::process::Command; use std::str::FromStr; use std::sync::Arc; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant}; use anyhow::{Context, Result, anyhow, bail}; -use compute_api::requests::ConfigurationRequest; +use compute_api::requests::{ComputeClaims, ConfigurationRequest}; use compute_api::responses::{ - ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse, + ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse, TlsConfig, }; use compute_api::spec::{ Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent, RemoteExtSpec, Role, }; +use jsonwebtoken::jwk::{ + AlgorithmParameters, CommonParameters, EllipticCurve, Jwk, JwkSet, KeyAlgorithm, KeyOperations, + OctetKeyPairParameters, OctetKeyPairType, PublicKeyUse, +}; use nix::sys::signal::{Signal, kill}; use pageserver_api::shard::ShardStripeSize; +use pem::Pem; +use pkcs8::der::Decode; use reqwest::header::CONTENT_TYPE; use safekeeper_api::membership::SafekeeperGeneration; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use tracing::debug; use url::Host; use utils::id::{NodeId, TenantId, TimelineId}; @@ -82,6 +89,7 @@ pub struct EndpointConf { drop_subscriptions_before_start: bool, features: Vec, cluster: Option, + compute_ctl_config: ComputeCtlConfig, } // @@ -137,6 +145,36 @@ impl ComputeControlPlane { .unwrap_or(self.base_port) } + /// Create a JSON Web Key Set. This ideally matches the way we create a JWKS + /// from the production control plane. + fn create_jwks_from_pem(pem: Pem) -> Result { + let document = pkcs8::Document::from_der(&pem.into_contents())?; + + let mut hasher = Sha256::new(); + hasher.update(&document); + let key_hash = hasher.finalize(); + + Ok(JwkSet { + keys: vec![Jwk { + common: CommonParameters { + public_key_use: Some(PublicKeyUse::Signature), + key_operations: Some(vec![KeyOperations::Verify]), + key_algorithm: Some(KeyAlgorithm::EdDSA), + key_id: Some(base64::encode_config(key_hash, base64::URL_SAFE_NO_PAD)), + x509_url: None::, + x509_chain: None::>, + x509_sha1_fingerprint: None::, + x509_sha256_fingerprint: None::, + }, + algorithm: AlgorithmParameters::OctetKeyPair(OctetKeyPairParameters { + key_type: OctetKeyPairType::OctetKeyPair, + curve: EllipticCurve::Ed25519, + x: base64::encode_config(&document, base64::URL_SAFE_NO_PAD), + }), + }], + }) + } + #[allow(clippy::too_many_arguments)] pub fn new_endpoint( &mut self, @@ -154,6 +192,10 @@ impl ComputeControlPlane { let pg_port = pg_port.unwrap_or_else(|| self.get_port()); let external_http_port = external_http_port.unwrap_or_else(|| self.get_port() + 1); let internal_http_port = internal_http_port.unwrap_or_else(|| external_http_port + 1); + let compute_ctl_config = ComputeCtlConfig { + jwks: Self::create_jwks_from_pem(self.env.read_public_key()?)?, + tls: None::, + }; let ep = Arc::new(Endpoint { endpoint_id: endpoint_id.to_owned(), pg_address: SocketAddr::new(IpAddr::from(Ipv4Addr::LOCALHOST), pg_port), @@ -181,6 +223,7 @@ impl ComputeControlPlane { reconfigure_concurrency: 1, features: vec![], cluster: None, + compute_ctl_config: compute_ctl_config.clone(), }); ep.create_endpoint_dir()?; @@ -200,6 +243,7 @@ impl ComputeControlPlane { reconfigure_concurrency: 1, features: vec![], cluster: None, + compute_ctl_config, })?, )?; std::fs::write( @@ -242,7 +286,6 @@ impl ComputeControlPlane { /////////////////////////////////////////////////////////////////////////////// -#[derive(Debug)] pub struct Endpoint { /// used as the directory name endpoint_id: String, @@ -271,6 +314,9 @@ pub struct Endpoint { features: Vec, // Cluster settings cluster: Option, + + /// The compute_ctl config for the endpoint's compute. + compute_ctl_config: ComputeCtlConfig, } #[derive(PartialEq, Eq)] @@ -333,6 +379,7 @@ impl Endpoint { drop_subscriptions_before_start: conf.drop_subscriptions_before_start, features: conf.features, cluster: conf.cluster, + compute_ctl_config: conf.compute_ctl_config, }) } @@ -580,6 +627,13 @@ impl Endpoint { Ok(safekeeper_connstrings) } + /// Generate a JWT with the correct claims. + pub fn generate_jwt(&self) -> Result { + self.env.generate_auth_token(&ComputeClaims { + compute_id: self.endpoint_id.clone(), + }) + } + #[allow(clippy::too_many_arguments)] pub async fn start( &self, @@ -706,7 +760,7 @@ impl Endpoint { ComputeConfig { spec: Some(spec), - compute_ctl_config: ComputeCtlConfig::default(), + compute_ctl_config: self.compute_ctl_config.clone(), } }; @@ -774,16 +828,7 @@ impl Endpoint { ]) // TODO: It would be nice if we generated compute IDs with the same // algorithm as the real control plane. - .args([ - "--compute-id", - &format!( - "compute-{}", - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs() - ), - ]) + .args(["--compute-id", &self.endpoint_id]) .stdin(std::process::Stdio::null()) .stderr(logfile.try_clone()?) .stdout(logfile); @@ -881,6 +926,7 @@ impl Endpoint { self.external_http_address.port() ), ) + .bearer_auth(self.generate_jwt()?) .send() .await?; @@ -957,6 +1003,7 @@ impl Endpoint { self.external_http_address.port() )) .header(CONTENT_TYPE.as_str(), "application/json") + .bearer_auth(self.generate_jwt()?) .body( serde_json::to_string(&ConfigurationRequest { spec, diff --git a/control_plane/src/local_env.rs b/control_plane/src/local_env.rs index fa10abe91a..b7906e5f81 100644 --- a/control_plane/src/local_env.rs +++ b/control_plane/src/local_env.rs @@ -12,6 +12,7 @@ use std::{env, fs}; use anyhow::{Context, bail}; use clap::ValueEnum; +use pem::Pem; use postgres_backend::AuthType; use reqwest::Url; use serde::{Deserialize, Serialize}; @@ -56,6 +57,7 @@ pub struct LocalEnv { // used to issue tokens during e.g pg start pub private_key_path: PathBuf, + /// Path to environment's public key pub public_key_path: PathBuf, pub broker: NeonBroker, @@ -758,11 +760,11 @@ impl LocalEnv { // this function is used only for testing purposes in CLI e g generate tokens during init pub fn generate_auth_token(&self, claims: &S) -> anyhow::Result { - let private_key_path = self.get_private_key_path(); - let key_data = fs::read(private_key_path)?; - encode_from_key_file(claims, &key_data) + let key = self.read_private_key()?; + encode_from_key_file(claims, &key) } + /// Get the path to the private key. pub fn get_private_key_path(&self) -> PathBuf { if self.private_key_path.is_absolute() { self.private_key_path.to_path_buf() @@ -771,6 +773,29 @@ impl LocalEnv { } } + /// Get the path to the public key. + pub fn get_public_key_path(&self) -> PathBuf { + if self.public_key_path.is_absolute() { + self.public_key_path.to_path_buf() + } else { + self.base_data_dir.join(&self.public_key_path) + } + } + + /// Read the contents of the private key file. + pub fn read_private_key(&self) -> anyhow::Result { + let private_key_path = self.get_private_key_path(); + let pem = pem::parse(fs::read(private_key_path)?)?; + Ok(pem) + } + + /// Read the contents of the public key file. + pub fn read_public_key(&self) -> anyhow::Result { + let public_key_path = self.get_public_key_path(); + let pem = pem::parse(fs::read(public_key_path)?)?; + Ok(pem) + } + /// Materialize the [`NeonLocalInitConf`] to disk. Called during [`neon_local init`]. pub fn init(conf: NeonLocalInitConf, force: &InitForceMode) -> anyhow::Result<()> { let base_path = base_path(); @@ -956,6 +981,7 @@ fn generate_auth_keys(private_key_path: &Path, public_key_path: &Path) -> anyhow String::from_utf8_lossy(&keygen_output.stderr) ); } + // Extract the public key from the private key file // // openssl pkey -in auth_private_key.pem -pubout -out auth_public_key.pem @@ -972,6 +998,7 @@ fn generate_auth_keys(private_key_path: &Path, public_key_path: &Path) -> anyhow String::from_utf8_lossy(&keygen_output.stderr) ); } + Ok(()) } diff --git a/control_plane/src/storage_controller.rs b/control_plane/src/storage_controller.rs index a4b56ae5c0..62ad5fa8d6 100644 --- a/control_plane/src/storage_controller.rs +++ b/control_plane/src/storage_controller.rs @@ -18,6 +18,7 @@ use pageserver_api::models::{ }; use pageserver_api::shard::TenantShardId; use pageserver_client::mgmt_api::ResponseErrorMessageExt; +use pem::Pem; use postgres_backend::AuthType; use reqwest::{Certificate, Method}; use serde::de::DeserializeOwned; @@ -34,8 +35,8 @@ use crate::local_env::{LocalEnv, NeonStorageControllerConf}; pub struct StorageController { env: LocalEnv, - private_key: Option>, - public_key: Option, + private_key: Option, + public_key: Option, client: reqwest::Client, config: NeonStorageControllerConf, @@ -116,7 +117,9 @@ impl StorageController { AuthType::Trust => (None, None), AuthType::NeonJWT => { let private_key_path = env.get_private_key_path(); - let private_key = fs::read(private_key_path).expect("failed to read private key"); + let private_key = + pem::parse(fs::read(private_key_path).expect("failed to read private key")) + .expect("failed to parse PEM file"); // If pageserver auth is enabled, this implicitly enables auth for this service, // using the same credentials. @@ -138,9 +141,13 @@ impl StorageController { .expect("Empty key dir") .expect("Error reading key dir"); - std::fs::read_to_string(dent.path()).expect("Can't read public key") + pem::parse(std::fs::read_to_string(dent.path()).expect("Can't read public key")) + .expect("Failed to parse PEM file") } else { - std::fs::read_to_string(&public_key_path).expect("Can't read public key") + pem::parse( + std::fs::read_to_string(&public_key_path).expect("Can't read public key"), + ) + .expect("Failed to parse PEM file") }; (Some(private_key), Some(public_key)) } diff --git a/libs/compute_api/src/responses.rs b/libs/compute_api/src/responses.rs index 353949736b..b7d6b7ca34 100644 --- a/libs/compute_api/src/responses.rs +++ b/libs/compute_api/src/responses.rs @@ -160,7 +160,7 @@ pub struct CatalogObjects { pub databases: Vec, } -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct ComputeCtlConfig { /// Set of JSON web keys that the compute can use to authenticate /// communication from the control plane. @@ -179,7 +179,7 @@ impl Default for ComputeCtlConfig { } } -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct TlsConfig { pub key_path: String, pub cert_path: String, diff --git a/libs/http-utils/Cargo.toml b/libs/http-utils/Cargo.toml index 5f6578f76e..ab9380089b 100644 --- a/libs/http-utils/Cargo.toml +++ b/libs/http-utils/Cargo.toml @@ -14,6 +14,7 @@ futures.workspace = true hyper0.workspace = true itertools.workspace = true jemalloc_pprof.workspace = true +jsonwebtoken.workspace = true once_cell.workspace = true pprof.workspace = true regex.workspace = true diff --git a/libs/http-utils/src/endpoint.rs b/libs/http-utils/src/endpoint.rs index 5588f6d87e..64147f2dd0 100644 --- a/libs/http-utils/src/endpoint.rs +++ b/libs/http-utils/src/endpoint.rs @@ -8,6 +8,7 @@ use bytes::{Bytes, BytesMut}; use hyper::header::{AUTHORIZATION, CONTENT_DISPOSITION, CONTENT_TYPE, HeaderName}; use hyper::http::HeaderValue; use hyper::{Body, Method, Request, Response}; +use jsonwebtoken::TokenData; use metrics::{Encoder, IntCounter, TextEncoder, register_int_counter}; use once_cell::sync::Lazy; use pprof::ProfilerGuardBuilder; @@ -618,7 +619,7 @@ pub fn auth_middleware( })?; let token = parse_token(header_value)?; - let data = auth.decode(token).map_err(|err| { + let data: TokenData = auth.decode(token).map_err(|err| { warn!("Authentication error: {err}"); // Rely on From for ApiError impl err diff --git a/libs/utils/Cargo.toml b/libs/utils/Cargo.toml index fd2fa63fd0..7b1dc56071 100644 --- a/libs/utils/Cargo.toml +++ b/libs/utils/Cargo.toml @@ -29,6 +29,7 @@ futures = { workspace = true } jsonwebtoken.workspace = true nix = { workspace = true, features = ["ioctl"] } once_cell.workspace = true +pem.workspace = true pin-project-lite.workspace = true regex.workspace = true serde.workspace = true diff --git a/libs/utils/src/auth.rs b/libs/utils/src/auth.rs index db4fc5685c..de3a964d23 100644 --- a/libs/utils/src/auth.rs +++ b/libs/utils/src/auth.rs @@ -11,7 +11,8 @@ use camino::Utf8Path; use jsonwebtoken::{ Algorithm, DecodingKey, EncodingKey, Header, TokenData, Validation, decode, encode, }; -use serde::{Deserialize, Serialize}; +use pem::Pem; +use serde::{Deserialize, Serialize, de::DeserializeOwned}; use crate::id::TenantId; @@ -73,7 +74,10 @@ impl SwappableJwtAuth { pub fn swap(&self, jwt_auth: JwtAuth) { self.0.swap(Arc::new(jwt_auth)); } - pub fn decode(&self, token: &str) -> std::result::Result, AuthError> { + pub fn decode( + &self, + token: &str, + ) -> std::result::Result, AuthError> { self.0.load().decode(token) } } @@ -148,7 +152,10 @@ impl JwtAuth { /// The function tries the stored decoding keys in succession, /// and returns the first yielding a successful result. /// If there is no working decoding key, it returns the last error. - pub fn decode(&self, token: &str) -> std::result::Result, AuthError> { + pub fn decode( + &self, + token: &str, + ) -> std::result::Result, AuthError> { let mut res = None; for decoding_key in &self.decoding_keys { res = Some(decode(token, decoding_key, &self.validation)); @@ -173,8 +180,8 @@ impl std::fmt::Debug for JwtAuth { } // this function is used only for testing purposes in CLI e g generate tokens during init -pub fn encode_from_key_file(claims: &S, key_data: &[u8]) -> Result { - let key = EncodingKey::from_ed_pem(key_data)?; +pub fn encode_from_key_file(claims: &S, pem: &Pem) -> Result { + let key = EncodingKey::from_ed_der(pem.contents()); Ok(encode(&Header::new(STORAGE_TOKEN_ALGORITHM), claims, &key)?) } @@ -188,13 +195,13 @@ mod tests { // // openssl genpkey -algorithm ed25519 -out ed25519-priv.pem // openssl pkey -in ed25519-priv.pem -pubout -out ed25519-pub.pem - const TEST_PUB_KEY_ED25519: &[u8] = br#" + const TEST_PUB_KEY_ED25519: &str = r#" -----BEGIN PUBLIC KEY----- MCowBQYDK2VwAyEARYwaNBayR+eGI0iXB4s3QxE3Nl2g1iWbr6KtLWeVD/w= -----END PUBLIC KEY----- "#; - const TEST_PRIV_KEY_ED25519: &[u8] = br#" + const TEST_PRIV_KEY_ED25519: &str = r#" -----BEGIN PRIVATE KEY----- MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH -----END PRIVATE KEY----- @@ -222,9 +229,9 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH // Check it can be validated with the public key let auth = JwtAuth::new(vec![ - DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519).unwrap(), + DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519.as_bytes()).unwrap(), ]); - let claims_from_token = auth.decode(encoded_eddsa).unwrap().claims; + let claims_from_token: Claims = auth.decode(encoded_eddsa).unwrap().claims; assert_eq!(claims_from_token, expected_claims); } @@ -235,13 +242,14 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH scope: Scope::Tenant, }; - let encoded = encode_from_key_file(&claims, TEST_PRIV_KEY_ED25519).unwrap(); + let pem = pem::parse(TEST_PRIV_KEY_ED25519).unwrap(); + let encoded = encode_from_key_file(&claims, &pem).unwrap(); // decode it back let auth = JwtAuth::new(vec![ - DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519).unwrap(), + DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519.as_bytes()).unwrap(), ]); - let decoded = auth.decode(&encoded).unwrap(); + let decoded: TokenData = auth.decode(&encoded).unwrap(); assert_eq!(decoded.claims, claims); } diff --git a/pageserver/Cargo.toml b/pageserver/Cargo.toml index 74f3fce6e5..5c5bab0642 100644 --- a/pageserver/Cargo.toml +++ b/pageserver/Cargo.toml @@ -35,6 +35,7 @@ humantime.workspace = true humantime-serde.workspace = true hyper0.workspace = true itertools.workspace = true +jsonwebtoken.workspace = true md5.workspace = true nix.workspace = true # hack to get the number of worker threads tokio uses diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index 7a62d8049b..560ac75f4a 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -15,6 +15,7 @@ use async_compression::tokio::write::GzipEncoder; use bytes::Buf; use futures::FutureExt; use itertools::Itertools; +use jsonwebtoken::TokenData; use once_cell::sync::OnceCell; use pageserver_api::config::{ PageServicePipeliningConfig, PageServicePipeliningConfigPipelined, @@ -2837,7 +2838,7 @@ where ) -> Result<(), QueryError> { // this unwrap is never triggered, because check_auth_jwt only called when auth_type is NeonJWT // which requires auth to be present - let data = self + let data: TokenData = self .auth .as_ref() .unwrap() diff --git a/safekeeper/Cargo.toml b/safekeeper/Cargo.toml index 965aa7504b..a0ba69aa34 100644 --- a/safekeeper/Cargo.toml +++ b/safekeeper/Cargo.toml @@ -27,6 +27,7 @@ humantime.workspace = true http.workspace = true hyper0.workspace = true itertools.workspace = true +jsonwebtoken.workspace = true futures.workspace = true once_cell.workspace = true parking_lot.workspace = true diff --git a/safekeeper/src/handler.rs b/safekeeper/src/handler.rs index 5ca3d1b7c2..b54bee8bfb 100644 --- a/safekeeper/src/handler.rs +++ b/safekeeper/src/handler.rs @@ -6,6 +6,7 @@ use std::str::{self, FromStr}; use std::sync::Arc; use anyhow::Context; +use jsonwebtoken::TokenData; use pageserver_api::models::ShardParameters; use pageserver_api::shard::{ShardIdentity, ShardStripeSize}; use postgres_backend::{PostgresBackend, QueryError}; @@ -278,7 +279,7 @@ impl postgres_backend::Handler .auth .as_ref() .expect("auth_type is configured but .auth of handler is missing"); - let data = auth + let data: TokenData = auth .decode(str::from_utf8(jwt_response).context("jwt response is not UTF-8")?) .map_err(|e| QueryError::Unauthorized(e.0))?; diff --git a/workspace_hack/Cargo.toml b/workspace_hack/Cargo.toml index b548a2a88a..2c37cebc27 100644 --- a/workspace_hack/Cargo.toml +++ b/workspace_hack/Cargo.toml @@ -70,6 +70,7 @@ num-traits = { version = "0.2", features = ["i128", "libm"] } once_cell = { version = "1" } p256 = { version = "0.13", features = ["jwk"] } parquet = { version = "53", default-features = false, features = ["zstd"] } +pkcs8 = { version = "0.10", default-features = false, features = ["pem", "std"] } prost = { version = "0.13", features = ["no-recursion-limit", "prost-derive"] } rand = { version = "0.8", features = ["small_rng"] } regex = { version = "1" } From cd9ad757975277a3ce20065c34d209747086be52 Mon Sep 17 00:00:00 2001 From: Tristan Partin Date: Tue, 15 Apr 2025 14:12:34 -0500 Subject: [PATCH 17/27] Remove compute_ctl authorization bypass on localhost (#11597) For whatever reason, this never worked in production computes anyway. Signed-off-by: Tristan Partin --- compute_tools/src/http/middleware/authorize.rs | 17 ++--------------- 1 file changed, 2 insertions(+), 15 deletions(-) diff --git a/compute_tools/src/http/middleware/authorize.rs b/compute_tools/src/http/middleware/authorize.rs index f1137de0ab..e6c3269b15 100644 --- a/compute_tools/src/http/middleware/authorize.rs +++ b/compute_tools/src/http/middleware/authorize.rs @@ -1,7 +1,7 @@ -use std::{collections::HashSet, net::SocketAddr}; +use std::collections::HashSet; use anyhow::{Result, anyhow}; -use axum::{RequestExt, body::Body, extract::ConnectInfo}; +use axum::{RequestExt, body::Body}; use axum_extra::{ TypedHeader, headers::{Authorization, authorization::Bearer}, @@ -64,19 +64,6 @@ impl AsyncAuthorizeRequest for Authorize { return Ok(request); } - let connect_info = request - .extract_parts::>() - .await - .unwrap(); - - // In the event the request is coming from the loopback interface, - // allow all requests - if connect_info.ip().is_loopback() { - warn!(%request_id, "Bypassed authorization because request is coming from the loopback interface"); - - return Ok(request); - } - let TypedHeader(Authorization(bearer)) = request .extract_parts::>>() .await From 35170656fe4c1c636c5b6715651b7f9064ecdb3e Mon Sep 17 00:00:00 2001 From: Konstantin Knizhnik Date: Tue, 15 Apr 2025 22:13:12 +0300 Subject: [PATCH 18/27] Allocate WalProposerConn using TopMemoryAllocator (#11577) ## Problem See https://neondb.slack.com/archives/C04DGM6SMTM/p1744659631698609 `WalProposerConn` is allocated using current memory context which life time is not long enough. ## Summary of changes Allocate `WalProposerConn` using `TopMemoryContext`. Co-authored-by: Konstantin Knizhnik --- pgxn/neon/walproposer_pg.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pgxn/neon/walproposer_pg.c b/pgxn/neon/walproposer_pg.c index 9c34c90002..a061639815 100644 --- a/pgxn/neon/walproposer_pg.c +++ b/pgxn/neon/walproposer_pg.c @@ -890,7 +890,7 @@ libpqwp_connect_start(char *conninfo) * palloc will exit on failure though, so there's not much we could do if * it *did* fail. */ - conn = palloc(sizeof(WalProposerConn)); + conn = (WalProposerConn*)MemoryContextAllocZero(TopMemoryContext, sizeof(WalProposerConn)); conn->pg_conn = pg_conn; conn->is_nonblocking = false; /* connections always start in blocking * mode */ From aa19f10e7e958fbe0e0641f2e8c5952ce3be44b3 Mon Sep 17 00:00:00 2001 From: "Alex Chi Z." <4198311+skyzh@users.noreply.github.com> Date: Tue, 15 Apr 2025 17:50:28 -0400 Subject: [PATCH 19/27] fix(test): allow shutdown warning in preempt tests (#11600) ## Problem test_gc_compaction_preempt is still flaky ## Summary of changes - allow shutdown warning logs Signed-off-by: Alex Chi Z --- test_runner/regress/test_compaction.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/test_runner/regress/test_compaction.py b/test_runner/regress/test_compaction.py index 84d37de9f1..001ddcdcb0 100644 --- a/test_runner/regress/test_compaction.py +++ b/test_runner/regress/test_compaction.py @@ -199,6 +199,8 @@ def test_pageserver_gc_compaction_preempt( conf = PREEMPT_GC_COMPACTION_TENANT_CONF.copy() env = neon_env_builder.init_start(initial_tenant_conf=conf) + env.pageserver.allowed_errors.append(".*The timeline or pageserver is shutting down.*") + tenant_id = env.initial_tenant timeline_id = env.initial_timeline From 96b46365e4520d650acc415f454decd6eed80971 Mon Sep 17 00:00:00 2001 From: Vlad Lazar Date: Wed, 16 Apr 2025 11:26:47 +0100 Subject: [PATCH 20/27] tests: attach final metrics to allure report (#11604) ## Problem Metrics are saved in https://github.com/neondatabase/neon/pull/11559, but the file is not matched by the attachment regex. ## Summary of changes Make attachment regex match the metrics file. --- test_runner/fixtures/neon_fixtures.py | 7 +------ test_runner/fixtures/utils.py | 2 +- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index 10bbb7020b..13bd74e05d 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -947,8 +947,6 @@ class NeonEnvBuilder: continue if SMALL_DB_FILE_NAME_REGEX.fullmatch(test_file.name): continue - if FINAL_METRICS_FILE_NAME == test_file.name: - continue log.debug(f"Removing large database {test_file} file") test_file.unlink() elif test_entry.is_dir(): @@ -2989,7 +2987,7 @@ class NeonPageserver(PgProtocol, LogUtils): return metrics = self.http_client().get_metrics_str() - metrics_snapshot_path = self.workdir / FINAL_METRICS_FILE_NAME + metrics_snapshot_path = self.workdir / "final_metrics.txt" with open(metrics_snapshot_path, "w") as f: f.write(metrics) @@ -5156,9 +5154,6 @@ SMALL_DB_FILE_NAME_REGEX: re.Pattern[str] = re.compile( r"config-v1|heatmap-v1|tenant-manifest|metadata|.+\.(?:toml|pid|json|sql|conf)" ) -FINAL_METRICS_FILE_NAME: str = "final_metrics.txt" - - SKIP_DIRS = frozenset( ( "pg_wal", diff --git a/test_runner/fixtures/utils.py b/test_runner/fixtures/utils.py index 13c2d320d1..0d7345cc82 100644 --- a/test_runner/fixtures/utils.py +++ b/test_runner/fixtures/utils.py @@ -258,7 +258,7 @@ def get_scale_for_db(size_mb: int) -> int: ATTACHMENT_NAME_REGEX: re.Pattern[str] = re.compile( - r"regression\.(diffs|out)|.+\.(?:log|stderr|stdout|filediff|metrics|html|walredo)" + r"regression\.(diffs|out)|.+\.(?:log|stderr|stdout|filediff|metrics|html|walredo)|final_metrics.txt" ) From b4e26a6284b8dedda229c3d087d706e62e19bb59 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Wed, 16 Apr 2025 15:34:18 +0300 Subject: [PATCH 21/27] Set last-written LSN as part of smgr_end_unlogged_build() (#11584) This way, the callers don't need to do it, reducing the footprint of changes we've had to made to various index AM's build functions. --- compute/patches/pgvector.patch | 27 ++++++--------------------- compute/patches/rum.patch | 18 +++--------------- pgxn/neon/pagestore_smgr.c | 25 +++++++++++++++++++++++-- vendor/postgres-v14 | 2 +- vendor/postgres-v15 | 2 +- vendor/postgres-v16 | 2 +- vendor/postgres-v17 | 2 +- vendor/revisions.json | 8 ++++---- 8 files changed, 40 insertions(+), 46 deletions(-) diff --git a/compute/patches/pgvector.patch b/compute/patches/pgvector.patch index 6fe3d073ed..6a203489fd 100644 --- a/compute/patches/pgvector.patch +++ b/compute/patches/pgvector.patch @@ -15,7 +15,7 @@ index 7a4b88c..56678af 100644 HEADERS = src/halfvec.h src/sparsevec.h src/vector.h diff --git a/src/hnswbuild.c b/src/hnswbuild.c -index b667478..dc95d89 100644 +index b667478..1298aa1 100644 --- a/src/hnswbuild.c +++ b/src/hnswbuild.c @@ -843,9 +843,17 @@ HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc) @@ -36,7 +36,7 @@ index b667478..dc95d89 100644 /* Close relations within worker */ index_close(indexRel, indexLockmode); table_close(heapRel, heapLockmode); -@@ -1100,12 +1108,39 @@ BuildIndex(Relation heap, Relation index, IndexInfo *indexInfo, +@@ -1100,13 +1108,25 @@ BuildIndex(Relation heap, Relation index, IndexInfo *indexInfo, SeedRandom(42); #endif @@ -48,32 +48,17 @@ index b667478..dc95d89 100644 BuildGraph(buildstate, forkNum); -- if (RelationNeedsWAL(index) || forkNum == INIT_FORKNUM) +#ifdef NEON_SMGR + smgr_finish_unlogged_build_phase_1(RelationGetSmgr(index)); +#endif + -+ if (RelationNeedsWAL(index) || forkNum == INIT_FORKNUM) { + if (RelationNeedsWAL(index) || forkNum == INIT_FORKNUM) log_newpage_range(index, forkNum, 0, RelationGetNumberOfBlocksInFork(index, forkNum), true); -+#ifdef NEON_SMGR -+ { -+#if PG_VERSION_NUM >= 160000 -+ RelFileLocator rlocator = RelationGetSmgr(index)->smgr_rlocator.locator; -+#else -+ RelFileNode rlocator = RelationGetSmgr(index)->smgr_rnode.node; -+#endif -+ if (set_lwlsn_block_range_hook) -+ set_lwlsn_block_range_hook(XactLastRecEnd, rlocator, -+ MAIN_FORKNUM, 0, RelationGetNumberOfBlocks(index)); -+ if (set_lwlsn_relation_hook) -+ set_lwlsn_relation_hook(XactLastRecEnd, rlocator, MAIN_FORKNUM); -+ } -+#endif -+ } -+ + +#ifdef NEON_SMGR + smgr_end_unlogged_build(RelationGetSmgr(index)); +#endif - ++ FreeBuildState(buildstate); } + diff --git a/compute/patches/rum.patch b/compute/patches/rum.patch index 5bc5d739b3..b45afe2874 100644 --- a/compute/patches/rum.patch +++ b/compute/patches/rum.patch @@ -1,5 +1,5 @@ diff --git a/src/ruminsert.c b/src/ruminsert.c -index 255e616..7a2240f 100644 +index 255e616..1c6edb7 100644 --- a/src/ruminsert.c +++ b/src/ruminsert.c @@ -628,6 +628,10 @@ rumbuild(Relation heap, Relation index, struct IndexInfo *indexInfo) @@ -24,24 +24,12 @@ index 255e616..7a2240f 100644 /* * Write index to xlog */ -@@ -713,6 +721,22 @@ rumbuild(Relation heap, Relation index, struct IndexInfo *indexInfo) +@@ -713,6 +721,10 @@ rumbuild(Relation heap, Relation index, struct IndexInfo *indexInfo) UnlockReleaseBuffer(buffer); } +#ifdef NEON_SMGR -+ { -+#if PG_VERSION_NUM >= 160000 -+ RelFileLocator rlocator = RelationGetSmgr(index)->smgr_rlocator.locator; -+#else -+ RelFileNode rlocator = RelationGetSmgr(index)->smgr_rnode.node; -+#endif -+ if (set_lwlsn_block_range_hook) -+ set_lwlsn_block_range_hook(XactLastRecEnd, rlocator, MAIN_FORKNUM, 0, RelationGetNumberOfBlocks(index)); -+ if (set_lwlsn_relation_hook) -+ set_lwlsn_relation_hook(XactLastRecEnd, rlocator, MAIN_FORKNUM); -+ -+ smgr_end_unlogged_build(index->rd_smgr); -+ } ++ smgr_end_unlogged_build(index->rd_smgr); +#endif + /* diff --git a/pgxn/neon/pagestore_smgr.c b/pgxn/neon/pagestore_smgr.c index ef6bd038bb..9fe085c558 100644 --- a/pgxn/neon/pagestore_smgr.c +++ b/pgxn/neon/pagestore_smgr.c @@ -2040,7 +2040,7 @@ neon_finish_unlogged_build_phase_1(SMgrRelation reln) /* * neon_end_unlogged_build() -- Finish an unlogged rel build. * - * Call this after you have finished WAL-logging an relation that was + * Call this after you have finished WAL-logging a relation that was * first populated without WAL-logging. * * This removes the local copy of the rel, since it's now been fully @@ -2059,14 +2059,35 @@ neon_end_unlogged_build(SMgrRelation reln) if (unlogged_build_phase != UNLOGGED_BUILD_NOT_PERMANENT) { + XLogRecPtr recptr; + BlockNumber nblocks; + Assert(unlogged_build_phase == UNLOGGED_BUILD_PHASE_2); Assert(reln->smgr_relpersistence == RELPERSISTENCE_UNLOGGED); + /* + * Update the last-written LSN cache. + * + * The relation is still on local disk so we can get the size by + * calling mdnblocks() directly. For the LSN, GetXLogInsertRecPtr() is + * very conservative. If we could assume that this function is called + * from the same backend that WAL-logged the contents, we could use + * XactLastRecEnd here. But better safe than sorry. + */ + nblocks = mdnblocks(reln, MAIN_FORKNUM); + recptr = GetXLogInsertRecPtr(); + + neon_set_lwlsn_block_range(recptr, + InfoFromNInfoB(rinfob), + MAIN_FORKNUM, 0, nblocks); + neon_set_lwlsn_relation(recptr, + InfoFromNInfoB(rinfob), + MAIN_FORKNUM); + /* Make the relation look permanent again */ reln->smgr_relpersistence = RELPERSISTENCE_PERMANENT; /* Remove local copy */ - rinfob = InfoBFromSMgrRel(reln); for (int forknum = 0; forknum <= MAX_FORKNUM; forknum++) { neon_log(SmgrTrace, "forgetting cached relsize for %u/%u/%u.%u", diff --git a/vendor/postgres-v14 b/vendor/postgres-v14 index a0391901a2..d3c9d61fb7 160000 --- a/vendor/postgres-v14 +++ b/vendor/postgres-v14 @@ -1 +1 @@ -Subproject commit a0391901a2af13aa029b905272a5b2024133c926 +Subproject commit d3c9d61fb7a362a165dac7060819dd9d6ad68c28 diff --git a/vendor/postgres-v15 b/vendor/postgres-v15 index aeb292eeac..8ecb12f21d 160000 --- a/vendor/postgres-v15 +++ b/vendor/postgres-v15 @@ -1 +1 @@ -Subproject commit aeb292eeace9072e07071254b6ffc7a74007d4d2 +Subproject commit 8ecb12f21d862dfa39f7204b8f5e1c00a2a225b3 diff --git a/vendor/postgres-v16 b/vendor/postgres-v16 index d56e79cd5d..37496f87b5 160000 --- a/vendor/postgres-v16 +++ b/vendor/postgres-v16 @@ -1 +1 @@ -Subproject commit d56e79cd5d6136c159b1d8d98acb7981d4b69364 +Subproject commit 37496f87b5324af53c56127e278ee5b1e8435253 diff --git a/vendor/postgres-v17 b/vendor/postgres-v17 index 66114c23bc..eab3a37834 160000 --- a/vendor/postgres-v17 +++ b/vendor/postgres-v17 @@ -1 +1 @@ -Subproject commit 66114c23bc61205b0e3fb1e77ee76a4abc1eb4b8 +Subproject commit eab3a37834cac6ec0719bf817ac918a201712d66 diff --git a/vendor/revisions.json b/vendor/revisions.json index d7eddf42b7..90d878d0f7 100644 --- a/vendor/revisions.json +++ b/vendor/revisions.json @@ -1,18 +1,18 @@ { "v17": [ "17.4", - "66114c23bc61205b0e3fb1e77ee76a4abc1eb4b8" + "eab3a37834cac6ec0719bf817ac918a201712d66" ], "v16": [ "16.8", - "d56e79cd5d6136c159b1d8d98acb7981d4b69364" + "37496f87b5324af53c56127e278ee5b1e8435253" ], "v15": [ "15.12", - "aeb292eeace9072e07071254b6ffc7a74007d4d2" + "8ecb12f21d862dfa39f7204b8f5e1c00a2a225b3" ], "v14": [ "14.17", - "a0391901a2af13aa029b905272a5b2024133c926" + "d3c9d61fb7a362a165dac7060819dd9d6ad68c28" ] } From edc11253b65e12a10843711bd88ad277511396d7 Mon Sep 17 00:00:00 2001 From: Tristan Partin Date: Wed, 16 Apr 2025 07:51:48 -0500 Subject: [PATCH 22/27] Fix neon_local public key parsing when create compute JWKS (#11602) Finally figured out the right incantation. I had had this in my original go, but due to some refactoring and apparently missed testing, I committed a mistake. The reason this doesn't currently break anything is that we bypass the authorization middleware when the "testing" cargo feature is enabled. Signed-off-by: Tristan Partin --- Cargo.lock | 3 +-- Cargo.toml | 2 +- control_plane/Cargo.toml | 2 +- control_plane/src/endpoint.rs | 14 ++++++++------ workspace_hack/Cargo.toml | 1 - 5 files changed, 11 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5c9170b7de..7ab9378853 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1432,7 +1432,6 @@ dependencies = [ "pageserver_api", "pageserver_client", "pem", - "pkcs8 0.10.2", "postgres_backend", "postgres_connection", "regex", @@ -1442,6 +1441,7 @@ dependencies = [ "serde", "serde_json", "sha2", + "spki 0.7.3", "storage_broker", "thiserror 1.0.69", "tokio", @@ -8469,7 +8469,6 @@ dependencies = [ "once_cell", "p256 0.13.2", "parquet", - "pkcs8 0.10.2", "prettyplease", "proc-macro2", "prost 0.13.3", diff --git a/Cargo.toml b/Cargo.toml index 8fac3bb46c..9d7904a787 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -143,7 +143,6 @@ parquet_derive = "53" pbkdf2 = { version = "0.12.1", features = ["simple", "std"] } pem = "3.0.3" pin-project-lite = "0.2" -pkcs8 = "0.10.2" pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] } procfs = "0.16" prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency @@ -176,6 +175,7 @@ signal-hook = "0.3" smallvec = "1.11" smol_str = { version = "0.2.0", features = ["serde"] } socket2 = "0.5" +spki = "0.7.3" strum = "0.26" strum_macros = "0.26" "subtle" = "2.5.0" diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index a0ea216d9c..92f0071bac 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -16,7 +16,6 @@ jsonwebtoken.workspace = true nix.workspace = true once_cell.workspace = true pem.workspace = true -pkcs8.workspace = true humantime-serde.workspace = true hyper0.workspace = true regex.workspace = true @@ -25,6 +24,7 @@ scopeguard.workspace = true serde.workspace = true serde_json.workspace = true sha2.workspace = true +spki.workspace = true thiserror.workspace = true toml.workspace = true toml_edit.workspace = true diff --git a/control_plane/src/endpoint.rs b/control_plane/src/endpoint.rs index 0fe6975a6e..b569b0fb8e 100644 --- a/control_plane/src/endpoint.rs +++ b/control_plane/src/endpoint.rs @@ -60,11 +60,12 @@ use jsonwebtoken::jwk::{ use nix::sys::signal::{Signal, kill}; use pageserver_api::shard::ShardStripeSize; use pem::Pem; -use pkcs8::der::Decode; use reqwest::header::CONTENT_TYPE; use safekeeper_api::membership::SafekeeperGeneration; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use spki::der::Decode; +use spki::{SubjectPublicKeyInfo, SubjectPublicKeyInfoRef}; use tracing::debug; use url::Host; use utils::id::{NodeId, TenantId, TimelineId}; @@ -147,11 +148,12 @@ impl ComputeControlPlane { /// Create a JSON Web Key Set. This ideally matches the way we create a JWKS /// from the production control plane. - fn create_jwks_from_pem(pem: Pem) -> Result { - let document = pkcs8::Document::from_der(&pem.into_contents())?; + fn create_jwks_from_pem(pem: &Pem) -> Result { + let spki: SubjectPublicKeyInfoRef = SubjectPublicKeyInfo::from_der(pem.contents())?; + let public_key = spki.subject_public_key.raw_bytes(); let mut hasher = Sha256::new(); - hasher.update(&document); + hasher.update(public_key); let key_hash = hasher.finalize(); Ok(JwkSet { @@ -169,7 +171,7 @@ impl ComputeControlPlane { algorithm: AlgorithmParameters::OctetKeyPair(OctetKeyPairParameters { key_type: OctetKeyPairType::OctetKeyPair, curve: EllipticCurve::Ed25519, - x: base64::encode_config(&document, base64::URL_SAFE_NO_PAD), + x: base64::encode_config(public_key, base64::URL_SAFE_NO_PAD), }), }], }) @@ -193,7 +195,7 @@ impl ComputeControlPlane { let external_http_port = external_http_port.unwrap_or_else(|| self.get_port() + 1); let internal_http_port = internal_http_port.unwrap_or_else(|| external_http_port + 1); let compute_ctl_config = ComputeCtlConfig { - jwks: Self::create_jwks_from_pem(self.env.read_public_key()?)?, + jwks: Self::create_jwks_from_pem(&self.env.read_public_key()?)?, tls: None::, }; let ep = Arc::new(Endpoint { diff --git a/workspace_hack/Cargo.toml b/workspace_hack/Cargo.toml index 2c37cebc27..b548a2a88a 100644 --- a/workspace_hack/Cargo.toml +++ b/workspace_hack/Cargo.toml @@ -70,7 +70,6 @@ num-traits = { version = "0.2", features = ["i128", "libm"] } once_cell = { version = "1" } p256 = { version = "0.13", features = ["jwk"] } parquet = { version = "53", default-features = false, features = ["zstd"] } -pkcs8 = { version = "0.10", default-features = false, features = ["pem", "std"] } prost = { version = "0.13", features = ["no-recursion-limit", "prost-derive"] } rand = { version = "0.8", features = ["small_rng"] } regex = { version = "1" } From 2a464261574aa141790564e6a9feb3ca11d63287 Mon Sep 17 00:00:00 2001 From: Matthias van de Meent Date: Wed, 16 Apr 2025 15:42:22 +0200 Subject: [PATCH 23/27] Update neon GUCs with new default settings (#11595) Staging and prod both have these settings configured like this, so let's update this so we can eventually drop the overrides in prod. --- pgxn/neon/communicator.c | 2 +- pgxn/neon/libpagestore.c | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/pgxn/neon/communicator.c b/pgxn/neon/communicator.c index 932034e22e..db3e053321 100644 --- a/pgxn/neon/communicator.c +++ b/pgxn/neon/communicator.c @@ -95,7 +95,7 @@ static uint32 local_request_counter; * Various settings related to prompt (fast) handling of PageStream responses * at any CHECK_FOR_INTERRUPTS point. */ -int readahead_getpage_pull_timeout_ms = 0; +int readahead_getpage_pull_timeout_ms = 50; static int PS_TIMEOUT_ID = 0; static bool timeout_set = false; static bool timeout_signaled = false; diff --git a/pgxn/neon/libpagestore.c b/pgxn/neon/libpagestore.c index dfabb6919e..19c14511bd 100644 --- a/pgxn/neon/libpagestore.c +++ b/pgxn/neon/libpagestore.c @@ -75,7 +75,7 @@ char *neon_auth_token; int readahead_buffer_size = 128; int flush_every_n_requests = 8; -int neon_protocol_version = 2; +int neon_protocol_version = 3; static int neon_compute_mode = 0; static int max_reconnect_attempts = 60; @@ -1432,7 +1432,7 @@ pg_init_libpagestore(void) "PageStream connection when we have pages which " "were read ahead but not yet received.", &readahead_getpage_pull_timeout_ms, - 0, 0, 5 * 60 * 1000, + 50, 0, 5 * 60 * 1000, PGC_USERSET, GUC_UNIT_MS, NULL, NULL, NULL); @@ -1440,7 +1440,7 @@ pg_init_libpagestore(void) "Version of compute<->page server protocol", NULL, &neon_protocol_version, - 2, /* use protocol version 2 */ + 3, /* use protocol version 3 */ 2, /* min */ 3, /* max */ PGC_SU_BACKEND, From 00eeff9b8d170c925821d051a32f9d38fba0ca69 Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Wed, 16 Apr 2025 16:41:02 +0200 Subject: [PATCH 24/27] pageserver: add `compaction_shard_ancestor` to disable shard ancestor compaction (#11608) ## Problem Splits of large tenants (several TB) can cause a huge amount of shard ancestor compaction work, which can overload Pageservers. Touches https://github.com/neondatabase/cloud/issues/22532. ## Summary of changes Add a setting `compaction_shard_ancestor` (default `true`) to disable shard ancestor compaction on a per-tenant basis. --- control_plane/src/pageserver.rs | 5 +++++ libs/pageserver_api/src/config.rs | 4 ++++ libs/pageserver_api/src/models.rs | 13 +++++++++++++ pageserver/src/tenant/timeline.rs | 8 ++++++++ pageserver/src/tenant/timeline/compaction.rs | 3 +-- test_runner/regress/test_attach_tenant_config.py | 1 + 6 files changed, 32 insertions(+), 2 deletions(-) diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs index 5c985e6dc8..b9257a27bf 100644 --- a/control_plane/src/pageserver.rs +++ b/control_plane/src/pageserver.rs @@ -413,6 +413,11 @@ impl PageServerNode { .map(serde_json::from_str) .transpose() .context("Failed to parse 'compaction_algorithm' json")?, + compaction_shard_ancestor: settings + .remove("compaction_shard_ancestor") + .map(|x| x.parse::()) + .transpose() + .context("Failed to parse 'compaction_shard_ancestor' as a bool")?, compaction_l0_first: settings .remove("compaction_l0_first") .map(|x| x.parse::()) diff --git a/libs/pageserver_api/src/config.rs b/libs/pageserver_api/src/config.rs index 53b68afb0f..e734b07c38 100644 --- a/libs/pageserver_api/src/config.rs +++ b/libs/pageserver_api/src/config.rs @@ -379,6 +379,8 @@ pub struct TenantConfigToml { /// size exceeds `compaction_upper_limit * checkpoint_distance`. pub compaction_upper_limit: usize, pub compaction_algorithm: crate::models::CompactionAlgorithmSettings, + /// If true, enable shard ancestor compaction (enabled by default). + pub compaction_shard_ancestor: bool, /// If true, compact down L0 across all tenant timelines before doing regular compaction. L0 /// compaction must be responsive to avoid read amp during heavy ingestion. Defaults to true. pub compaction_l0_first: bool, @@ -677,6 +679,7 @@ pub mod tenant_conf_defaults { pub const DEFAULT_COMPACTION_PERIOD: &str = "20 s"; pub const DEFAULT_COMPACTION_THRESHOLD: usize = 10; + pub const DEFAULT_COMPACTION_SHARD_ANCESTOR: bool = true; // This value needs to be tuned to avoid OOM. We have 3/4*CPUs threads for L0 compaction, that's // 3/4*16=9 on most of our pageservers. Compacting 20 layers requires about 1 GB memory (could @@ -734,6 +737,7 @@ impl Default for TenantConfigToml { compaction_algorithm: crate::models::CompactionAlgorithmSettings { kind: DEFAULT_COMPACTION_ALGORITHM, }, + compaction_shard_ancestor: DEFAULT_COMPACTION_SHARD_ANCESTOR, compaction_l0_first: DEFAULT_COMPACTION_L0_FIRST, compaction_l0_semaphore: DEFAULT_COMPACTION_L0_SEMAPHORE, l0_flush_delay_threshold: None, diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs index f491ed10e1..ea5456e04b 100644 --- a/libs/pageserver_api/src/models.rs +++ b/libs/pageserver_api/src/models.rs @@ -526,6 +526,8 @@ pub struct TenantConfigPatch { #[serde(skip_serializing_if = "FieldPatch::is_noop")] pub compaction_algorithm: FieldPatch, #[serde(skip_serializing_if = "FieldPatch::is_noop")] + pub compaction_shard_ancestor: FieldPatch, + #[serde(skip_serializing_if = "FieldPatch::is_noop")] pub compaction_l0_first: FieldPatch, #[serde(skip_serializing_if = "FieldPatch::is_noop")] pub compaction_l0_semaphore: FieldPatch, @@ -615,6 +617,9 @@ pub struct TenantConfig { #[serde(skip_serializing_if = "Option::is_none")] pub compaction_algorithm: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub compaction_shard_ancestor: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub compaction_l0_first: Option, @@ -724,6 +729,7 @@ impl TenantConfig { mut compaction_threshold, mut compaction_upper_limit, mut compaction_algorithm, + mut compaction_shard_ancestor, mut compaction_l0_first, mut compaction_l0_semaphore, mut l0_flush_delay_threshold, @@ -772,6 +778,9 @@ impl TenantConfig { .compaction_upper_limit .apply(&mut compaction_upper_limit); patch.compaction_algorithm.apply(&mut compaction_algorithm); + patch + .compaction_shard_ancestor + .apply(&mut compaction_shard_ancestor); patch.compaction_l0_first.apply(&mut compaction_l0_first); patch .compaction_l0_semaphore @@ -860,6 +869,7 @@ impl TenantConfig { compaction_threshold, compaction_upper_limit, compaction_algorithm, + compaction_shard_ancestor, compaction_l0_first, compaction_l0_semaphore, l0_flush_delay_threshold, @@ -920,6 +930,9 @@ impl TenantConfig { .as_ref() .unwrap_or(&global_conf.compaction_algorithm) .clone(), + compaction_shard_ancestor: self + .compaction_shard_ancestor + .unwrap_or(global_conf.compaction_shard_ancestor), compaction_l0_first: self .compaction_l0_first .unwrap_or(global_conf.compaction_l0_first), diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 613834dc88..bc54c85119 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -2702,6 +2702,14 @@ impl Timeline { .clone() } + pub fn get_compaction_shard_ancestor(&self) -> bool { + let tenant_conf = self.tenant_conf.load(); + tenant_conf + .tenant_conf + .compaction_shard_ancestor + .unwrap_or(self.conf.default_tenant_conf.compaction_shard_ancestor) + } + fn get_eviction_policy(&self) -> EvictionPolicy { let tenant_conf = self.tenant_conf.load(); tenant_conf diff --git a/pageserver/src/tenant/timeline/compaction.rs b/pageserver/src/tenant/timeline/compaction.rs index 76c153d60f..92b24a73c9 100644 --- a/pageserver/src/tenant/timeline/compaction.rs +++ b/pageserver/src/tenant/timeline/compaction.rs @@ -1239,8 +1239,7 @@ impl Timeline { let partition_count = self.partitioning.read().0.0.parts.len(); // 4. Shard ancestor compaction - - if self.shard_identity.count >= ShardCount::new(2) { + if self.get_compaction_shard_ancestor() && self.shard_identity.count >= ShardCount::new(2) { // Limit the number of layer rewrites to the number of partitions: this means its // runtime should be comparable to a full round of image layer creations, rather than // being potentially much longer. diff --git a/test_runner/regress/test_attach_tenant_config.py b/test_runner/regress/test_attach_tenant_config.py index 9b6930695c..ee408e3c65 100644 --- a/test_runner/regress/test_attach_tenant_config.py +++ b/test_runner/regress/test_attach_tenant_config.py @@ -155,6 +155,7 @@ def test_fully_custom_config(positive_env: NeonEnv): "compaction_algorithm": { "kind": "tiered", }, + "compaction_shard_ancestor": False, "eviction_policy": { "kind": "LayerAccessThreshold", "period": "20s", From 46100717ad4a7d9d7844233933833320b003db9f Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Wed, 16 Apr 2025 17:38:10 +0200 Subject: [PATCH 25/27] pageserver: add `VectoredBlob::raw_with_header` (#11607) ## Problem To avoid recompressing page images during layer filtering, we need access to the raw header and data from vectored reads such that we can pass them through to the target layer. Touches #11562. ## Summary of changes Adds `VectoredBlob::raw_with_header()` to return a raw view of the header+data, and updates `read()` to track it. Also adds `blob_io::Header` with header metadata and decode logic, to reuse for tests and assertions. This isn't yet widely used. --- pageserver/src/tenant/blob_io.rs | 57 +++++++++++++++++ pageserver/src/tenant/vectored_blob_io.rs | 77 +++++++++++------------ 2 files changed, 93 insertions(+), 41 deletions(-) diff --git a/pageserver/src/tenant/blob_io.rs b/pageserver/src/tenant/blob_io.rs index abeaa166a4..d1dd105b13 100644 --- a/pageserver/src/tenant/blob_io.rs +++ b/pageserver/src/tenant/blob_io.rs @@ -37,6 +37,63 @@ pub struct CompressionInfo { pub compressed_size: Option, } +/// A blob header, with header+data length and compression info. +/// +/// TODO: use this more widely, and add an encode() method too. +/// TODO: document the header format. +#[derive(Clone, Copy, Default)] +pub struct Header { + pub header_len: usize, + pub data_len: usize, + pub compression_bits: u8, +} + +impl Header { + /// Decodes a header from a byte slice. + pub fn decode(bytes: &[u8]) -> Result { + let Some(&first_header_byte) = bytes.first() else { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "zero-length blob header", + )); + }; + + // If the first bit is 0, this is just a 1-byte length prefix up to 128 bytes. + if first_header_byte < 0x80 { + return Ok(Self { + header_len: 1, // by definition + data_len: first_header_byte as usize, + compression_bits: BYTE_UNCOMPRESSED, + }); + } + + // Otherwise, this is a 4-byte header containing compression information and length. + const HEADER_LEN: usize = 4; + let mut header_buf: [u8; HEADER_LEN] = bytes[0..HEADER_LEN].try_into().map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("blob header too short: {bytes:?}"), + ) + })?; + + // TODO: verify the compression bits and convert to an enum. + let compression_bits = header_buf[0] & LEN_COMPRESSION_BIT_MASK; + header_buf[0] &= !LEN_COMPRESSION_BIT_MASK; + let data_len = u32::from_be_bytes(header_buf) as usize; + + Ok(Self { + header_len: HEADER_LEN, + data_len, + compression_bits, + }) + } + + /// Returns the total header+data length. + pub fn total_len(&self) -> usize { + self.header_len + self.data_len + } +} + impl BlockCursor<'_> { /// Read a blob into a new buffer. pub async fn read_blob( diff --git a/pageserver/src/tenant/vectored_blob_io.rs b/pageserver/src/tenant/vectored_blob_io.rs index 166917d674..8e535a55d7 100644 --- a/pageserver/src/tenant/vectored_blob_io.rs +++ b/pageserver/src/tenant/vectored_blob_io.rs @@ -26,7 +26,7 @@ use utils::lsn::Lsn; use utils::vec_map::VecMap; use crate::context::RequestContext; -use crate::tenant::blob_io::{BYTE_UNCOMPRESSED, BYTE_ZSTD, LEN_COMPRESSION_BIT_MASK}; +use crate::tenant::blob_io::{BYTE_UNCOMPRESSED, BYTE_ZSTD, Header}; use crate::virtual_file::{self, IoBufferMut, VirtualFile}; /// Metadata bundled with the start and end offset of a blob. @@ -111,18 +111,20 @@ impl From for BufView<'_> { pub struct VectoredBlob { /// Blob metadata. pub meta: BlobMeta, - /// Start offset. - start: usize, + /// Header start offset. + header_start: usize, + /// Data start offset. + data_start: usize, /// End offset. end: usize, - /// Compression used on the the blob. + /// Compression used on the data, extracted from the header. compression_bits: u8, } impl VectoredBlob { /// Reads a decompressed view of the blob. pub(crate) async fn read<'a>(&self, buf: &BufView<'a>) -> Result, std::io::Error> { - let view = buf.view(self.start..self.end); + let view = buf.view(self.data_start..self.end); match self.compression_bits { BYTE_UNCOMPRESSED => Ok(view), @@ -140,13 +142,19 @@ impl VectoredBlob { std::io::ErrorKind::InvalidData, format!( "Failed to decompress blob for {}@{}, {}..{}: invalid compression byte {bits:x}", - self.meta.key, self.meta.lsn, self.start, self.end + self.meta.key, self.meta.lsn, self.data_start, self.end ), ); Err(error) } } } + + /// Returns the raw blob including header. + #[allow(unused)] + pub(crate) fn raw_with_header<'a>(&self, buf: &BufView<'a>) -> BufView<'a> { + buf.view(self.header_start..self.end) + } } impl std::fmt::Display for VectoredBlob { @@ -154,7 +162,7 @@ impl std::fmt::Display for VectoredBlob { write!( f, "{}@{}, {}..{}", - self.meta.key, self.meta.lsn, self.start, self.end + self.meta.key, self.meta.lsn, self.data_start, self.end ) } } @@ -493,50 +501,28 @@ impl<'a> VectoredBlobReader<'a> { let blobs_at = read.blobs_at.as_slice(); - let start_offset = read.start; - - let mut metas = Vec::with_capacity(blobs_at.len()); + let mut blobs = Vec::with_capacity(blobs_at.len()); // Blobs in `read` only provide their starting offset. The end offset // of a blob is implicit: the start of the next blob if one exists // or the end of the read. - for (blob_start, meta) in blobs_at { - let blob_start_in_buf = blob_start - start_offset; - let first_len_byte = buf[blob_start_in_buf as usize]; + for (blob_start, meta) in blobs_at.iter().copied() { + let header_start = (blob_start - read.start) as usize; + let header = Header::decode(&buf[header_start..])?; + let data_start = header_start + header.header_len; + let end = data_start + header.data_len; + let compression_bits = header.compression_bits; - // Each blob is prefixed by a header containing its size and compression information. - // Extract the size and skip that header to find the start of the data. - // The size can be 1 or 4 bytes. The most significant bit is 0 in the - // 1 byte case and 1 in the 4 byte case. - let (size_length, blob_size, compression_bits) = if first_len_byte < 0x80 { - (1, first_len_byte as u64, BYTE_UNCOMPRESSED) - } else { - let mut blob_size_buf = [0u8; 4]; - let offset_in_buf = blob_start_in_buf as usize; - - blob_size_buf.copy_from_slice(&buf[offset_in_buf..offset_in_buf + 4]); - blob_size_buf[0] &= !LEN_COMPRESSION_BIT_MASK; - - let compression_bits = first_len_byte & LEN_COMPRESSION_BIT_MASK; - ( - 4, - u32::from_be_bytes(blob_size_buf) as u64, - compression_bits, - ) - }; - - let start = (blob_start_in_buf + size_length) as usize; - let end = start + blob_size as usize; - - metas.push(VectoredBlob { - start, + blobs.push(VectoredBlob { + header_start, + data_start, end, - meta: *meta, + meta, compression_bits, }); } - Ok(VectoredBlobsBuf { buf, blobs: metas }) + Ok(VectoredBlobsBuf { buf, blobs }) } } @@ -997,6 +983,15 @@ mod tests { &read_buf[..], "mismatch for idx={idx} at offset={offset}" ); + + // Check that raw_with_header returns a valid header. + let raw = read_blob.raw_with_header(&view); + let header = Header::decode(&raw)?; + if !compression || header.header_len == 1 { + assert_eq!(header.compression_bits, BYTE_UNCOMPRESSED); + } + assert_eq!(raw.len(), header.total_len()); + buf = result.buf; } Ok(()) From 7747a9619f2c2b5c91e9f7581e7f1c8adf70b035 Mon Sep 17 00:00:00 2001 From: Anastasia Lubennikova Date: Wed, 16 Apr 2025 16:55:11 +0100 Subject: [PATCH 26/27] compute: fix copy-paste typo for neon GUC parameters check (#11610) fix for commit [5063151](https://github.com/neondatabase/neon/commit/50631512710d8c5fd9c4c681d7882e16f4df93f3) --- pgxn/neon/libpagestore.c | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pgxn/neon/libpagestore.c b/pgxn/neon/libpagestore.c index 19c14511bd..64d38e7913 100644 --- a/pgxn/neon/libpagestore.c +++ b/pgxn/neon/libpagestore.c @@ -1362,7 +1362,7 @@ pg_init_libpagestore(void) "", PGC_POSTMASTER, 0, /* no flags required */ - check_neon_id, NULL, NULL); + NULL, NULL, NULL); DefineCustomStringVariable("neon.branch_id", "Neon branch_id the server is running on", NULL, @@ -1370,7 +1370,7 @@ pg_init_libpagestore(void) "", PGC_POSTMASTER, 0, /* no flags required */ - check_neon_id, NULL, NULL); + NULL, NULL, NULL); DefineCustomStringVariable("neon.endpoint_id", "Neon endpoint_id the server is running on", NULL, @@ -1378,7 +1378,7 @@ pg_init_libpagestore(void) "", PGC_POSTMASTER, 0, /* no flags required */ - check_neon_id, NULL, NULL); + NULL, NULL, NULL); DefineCustomIntVariable("neon.stripe_size", "sharding stripe size", From 0e00faf52891f36db29b471a5a7ad89b8a123b21 Mon Sep 17 00:00:00 2001 From: Vlad Lazar Date: Wed, 16 Apr 2025 17:31:23 +0100 Subject: [PATCH 27/27] tests: stability fixes for `test_migration_to_cold_secondary` (#11606) 1. Compute may generate WAL on shutdown. The test assumes that after shutdown, no further ingest happens. Tweak the compute shutdown to make the assumption true. 2. Assertion of local layer count post cold migration is not right since we may have downloaded layers due to ingest. Remove it. Closes https://github.com/neondatabase/neon/issues/11587 --- test_runner/regress/test_pageserver_secondary.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/test_runner/regress/test_pageserver_secondary.py b/test_runner/regress/test_pageserver_secondary.py index d48e731394..3aa0c63979 100644 --- a/test_runner/regress/test_pageserver_secondary.py +++ b/test_runner/regress/test_pageserver_secondary.py @@ -14,6 +14,7 @@ from fixtures.neon_fixtures import ( NeonEnvBuilder, NeonPageserver, StorageControllerMigrationConfig, + flush_ep_to_pageserver, ) from fixtures.pageserver.common_types import parse_layer_file_name from fixtures.pageserver.utils import ( @@ -997,10 +998,6 @@ def test_migration_to_cold_secondary(neon_env_builder: NeonEnvBuilder): ps_secondary.http_client().tenant_heatmap_upload(tenant_id) heatmap_after_migration = timeline_heatmap(timeline_id) - local_layers = ps_secondary.list_layers(tenant_id, timeline_id) - # We download 1 layer per second and give up within 5 seconds. - assert len(local_layers) < 10 - after_migration_heatmap_layers_count = len(heatmap_after_migration["layers"]) log.info(f"Heatmap size after cold migration is {after_migration_heatmap_layers_count}") @@ -1038,9 +1035,14 @@ def test_migration_to_cold_secondary(neon_env_builder: NeonEnvBuilder): .value ) - workload.stop() assert before == after + # Stop the endpoint and wait until any finally written WAL propagates to + # the pageserver and is uploaded to remote storage. + flush_ep_to_pageserver(env, workload.endpoint(), tenant_id, timeline_id) + ps_secondary.http_client().timeline_checkpoint(tenant_id, timeline_id, wait_until_uploaded=True) + workload.stop() + # Now simulate the case where a child timeline is archived, parent layers # are evicted and the child is unarchived. When the child is unarchived, # itself and the parent update their heatmaps to contain layers needed by the