From 2de2b26c62016cce48cbc5449d44e3259f237b56 Mon Sep 17 00:00:00 2001 From: "Alex Chi Z." <4198311+skyzh@users.noreply.github.com> Date: Mon, 20 Jan 2025 15:44:12 -0500 Subject: [PATCH 01/14] feat(pageserver): add reldir migration configs (#10439) ## Problem Part of #9516 per RFC at https://github.com/neondatabase/neon/pull/10412 ## Summary of changes Adding the necessary config items and index_part items for the large relation count work. --------- Signed-off-by: Alex Chi Z --- control_plane/src/pageserver.rs | 5 + libs/pageserver_api/src/config.rs | 5 + libs/pageserver_api/src/models.rs | 6 + pageserver/src/tenant.rs | 1 + pageserver/src/tenant/config.rs | 8 ++ .../tenant/remote_timeline_client/index.rs | 115 +++++++++++++++++- .../regress/test_attach_tenant_config.py | 1 + 7 files changed, 138 insertions(+), 3 deletions(-) diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs index ef5b3d6593..df81b44f2d 100644 --- a/control_plane/src/pageserver.rs +++ b/control_plane/src/pageserver.rs @@ -418,6 +418,11 @@ impl PageServerNode { .map(serde_json::from_str) .transpose() .context("parse `wal_receiver_protocol_override` from json")?, + rel_size_v2_enabled: settings + .remove("rel_size_v2_enabled") + .map(|x| x.parse::()) + .transpose() + .context("Failed to parse 'rel_size_v2_enabled' as bool")?, }; if !settings.is_empty() { bail!("Unrecognized tenant settings: {settings:?}") diff --git a/libs/pageserver_api/src/config.rs b/libs/pageserver_api/src/config.rs index 09cfbc55fd..7fb7a9d54e 100644 --- a/libs/pageserver_api/src/config.rs +++ b/libs/pageserver_api/src/config.rs @@ -301,6 +301,10 @@ pub struct TenantConfigToml { pub timeline_offloading: bool, pub wal_receiver_protocol_override: Option, + + /// Enable rel_size_v2 for this tenant. Once enabled, the tenant will persist this information into + /// `index_part.json`, and it cannot be reversed. + pub rel_size_v2_enabled: Option, } pub mod defaults { @@ -538,6 +542,7 @@ impl Default for TenantConfigToml { lsn_lease_length_for_ts: LsnLease::DEFAULT_LENGTH_FOR_TS, timeline_offloading: false, wal_receiver_protocol_override: None, + rel_size_v2_enabled: None, } } } diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs index c38af9cb80..1538134c96 100644 --- a/libs/pageserver_api/src/models.rs +++ b/libs/pageserver_api/src/models.rs @@ -497,6 +497,8 @@ pub struct TenantConfigPatch { pub timeline_offloading: FieldPatch, #[serde(skip_serializing_if = "FieldPatch::is_noop")] pub wal_receiver_protocol_override: FieldPatch, + #[serde(skip_serializing_if = "FieldPatch::is_noop")] + pub rel_size_v2_enabled: FieldPatch, } /// An alternative representation of `pageserver::tenant::TenantConf` with @@ -528,6 +530,7 @@ pub struct TenantConfig { pub lsn_lease_length_for_ts: Option, pub timeline_offloading: Option, pub wal_receiver_protocol_override: Option, + pub rel_size_v2_enabled: Option, } impl TenantConfig { @@ -557,6 +560,7 @@ impl TenantConfig { mut lsn_lease_length_for_ts, mut timeline_offloading, mut wal_receiver_protocol_override, + mut rel_size_v2_enabled, } = self; patch.checkpoint_distance.apply(&mut checkpoint_distance); @@ -601,6 +605,7 @@ impl TenantConfig { patch .wal_receiver_protocol_override .apply(&mut wal_receiver_protocol_override); + patch.rel_size_v2_enabled.apply(&mut rel_size_v2_enabled); Self { checkpoint_distance, @@ -627,6 +632,7 @@ impl TenantConfig { lsn_lease_length_for_ts, timeline_offloading, wal_receiver_protocol_override, + rel_size_v2_enabled, } } } diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index bb1b36aed6..05a311391c 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -5475,6 +5475,7 @@ pub(crate) mod harness { lsn_lease_length_for_ts: Some(tenant_conf.lsn_lease_length_for_ts), timeline_offloading: Some(tenant_conf.timeline_offloading), wal_receiver_protocol_override: tenant_conf.wal_receiver_protocol_override, + rel_size_v2_enabled: tenant_conf.rel_size_v2_enabled, } } } diff --git a/pageserver/src/tenant/config.rs b/pageserver/src/tenant/config.rs index edf2e6a3aa..14d8e9ccd4 100644 --- a/pageserver/src/tenant/config.rs +++ b/pageserver/src/tenant/config.rs @@ -357,6 +357,9 @@ pub struct TenantConfOpt { #[serde(skip_serializing_if = "Option::is_none")] pub wal_receiver_protocol_override: Option, + + #[serde(skip_serializing_if = "Option::is_none")] + pub rel_size_v2_enabled: Option, } impl TenantConfOpt { @@ -425,6 +428,7 @@ impl TenantConfOpt { wal_receiver_protocol_override: self .wal_receiver_protocol_override .or(global_conf.wal_receiver_protocol_override), + rel_size_v2_enabled: self.rel_size_v2_enabled.or(global_conf.rel_size_v2_enabled), } } @@ -454,6 +458,7 @@ impl TenantConfOpt { mut lsn_lease_length_for_ts, mut timeline_offloading, mut wal_receiver_protocol_override, + mut rel_size_v2_enabled, } = self; patch.checkpoint_distance.apply(&mut checkpoint_distance); @@ -522,6 +527,7 @@ impl TenantConfOpt { patch .wal_receiver_protocol_override .apply(&mut wal_receiver_protocol_override); + patch.rel_size_v2_enabled.apply(&mut rel_size_v2_enabled); Ok(Self { checkpoint_distance, @@ -548,6 +554,7 @@ impl TenantConfOpt { lsn_lease_length_for_ts, timeline_offloading, wal_receiver_protocol_override, + rel_size_v2_enabled, }) } } @@ -603,6 +610,7 @@ impl From for models::TenantConfig { lsn_lease_length_for_ts: value.lsn_lease_length_for_ts.map(humantime), timeline_offloading: value.timeline_offloading, wal_receiver_protocol_override: value.wal_receiver_protocol_override, + rel_size_v2_enabled: value.rel_size_v2_enabled, } } } diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs index 08e94ae197..30b6b07ca3 100644 --- a/pageserver/src/tenant/remote_timeline_client/index.rs +++ b/pageserver/src/tenant/remote_timeline_client/index.rs @@ -79,6 +79,24 @@ pub struct IndexPart { /// when this flag is introduced. #[serde(skip_serializing_if = "Option::is_none", default)] pub(crate) last_aux_file_policy: Option, + + #[serde(skip_serializing_if = "Option::is_none", default)] + pub(crate) rel_size_migration: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum RelSizeMigration { + /// The tenant is using the old rel_size format. + /// Note that this enum is persisted as `Option` in the index part, so + /// `None` is the same as `Some(RelSizeMigration::Legacy)`. + Legacy, + /// The tenant is migrating to the new rel_size format. Both old and new rel_size format are + /// persisted in the index part. The read path will read both formats and merge them. + Migrating, + /// The tenant has migrated to the new rel_size format. Only the new rel_size format is persisted + /// in the index part, and the read path will not read the old format. + Migrated, } impl IndexPart { @@ -97,10 +115,11 @@ impl IndexPart { /// - 8: added `archived_at` /// - 9: +gc_blocking /// - 10: +import_pgdata - const LATEST_VERSION: usize = 10; + /// - 11: +rel_size_migration + const LATEST_VERSION: usize = 11; // Versions we may see when reading from a bucket. - pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; + pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]; pub const FILE_NAME: &'static str = "index_part.json"; @@ -116,6 +135,7 @@ impl IndexPart { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, } } @@ -416,6 +436,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -461,6 +482,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -507,6 +529,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -556,6 +579,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, }; let empty_layers_parsed = IndexPart::from_json_bytes(empty_layers_json.as_bytes()).unwrap(); @@ -600,6 +624,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -647,6 +672,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -699,6 +725,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: Some(AuxFilePolicy::V2), import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -756,6 +783,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: Default::default(), import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -814,6 +842,7 @@ mod tests { gc_blocking: None, last_aux_file_policy: Default::default(), import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -877,6 +906,7 @@ mod tests { last_aux_file_policy: Default::default(), archived_at: None, import_pgdata: None, + rel_size_migration: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -952,7 +982,86 @@ mod tests { started_at: parse_naive_datetime("2024-11-13T09:23:42.123000000"), finished_at: parse_naive_datetime("2024-11-13T09:42:23.123000000"), idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()), - }))) + }))), + rel_size_migration: None, + }; + + let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); + assert_eq!(part, expected); + } + + #[test] + fn v11_rel_size_migration_is_parsed() { + let example = r#"{ + "version": 11, + "layer_metadata":{ + "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9": { "file_size": 25600000 }, + "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51": { "file_size": 9007199254741001 } + }, + "disk_consistent_lsn":"0/16960E8", + "metadata": { + "disk_consistent_lsn": "0/16960E8", + "prev_record_lsn": "0/1696070", + "ancestor_timeline": "e45a7f37d3ee2ff17dc14bf4f4e3f52e", + "ancestor_lsn": "0/0", + "latest_gc_cutoff_lsn": "0/1696070", + "initdb_lsn": "0/1696070", + "pg_version": 14 + }, + "gc_blocking": { + "started_at": "2024-07-19T09:00:00.123", + "reasons": ["DetachAncestor"] + }, + "import_pgdata": { + "V1": { + "Done": { + "idempotency_key": "specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5", + "started_at": "2024-11-13T09:23:42.123", + "finished_at": "2024-11-13T09:42:23.123" + } + } + }, + "rel_size_migration": "legacy" + }"#; + + let expected = IndexPart { + version: 11, + layer_metadata: HashMap::from([ + ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), LayerFileMetadata { + file_size: 25600000, + generation: Generation::none(), + shard: ShardIndex::unsharded() + }), + ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap(), LayerFileMetadata { + file_size: 9007199254741001, + generation: Generation::none(), + shard: ShardIndex::unsharded() + }) + ]), + disk_consistent_lsn: "0/16960E8".parse::().unwrap(), + metadata: TimelineMetadata::new( + Lsn::from_str("0/16960E8").unwrap(), + Some(Lsn::from_str("0/1696070").unwrap()), + Some(TimelineId::from_str("e45a7f37d3ee2ff17dc14bf4f4e3f52e").unwrap()), + Lsn::INVALID, + Lsn::from_str("0/1696070").unwrap(), + Lsn::from_str("0/1696070").unwrap(), + 14, + ).with_recalculated_checksum().unwrap(), + deleted_at: None, + lineage: Default::default(), + gc_blocking: Some(GcBlocking { + started_at: parse_naive_datetime("2024-07-19T09:00:00.123000000"), + reasons: enumset::EnumSet::from_iter([GcBlockingReason::DetachAncestor]), + }), + last_aux_file_policy: Default::default(), + archived_at: None, + import_pgdata: Some(import_pgdata::index_part_format::Root::V1(import_pgdata::index_part_format::V1::Done(import_pgdata::index_part_format::Done{ + started_at: parse_naive_datetime("2024-11-13T09:23:42.123000000"), + finished_at: parse_naive_datetime("2024-11-13T09:42:23.123000000"), + idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()), + }))), + rel_size_migration: Some(RelSizeMigration::Legacy), }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); diff --git a/test_runner/regress/test_attach_tenant_config.py b/test_runner/regress/test_attach_tenant_config.py index 45112fd67e..b34dbddc80 100644 --- a/test_runner/regress/test_attach_tenant_config.py +++ b/test_runner/regress/test_attach_tenant_config.py @@ -176,6 +176,7 @@ def test_fully_custom_config(positive_env: NeonEnv): "type": "interpreted", "args": {"format": "bincode", "compression": {"zstd": {"level": 1}}}, }, + "rel_size_v2_enabled": True, } vps_http = env.storage_controller.pageserver_api() From 2ab9f6982590cd8570f505df23c134cc71a1a576 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arpad=20M=C3=BCller?= Date: Mon, 20 Jan 2025 22:57:15 +0100 Subject: [PATCH 02/14] Simplify pageserver_physical_gc function (#10104) This simplifies the code in `pageserver_physical_gc` a little bit after the feedback in #10007 that the code is too complicated. Most importantly, we don't pass around `GcSummary` any more in a complicated fashion, and we save on async stream-combinator-inception in one place in favour of `try_stream!{}`. Follow-up of #10007 --- .../src/pageserver_physical_gc.rs | 86 +++++++++---------- 1 file changed, 40 insertions(+), 46 deletions(-) diff --git a/storage_scrubber/src/pageserver_physical_gc.rs b/storage_scrubber/src/pageserver_physical_gc.rs index a997373375..063c6bcfb9 100644 --- a/storage_scrubber/src/pageserver_physical_gc.rs +++ b/storage_scrubber/src/pageserver_physical_gc.rs @@ -8,6 +8,8 @@ use crate::checks::{ }; use crate::metadata_stream::{stream_tenant_timelines, stream_tenants}; use crate::{init_remote, BucketConfig, NodeKind, RootTarget, TenantShardTimelineId, MAX_RETRIES}; +use async_stream::try_stream; +use futures::future::Either; use futures_util::{StreamExt, TryStreamExt}; use pageserver::tenant::remote_timeline_client::index::LayerFileMetadata; use pageserver::tenant::remote_timeline_client::manifest::OffloadedTimelineManifest; @@ -578,7 +580,7 @@ async fn gc_timeline( target: &RootTarget, mode: GcMode, ttid: TenantShardTimelineId, - accumulator: &Arc>, + accumulator: &std::sync::Mutex, tenant_manifest_info: Arc>, ) -> anyhow::Result { let mut summary = GcSummary::default(); @@ -721,9 +723,9 @@ pub async fn pageserver_physical_gc( let remote_client = Arc::new(remote_client); let tenants = if tenant_shard_ids.is_empty() { - futures::future::Either::Left(stream_tenants(&remote_client, &target)) + Either::Left(stream_tenants(&remote_client, &target)) } else { - futures::future::Either::Right(futures::stream::iter(tenant_shard_ids.into_iter().map(Ok))) + Either::Right(futures::stream::iter(tenant_shard_ids.into_iter().map(Ok))) }; // How many tenants to process in parallel. We need to be mindful of pageservers @@ -731,16 +733,16 @@ pub async fn pageserver_physical_gc( const CONCURRENCY: usize = 32; // Accumulate information about each tenant for cross-shard GC step we'll do at the end - let accumulator = Arc::new(std::sync::Mutex::new(TenantRefAccumulator::default())); + let accumulator = std::sync::Mutex::new(TenantRefAccumulator::default()); + + // Accumulate information about how many manifests we have GCd + let manifest_gc_summary = std::sync::Mutex::new(GcSummary::default()); // Generate a stream of TenantTimelineId - enum GcSummaryOrContent { - Content(T), - GcSummary(GcSummary), - } let timelines = tenants.map_ok(|tenant_shard_id| { let target_ref = ⌖ let remote_client_ref = &remote_client; + let manifest_gc_summary_ref = &manifest_gc_summary; async move { let gc_manifest_result = gc_tenant_manifests( remote_client_ref, @@ -757,55 +759,48 @@ pub async fn pageserver_physical_gc( (GcSummary::default(), None) } }; + manifest_gc_summary_ref + .lock() + .unwrap() + .merge(summary_from_manifest); let tenant_manifest_arc = Arc::new(tenant_manifest_opt); - let summary_from_manifest = Ok(GcSummaryOrContent::<(_, _)>::GcSummary( - summary_from_manifest, - )); - stream_tenant_timelines(remote_client_ref, target_ref, tenant_shard_id) - .await - .map(|stream| { - stream - .zip(futures::stream::iter(std::iter::repeat( - tenant_manifest_arc, - ))) - .map(|(ttid_res, tenant_manifest_arc)| { - ttid_res.map(move |ttid| { - GcSummaryOrContent::Content((ttid, tenant_manifest_arc)) - }) - }) - .chain(futures::stream::iter([summary_from_manifest].into_iter())) - }) + let mut timelines = Box::pin( + stream_tenant_timelines(remote_client_ref, target_ref, tenant_shard_id).await?, + ); + Ok(try_stream! { + while let Some(ttid_res) = timelines.next().await { + let ttid = ttid_res?; + yield (ttid, tenant_manifest_arc.clone()); + } + }) } }); - let timelines = std::pin::pin!(timelines.try_buffered(CONCURRENCY)); - let timelines = timelines.try_flatten(); let mut summary = GcSummary::default(); - - // Drain futures for per-shard GC, populating accumulator as a side effect { - let timelines = timelines.map_ok(|summary_or_ttid| match summary_or_ttid { - GcSummaryOrContent::Content((ttid, tenant_manifest_arc)) => { - futures::future::Either::Left(gc_timeline( - &remote_client, - &min_age, - &target, - mode, - ttid, - &accumulator, - tenant_manifest_arc, - )) - } - GcSummaryOrContent::GcSummary(gc_summary) => { - futures::future::Either::Right(futures::future::ok(gc_summary)) - } + let timelines = std::pin::pin!(timelines.try_buffered(CONCURRENCY)); + let timelines = timelines.try_flatten(); + + let timelines = timelines.map_ok(|(ttid, tenant_manifest_arc)| { + gc_timeline( + &remote_client, + &min_age, + &target, + mode, + ttid, + &accumulator, + tenant_manifest_arc, + ) }); let mut timelines = std::pin::pin!(timelines.try_buffered(CONCURRENCY)); + // Drain futures for per-shard GC, populating accumulator as a side effect while let Some(i) = timelines.next().await { summary.merge(i?); } } + // Streams are lazily evaluated, so only now do we have access to the inner object + summary.merge(manifest_gc_summary.into_inner().unwrap()); // Execute cross-shard GC, using the accumulator's full view of all the shards built in the per-shard GC let Some(client) = controller_client else { @@ -813,8 +808,7 @@ pub async fn pageserver_physical_gc( return Ok(summary); }; - let (ancestor_shards, ancestor_refs) = Arc::into_inner(accumulator) - .unwrap() + let (ancestor_shards, ancestor_refs) = accumulator .into_inner() .unwrap() .into_gc_ancestors(client, &mut summary) From 624a5075444a92a199378e80182fad4eecf8e509 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?JC=20Gr=C3=BCnhage?= Date: Tue, 21 Jan 2025 13:45:21 +0100 Subject: [PATCH 03/14] Create Github releases with empty body for now (#10448) ## Problem When releasing `release-7574`, the Github Release creation failed with "body is too long" (see https://github.com/neondatabase/neon/actions/runs/12834025431/job/35792346745#step:5:77). There's lots of room for improvement of the release notes, but for now we'll disable them instead. ## Summary of changes - Disable automatic generation of release notes for Github releases - Enable creation of Github releases for proxy/compute --- .github/workflows/build_and_test.yml | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index b0e07535b3..4fc81dccaa 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -1078,12 +1078,6 @@ jobs: console.log(`Tag ${tag} created successfully.`); } - // TODO: check how GitHub releases looks for proxy/compute releases and enable them if they're ok - if (context.ref !== 'refs/heads/release') { - console.log(`GitHub release skipped for ${context.ref}.`); - return; - } - try { const existingRelease = await github.rest.repos.getReleaseByTag({ owner: context.repo.owner, @@ -1102,7 +1096,8 @@ jobs: owner: context.repo.owner, repo: context.repo.repo, tag_name: tag, - generate_release_notes: true, + // TODO: Automate release notes properly + generate_release_notes: false, }); console.log(`Release for tag ${tag} created successfully.`); } From 7e4a39ea539abedf78e885797e23923f1d5e2873 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arpad=20M=C3=BCller?= Date: Tue, 21 Jan 2025 16:40:04 +0100 Subject: [PATCH 04/14] Fix two flakiness sources in test_scrubber_physical_gc_ancestors (#10457) We currently have some flakiness in `test_scrubber_physical_gc_ancestors`, see #10391. The first flakiness kind is about the reconciler not actually becoming idle within the timeout of 30 seconds. We see continuous forward progress so this is likely not a hang. We also see this happen in parallel to a test failure, so is likely due to runners being overloaded. Therefore, we increase the timeout. The second flakiness kind is an assertion failure. This one is a little bit more tricky, but we saw in the successful run that there was some advance of the lsn between the compaction ran (which created layer files) and the gc run. Apparently gc rejects reductions to the single image layer setting if the cutoff lsn is the same as the lsn of the image layer: it will claim that that layer is newer than the space cutoff and therefore skip it, while thinking the old layer (that we want to delete) is the latest one (so it's not deleted). We address the second flakiness kind by inserting a tiny amount of WAL between the compaction and gc. This should hopefully fix things. Related issue: #10391 (not closing it with the merger of the PR as we'll need to validate that these changes had the intended effect). Thanks to Chi for going over this together with me in a call. --- test_runner/regress/test_storage_scrubber.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/test_runner/regress/test_storage_scrubber.py b/test_runner/regress/test_storage_scrubber.py index 220c428531..a782e85567 100644 --- a/test_runner/regress/test_storage_scrubber.py +++ b/test_runner/regress/test_storage_scrubber.py @@ -227,7 +227,9 @@ def test_scrubber_physical_gc_ancestors(neon_env_builder: NeonEnvBuilder, shard_ new_shard_count = 4 assert shard_count is None or new_shard_count > shard_count shards = env.storage_controller.tenant_shard_split(tenant_id, shard_count=new_shard_count) - env.storage_controller.reconcile_until_idle() # Move shards to their final locations immediately + env.storage_controller.reconcile_until_idle( + timeout_secs=120 + ) # Move shards to their final locations immediately # Create a timeline after split, to ensure scrubber can handle timelines that exist in child shards but not ancestors env.storage_controller.pageserver_api().timeline_create( @@ -269,6 +271,8 @@ def test_scrubber_physical_gc_ancestors(neon_env_builder: NeonEnvBuilder, shard_ ps.http_client().timeline_compact( shard, timeline_id, force_image_layer_creation=True, wait_until_uploaded=True ) + # Add some WAL so that we don't gc at the latest remote consistent lsn + workload.churn_rows(1) ps.http_client().timeline_gc(shard, timeline_id, 0) # We will use a min_age_secs=1 threshold for deletion, let it pass From 19bf7b78a0c38508753d3f54e8faea695f686d19 Mon Sep 17 00:00:00 2001 From: Gleb Novikov Date: Tue, 21 Jan 2025 16:50:44 +0000 Subject: [PATCH 05/14] fast import: basic python test (#10271) We did not have any tests on fast_import binary yet. In this PR I have introduced: - `FastImport` class and tools for testing in python - basic test that runs fast import against vanilla postgres and checks that data is there Should be merged after https://github.com/neondatabase/neon/pull/10251 --- compute_tools/src/bin/fast_import.rs | 21 ++++- test_runner/conftest.py | 1 + test_runner/fixtures/fast_import.py | 104 ++++++++++++++++++++++ test_runner/regress/test_import_pgdata.py | 42 ++++++++- 4 files changed, 165 insertions(+), 3 deletions(-) create mode 100644 test_runner/fixtures/fast_import.py diff --git a/compute_tools/src/bin/fast_import.rs b/compute_tools/src/bin/fast_import.rs index 5b008f8182..c8440afb64 100644 --- a/compute_tools/src/bin/fast_import.rs +++ b/compute_tools/src/bin/fast_import.rs @@ -58,6 +58,8 @@ struct Args { pg_bin_dir: Utf8PathBuf, #[clap(long)] pg_lib_dir: Utf8PathBuf, + #[clap(long)] + pg_port: Option, // port to run postgres on, 5432 is default } #[serde_with::serde_as] @@ -74,6 +76,13 @@ enum EncryptionSecret { KMS { key_id: String }, } +// copied from pageserver_api::config::defaults::DEFAULT_LOCALE to avoid dependency just for a constant +const DEFAULT_LOCALE: &str = if cfg!(target_os = "macos") { + "C" +} else { + "C.UTF-8" +}; + #[tokio::main] pub(crate) async fn main() -> anyhow::Result<()> { utils::logging::init( @@ -97,6 +106,10 @@ pub(crate) async fn main() -> anyhow::Result<()> { let working_directory = args.working_directory; let pg_bin_dir = args.pg_bin_dir; let pg_lib_dir = args.pg_lib_dir; + let pg_port = args.pg_port.unwrap_or_else(|| { + info!("pg_port not specified, using default 5432"); + 5432 + }); // Initialize AWS clients only if s3_prefix is specified let (aws_config, kms_client) = if args.s3_prefix.is_some() { @@ -180,7 +193,7 @@ pub(crate) async fn main() -> anyhow::Result<()> { let superuser = "cloud_admin"; // XXX: this shouldn't be hard-coded postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs { superuser, - locale: "en_US.UTF-8", // XXX: this shouldn't be hard-coded, + locale: DEFAULT_LOCALE, // XXX: this shouldn't be hard-coded, pg_version, initdb_bin: pg_bin_dir.join("initdb").as_ref(), library_search_path: &pg_lib_dir, // TODO: is this right? Prob works in compute image, not sure about neon_local. @@ -197,6 +210,7 @@ pub(crate) async fn main() -> anyhow::Result<()> { let mut postgres_proc = tokio::process::Command::new(pgbin) .arg("-D") .arg(&pgdata_dir) + .args(["-p", &format!("{pg_port}")]) .args(["-c", "wal_level=minimal"]) .args(["-c", "shared_buffers=10GB"]) .args(["-c", "max_wal_senders=0"]) @@ -216,6 +230,7 @@ pub(crate) async fn main() -> anyhow::Result<()> { ), ]) .env_clear() + .env("LD_LIBRARY_PATH", &pg_lib_dir) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .spawn() @@ -232,7 +247,7 @@ pub(crate) async fn main() -> anyhow::Result<()> { // Create neondb database in the running postgres let restore_pg_connstring = - format!("host=localhost port=5432 user={superuser} dbname=postgres"); + format!("host=localhost port={pg_port} user={superuser} dbname=postgres"); let start_time = std::time::Instant::now(); @@ -314,6 +329,7 @@ pub(crate) async fn main() -> anyhow::Result<()> { .arg(&source_connection_string) // how we run it .env_clear() + .env("LD_LIBRARY_PATH", &pg_lib_dir) .kill_on_drop(true) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) @@ -347,6 +363,7 @@ pub(crate) async fn main() -> anyhow::Result<()> { .arg(&dumpdir) // how we run it .env_clear() + .env("LD_LIBRARY_PATH", &pg_lib_dir) .kill_on_drop(true) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) diff --git a/test_runner/conftest.py b/test_runner/conftest.py index 9e32469d69..4b591d3316 100644 --- a/test_runner/conftest.py +++ b/test_runner/conftest.py @@ -15,4 +15,5 @@ pytest_plugins = ( "fixtures.compare_fixtures", "fixtures.slow", "fixtures.reruns", + "fixtures.fast_import", ) diff --git a/test_runner/fixtures/fast_import.py b/test_runner/fixtures/fast_import.py new file mode 100644 index 0000000000..33248132ab --- /dev/null +++ b/test_runner/fixtures/fast_import.py @@ -0,0 +1,104 @@ +import os +import shutil +import subprocess +import tempfile +from collections.abc import Iterator +from pathlib import Path + +import pytest + +from fixtures.log_helper import log +from fixtures.neon_cli import AbstractNeonCli +from fixtures.pg_version import PgVersion + + +class FastImport(AbstractNeonCli): + COMMAND = "fast_import" + cmd: subprocess.CompletedProcess[str] | None = None + + def __init__( + self, + extra_env: dict[str, str] | None, + binpath: Path, + pg_distrib_dir: Path, + pg_version: PgVersion, + workdir: Path, + ): + if extra_env is None: + env_vars = {} + else: + env_vars = extra_env.copy() + + if not (binpath / self.COMMAND).exists(): + raise Exception(f"{self.COMMAND} binary not found at '{binpath}'") + super().__init__(env_vars, binpath) + + pg_dir = pg_distrib_dir / pg_version.v_prefixed + self.pg_distrib_dir = pg_distrib_dir + self.pg_version = pg_version + self.pg_bin = pg_dir / "bin" + if not (self.pg_bin / "postgres").exists(): + raise Exception(f"postgres binary was not found at '{self.pg_bin}'") + self.pg_lib = pg_dir / "lib" + if env_vars.get("LD_LIBRARY_PATH") is not None: + self.pg_lib = Path(env_vars["LD_LIBRARY_PATH"]) + elif os.getenv("LD_LIBRARY_PATH") is not None: + self.pg_lib = Path(str(os.getenv("LD_LIBRARY_PATH"))) + if not workdir.exists(): + raise Exception(f"Working directory '{workdir}' does not exist") + self.workdir = workdir + + def run( + self, + pg_port: int, + source_connection_string: str | None = None, + s3prefix: str | None = None, + interactive: bool = False, + ) -> subprocess.CompletedProcess[str]: + if self.cmd is not None: + raise Exception("Command already executed") + args = [ + f"--pg-bin-dir={self.pg_bin}", + f"--pg-lib-dir={self.pg_lib}", + f"--pg-port={pg_port}", + f"--working-directory={self.workdir}", + ] + if source_connection_string is not None: + args.append(f"--source-connection-string={source_connection_string}") + if s3prefix is not None: + args.append(f"--s3-prefix={s3prefix}") + if interactive: + args.append("--interactive") + + self.cmd = self.raw_cli(args) + return self.cmd + + def __enter__(self): + return self + + def __exit__(self, *args): + if self.workdir.exists(): + shutil.rmtree(self.workdir) + + +@pytest.fixture(scope="function") +def fast_import( + pg_version: PgVersion, + test_output_dir: Path, + neon_binpath: Path, + pg_distrib_dir: Path, +) -> Iterator[FastImport]: + workdir = Path(tempfile.mkdtemp()) + with FastImport(None, neon_binpath, pg_distrib_dir, pg_version, workdir) as fi: + yield fi + + if fi.cmd is None: + return + + # dump stdout & stderr into test log dir + with open(test_output_dir / "fast_import.stdout", "w") as f: + f.write(fi.cmd.stdout) + with open(test_output_dir / "fast_import.stderr", "w") as f: + f.write(fi.cmd.stderr) + + log.info("Written logs to %s", test_output_dir) diff --git a/test_runner/regress/test_import_pgdata.py b/test_runner/regress/test_import_pgdata.py index 6ea2393a9d..d02a9d19db 100644 --- a/test_runner/regress/test_import_pgdata.py +++ b/test_runner/regress/test_import_pgdata.py @@ -7,13 +7,15 @@ import psycopg2 import psycopg2.errors import pytest from fixtures.common_types import Lsn, TenantId, TenantShardId, TimelineId +from fixtures.fast_import import FastImport from fixtures.log_helper import log -from fixtures.neon_fixtures import NeonEnvBuilder, VanillaPostgres +from fixtures.neon_fixtures import NeonEnvBuilder, PgBin, PgProtocol, VanillaPostgres from fixtures.pageserver.http import ( ImportPgdataIdemptencyKey, PageserverApiException, ) from fixtures.pg_version import PgVersion +from fixtures.port_distributor import PortDistributor from fixtures.remote_storage import RemoteStorageKind from fixtures.utils import run_only_on_postgres from pytest_httpserver import HTTPServer @@ -313,3 +315,41 @@ def test_pgdata_import_smoke( validate_vanilla_equivalence(br_initdb_endpoint) with pytest.raises(psycopg2.errors.UndefinedTable): br_initdb_endpoint.safe_psql("select * from othertable") + + +@run_only_on_postgres( + [PgVersion.V14, PgVersion.V15, PgVersion.V16], + "newer control file catalog version and struct format isn't supported", +) +def test_fast_import_binary( + test_output_dir, + vanilla_pg: VanillaPostgres, + port_distributor: PortDistributor, + fast_import: FastImport, +): + vanilla_pg.start() + vanilla_pg.safe_psql("CREATE TABLE foo (a int); INSERT INTO foo SELECT generate_series(1, 10);") + + pg_port = port_distributor.get_port() + fast_import.run(pg_port, vanilla_pg.connstr()) + vanilla_pg.stop() + + pgbin = PgBin(test_output_dir, fast_import.pg_distrib_dir, fast_import.pg_version) + with VanillaPostgres( + fast_import.workdir / "pgdata", pgbin, pg_port, False + ) as new_pgdata_vanilla_pg: + new_pgdata_vanilla_pg.start() + + # database name and user are hardcoded in fast_import binary, and they are different from normal vanilla postgres + conn = PgProtocol(dsn=f"postgresql://cloud_admin@localhost:{pg_port}/neondb") + res = conn.safe_psql("SELECT count(*) FROM foo;") + log.info(f"Result: {res}") + assert res[0][0] == 10 + + +# TODO: Maybe test with pageserver? +# 1. run whole neon env +# 2. create timeline with some s3 path??? +# 3. run fast_import with s3 prefix +# 4. ??? mock http where pageserver will report progress +# 5. run compute on this timeline and check if data is there From 737888e5c99474b2c411418a97f2eb16d825aa07 Mon Sep 17 00:00:00 2001 From: a-masterov <72613290+a-masterov@users.noreply.github.com> Date: Tue, 21 Jan 2025 20:17:14 +0100 Subject: [PATCH 06/14] Remove the tests for `pg_anon` (#10382) ## Problem We are removing the `pg_anon` v1 extension from Neon. So we don't need to test it anymore and can remove the code for simplicity. ## Summary of changes The code required for testing `pg_anon` is removed. --- compute/compute-node.Dockerfile | 6 ----- docker-compose/compute_wrapper/Dockerfile | 3 --- docker-compose/docker_compose_test.sh | 30 +---------------------- 3 files changed, 1 insertion(+), 38 deletions(-) diff --git a/compute/compute-node.Dockerfile b/compute/compute-node.Dockerfile index 8c7200c5cb..dbe7de046b 100644 --- a/compute/compute-node.Dockerfile +++ b/compute/compute-node.Dockerfile @@ -1347,9 +1347,6 @@ COPY --from=pg-roaringbitmap-pg-build /pg_roaringbitmap.tar.gz /ext-src COPY --from=pg-semver-pg-build /pg_semver.tar.gz /ext-src #COPY --from=pg-embedding-pg-build /home/nonroot/pg_embedding-src/ /ext-src #COPY --from=wal2json-pg-build /wal2json_2_5.tar.gz /ext-src -#pg_anon is not supported yet for pg v17 so, don't fail if nothing found -COPY --from=pg-anon-pg-build /pg_anon.tar.g? /ext-src -COPY compute/patches/pg_anon.patch /ext-src COPY --from=pg-ivm-build /pg_ivm.tar.gz /ext-src COPY --from=pg-partman-build /pg_partman.tar.gz /ext-src RUN cd /ext-src/ && for f in *.tar.gz; \ @@ -1360,9 +1357,6 @@ RUN cd /ext-src/rum-src && patch -p1 <../rum.patch RUN cd /ext-src/pgvector-src && patch -p1 <../pgvector.patch RUN cd /ext-src/pg_hint_plan-src && patch -p1 < /ext-src/pg_hint_plan_${PG_VERSION}.patch COPY --chmod=755 docker-compose/run-tests.sh /run-tests.sh -RUN case "${PG_VERSION}" in "v17") \ - echo "postgresql_anonymizer does not yet support PG17" && exit 0;; \ - esac && patch -p1 "${SPEC_PATH}/spec.json" - fi PG_VERSION=$pg_version PG_TEST_VERSION=$PG_TEST_VERSION docker compose --profile test-extensions -f $COMPOSE_FILE up --build -d echo "wait until the compute is ready. timeout after 60s. " @@ -62,27 +53,12 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do done if [ $pg_version -ge 16 ]; then - echo Enabling trust connection - docker exec $COMPUTE_CONTAINER_NAME bash -c "sed -i '\$d' /var/db/postgres/compute/pg_hba.conf && echo -e 'host\t all\t all\t all\t trust' >> /var/db/postgres/compute/pg_hba.conf && psql $PSQL_OPTION -c 'select pg_reload_conf()' " - echo Adding postgres role - docker exec $COMPUTE_CONTAINER_NAME psql $PSQL_OPTION -c "CREATE ROLE postgres SUPERUSER LOGIN" # This is required for the pg_hint_plan test, to prevent flaky log message causing the test to fail # It cannot be moved to Dockerfile now because the database directory is created after the start of the container echo Adding dummy config docker exec $COMPUTE_CONTAINER_NAME touch /var/db/postgres/compute/compute_ctl_temp_override.conf - # This block is required for the pg_anon extension test. - # The test assumes that it is running on the same host with the postgres engine. - # In our case it's not true, that's why we are copying files to the compute node + # The following block copies the files for the pg_hintplan test to the compute node for the extension test in an isolated docker-compose environment TMPDIR=$(mktemp -d) - # Add support for pg_anon for pg_v16 - if [ $pg_version -ne 17 ]; then - docker cp $TEST_CONTAINER_NAME:/ext-src/pg_anon-src/data $TMPDIR/data - echo -e '1\t too \t many \t tabs' > $TMPDIR/data/bad.csv - docker cp $TMPDIR/data $COMPUTE_CONTAINER_NAME:/tmp/tmp_anon_alternate_data - rm -rf $TMPDIR - fi - TMPDIR=$(mktemp -d) - # The following block does the same for the pg_hintplan test docker cp $TEST_CONTAINER_NAME:/ext-src/pg_hint_plan-src/data $TMPDIR/data docker cp $TMPDIR/data $COMPUTE_CONTAINER_NAME:/ext-src/pg_hint_plan-src/ rm -rf $TMPDIR @@ -106,8 +82,4 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do fi fi cleanup - # Restore the original spec.json - if [ "$pg_version" -ne 17 ]; then - mv "$SPEC_PATH/spec.bak" "$SPEC_PATH/spec.json" - fi done From 7d4bfcdc4795654100b87a02d398e41373027e44 Mon Sep 17 00:00:00 2001 From: "Alex Chi Z." <4198311+skyzh@users.noreply.github.com> Date: Tue, 21 Jan 2025 14:29:38 -0500 Subject: [PATCH 07/14] feat(pageserver): add config items for gc-compaction auto trigger (#10455) ## Problem part of https://github.com/neondatabase/neon/issues/9114 The automatic trigger is already implemented at https://github.com/neondatabase/neon/pull/10221 but I need to write some tests and finish my experiments in staging before I can merge it with confidence. Given that I have some other patches that will modify the config items, I'd like to get the config items merged first to reduce conflicts. ## Summary of changes * add `l2_lsn` to index_part.json -- below that LSN, data have been processed by gc-compaction * add a set of gc-compaction auto trigger control items into the config --------- Signed-off-by: Alex Chi Z --- control_plane/src/pageserver.rs | 15 +++ libs/pageserver_api/src/config.rs | 16 +++ libs/pageserver_api/src/models.rs | 24 ++++ pageserver/src/tenant.rs | 5 + pageserver/src/tenant/config.rs | 36 ++++++ .../tenant/remote_timeline_client/index.rs | 104 +++++++++++++++++- .../regress/test_attach_tenant_config.py | 3 + 7 files changed, 201 insertions(+), 2 deletions(-) diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs index df81b44f2d..b33b2877b3 100644 --- a/control_plane/src/pageserver.rs +++ b/control_plane/src/pageserver.rs @@ -423,6 +423,21 @@ impl PageServerNode { .map(|x| x.parse::()) .transpose() .context("Failed to parse 'rel_size_v2_enabled' as bool")?, + gc_compaction_enabled: settings + .remove("gc_compaction_enabled") + .map(|x| x.parse::()) + .transpose() + .context("Failed to parse 'gc_compaction_enabled' as bool")?, + gc_compaction_initial_threshold_kb: settings + .remove("gc_compaction_initial_threshold_kb") + .map(|x| x.parse::()) + .transpose() + .context("Failed to parse 'gc_compaction_initial_threshold_kb' as integer")?, + gc_compaction_ratio_percent: settings + .remove("gc_compaction_ratio_percent") + .map(|x| x.parse::()) + .transpose() + .context("Failed to parse 'gc_compaction_ratio_percent' as integer")?, }; if !settings.is_empty() { bail!("Unrecognized tenant settings: {settings:?}") diff --git a/libs/pageserver_api/src/config.rs b/libs/pageserver_api/src/config.rs index 7fb7a9d54e..f0aeb00736 100644 --- a/libs/pageserver_api/src/config.rs +++ b/libs/pageserver_api/src/config.rs @@ -305,6 +305,16 @@ pub struct TenantConfigToml { /// Enable rel_size_v2 for this tenant. Once enabled, the tenant will persist this information into /// `index_part.json`, and it cannot be reversed. pub rel_size_v2_enabled: Option, + + // gc-compaction related configs + /// Enable automatic gc-compaction trigger on this tenant. + pub gc_compaction_enabled: bool, + /// The initial threshold for gc-compaction in KB. Once the total size of layers below the gc-horizon is above this threshold, + /// gc-compaction will be triggered. + pub gc_compaction_initial_threshold_kb: u64, + /// The ratio that triggers the auto gc-compaction. If (the total size of layers between L2 LSN and gc-horizon) / (size below the L2 LSN) + /// is above this ratio, gc-compaction will be triggered. + pub gc_compaction_ratio_percent: u64, } pub mod defaults { @@ -498,6 +508,9 @@ pub mod tenant_conf_defaults { // By default ingest enough WAL for two new L0 layers before checking if new image // image layers should be created. pub const DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD: u8 = 2; + pub const DEFAULT_GC_COMPACTION_ENABLED: bool = false; + pub const DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB: u64 = 10240000; + pub const DEFAULT_GC_COMPACTION_RATIO_PERCENT: u64 = 100; } impl Default for TenantConfigToml { @@ -543,6 +556,9 @@ impl Default for TenantConfigToml { timeline_offloading: false, wal_receiver_protocol_override: None, rel_size_v2_enabled: None, + gc_compaction_enabled: DEFAULT_GC_COMPACTION_ENABLED, + gc_compaction_initial_threshold_kb: DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB, + gc_compaction_ratio_percent: DEFAULT_GC_COMPACTION_RATIO_PERCENT, } } } diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs index 1538134c96..fd4879087f 100644 --- a/libs/pageserver_api/src/models.rs +++ b/libs/pageserver_api/src/models.rs @@ -499,6 +499,12 @@ pub struct TenantConfigPatch { pub wal_receiver_protocol_override: FieldPatch, #[serde(skip_serializing_if = "FieldPatch::is_noop")] pub rel_size_v2_enabled: FieldPatch, + #[serde(skip_serializing_if = "FieldPatch::is_noop")] + pub gc_compaction_enabled: FieldPatch, + #[serde(skip_serializing_if = "FieldPatch::is_noop")] + pub gc_compaction_initial_threshold_kb: FieldPatch, + #[serde(skip_serializing_if = "FieldPatch::is_noop")] + pub gc_compaction_ratio_percent: FieldPatch, } /// An alternative representation of `pageserver::tenant::TenantConf` with @@ -531,6 +537,9 @@ pub struct TenantConfig { pub timeline_offloading: Option, pub wal_receiver_protocol_override: Option, pub rel_size_v2_enabled: Option, + pub gc_compaction_enabled: Option, + pub gc_compaction_initial_threshold_kb: Option, + pub gc_compaction_ratio_percent: Option, } impl TenantConfig { @@ -561,6 +570,9 @@ impl TenantConfig { mut timeline_offloading, mut wal_receiver_protocol_override, mut rel_size_v2_enabled, + mut gc_compaction_enabled, + mut gc_compaction_initial_threshold_kb, + mut gc_compaction_ratio_percent, } = self; patch.checkpoint_distance.apply(&mut checkpoint_distance); @@ -606,6 +618,15 @@ impl TenantConfig { .wal_receiver_protocol_override .apply(&mut wal_receiver_protocol_override); patch.rel_size_v2_enabled.apply(&mut rel_size_v2_enabled); + patch + .gc_compaction_enabled + .apply(&mut gc_compaction_enabled); + patch + .gc_compaction_initial_threshold_kb + .apply(&mut gc_compaction_initial_threshold_kb); + patch + .gc_compaction_ratio_percent + .apply(&mut gc_compaction_ratio_percent); Self { checkpoint_distance, @@ -633,6 +654,9 @@ impl TenantConfig { timeline_offloading, wal_receiver_protocol_override, rel_size_v2_enabled, + gc_compaction_enabled, + gc_compaction_initial_threshold_kb, + gc_compaction_ratio_percent, } } } diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index 05a311391c..e45ba2ca3b 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -5476,6 +5476,11 @@ pub(crate) mod harness { timeline_offloading: Some(tenant_conf.timeline_offloading), wal_receiver_protocol_override: tenant_conf.wal_receiver_protocol_override, rel_size_v2_enabled: tenant_conf.rel_size_v2_enabled, + gc_compaction_enabled: Some(tenant_conf.gc_compaction_enabled), + gc_compaction_initial_threshold_kb: Some( + tenant_conf.gc_compaction_initial_threshold_kb, + ), + gc_compaction_ratio_percent: Some(tenant_conf.gc_compaction_ratio_percent), } } } diff --git a/pageserver/src/tenant/config.rs b/pageserver/src/tenant/config.rs index 14d8e9ccd4..3db1445f6e 100644 --- a/pageserver/src/tenant/config.rs +++ b/pageserver/src/tenant/config.rs @@ -360,6 +360,15 @@ pub struct TenantConfOpt { #[serde(skip_serializing_if = "Option::is_none")] pub rel_size_v2_enabled: Option, + + #[serde(skip_serializing_if = "Option::is_none")] + pub gc_compaction_enabled: Option, + + #[serde(skip_serializing_if = "Option::is_none")] + pub gc_compaction_initial_threshold_kb: Option, + + #[serde(skip_serializing_if = "Option::is_none")] + pub gc_compaction_ratio_percent: Option, } impl TenantConfOpt { @@ -429,6 +438,15 @@ impl TenantConfOpt { .wal_receiver_protocol_override .or(global_conf.wal_receiver_protocol_override), rel_size_v2_enabled: self.rel_size_v2_enabled.or(global_conf.rel_size_v2_enabled), + gc_compaction_enabled: self + .gc_compaction_enabled + .unwrap_or(global_conf.gc_compaction_enabled), + gc_compaction_initial_threshold_kb: self + .gc_compaction_initial_threshold_kb + .unwrap_or(global_conf.gc_compaction_initial_threshold_kb), + gc_compaction_ratio_percent: self + .gc_compaction_ratio_percent + .unwrap_or(global_conf.gc_compaction_ratio_percent), } } @@ -459,6 +477,9 @@ impl TenantConfOpt { mut timeline_offloading, mut wal_receiver_protocol_override, mut rel_size_v2_enabled, + mut gc_compaction_enabled, + mut gc_compaction_initial_threshold_kb, + mut gc_compaction_ratio_percent, } = self; patch.checkpoint_distance.apply(&mut checkpoint_distance); @@ -528,6 +549,15 @@ impl TenantConfOpt { .wal_receiver_protocol_override .apply(&mut wal_receiver_protocol_override); patch.rel_size_v2_enabled.apply(&mut rel_size_v2_enabled); + patch + .gc_compaction_enabled + .apply(&mut gc_compaction_enabled); + patch + .gc_compaction_initial_threshold_kb + .apply(&mut gc_compaction_initial_threshold_kb); + patch + .gc_compaction_ratio_percent + .apply(&mut gc_compaction_ratio_percent); Ok(Self { checkpoint_distance, @@ -555,6 +585,9 @@ impl TenantConfOpt { timeline_offloading, wal_receiver_protocol_override, rel_size_v2_enabled, + gc_compaction_enabled, + gc_compaction_initial_threshold_kb, + gc_compaction_ratio_percent, }) } } @@ -611,6 +644,9 @@ impl From for models::TenantConfig { timeline_offloading: value.timeline_offloading, wal_receiver_protocol_override: value.wal_receiver_protocol_override, rel_size_v2_enabled: value.rel_size_v2_enabled, + gc_compaction_enabled: value.gc_compaction_enabled, + gc_compaction_initial_threshold_kb: value.gc_compaction_initial_threshold_kb, + gc_compaction_ratio_percent: value.gc_compaction_ratio_percent, } } } diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs index 30b6b07ca3..3824bc8f11 100644 --- a/pageserver/src/tenant/remote_timeline_client/index.rs +++ b/pageserver/src/tenant/remote_timeline_client/index.rs @@ -77,11 +77,17 @@ pub struct IndexPart { /// /// None means no aux files have been written to the storage before the point /// when this flag is introduced. + /// + /// This flag is not used any more as all tenants have been transitioned to the new aux file policy. #[serde(skip_serializing_if = "Option::is_none", default)] pub(crate) last_aux_file_policy: Option, #[serde(skip_serializing_if = "Option::is_none", default)] pub(crate) rel_size_migration: Option, + + /// The LSN of gc-compaction horizon. Once gc-compaction is finished for all layer files below an LSN, this LSN will be updated. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub(crate) l2_lsn: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -116,10 +122,11 @@ impl IndexPart { /// - 9: +gc_blocking /// - 10: +import_pgdata /// - 11: +rel_size_migration - const LATEST_VERSION: usize = 11; + /// - 12: +l2_lsn + const LATEST_VERSION: usize = 12; // Versions we may see when reading from a bucket. - pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]; + pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]; pub const FILE_NAME: &'static str = "index_part.json"; @@ -136,6 +143,7 @@ impl IndexPart { last_aux_file_policy: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, } } @@ -437,6 +445,7 @@ mod tests { last_aux_file_policy: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -483,6 +492,7 @@ mod tests { last_aux_file_policy: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -530,6 +540,7 @@ mod tests { last_aux_file_policy: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -580,6 +591,7 @@ mod tests { last_aux_file_policy: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let empty_layers_parsed = IndexPart::from_json_bytes(empty_layers_json.as_bytes()).unwrap(); @@ -625,6 +637,7 @@ mod tests { last_aux_file_policy: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -673,6 +686,7 @@ mod tests { last_aux_file_policy: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -726,6 +740,7 @@ mod tests { last_aux_file_policy: Some(AuxFilePolicy::V2), import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -784,6 +799,7 @@ mod tests { last_aux_file_policy: Default::default(), import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -843,6 +859,7 @@ mod tests { last_aux_file_policy: Default::default(), import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -907,6 +924,7 @@ mod tests { archived_at: None, import_pgdata: None, rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -984,6 +1002,7 @@ mod tests { idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()), }))), rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -1062,6 +1081,87 @@ mod tests { idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()), }))), rel_size_migration: Some(RelSizeMigration::Legacy), + l2_lsn: None, + }; + + let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); + assert_eq!(part, expected); + } + + #[test] + fn v12_l2_lsn_is_parsed() { + let example = r#"{ + "version": 12, + "layer_metadata":{ + "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9": { "file_size": 25600000 }, + "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51": { "file_size": 9007199254741001 } + }, + "disk_consistent_lsn":"0/16960E8", + "metadata": { + "disk_consistent_lsn": "0/16960E8", + "prev_record_lsn": "0/1696070", + "ancestor_timeline": "e45a7f37d3ee2ff17dc14bf4f4e3f52e", + "ancestor_lsn": "0/0", + "latest_gc_cutoff_lsn": "0/1696070", + "initdb_lsn": "0/1696070", + "pg_version": 14 + }, + "gc_blocking": { + "started_at": "2024-07-19T09:00:00.123", + "reasons": ["DetachAncestor"] + }, + "import_pgdata": { + "V1": { + "Done": { + "idempotency_key": "specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5", + "started_at": "2024-11-13T09:23:42.123", + "finished_at": "2024-11-13T09:42:23.123" + } + } + }, + "rel_size_migration": "legacy", + "l2_lsn": "0/16960E8" + }"#; + + let expected = IndexPart { + version: 12, + layer_metadata: HashMap::from([ + ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), LayerFileMetadata { + file_size: 25600000, + generation: Generation::none(), + shard: ShardIndex::unsharded() + }), + ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap(), LayerFileMetadata { + file_size: 9007199254741001, + generation: Generation::none(), + shard: ShardIndex::unsharded() + }) + ]), + disk_consistent_lsn: "0/16960E8".parse::().unwrap(), + metadata: TimelineMetadata::new( + Lsn::from_str("0/16960E8").unwrap(), + Some(Lsn::from_str("0/1696070").unwrap()), + Some(TimelineId::from_str("e45a7f37d3ee2ff17dc14bf4f4e3f52e").unwrap()), + Lsn::INVALID, + Lsn::from_str("0/1696070").unwrap(), + Lsn::from_str("0/1696070").unwrap(), + 14, + ).with_recalculated_checksum().unwrap(), + deleted_at: None, + lineage: Default::default(), + gc_blocking: Some(GcBlocking { + started_at: parse_naive_datetime("2024-07-19T09:00:00.123000000"), + reasons: enumset::EnumSet::from_iter([GcBlockingReason::DetachAncestor]), + }), + last_aux_file_policy: Default::default(), + archived_at: None, + import_pgdata: Some(import_pgdata::index_part_format::Root::V1(import_pgdata::index_part_format::V1::Done(import_pgdata::index_part_format::Done{ + started_at: parse_naive_datetime("2024-11-13T09:23:42.123000000"), + finished_at: parse_naive_datetime("2024-11-13T09:42:23.123000000"), + idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()), + }))), + rel_size_migration: Some(RelSizeMigration::Legacy), + l2_lsn: Some("0/16960E8".parse::().unwrap()), }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); diff --git a/test_runner/regress/test_attach_tenant_config.py b/test_runner/regress/test_attach_tenant_config.py index b34dbddc80..b8d47346a3 100644 --- a/test_runner/regress/test_attach_tenant_config.py +++ b/test_runner/regress/test_attach_tenant_config.py @@ -177,6 +177,9 @@ def test_fully_custom_config(positive_env: NeonEnv): "args": {"format": "bincode", "compression": {"zstd": {"level": 1}}}, }, "rel_size_v2_enabled": True, + "gc_compaction_enabled": True, + "gc_compaction_initial_threshold_kb": 1024000, + "gc_compaction_ratio_percent": 200, } vps_http = env.storage_controller.pageserver_api() From a75e11cc002aae79efd3d8b46fe9c7be96eca326 Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Tue, 21 Jan 2025 21:56:34 +0100 Subject: [PATCH 08/14] pageserver: return duration from `StorageTimeMetricsTimer` (#10468) ## Problem It's sometimes useful to obtain the elapsed duration from a `StorageTimeMetricsTimer` for purposes beyond just recording it in metrics (e.g. to log it). Extracted from #10405. ## Summary of changes Add `StorageTimeMetricsTimer.elapsed()` and return the duration from `stop_and_record()`. --- pageserver/src/metrics.rs | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/pageserver/src/metrics.rs b/pageserver/src/metrics.rs index 4758aaf230..252e566f70 100644 --- a/pageserver/src/metrics.rs +++ b/pageserver/src/metrics.rs @@ -2550,12 +2550,19 @@ impl StorageTimeMetricsTimer { } } - /// Record the time from creation to now. - pub fn stop_and_record(self) { - let duration = self.start.elapsed().as_secs_f64(); - self.metrics.timeline_sum.inc_by(duration); + /// Returns the elapsed duration of the timer. + pub fn elapsed(&self) -> Duration { + self.start.elapsed() + } + + /// Record the time from creation to now and return it. + pub fn stop_and_record(self) -> Duration { + let duration = self.elapsed(); + let seconds = duration.as_secs_f64(); + self.metrics.timeline_sum.inc_by(seconds); self.metrics.timeline_count.inc(); - self.metrics.global_histogram.observe(duration); + self.metrics.global_histogram.observe(seconds); + duration } /// Turns this timer into a timer, which will always record -- usually this means recording @@ -2575,6 +2582,14 @@ impl Drop for AlwaysRecordingStorageTimeMetricsTimer { } } +impl AlwaysRecordingStorageTimeMetricsTimer { + /// Returns the elapsed duration of the timer. + #[allow(unused)] + pub fn elapsed(&self) -> Duration { + self.0.as_ref().expect("not dropped yet").elapsed() + } +} + /// Timing facilities for an globally histogrammed metric, which is supported by per tenant and /// timeline total sum and count. #[derive(Clone, Debug)] From 8a8c656c0646895fbfc7a7bc5df4a93917071032 Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Tue, 21 Jan 2025 22:18:09 +0100 Subject: [PATCH 09/14] pageserver: add `LayerMap::watch_layer0_deltas()` (#10470) ## Problem For compaction backpressure, we need a mechanism to signal when compaction has reduced the L0 delta layer count below the backpressure threshold. Extracted from #10405. ## Summary of changes Add `LayerMap::watch_level0_deltas()` which returns a `tokio::sync::watch::Receiver` signalling the current L0 delta layer count. --- pageserver/src/tenant/layer_map.rs | 29 ++++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/pageserver/src/tenant/layer_map.rs b/pageserver/src/tenant/layer_map.rs index 1b6924425c..a69cce932e 100644 --- a/pageserver/src/tenant/layer_map.rs +++ b/pageserver/src/tenant/layer_map.rs @@ -57,6 +57,7 @@ use std::collections::{HashMap, VecDeque}; use std::iter::Peekable; use std::ops::Range; use std::sync::Arc; +use tokio::sync::watch; use utils::lsn::Lsn; use historic_layer_coverage::BufferedHistoricLayerCoverage; @@ -67,7 +68,6 @@ use super::storage_layer::{LayerVisibilityHint, PersistentLayerDesc}; /// /// LayerMap tracks what layers exist on a timeline. /// -#[derive(Default)] pub struct LayerMap { // // 'open_layer' holds the current InMemoryLayer that is accepting new @@ -93,7 +93,25 @@ pub struct LayerMap { /// L0 layers have key range Key::MIN..Key::MAX, and locating them using R-Tree search is very inefficient. /// So L0 layers are held in l0_delta_layers vector, in addition to the R-tree. + /// + /// NB: make sure to notify `watch_l0_deltas` on changes. l0_delta_layers: Vec>, + + /// Notifies about L0 delta layer changes, sending the current number of L0 layers. + watch_l0_deltas: watch::Sender, +} + +impl Default for LayerMap { + fn default() -> Self { + Self { + open_layer: Default::default(), + next_open_layer_at: Default::default(), + frozen_layers: Default::default(), + historic: Default::default(), + l0_delta_layers: Default::default(), + watch_l0_deltas: watch::channel(0).0, + } + } } /// The primary update API for the layer map. @@ -466,6 +484,8 @@ impl LayerMap { if Self::is_l0(&layer_desc.key_range, layer_desc.is_delta) { self.l0_delta_layers.push(layer_desc.clone().into()); + self.watch_l0_deltas + .send_replace(self.l0_delta_layers.len()); } self.historic.insert( @@ -488,6 +508,8 @@ impl LayerMap { let mut l0_delta_layers = std::mem::take(&mut self.l0_delta_layers); l0_delta_layers.retain(|other| other.key() != layer_key); self.l0_delta_layers = l0_delta_layers; + self.watch_l0_deltas + .send_replace(self.l0_delta_layers.len()); // this assertion is related to use of Arc::ptr_eq in Self::compare_arced_layers, // there's a chance that the comparison fails at runtime due to it comparing (pointer, // vtable) pairs. @@ -850,6 +872,11 @@ impl LayerMap { &self.l0_delta_layers } + /// Subscribes to L0 delta layer changes, sending the current number of L0 delta layers. + pub fn watch_level0_deltas(&self) -> watch::Receiver { + self.watch_l0_deltas.subscribe() + } + /// debugging function to print out the contents of the layer map #[allow(unused)] pub async fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> { From 14e1f89053b87a38f0dd697f5c32721380d19a9a Mon Sep 17 00:00:00 2001 From: Erik Grinaker Date: Tue, 21 Jan 2025 23:01:27 +0100 Subject: [PATCH 10/14] pageserver: eagerly notify flush waiters (#10469) ## Problem Currently, the layer flush loop will continue flushing layers as long as any are pending, and only notify waiters once there are no further layers to flush. This can cause waiters to wait longer than necessary, and potentially starve them if pending layers keep arriving faster than they can be flushed. The impact of this will increase when we add compaction backpressure and propagate it up into the WAL receiver. Extracted from #10405. ## Summary of changes Break out of the layer flush loop once we've flushed up to the requested LSN. If further flush requests have arrived in the meanwhile, flushing will resume immediately after. --- pageserver/src/tenant/timeline.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 5f4272fb2b..3245f23a28 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -3617,6 +3617,12 @@ impl Timeline { return; } + // Break to notify potential waiters as soon as we've flushed the requested LSN. If + // more requests have arrived in the meanwhile, we'll resume flushing afterwards. + if flushed_to_lsn >= frozen_to_lsn { + break Ok(()); + } + let timer = self.metrics.flush_time_histo.start_timer(); let num_frozen_layers; From 2b49d6ee050f41cb67d34e4f117196b3d01a2394 Mon Sep 17 00:00:00 2001 From: Conrad Ludgate Date: Wed, 22 Jan 2025 09:15:52 +0000 Subject: [PATCH 11/14] feat: adjust the tonic features to remove axum dependency (#10348) To help facilitate an upgrade to axum 0.8 (https://github.com/neondatabase/neon/pull/10332#pullrequestreview-2541989619) this massages the tonic dependency features so that tonic does not depend on axum. --- Cargo.lock | 8 ------- Cargo.toml | 2 +- libs/wal_decoder/Cargo.toml | 1 - libs/wal_decoder/src/models.rs | 2 +- storage_broker/src/bin/storage_broker.rs | 29 ++++++++++++++---------- workspace_hack/Cargo.toml | 11 ++++----- 6 files changed, 23 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 02b02a09c1..2020c417f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7004,12 +7004,9 @@ version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52" dependencies = [ - "async-stream", "async-trait", - "axum", "base64 0.22.1", "bytes", - "h2 0.4.4", "http 1.1.0", "http-body 1.0.0", "http-body-util", @@ -7021,7 +7018,6 @@ dependencies = [ "prost", "rustls-native-certs 0.8.0", "rustls-pemfile 2.1.1", - "socket2", "tokio", "tokio-rustls 0.26.0", "tokio-stream", @@ -7582,7 +7578,6 @@ dependencies = [ "tikv-jemallocator", "tokio", "tokio-util", - "tonic", "tonic-build", "tracing", "utils", @@ -7991,8 +7986,6 @@ version = "0.1.0" dependencies = [ "ahash", "anyhow", - "axum", - "axum-core", "base64 0.13.1", "base64 0.21.1", "base64ct", @@ -8073,7 +8066,6 @@ dependencies = [ "toml_edit", "tonic", "tower 0.4.13", - "tower 0.5.2", "tracing", "tracing-core", "url", diff --git a/Cargo.toml b/Cargo.toml index a4e601bb58..6e1e288895 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -187,7 +187,7 @@ tokio-tar = "0.3" tokio-util = { version = "0.7.10", features = ["io", "rt"] } toml = "0.8" toml_edit = "0.22" -tonic = {version = "0.12.3", features = ["tls", "tls-roots"]} +tonic = {version = "0.12.3", default-features = false, features = ["channel", "tls", "tls-roots"]} tower = { version = "0.5.2", default-features = false } tower-http = { version = "0.6.2", features = ["request-id", "trace"] } tower-service = "0.3.3" diff --git a/libs/wal_decoder/Cargo.toml b/libs/wal_decoder/Cargo.toml index 09c4afb18a..cb0ef4b00d 100644 --- a/libs/wal_decoder/Cargo.toml +++ b/libs/wal_decoder/Cargo.toml @@ -17,7 +17,6 @@ postgres_ffi.workspace = true serde.workspace = true thiserror.workspace = true tokio = { workspace = true, features = ["io-util"] } -tonic.workspace = true tracing.workspace = true utils.workspace = true workspace_hack = { version = "0.1", path = "../../workspace_hack" } diff --git a/libs/wal_decoder/src/models.rs b/libs/wal_decoder/src/models.rs index c2f9125b21..51bf7e44ab 100644 --- a/libs/wal_decoder/src/models.rs +++ b/libs/wal_decoder/src/models.rs @@ -45,7 +45,7 @@ pub mod proto { #![allow(clippy::derive_partial_eq_without_eq)] // The generated ValueMeta has a `len` method generate for its `len` field. #![allow(clippy::len_without_is_empty)] - tonic::include_proto!("interpreted_wal"); + include!(concat!(env!("OUT_DIR"), concat!("/interpreted_wal.rs"))); } #[derive(Copy, Clone, Serialize, Deserialize)] diff --git a/storage_broker/src/bin/storage_broker.rs b/storage_broker/src/bin/storage_broker.rs index 1fbb651656..9d4c22484c 100644 --- a/storage_broker/src/bin/storage_broker.rs +++ b/storage_broker/src/bin/storage_broker.rs @@ -32,7 +32,6 @@ use tokio::sync::broadcast::error::RecvError; use tokio::time; use tonic::body::{self, empty_body, BoxBody}; use tonic::codegen::Service; -use tonic::transport::server::Connected; use tonic::Code; use tonic::{Request, Response, Status}; use tracing::*; @@ -459,9 +458,10 @@ impl BrokerService for Broker { &self, request: Request>, ) -> Result, Status> { - let remote_addr = request - .remote_addr() - .expect("TCPConnectInfo inserted by handler"); + let &RemoteAddr(remote_addr) = request + .extensions() + .get() + .expect("RemoteAddr inserted by handler"); let mut publisher = self.registry.register_publisher(remote_addr); let mut stream = request.into_inner(); @@ -484,9 +484,10 @@ impl BrokerService for Broker { &self, request: Request, ) -> Result, Status> { - let remote_addr = request - .remote_addr() - .expect("TCPConnectInfo inserted by handler"); + let &RemoteAddr(remote_addr) = request + .extensions() + .get() + .expect("RemoteAddr inserted by handler"); let proto_key = request .into_inner() .subscription_key @@ -537,9 +538,10 @@ impl BrokerService for Broker { &self, request: Request, ) -> std::result::Result, Status> { - let remote_addr = request - .remote_addr() - .expect("TCPConnectInfo inserted by handler"); + let &RemoteAddr(remote_addr) = request + .extensions() + .get() + .expect("RemoteAddr inserted by handler"); let proto_filter = request.into_inner(); let ttid_filter = proto_filter.tenant_timeline_id.as_ref(); @@ -628,6 +630,9 @@ async fn http1_handler( Ok(resp) } +#[derive(Clone, Copy)] +struct RemoteAddr(SocketAddr); + #[tokio::main] async fn main() -> Result<(), Box> { let args = Args::parse(); @@ -687,13 +692,13 @@ async fn main() -> Result<(), Box> { .max_concurrent_streams(None); let storage_broker_server_cloned = storage_broker_server.clone(); - let connect_info = stream.connect_info(); + let remote_addr = RemoteAddr(addr); let service_fn_ = async move { service_fn(move |mut req| { // That's what tonic's MakeSvc.call does to pass conninfo to // the request handler (and where its request.remote_addr() // expects it to find). - req.extensions_mut().insert(connect_info.clone()); + req.extensions_mut().insert(remote_addr); // Technically this second clone is not needed, but consume // by async block is apparently unavoidable. BTW, error diff --git a/workspace_hack/Cargo.toml b/workspace_hack/Cargo.toml index 0ffeeead18..a3dffa8f19 100644 --- a/workspace_hack/Cargo.toml +++ b/workspace_hack/Cargo.toml @@ -17,8 +17,6 @@ license.workspace = true [dependencies] ahash = { version = "0.8" } anyhow = { version = "1", features = ["backtrace"] } -axum = { version = "0.7", features = ["ws"] } -axum-core = { version = "0.4", default-features = false, features = ["tracing"] } base64-594e8ee84c453af0 = { package = "base64", version = "0.13", features = ["alloc"] } base64-647d43efb71741da = { package = "base64", version = "0.21", features = ["alloc"] } base64ct = { version = "1", default-features = false, features = ["std"] } @@ -46,7 +44,7 @@ hex = { version = "0.4", features = ["serde"] } hmac = { version = "0.12", default-features = false, features = ["reset"] } hyper-582f2526e08bb6a0 = { package = "hyper", version = "0.14", features = ["full"] } hyper-dff4ba8e3ae991db = { package = "hyper", version = "1", features = ["full"] } -hyper-util = { version = "0.1", features = ["client-legacy", "server-auto", "service"] } +hyper-util = { version = "0.1", features = ["client-legacy", "http1", "http2", "server", "service"] } indexmap-dff4ba8e3ae991db = { package = "indexmap", version = "1", default-features = false, features = ["std"] } indexmap-f595c2ba2a3f28df = { package = "indexmap", version = "2", features = ["serde"] } itertools = { version = "0.12" } @@ -87,12 +85,11 @@ tikv-jemalloc-sys = { version = "0.6", features = ["profiling", "stats", "unpref time = { version = "0.3", features = ["macros", "serde-well-known"] } tokio = { version = "1", features = ["full", "test-util"] } tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "ring", "tls12"] } -tokio-stream = { version = "0.1", features = ["net"] } +tokio-stream = { version = "0.1" } tokio-util = { version = "0.7", features = ["codec", "compat", "io", "rt"] } toml_edit = { version = "0.22", features = ["serde"] } -tonic = { version = "0.12", features = ["tls-roots"] } -tower-9fbad63c4bcf4a8f = { package = "tower", version = "0.4", default-features = false, features = ["balance", "buffer", "limit", "util"] } -tower-d8f496e17d97b5cb = { package = "tower", version = "0.5", default-features = false, features = ["log", "make", "util"] } +tonic = { version = "0.12", default-features = false, features = ["codegen", "prost", "tls-roots"] } +tower = { version = "0.4", default-features = false, features = ["balance", "buffer", "limit", "util"] } tracing = { version = "0.1", features = ["log"] } tracing-core = { version = "0.1" } url = { version = "2", features = ["serde"] } From b4d87b9dfedc3d08d00091e2407d0996a9ea2026 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Wed, 22 Jan 2025 11:10:43 +0100 Subject: [PATCH 12/14] fix(tests): actually enable pipelinig by default in the test suite (#10472) ## Problem PR #9993 was supposed to enable `page_service_pipelining` by default for all `NeonEnv`s, but this was ineffective in our CI environment. Thus, CI Python-based tests and benchmarks, unless explicitly configuring pipelining, were still using serial protocol handling. ## Analysis The root cause was that in our CI environment, `config.compatibility_neon_binpath` is always Truthy. It's not in local environments, which is why this slipped through in local testing. Lesson: always add a log line ot pageserver startup and spot-check tests to ensure the intended default is picked up. ## Summary of changes Fix it. Since enough time has passed, the compatiblity snapshot contains a recent enough software version so we don't need to worry about `compatibility_neon_binpath` anymore. ## Future Work The question how to add a new default except for compatibliity tests, which is what the broken code was supposed to do, is still unsolved. Slack discussion: https://neondb.slack.com/archives/C059ZC138NR/p1737490501941309 --- test_runner/fixtures/neon_fixtures.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index a01cb47984..d79c2a5ea8 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -1115,13 +1115,11 @@ class NeonEnv: # Batching (https://github.com/neondatabase/neon/issues/9377): # enable batching by default in tests and benchmarks. - # Compat tests are exempt because old versions fail to parse the new config. - if not config.compatibility_neon_binpath: - ps_cfg["page_service_pipelining"] = { - "mode": "pipelined", - "execution": "concurrent-futures", - "max_batch_size": 32, - } + ps_cfg["page_service_pipelining"] = { + "mode": "pipelined", + "execution": "concurrent-futures", + "max_batch_size": 32, + } if self.pageserver_virtual_file_io_engine is not None: ps_cfg["virtual_file_io_engine"] = self.pageserver_virtual_file_io_engine From b31ce14083d3d4130e1190daadbf085082ac7280 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Wed, 22 Jan 2025 13:28:26 +0100 Subject: [PATCH 13/14] initial logical size calculation: always poll to completion (#10471) # Refs - extracted from https://github.com/neondatabase/neon/pull/9353 # Problem Before this PR, when task_mgr shutdown is signalled, e.g. during pageserver shutdown or Tenant shutdown, initial logical size calculation stops polling and drops the future that represents the calculation. This is against the current policy that we poll all futures to completion. This became apparent during development of concurrent IO which warns if we drop a `Timeline::get_vectored` future that still has in-flight IOs. We may revise the policy in the future, but, right now initial logical size calculation is the only part of the codebase that doesn't adhere to the policy, so let's fix it. ## Code Changes - make sensitive exclusively to `Timeline::cancel` - This should be sufficient for all cases of shutdowns; the sensitivity to task_mgr shutdown is unnecessary. - this broke the various cancel tests in `test_timeline_size.py`, e.g., `test_timeline_initial_logical_size_calculation_cancellation` - the tests would time out because the await point was not sensitive to cancellation - to fix this, refactor `pausable_failpoint` so that it accepts a cancellation token - side note: we _really_ should write our own failpoint library; maybe after we get heap-allocated RequestContext, we can plumb failpoints through there. --- libs/utils/src/failpoint_support.rs | 54 ++++++++++++++++------- pageserver/src/tenant/tasks.rs | 7 ++- pageserver/src/tenant/timeline.rs | 67 +++++++++++------------------ 3 files changed, 66 insertions(+), 62 deletions(-) diff --git a/libs/utils/src/failpoint_support.rs b/libs/utils/src/failpoint_support.rs index 701ba2d42c..272c6ebb26 100644 --- a/libs/utils/src/failpoint_support.rs +++ b/libs/utils/src/failpoint_support.rs @@ -11,31 +11,55 @@ use tracing::*; /// Declare a failpoint that can use to `pause` failpoint action. /// We don't want to block the executor thread, hence, spawn_blocking + await. +/// +/// Optionally pass a cancellation token, and this failpoint will drop out of +/// its pause when the cancellation token fires. This is useful for testing +/// cases where we would like to block something, but test its clean shutdown behavior. +/// The macro evaluates to a Result in that case, where Ok(()) is the case +/// where the failpoint was not paused, and Err() is the case where cancellation +/// token fired while evaluating the failpoint. +/// +/// Remember to unpause the failpoint in the test; until that happens, one of the +/// limited number of spawn_blocking thread pool threads is leaked. #[macro_export] macro_rules! pausable_failpoint { - ($name:literal) => { + ($name:literal) => {{ if cfg!(feature = "testing") { - tokio::task::spawn_blocking({ - let current = tracing::Span::current(); + let cancel = ::tokio_util::sync::CancellationToken::new(); + let _ = $crate::pausable_failpoint!($name, &cancel); + } + }}; + ($name:literal, $cancel:expr) => {{ + if cfg!(feature = "testing") { + let failpoint_fut = ::tokio::task::spawn_blocking({ + let current = ::tracing::Span::current(); move || { let _entered = current.entered(); - tracing::info!("at failpoint {}", $name); - fail::fail_point!($name); + ::tracing::info!("at failpoint {}", $name); + ::fail::fail_point!($name); + } + }); + let cancel_fut = async move { + $cancel.cancelled().await; + }; + ::tokio::select! { + res = failpoint_fut => { + res.expect("spawn_blocking"); + // continue with execution + Ok(()) + }, + _ = cancel_fut => { + Err(()) } - }) - .await - .expect("spawn_blocking"); - } - }; - ($name:literal, $cond:expr) => { - if cfg!(feature = "testing") { - if $cond { - pausable_failpoint!($name) } + } else { + Ok(()) } - }; + }}; } +pub use pausable_failpoint; + /// use with fail::cfg("$name", "return(2000)") /// /// The effect is similar to a "sleep(2000)" action, i.e. we sleep for the diff --git a/pageserver/src/tenant/tasks.rs b/pageserver/src/tenant/tasks.rs index 0118a5ce5f..3725e2f7fc 100644 --- a/pageserver/src/tenant/tasks.rs +++ b/pageserver/src/tenant/tasks.rs @@ -67,10 +67,9 @@ pub(crate) async fn concurrent_background_tasks_rate_limit_permit( ) -> tokio::sync::SemaphorePermit<'static> { let _guard = crate::metrics::BACKGROUND_LOOP_SEMAPHORE.measure_acquisition(loop_kind); - pausable_failpoint!( - "initial-size-calculation-permit-pause", - loop_kind == BackgroundLoopKind::InitialLogicalSizeCalculation - ); + if loop_kind == BackgroundLoopKind::InitialLogicalSizeCalculation { + pausable_failpoint!("initial-size-calculation-permit-pause"); + } // TODO: assert that we run on BACKGROUND_RUNTIME; requires tokio_unstable Handle::id(); match CONCURRENT_BACKGROUND_TASKS.acquire().await { diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 3245f23a28..e83b516d79 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -60,6 +60,7 @@ use utils::{ use wal_decoder::serialized_batch::{SerializedValueBatch, ValueMeta}; use std::sync::atomic::Ordering as AtomicOrdering; +use std::sync::OnceLock; use std::sync::{Arc, Mutex, RwLock, Weak}; use std::time::{Duration, Instant, SystemTime}; use std::{ @@ -72,7 +73,6 @@ use std::{ collections::btree_map::Entry, ops::{Deref, Range}, }; -use std::{pin::pin, sync::OnceLock}; use crate::{ aux_file::AuxFileSizeEstimator, @@ -2804,12 +2804,10 @@ impl Timeline { "initial size calculation", // NB: don't log errors here, task_mgr will do that. async move { - let cancel = task_mgr::shutdown_token(); self_clone .initial_logical_size_calculation_task( initial_part_end, cancel_wait_for_background_loop_concurrency_limit_semaphore, - cancel, background_ctx, ) .await; @@ -2819,11 +2817,21 @@ impl Timeline { ); } + /// # Cancellation + /// + /// This method is sensitive to `Timeline::cancel`. + /// + /// It is _not_ sensitive to task_mgr::shutdown_token(). + /// + /// # Cancel-Safety + /// + /// It does Timeline IO, hence this should be polled to completion because + /// we could be leaving in-flight IOs behind, which is safe, but annoying + /// to reason about. async fn initial_logical_size_calculation_task( self: Arc, initial_part_end: Lsn, skip_concurrency_limiter: CancellationToken, - cancel: CancellationToken, background_ctx: RequestContext, ) { scopeguard::defer! { @@ -2836,7 +2844,6 @@ impl Timeline { let self_ref = &self; let skip_concurrency_limiter = &skip_concurrency_limiter; async move { - let cancel = task_mgr::shutdown_token(); let wait_for_permit = super::tasks::concurrent_background_tasks_rate_limit_permit( BackgroundLoopKind::InitialLogicalSizeCalculation, background_ctx, @@ -2850,9 +2857,6 @@ impl Timeline { _ = self_ref.cancel.cancelled() => { return Err(CalculateLogicalSizeError::Cancelled); } - _ = cancel.cancelled() => { - return Err(CalculateLogicalSizeError::Cancelled); - }, () = skip_concurrency_limiter.cancelled() => { // Some action that is part of a end user interaction requested logical size // => break out of the rate limit @@ -2911,22 +2915,18 @@ impl Timeline { ) .expect("10min < 1hour"), ); - tokio::time::sleep(sleep_duration).await; + tokio::select! { + _ = tokio::time::sleep(sleep_duration) => {} + _ = self.cancel.cancelled() => return ControlFlow::Break(()), + } } } } }; - let (calculated_size, metrics_guard) = tokio::select! { - res = retrying => { - match res { - ControlFlow::Continue(calculated_size) => calculated_size, - ControlFlow::Break(()) => return, - } - } - _ = cancel.cancelled() => { - return; - } + let (calculated_size, metrics_guard) = match retrying.await { + ControlFlow::Continue(calculated_size) => calculated_size, + ControlFlow::Break(()) => return, }; // we cannot query current_logical_size.current_size() to know the current @@ -2982,9 +2982,6 @@ impl Timeline { receiver } - /// # Cancel-Safety - /// - /// This method is cancellation-safe. #[instrument(skip_all)] async fn logical_size_calculation_task( self: &Arc, @@ -3002,32 +2999,13 @@ impl Timeline { .enter() .map_err(|_| CalculateLogicalSizeError::Cancelled)?; - let self_calculation = Arc::clone(self); - - let mut calculation = pin!(async { - let ctx = ctx.attached_child(); - self_calculation - .calculate_logical_size(lsn, cause, &guard, &ctx) - .await - }); - - tokio::select! { - res = &mut calculation => { res } - _ = self.cancel.cancelled() => { - debug!("cancelling logical size calculation for timeline shutdown"); - calculation.await - } - } + self.calculate_logical_size(lsn, cause, &guard, ctx).await } /// Calculate the logical size of the database at the latest LSN. /// /// NOTE: counted incrementally, includes ancestors. This can be a slow operation, /// especially if we need to download remote layers. - /// - /// # Cancel-Safety - /// - /// This method is cancellation-safe. async fn calculate_logical_size( &self, up_to_lsn: Lsn, @@ -3040,7 +3018,10 @@ impl Timeline { self.timeline_id, up_to_lsn ); - pausable_failpoint!("timeline-calculate-logical-size-pause"); + if let Err(()) = pausable_failpoint!("timeline-calculate-logical-size-pause", &self.cancel) + { + return Err(CalculateLogicalSizeError::Cancelled); + } // See if we've already done the work for initial size calculation. // This is a short-cut for timelines that are mostly unused. From 881e351f693ae1d177d57b171569d327d0abcfe8 Mon Sep 17 00:00:00 2001 From: Alexey Kondratov Date: Wed, 22 Jan 2025 13:38:23 +0100 Subject: [PATCH 14/14] feat(compute): Allow installing both 0.8.0 and 0.7.4 pgvector (#10345) ## Problem Both these versions are binary compatible, but the way pgvector structures the SQL files forbids installing 0.7.4 if you have a 0.8.0 distribution. Yet, some users may need a previous version for backward compatibility, e.g., restoring the dump. See this thread for discussion https://neondb.slack.com/archives/C04DGM6SMTM/p1735911490242919?thread_ts=1731343604.259169&cid=C04DGM6SMTM ## Summary of changes Put `vector--0.7.4.sql` file into compute image to allow installing this version as well. Tested on staging and it seems to be working as expected: ```sql select * from pg_available_extensions where name = 'vector'; name | default_version | installed_version | comment --------+-----------------+-------------------+------------------------------------------------------ vector | 0.8.0 | (null) | vector data type and ivfflat and hnsw access methods create extension vector version '0.7.4'; select * from pg_available_extensions where name = 'vector'; name | default_version | installed_version | comment --------+-----------------+-------------------+------------------------------------------------------ vector | 0.8.0 | 0.7.4 | vector data type and ivfflat and hnsw access methods alter extension vector update; select * from pg_available_extensions where name = 'vector'; name | default_version | installed_version | comment --------+-----------------+-------------------+------------------------------------------------------ vector | 0.8.0 | 0.8.0 | vector data type and ivfflat and hnsw access methods drop extension vector; create extension vector; select * from pg_available_extensions where name = 'vector'; name | default_version | installed_version | comment --------+-----------------+-------------------+------------------------------------------------------ vector | 0.8.0 | 0.8.0 | vector data type and ivfflat and hnsw access methods ``` If we find out it's a good approach, we can adopt the same for other extensions with a stable ABI -- support both `current` and `current - 1` releases. --- compute/compute-node.Dockerfile | 2 ++ compute/patches/pgvector.patch | 22 +++++++++++++++++++--- 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/compute/compute-node.Dockerfile b/compute/compute-node.Dockerfile index dbe7de046b..706c947008 100644 --- a/compute/compute-node.Dockerfile +++ b/compute/compute-node.Dockerfile @@ -360,6 +360,8 @@ COPY compute/patches/pgvector.patch /pgvector.patch RUN wget https://github.com/pgvector/pgvector/archive/refs/tags/v0.8.0.tar.gz -O pgvector.tar.gz && \ echo "867a2c328d4928a5a9d6f052cd3bc78c7d60228a9b914ad32aa3db88e9de27b0 pgvector.tar.gz" | sha256sum --check && \ mkdir pgvector-src && cd pgvector-src && tar xzf ../pgvector.tar.gz --strip-components=1 -C . && \ + wget https://github.com/pgvector/pgvector/raw/refs/tags/v0.7.4/sql/vector.sql -O ./sql/vector--0.7.4.sql && \ + echo "10218d05dc02299562252a9484775178b14a1d8edb92a2d1672ef488530f7778 ./sql/vector--0.7.4.sql" | sha256sum --check && \ patch -p1 < /pgvector.patch && \ make -j $(getconf _NPROCESSORS_ONLN) OPTFLAGS="" && \ make -j $(getconf _NPROCESSORS_ONLN) OPTFLAGS="" install && \ diff --git a/compute/patches/pgvector.patch b/compute/patches/pgvector.patch index 3e1ffcaaaf..da41c86140 100644 --- a/compute/patches/pgvector.patch +++ b/compute/patches/pgvector.patch @@ -1,8 +1,24 @@ +diff --git a/Makefile b/Makefile +index 7a4b88c..56678af 100644 +--- a/Makefile ++++ b/Makefile +@@ -3,7 +3,10 @@ EXTVERSION = 0.8.0 + + MODULE_big = vector + DATA = $(wildcard sql/*--*--*.sql) +-DATA_built = sql/$(EXTENSION)--$(EXTVERSION).sql ++# This change is needed to install different per-version SQL files ++# like pgvector--0.8.0.sql and pgvector--0.7.4.sql ++# The corresponding file is downloaded during the Docker image build process ++DATA_built = sql/$(EXTENSION)--$(EXTVERSION).sql sql/vector--0.7.4.sql + OBJS = src/bitutils.o src/bitvec.o src/halfutils.o src/halfvec.o src/hnsw.o src/hnswbuild.o src/hnswinsert.o src/hnswscan.o src/hnswutils.o src/hnswvacuum.o src/ivfbuild.o src/ivfflat.o src/ivfinsert.o src/ivfkmeans.o src/ivfscan.o src/ivfutils.o src/ivfvacuum.o src/sparsevec.o src/vector.o + HEADERS = src/halfvec.h src/sparsevec.h src/vector.h + diff --git a/src/hnswbuild.c b/src/hnswbuild.c -index dcfb2bd..d5189ee 100644 +index b667478..fc1897c 100644 --- a/src/hnswbuild.c +++ b/src/hnswbuild.c -@@ -860,9 +860,17 @@ HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc) +@@ -843,9 +843,17 @@ HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc) hnswarea = shm_toc_lookup(toc, PARALLEL_KEY_HNSW_AREA, false); @@ -20,7 +36,7 @@ index dcfb2bd..d5189ee 100644 /* Close relations within worker */ index_close(indexRel, indexLockmode); table_close(heapRel, heapLockmode); -@@ -1117,12 +1125,38 @@ BuildIndex(Relation heap, Relation index, IndexInfo *indexInfo, +@@ -1100,12 +1108,38 @@ BuildIndex(Relation heap, Relation index, IndexInfo *indexInfo, SeedRandom(42); #endif