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.`); } diff --git a/Cargo.lock b/Cargo.lock index c823b76ed8..1f090a27e4 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/compute/compute-node.Dockerfile b/compute/compute-node.Dockerfile index 8c7200c5cb..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 && \ @@ -1347,9 +1349,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 +1359,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 , // 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/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs index ef5b3d6593..b33b2877b3 100644 --- a/control_plane/src/pageserver.rs +++ b/control_plane/src/pageserver.rs @@ -418,6 +418,26 @@ 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")?, + 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/docker-compose/compute_wrapper/Dockerfile b/docker-compose/compute_wrapper/Dockerfile index 05a2cf124c..e2e5bc7248 100644 --- a/docker-compose/compute_wrapper/Dockerfile +++ b/docker-compose/compute_wrapper/Dockerfile @@ -10,10 +10,7 @@ USER root RUN apt-get update && \ apt-get install -y curl \ jq \ - python3-pip \ netcat-openbsd -#Faker is required for the pg_anon test -RUN case $COMPUTE_IMAGE in compute-node-v17) OPT="--break-system-packages";; *) OPT= ;; esac && pip3 install $OPT Faker #This is required for the pg_hintplan test RUN mkdir -p /ext-src/pg_hint_plan-src && chown postgres /ext-src/pg_hint_plan-src diff --git a/docker-compose/docker_compose_test.sh b/docker-compose/docker_compose_test.sh index 063664d0c6..4f1ae64873 100755 --- a/docker-compose/docker_compose_test.sh +++ b/docker-compose/docker_compose_test.sh @@ -18,9 +18,6 @@ cd $(dirname $0) COMPUTE_CONTAINER_NAME=docker-compose-compute-1 TEST_CONTAINER_NAME=docker-compose-neon-test-extensions-1 PSQL_OPTION="-h localhost -U cloud_admin -p 55433 -d postgres" -: ${http_proxy:=} -: ${https_proxy:=} -export http_proxy https_proxy cleanup() { echo "show container information" @@ -35,12 +32,6 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do echo "clean up containers if exists" cleanup PG_TEST_VERSION=$((pg_version < 16 ? 16 : pg_version)) - # The support of pg_anon not yet added to PG17, so we have to add the corresponding option for other PG versions - if [ "${pg_version}" -ne 17 ]; then - SPEC_PATH="compute_wrapper/var/db/postgres/specs" - mv $SPEC_PATH/spec.json $SPEC_PATH/spec.bak - jq '.cluster.settings += [{"name": "session_preload_libraries","value": "anon","vartype": "string"}]' "${SPEC_PATH}/spec.bak" > "${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 diff --git a/libs/pageserver_api/src/config.rs b/libs/pageserver_api/src/config.rs index a0f3c00c3c..4982c6233d 100644 --- a/libs/pageserver_api/src/config.rs +++ b/libs/pageserver_api/src/config.rs @@ -321,6 +321,20 @@ 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, + + // 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 { @@ -519,6 +533,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 { @@ -563,6 +580,10 @@ 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, + 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 c38af9cb80..fd4879087f 100644 --- a/libs/pageserver_api/src/models.rs +++ b/libs/pageserver_api/src/models.rs @@ -497,6 +497,14 @@ 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, + #[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 @@ -528,6 +536,10 @@ 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, + pub gc_compaction_enabled: Option, + pub gc_compaction_initial_threshold_kb: Option, + pub gc_compaction_ratio_percent: Option, } impl TenantConfig { @@ -557,6 +569,10 @@ impl TenantConfig { mut lsn_lease_length_for_ts, 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); @@ -601,6 +617,16 @@ impl TenantConfig { patch .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, @@ -627,6 +653,10 @@ impl TenantConfig { lsn_lease_length_for_ts, 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/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/pageserver/src/metrics.rs b/pageserver/src/metrics.rs index 7d383379db..02467cb6f7 100644 --- a/pageserver/src/metrics.rs +++ b/pageserver/src/metrics.rs @@ -2483,12 +2483,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 @@ -2508,6 +2515,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)] diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index b948fb6c55..a273ef5d01 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -5475,6 +5475,12 @@ 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, + 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 edf2e6a3aa..3db1445f6e 100644 --- a/pageserver/src/tenant/config.rs +++ b/pageserver/src/tenant/config.rs @@ -357,6 +357,18 @@ 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, + + #[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 { @@ -425,6 +437,16 @@ 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), + 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), } } @@ -454,6 +476,10 @@ impl TenantConfOpt { mut lsn_lease_length_for_ts, 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); @@ -522,6 +548,16 @@ impl TenantConfOpt { patch .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, @@ -548,6 +584,10 @@ impl TenantConfOpt { lsn_lease_length_for_ts, timeline_offloading, wal_receiver_protocol_override, + rel_size_v2_enabled, + gc_compaction_enabled, + gc_compaction_initial_threshold_kb, + gc_compaction_ratio_percent, }) } } @@ -603,6 +643,10 @@ 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, + 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/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<()> { diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs index 08e94ae197..3824bc8f11 100644 --- a/pageserver/src/tenant/remote_timeline_client/index.rs +++ b/pageserver/src/tenant/remote_timeline_client/index.rs @@ -77,8 +77,32 @@ 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)] +#[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 +121,12 @@ impl IndexPart { /// - 8: added `archived_at` /// - 9: +gc_blocking /// - 10: +import_pgdata - const LATEST_VERSION: usize = 10; + /// - 11: +rel_size_migration + /// - 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]; + 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"; @@ -116,6 +142,8 @@ impl IndexPart { gc_blocking: None, last_aux_file_policy: None, import_pgdata: None, + rel_size_migration: None, + l2_lsn: None, } } @@ -416,6 +444,8 @@ mod tests { gc_blocking: None, 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(); @@ -461,6 +491,8 @@ mod tests { gc_blocking: None, 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(); @@ -507,6 +539,8 @@ mod tests { gc_blocking: None, 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(); @@ -556,6 +590,8 @@ mod tests { gc_blocking: None, 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(); @@ -600,6 +636,8 @@ mod tests { gc_blocking: None, 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(); @@ -647,6 +685,8 @@ mod tests { gc_blocking: None, 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(); @@ -699,6 +739,8 @@ mod tests { gc_blocking: None, 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(); @@ -756,6 +798,8 @@ mod tests { gc_blocking: None, 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(); @@ -814,6 +858,8 @@ mod tests { gc_blocking: None, 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(); @@ -877,6 +923,8 @@ mod tests { last_aux_file_policy: Default::default(), archived_at: None, import_pgdata: None, + rel_size_migration: None, + l2_lsn: None, }; let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap(); @@ -952,7 +1000,168 @@ 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, + l2_lsn: 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), + 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/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 2dacd5a383..5d348ac474 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -2841,9 +2841,6 @@ impl Timeline { /// /// It is _not_ sensitive to task_mgr::shutdown_token(). /// - /// The rationale is that we spawn initial logical size calculation - /// during activation, hence - /// /// # Cancel-Safety /// /// It does Timeline IO, hence this should be polled to completion because @@ -3631,6 +3628,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; 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/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) 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_attach_tenant_config.py b/test_runner/regress/test_attach_tenant_config.py index 45112fd67e..b8d47346a3 100644 --- a/test_runner/regress/test_attach_tenant_config.py +++ b/test_runner/regress/test_attach_tenant_config.py @@ -176,6 +176,10 @@ def test_fully_custom_config(positive_env: NeonEnv): "type": "interpreted", "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() 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 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 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"] }