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/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 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: | diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 46c8cd6fc9..80c4511b36 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: @@ -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: diff --git a/Cargo.lock b/Cargo.lock index 5d2cdcea27..7ab9378853 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,12 @@ dependencies = [ "humantime", "humantime-serde", "hyper 0.14.30", + "jsonwebtoken", "nix 0.27.1", "once_cell", "pageserver_api", "pageserver_client", + "pem", "postgres_backend", "postgres_connection", "regex", @@ -1437,6 +1440,8 @@ dependencies = [ "scopeguard", "serde", "serde_json", + "sha2", + "spki 0.7.3", "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", diff --git a/Cargo.toml b/Cargo.toml index d957fa9070..9d7904a787 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -141,6 +141,7 @@ 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" pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] } procfs = "0.16" @@ -174,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/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/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, }) } diff --git a/compute_tools/src/http/middleware/authorize.rs b/compute_tools/src/http/middleware/authorize.rs index ee3a5cb953..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}, @@ -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}; @@ -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") { @@ -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 @@ -92,7 +79,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 +99,16 @@ impl Authorize { token: &str, validation: &Validation, ) -> Result> { + debug_assert!(!jwks.keys.is_empty()); + + 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 +121,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 +131,6 @@ impl Authorize { } } - Err(anyhow!("Failed to verify authorization token")) + Err(anyhow!("failed to verify authorization token")) } } diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 162c49ec7c..92f0071bac 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -6,13 +6,16 @@ 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 humantime-serde.workspace = true hyper0.workspace = true regex.workspace = true @@ -20,6 +23,8 @@ reqwest = { workspace = true, features = ["blocking", "json"] } 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/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..b569b0fb8e 100644 --- a/control_plane/src/endpoint.rs +++ b/control_plane/src/endpoint.rs @@ -42,22 +42,30 @@ 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 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}; @@ -82,6 +90,7 @@ pub struct EndpointConf { drop_subscriptions_before_start: bool, features: Vec, cluster: Option, + compute_ctl_config: ComputeCtlConfig, } // @@ -137,6 +146,37 @@ 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 spki: SubjectPublicKeyInfoRef = SubjectPublicKeyInfo::from_der(pem.contents())?; + let public_key = spki.subject_public_key.raw_bytes(); + + let mut hasher = Sha256::new(); + hasher.update(public_key); + 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(public_key, base64::URL_SAFE_NO_PAD), + }), + }], + }) + } + #[allow(clippy::too_many_arguments)] pub fn new_endpoint( &mut self, @@ -154,6 +194,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 +225,7 @@ impl ComputeControlPlane { reconfigure_concurrency: 1, features: vec![], cluster: None, + compute_ctl_config: compute_ctl_config.clone(), }); ep.create_endpoint_dir()?; @@ -200,6 +245,7 @@ impl ComputeControlPlane { reconfigure_concurrency: 1, features: vec![], cluster: None, + compute_ctl_config, })?, )?; std::fs::write( @@ -242,7 +288,6 @@ impl ComputeControlPlane { /////////////////////////////////////////////////////////////////////////////// -#[derive(Debug)] pub struct Endpoint { /// used as the directory name endpoint_id: String, @@ -271,6 +316,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 +381,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 +629,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 +762,7 @@ impl Endpoint { ComputeConfig { spec: Some(spec), - compute_ctl_config: ComputeCtlConfig::default(), + compute_ctl_config: self.compute_ctl_config.clone(), } }; @@ -774,16 +830,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 +928,7 @@ impl Endpoint { self.external_http_address.port() ), ) + .bearer_auth(self.generate_jwt()?) .send() .await?; @@ -957,6 +1005,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/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/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/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 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/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/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 56d97bf8a9..5c5bab0642 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 @@ -33,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 d13dd3c77b..b000a25376 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, @@ -2833,7 +2834,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/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/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/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.rs b/pageserver/src/tenant/timeline.rs index 204bdb5eee..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 @@ -4026,7 +4034,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 } => { @@ -5702,6 +5710,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 diff --git a/pageserver/src/tenant/timeline/compaction.rs b/pageserver/src/tenant/timeline/compaction.rs index 91cc8ca10c..92b24a73c9 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::{ @@ -69,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); @@ -819,7 +827,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 +876,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 +889,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 +910,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(()) } } @@ -1222,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. @@ -1273,7 +1289,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. @@ -1317,14 +1336,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", ); } @@ -1336,12 +1356,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 @@ -1371,7 +1398,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(), 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(()) 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..64d38e7913 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; @@ -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", @@ -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, 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/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 */ 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/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 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/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/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" ) 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", 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", 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 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 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" ] }