diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 0ceaa96fb0..864abad574 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -986,6 +986,7 @@ jobs: - name: Verify docker-compose example and test extensions timeout-minutes: 60 env: + PARALLEL_COMPUTES: 3 TAG: >- ${{ needs.meta.outputs.run-kind == 'compute-rc-pr' diff --git a/.gitignore b/.gitignore index 6574d7b9de..4857972f1d 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,7 @@ neon.iml /.neon /integration_tests/.neon compaction-suite-results.* +docker-compose/docker-compose-parallel.yml # Coverage *.profraw diff --git a/compute/compute-node.Dockerfile b/compute/compute-node.Dockerfile index 0dd32011fb..39136fe573 100644 --- a/compute/compute-node.Dockerfile +++ b/compute/compute-node.Dockerfile @@ -1915,10 +1915,10 @@ RUN cd /ext-src/pg_repack-src && patch -p1 /etc/ld.so.conf.d/00-neon.conf && /sbin/ldconfig -RUN apt-get update && apt-get install -y libtap-parser-sourcehandler-pgtap-perl jq \ +RUN apt-get update && apt-get install -y libtap-parser-sourcehandler-pgtap-perl jq parallel \ && apt clean && rm -rf /ext-src/*.tar.gz /ext-src/*.patch /var/lib/apt/lists/* ENV PATH=/usr/local/pgsql/bin:$PATH -ENV PGHOST=compute +ENV PGHOST=compute1 ENV PGPORT=55433 ENV PGUSER=cloud_admin ENV PGDATABASE=postgres diff --git a/compute_tools/src/compute_prewarm.rs b/compute_tools/src/compute_prewarm.rs index 4190580e5e..3f6f9a7ecc 100644 --- a/compute_tools/src/compute_prewarm.rs +++ b/compute_tools/src/compute_prewarm.rs @@ -105,7 +105,8 @@ impl ComputeNode { cloned.state.lock().unwrap().lfc_prewarm_state = LfcPrewarmState::Completed; return; }; - error!(%err); + crate::metrics::LFC_PREWARM_ERRORS.inc(); + error!(%err, "prewarming lfc"); cloned.state.lock().unwrap().lfc_prewarm_state = LfcPrewarmState::Failed { error: err.to_string(), }; @@ -180,7 +181,8 @@ impl ComputeNode { self.state.lock().unwrap().lfc_offload_state = LfcOffloadState::Completed; return; }; - error!(%err); + crate::metrics::LFC_OFFLOAD_ERRORS.inc(); + error!(%err, "offloading lfc"); self.state.lock().unwrap().lfc_offload_state = LfcOffloadState::Failed { error: err.to_string(), }; diff --git a/compute_tools/src/metrics.rs b/compute_tools/src/metrics.rs index 8f81675c49..91dedbb42a 100644 --- a/compute_tools/src/metrics.rs +++ b/compute_tools/src/metrics.rs @@ -105,6 +105,14 @@ pub(crate) static LFC_PREWARMS: Lazy = Lazy::new(|| { .expect("failed to define a metric") }); +pub(crate) static LFC_PREWARM_ERRORS: Lazy = Lazy::new(|| { + register_int_counter!( + "compute_ctl_lfc_prewarm_errors_total", + "Total number of LFC prewarms errors requested by compute_ctl or autoprewarm option", + ) + .expect("failed to define a metric") +}); + pub(crate) static LFC_OFFLOADS: Lazy = Lazy::new(|| { register_int_counter!( "compute_ctl_lfc_offloads_total", @@ -113,6 +121,14 @@ pub(crate) static LFC_OFFLOADS: Lazy = Lazy::new(|| { .expect("failed to define a metric") }); +pub(crate) static LFC_OFFLOAD_ERRORS: Lazy = Lazy::new(|| { + register_int_counter!( + "compute_ctl_lfc_offload_errors_total", + "Total number of LFC offload errors requested by compute_ctl or lfc_offload_period_seconds option", + ) + .expect("failed to define a metric") +}); + pub fn collect() -> Vec { let mut metrics = COMPUTE_CTL_UP.collect(); metrics.extend(INSTALLED_EXTENSIONS.collect()); @@ -123,6 +139,8 @@ pub fn collect() -> Vec { metrics.extend(PG_CURR_DOWNTIME_MS.collect()); metrics.extend(PG_TOTAL_DOWNTIME_MS.collect()); metrics.extend(LFC_PREWARMS.collect()); + metrics.extend(LFC_PREWARM_ERRORS.collect()); metrics.extend(LFC_OFFLOADS.collect()); + metrics.extend(LFC_OFFLOAD_ERRORS.collect()); metrics } diff --git a/docker-compose/compute_wrapper/shell/compute.sh b/docker-compose/compute_wrapper/shell/compute.sh index 1e62e91fd0..6f36b4358e 100755 --- a/docker-compose/compute_wrapper/shell/compute.sh +++ b/docker-compose/compute_wrapper/shell/compute.sh @@ -54,14 +54,16 @@ else printf '%s\n' "${result}" | jq . fi - echo "Check if a timeline present" - PARAMS=( - -X GET - -H "Content-Type: application/json" - "http://pageserver:9898/v1/tenant/${tenant_id}/timeline" - ) - timeline_id=$(curl "${PARAMS[@]}" | jq -r .[0].timeline_id) - if [[ -z "${timeline_id}" || "${timeline_id}" = null ]]; then + if [[ "${RUN_PARALLEL:-false}" != "true" ]]; then + echo "Check if a timeline present" + PARAMS=( + -X GET + -H "Content-Type: application/json" + "http://pageserver:9898/v1/tenant/${tenant_id}/timeline" + ) + timeline_id=$(curl "${PARAMS[@]}" | jq -r .[0].timeline_id) + fi + if [[ -z "${timeline_id:-}" || "${timeline_id:-}" = null ]]; then generate_id timeline_id PARAMS=( -sbf diff --git a/docker-compose/docker-compose.yml b/docker-compose/docker-compose.yml index 2519b75c7f..19c3bc74e2 100644 --- a/docker-compose/docker-compose.yml +++ b/docker-compose/docker-compose.yml @@ -142,7 +142,7 @@ services: - "storage_broker" - "--listen-addr=0.0.0.0:50051" - compute: + compute1: restart: always build: context: ./compute_wrapper/ @@ -152,6 +152,7 @@ services: - TAG=${COMPUTE_TAG:-${TAG:-latest}} - http_proxy=${http_proxy:-} - https_proxy=${https_proxy:-} + image: built-compute environment: - PG_VERSION=${PG_VERSION:-16} - TENANT_ID=${TENANT_ID:-} @@ -166,6 +167,11 @@ services: - 3080:3080 # http endpoints entrypoint: - "/shell/compute.sh" + # Ad an alias for compute1 for compatibility + networks: + default: + aliases: + - compute depends_on: - safekeeper1 - safekeeper2 @@ -174,15 +180,20 @@ services: compute_is_ready: image: postgres:latest + environment: + - PARALLEL_COMPUTES=1 entrypoint: - - "/bin/bash" + - "/bin/sh" - "-c" command: - - "until pg_isready -h compute -p 55433 -U cloud_admin ; do - echo 'Waiting to start compute...' && sleep 1; - done" + - "for i in $(seq 1 $${PARALLEL_COMPUTES}); do + until pg_isready -h compute$$i -p 55433 -U cloud_admin ; do + sleep 1; + done; + done; + echo All computes are started" depends_on: - - compute + - compute1 neon-test-extensions: profiles: ["test-extensions"] @@ -196,4 +207,4 @@ services: command: - sleep 3600 depends_on: - - compute + - compute1 diff --git a/docker-compose/docker_compose_test.sh b/docker-compose/docker_compose_test.sh index 6edf90ca8d..063b8dee85 100755 --- a/docker-compose/docker_compose_test.sh +++ b/docker-compose/docker_compose_test.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # A basic test to ensure Docker images are built correctly. # Build a wrapper around the compute, start all services and runs a simple SQL query. @@ -13,9 +13,36 @@ # set -eux -o pipefail +cd "$(dirname "${0}")" export COMPOSE_FILE='docker-compose.yml' export COMPOSE_PROFILES=test-extensions -cd "$(dirname "${0}")" +export PARALLEL_COMPUTES=${PARALLEL_COMPUTES:-1} +READY_MESSAGE="All computes are started" +COMPUTES=() +for i in $(seq 1 "${PARALLEL_COMPUTES}"); do + COMPUTES+=("compute${i}") +done +CURRENT_TMPDIR=$(mktemp -d) +trap 'rm -rf ${CURRENT_TMPDIR} docker-compose-parallel.yml' EXIT +if [[ ${PARALLEL_COMPUTES} -gt 1 ]]; then + export COMPOSE_FILE=docker-compose-parallel.yml + cp docker-compose.yml docker-compose-parallel.yml + # Replace the environment variable PARALLEL_COMPUTES with the actual value + yq eval -i ".services.compute_is_ready.environment |= map(select(. | test(\"^PARALLEL_COMPUTES=\") | not)) + [\"PARALLEL_COMPUTES=${PARALLEL_COMPUTES}\"]" ${COMPOSE_FILE} + for i in $(seq 2 "${PARALLEL_COMPUTES}"); do + # Duplicate compute1 as compute${i} for parallel execution + yq eval -i ".services.compute${i} = .services.compute1" ${COMPOSE_FILE} + # We don't need these sections, so delete them + yq eval -i "(del .services.compute${i}.build) | (del .services.compute${i}.ports) | (del .services.compute${i}.networks)" ${COMPOSE_FILE} + # Let the compute 1 be the only dependence + yq eval -i ".services.compute${i}.depends_on = [\"compute1\"]" ${COMPOSE_FILE} + # Set RUN_PARALLEL=true for compute2. They will generate tenant_id and timeline_id to avoid using the same as other computes + yq eval -i ".services.compute${i}.environment += [\"RUN_PARALLEL=true\"]" ${COMPOSE_FILE} + # Remove TENANT_ID and TIMELINE_ID from the environment variables of the generated computes + # They will create new TENANT_ID and TIMELINE_ID anyway. + yq eval -i ".services.compute${i}.environment |= map(select(. | (test(\"^TENANT_ID=\") or test(\"^TIMELINE_ID=\")) | not))" ${COMPOSE_FILE} + done +fi PSQL_OPTION="-h localhost -U cloud_admin -p 55433 -d postgres" function cleanup() { @@ -27,11 +54,11 @@ function cleanup() { for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do pg_version=${pg_version/v/} - echo "clean up containers if exists" + echo "clean up containers if exist" cleanup PG_TEST_VERSION=$((pg_version < 16 ? 16 : pg_version)) - PG_VERSION=${pg_version} PG_TEST_VERSION=${PG_TEST_VERSION} docker compose up --quiet-pull --build -d - + PG_VERSION=${pg_version} PG_TEST_VERSION=${PG_TEST_VERSION} docker compose build compute1 + PG_VERSION=${pg_version} PG_TEST_VERSION=${PG_TEST_VERSION} docker compose up --quiet-pull -d echo "wait until the compute is ready. timeout after 60s. " cnt=0 while sleep 3; do @@ -41,45 +68,50 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do echo "timeout before the compute is ready." exit 1 fi - if docker compose logs "compute_is_ready" | grep -q "accepting connections"; then + if docker compose logs compute_is_ready | grep -q "${READY_MESSAGE}"; then echo "OK. The compute is ready to connect." echo "execute simple queries." - docker compose exec compute /bin/bash -c "psql ${PSQL_OPTION} -c 'SELECT 1'" + for compute in "${COMPUTES[@]}"; do + docker compose exec "${compute}" /bin/bash -c "psql ${PSQL_OPTION} -c 'SELECT 1'" + done break fi done if [[ ${pg_version} -ge 16 ]]; then - # 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 compose exec compute touch /var/db/postgres/compute/compute_ctl_temp_override.conf - # Prepare for the PostGIS test - docker compose exec compute mkdir -p /tmp/pgis_reg/pgis_reg_tmp - TMPDIR=$(mktemp -d) - docker compose cp neon-test-extensions:/ext-src/postgis-src/raster/test "${TMPDIR}" - docker compose cp neon-test-extensions:/ext-src/postgis-src/regress/00-regress-install "${TMPDIR}" - docker compose exec compute mkdir -p /ext-src/postgis-src/raster /ext-src/postgis-src/regress /ext-src/postgis-src/regress/00-regress-install - docker compose cp "${TMPDIR}/test" compute:/ext-src/postgis-src/raster/test - docker compose cp "${TMPDIR}/00-regress-install" compute:/ext-src/postgis-src/regress - rm -rf "${TMPDIR}" - # 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) - docker compose cp neon-test-extensions:/ext-src/pg_hint_plan-src/data "${TMPDIR}/data" - docker compose cp "${TMPDIR}/data" compute:/ext-src/pg_hint_plan-src/ - rm -rf "${TMPDIR}" - # The following block does the same for the contrib/file_fdw test - TMPDIR=$(mktemp -d) - docker compose cp neon-test-extensions:/postgres/contrib/file_fdw/data "${TMPDIR}/data" - docker compose cp "${TMPDIR}/data" compute:/postgres/contrib/file_fdw/data - rm -rf "${TMPDIR}" + mkdir "${CURRENT_TMPDIR}"/{pg_hint_plan-src,file_fdw,postgis-src} + docker compose cp neon-test-extensions:/ext-src/postgis-src/raster/test "${CURRENT_TMPDIR}/postgis-src/test" + docker compose cp neon-test-extensions:/ext-src/postgis-src/regress/00-regress-install "${CURRENT_TMPDIR}/postgis-src/00-regress-install" + docker compose cp neon-test-extensions:/ext-src/pg_hint_plan-src/data "${CURRENT_TMPDIR}/pg_hint_plan-src/data" + docker compose cp neon-test-extensions:/postgres/contrib/file_fdw/data "${CURRENT_TMPDIR}/file_fdw/data" + + for compute in "${COMPUTES[@]}"; do + # 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 on "${compute}" + docker compose exec "${compute}" touch /var/db/postgres/compute/compute_ctl_temp_override.conf + # Prepare for the PostGIS test + docker compose exec "${compute}" mkdir -p /tmp/pgis_reg/pgis_reg_tmp /ext-src/postgis-src/raster /ext-src/postgis-src/regress /ext-src/postgis-src/regress/00-regress-install + docker compose cp "${CURRENT_TMPDIR}/postgis-src/test" "${compute}":/ext-src/postgis-src/raster/test + docker compose cp "${CURRENT_TMPDIR}/postgis-src/00-regress-install" "${compute}":/ext-src/postgis-src/regress + # 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 + docker compose cp "${CURRENT_TMPDIR}/pg_hint_plan-src/data" "${compute}":/ext-src/pg_hint_plan-src/ + # The following block does the same for the contrib/file_fdw test + docker compose cp "${CURRENT_TMPDIR}/file_fdw/data" "${compute}":/postgres/contrib/file_fdw/data + done # Apply patches docker compose exec -T neon-test-extensions bash -c "(cd /postgres && patch -p1)" <"../compute/patches/contrib_pg${pg_version}.patch" # We are running tests now rm -f testout.txt testout_contrib.txt + # We want to run the longest tests first to better utilize parallelization and reduce overall test time. + # Tests listed in the RUN_FIRST variable will be run before others. + # If parallelization is not used, this environment variable will be ignored. + docker compose exec -e USE_PGXS=1 -e SKIP=timescaledb-src,rdkit-src,pg_jsonschema-src,kq_imcx-src,wal2json_2_5-src,rag_jina_reranker_v1_tiny_en-src,rag_bge_small_en_v15-src \ + -e RUN_FIRST=hll-src,postgis-src,pgtap-src -e PARALLEL_COMPUTES="${PARALLEL_COMPUTES}" \ neon-test-extensions /run-tests.sh /ext-src | tee testout.txt && EXT_SUCCESS=1 || EXT_SUCCESS=0 docker compose exec -e SKIP=start-scripts,postgres_fdw,ltree_plpython,jsonb_plpython,jsonb_plperl,hstore_plpython,hstore_plperl,dblink,bool_plperl \ + -e PARALLEL_COMPUTES="${PARALLEL_COMPUTES}" \ neon-test-extensions /run-tests.sh /postgres/contrib | tee testout_contrib.txt && CONTRIB_SUCCESS=1 || CONTRIB_SUCCESS=0 if [[ ${EXT_SUCCESS} -eq 0 || ${CONTRIB_SUCCESS} -eq 0 ]]; then CONTRIB_FAILED= diff --git a/docker-compose/run-tests.sh b/docker-compose/run-tests.sh index 930402ce66..b37b9363fa 100755 --- a/docker-compose/run-tests.sh +++ b/docker-compose/run-tests.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash set -x if [[ -v BENCHMARK_CONNSTR ]]; then @@ -26,8 +26,9 @@ if [[ -v BENCHMARK_CONNSTR ]]; then fi fi REGULAR_USER=false -while getopts r arg; do - case $arg in +PARALLEL_COMPUTES=${PARALLEL_COMPUTES:-1} +while getopts pr arg; do + case ${arg} in r) REGULAR_USER=true shift $((OPTIND-1)) @@ -41,26 +42,49 @@ extdir=${1} cd "${extdir}" || exit 2 FAILED= -LIST=$( (echo -e "${SKIP//","/"\n"}"; ls) | sort | uniq -u) -for d in ${LIST}; do - [ -d "${d}" ] || continue - if ! psql -w -c "select 1" >/dev/null; then - FAILED="${d} ${FAILED}" - break - fi - if [[ ${REGULAR_USER} = true ]] && [ -f "${d}"/regular-test.sh ]; then - "${d}/regular-test.sh" || FAILED="${d} ${FAILED}" - continue - fi +export FAILED_FILE=/tmp/failed +rm -f ${FAILED_FILE} +mapfile -t LIST < <( (echo -e "${SKIP//","/"\n"}"; ls) | sort | uniq -u) +if [[ ${PARALLEL_COMPUTES} -gt 1 ]]; then + # Avoid errors if RUN_FIRST is not defined + RUN_FIRST=${RUN_FIRST:-} + # Move entries listed in the RUN_FIRST variable to the beginning + ORDERED_LIST=$(printf "%s\n" "${LIST[@]}" | grep -x -Ff <(echo -e "${RUN_FIRST//,/$'\n'}"); printf "%s\n" "${LIST[@]}" | grep -vx -Ff <(echo -e "${RUN_FIRST//,/$'\n'}")) + parallel -j"${PARALLEL_COMPUTES}" "[[ -d {} ]] || exit 0 + export PGHOST=compute{%} + if ! psql -c 'select 1'>/dev/null; then + exit 1 + fi + echo Running on \${PGHOST} + if [[ -f ${extdir}/{}/neon-test.sh ]]; then + echo Running from script + ${extdir}/{}/neon-test.sh || echo {} >> ${FAILED_FILE}; + else + echo Running using make; + USE_PGXS=1 make -C {} installcheck || echo {} >> ${FAILED_FILE}; + fi" ::: ${ORDERED_LIST} + [[ ! -f ${FAILED_FILE} ]] && exit 0 +else + for d in "${LIST[@]}"; do + [ -d "${d}" ] || continue + if ! psql -w -c "select 1" >/dev/null; then + FAILED="${d} ${FAILED}" + break + fi + if [[ ${REGULAR_USER} = true ]] && [ -f "${d}"/regular-test.sh ]; then + "${d}/regular-test.sh" || FAILED="${d} ${FAILED}" + continue + fi - if [ -f "${d}/neon-test.sh" ]; then - "${d}/neon-test.sh" || FAILED="${d} ${FAILED}" - else - USE_PGXS=1 make -C "${d}" installcheck || FAILED="${d} ${FAILED}" - fi -done -[ -z "${FAILED}" ] && exit 0 -for d in ${FAILED}; do + if [ -f "${d}/neon-test.sh" ]; then + "${d}/neon-test.sh" || FAILED="${d} ${FAILED}" + else + USE_PGXS=1 make -C "${d}" installcheck || FAILED="${d} ${FAILED}" + fi + done + [[ -z ${FAILED} ]] && exit 0 +fi +for d in ${FAILED} $([[ ! -f ${FAILED_FILE} ]] || cat ${FAILED_FILE}); do cat "$(find $d -name regression.diffs)" done for postgis_diff in /tmp/pgis_reg/*_diff; do @@ -68,4 +92,5 @@ for postgis_diff in /tmp/pgis_reg/*_diff; do cat "${postgis_diff}" done echo "${FAILED}" +cat ${FAILED_FILE} exit 1 diff --git a/docker-compose/test_extensions_upgrade.sh b/docker-compose/test_extensions_upgrade.sh index f1cf17f531..1d39fc029e 100755 --- a/docker-compose/test_extensions_upgrade.sh +++ b/docker-compose/test_extensions_upgrade.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash set -eux -o pipefail cd "$(dirname "${0}")" # Takes a variable name as argument. The result is stored in that variable. @@ -60,8 +60,8 @@ function check_timeline() { # Restarts the compute node with the required compute tag and timeline. # Accepts the tag for the compute node and the timeline as parameters. function restart_compute() { - docker compose down compute compute_is_ready - COMPUTE_TAG=${1} TENANT_ID=${tenant_id} TIMELINE_ID=${2} docker compose up --quiet-pull -d --build compute compute_is_ready + docker compose down compute1 compute_is_ready + COMPUTE_TAG=${1} TENANT_ID=${tenant_id} TIMELINE_ID=${2} docker compose up --quiet-pull -d --build compute1 compute_is_ready wait_for_ready check_timeline ${2} } diff --git a/pageserver/src/config.rs b/pageserver/src/config.rs index 75b41b9b60..6e22f9f36e 100644 --- a/pageserver/src/config.rs +++ b/pageserver/src/config.rs @@ -635,7 +635,7 @@ impl PageServerConf { pub fn dummy_conf(repo_dir: Utf8PathBuf) -> Self { let pg_distrib_dir = Utf8PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../pg_install"); - let config_toml = pageserver_api::config::ConfigToml { + let mut config_toml = pageserver_api::config::ConfigToml { wait_lsn_timeout: Duration::from_secs(60), wal_redo_timeout: Duration::from_secs(60), pg_distrib_dir: Some(pg_distrib_dir), @@ -647,6 +647,15 @@ impl PageServerConf { control_plane_api: Some(Url::parse("http://localhost:6666").unwrap()), ..Default::default() }; + + // Test authors tend to forget about the default 10min initial lease deadline + // when writing tests, which turns their immediate gc requests via mgmt API + // into no-ops. Override the binary default here, such that there is no initial + // lease deadline by default in tests. Tests that care can always override it + // themselves. + // Cf https://databricks.atlassian.net/browse/LKB-92?focusedCommentId=6722329 + config_toml.tenant_config.lsn_lease_length = Duration::from_secs(0); + PageServerConf::parse_and_validate(NodeId(0), config_toml, &repo_dir).unwrap() } } diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index f4877fd763..9aabd6341f 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -34,7 +34,7 @@ use once_cell::sync::Lazy; pub use pageserver_api::models::TenantState; use pageserver_api::models::{self, RelSizeMigration}; use pageserver_api::models::{ - CompactInfoResponse, LsnLease, TimelineArchivalState, TimelineState, TopTenantShardItem, + CompactInfoResponse, TimelineArchivalState, TimelineState, TopTenantShardItem, WalRedoManagerStatus, }; use pageserver_api::shard::{ShardIdentity, ShardStripeSize, TenantShardId}; @@ -180,6 +180,7 @@ pub(super) struct AttachedTenantConf { impl AttachedTenantConf { fn new( + conf: &'static PageServerConf, tenant_conf: pageserver_api::models::TenantConfig, location: AttachedLocationConfig, ) -> Self { @@ -191,9 +192,7 @@ impl AttachedTenantConf { let lsn_lease_deadline = if location.attach_mode == AttachmentMode::Single { Some( tokio::time::Instant::now() - + tenant_conf - .lsn_lease_length - .unwrap_or(LsnLease::DEFAULT_LENGTH), + + TenantShard::get_lsn_lease_length_impl(conf, &tenant_conf), ) } else { // We don't use `lsn_lease_deadline` to delay GC in AttachedMulti and AttachedStale @@ -208,10 +207,13 @@ impl AttachedTenantConf { } } - fn try_from(location_conf: LocationConf) -> anyhow::Result { + fn try_from( + conf: &'static PageServerConf, + location_conf: LocationConf, + ) -> anyhow::Result { match &location_conf.mode { LocationMode::Attached(attach_conf) => { - Ok(Self::new(location_conf.tenant_conf, *attach_conf)) + Ok(Self::new(conf, location_conf.tenant_conf, *attach_conf)) } LocationMode::Secondary(_) => { anyhow::bail!( @@ -4205,10 +4207,16 @@ impl TenantShard { } pub fn get_lsn_lease_length(&self) -> Duration { - let tenant_conf = self.tenant_conf.load().tenant_conf.clone(); + Self::get_lsn_lease_length_impl(self.conf, &self.tenant_conf.load().tenant_conf) + } + + pub fn get_lsn_lease_length_impl( + conf: &'static PageServerConf, + tenant_conf: &pageserver_api::models::TenantConfig, + ) -> Duration { tenant_conf .lsn_lease_length - .unwrap_or(self.conf.default_tenant_conf.lsn_lease_length) + .unwrap_or(conf.default_tenant_conf.lsn_lease_length) } pub fn get_timeline_offloading_enabled(&self) -> bool { @@ -6020,11 +6028,14 @@ pub(crate) mod harness { let tenant = Arc::new(TenantShard::new( TenantState::Attaching, self.conf, - AttachedTenantConf::try_from(LocationConf::attached_single( - self.tenant_conf.clone(), - self.generation, - ShardParameters::default(), - )) + AttachedTenantConf::try_from( + self.conf, + LocationConf::attached_single( + self.tenant_conf.clone(), + self.generation, + ShardParameters::default(), + ), + ) .unwrap(), self.shard_identity, Some(walredo_mgr), @@ -6125,7 +6136,7 @@ mod tests { use pageserver_api::keyspace::KeySpace; #[cfg(feature = "testing")] use pageserver_api::keyspace::KeySpaceRandomAccum; - use pageserver_api::models::{CompactionAlgorithm, CompactionAlgorithmSettings}; + use pageserver_api::models::{CompactionAlgorithm, CompactionAlgorithmSettings, LsnLease}; use pageserver_compaction::helpers::overlaps_with; #[cfg(feature = "testing")] use rand::SeedableRng; @@ -6675,17 +6686,13 @@ mod tests { tline.freeze_and_flush().await.map_err(|e| e.into()) } - #[tokio::test(start_paused = true)] + #[tokio::test] async fn test_prohibit_branch_creation_on_garbage_collected_data() -> anyhow::Result<()> { let (tenant, ctx) = TenantHarness::create("test_prohibit_branch_creation_on_garbage_collected_data") .await? .load() .await; - // Advance to the lsn lease deadline so that GC is not blocked by - // initial transition into AttachedSingle. - tokio::time::advance(tenant.get_lsn_lease_length()).await; - tokio::time::resume(); let tline = tenant .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx) .await?; @@ -9384,17 +9391,21 @@ mod tests { Ok(()) } - #[tokio::test(start_paused = true)] + #[tokio::test] async fn test_lsn_lease() -> anyhow::Result<()> { let (tenant, ctx) = TenantHarness::create("test_lsn_lease") .await .unwrap() .load() .await; - // Advance to the lsn lease deadline so that GC is not blocked by - // initial transition into AttachedSingle. - tokio::time::advance(tenant.get_lsn_lease_length()).await; - tokio::time::resume(); + // set a non-zero lease length to test the feature + tenant + .update_tenant_config(|mut conf| { + conf.lsn_lease_length = Some(LsnLease::DEFAULT_LENGTH); + Ok(conf) + }) + .unwrap(); + let key = Key::from_hex("010000000033333333444444445500000000").unwrap(); let end_lsn = Lsn(0x100); diff --git a/pageserver/src/tenant/mgr.rs b/pageserver/src/tenant/mgr.rs index 95f5c60170..be18b40862 100644 --- a/pageserver/src/tenant/mgr.rs +++ b/pageserver/src/tenant/mgr.rs @@ -664,7 +664,7 @@ pub async fn init_tenant_mgr( tenant_shard_id, &tenant_dir_path, resources.clone(), - AttachedTenantConf::new(location_conf.tenant_conf, attached_conf), + AttachedTenantConf::new(conf, location_conf.tenant_conf, attached_conf), shard_identity, Some(init_order.clone()), SpawnMode::Lazy, @@ -842,8 +842,11 @@ impl TenantManager { // take our fast path and just provide the updated configuration // to the tenant. tenant.set_new_location_config( - AttachedTenantConf::try_from(new_location_config.clone()) - .map_err(UpsertLocationError::BadRequest)?, + AttachedTenantConf::try_from( + self.conf, + new_location_config.clone(), + ) + .map_err(UpsertLocationError::BadRequest)?, ); Some(FastPathModified::Attached(tenant.clone())) @@ -1046,7 +1049,7 @@ impl TenantManager { // Testing hack: if we are configured with no control plane, then drop the generation // from upserts. This enables creating generation-less tenants even though neon_local // always uses generations when calling the location conf API. - let attached_conf = AttachedTenantConf::try_from(new_location_config) + let attached_conf = AttachedTenantConf::try_from(self.conf, new_location_config) .map_err(UpsertLocationError::BadRequest)?; let tenant = tenant_spawn( @@ -1250,7 +1253,7 @@ impl TenantManager { tenant_shard_id, &tenant_path, self.resources.clone(), - AttachedTenantConf::try_from(config)?, + AttachedTenantConf::try_from(self.conf, config)?, shard_identity, None, SpawnMode::Eager, @@ -2131,7 +2134,7 @@ impl TenantManager { tenant_shard_id, &tenant_path, self.resources.clone(), - AttachedTenantConf::try_from(config).map_err(Error::DetachReparent)?, + AttachedTenantConf::try_from(self.conf, config).map_err(Error::DetachReparent)?, shard_identity, None, SpawnMode::Eager, diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index f2ec022666..f54d5be635 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -1315,6 +1315,14 @@ class NeonEnv: # This feature is pending rollout. # tenant_config["rel_size_v2_enabled"] = True + # Test authors tend to forget about the default 10min initial lease deadline + # when writing tests, which turns their immediate gc requests via mgmt API + # into no-ops. Override the binary default here, such that there is no initial + # lease deadline by default in tests. Tests that care can always override it + # themselves. + # Cf https://databricks.atlassian.net/browse/LKB-92?focusedCommentId=6722329 + tenant_config["lsn_lease_length"] = "0s" + if self.pageserver_remote_storage is not None: ps_cfg["remote_storage"] = remote_storage_to_toml_dict( self.pageserver_remote_storage diff --git a/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py b/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py index 68bfa81b25..bf998a2a0a 100644 --- a/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py +++ b/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py @@ -104,9 +104,7 @@ def setup_and_run_pagebench_benchmark( # configure cache sizes like in prod page_cache_size = 16384 max_file_descriptors = 500000 - neon_env_builder.pageserver_config_override = ( - f"page_cache_size={page_cache_size}; max_file_descriptors={max_file_descriptors}" - ) + neon_env_builder.pageserver_config_override = f"page_cache_size={page_cache_size}; max_file_descriptors={max_file_descriptors}; disk_usage_based_eviction={{max_usage_pct=99, min_avail_bytes=0, period = '999y'}}" tracing_config = PageserverTracingConfig( sampling_ratio=(0, 1000), diff --git a/test_runner/random_ops/test_random_ops.py b/test_runner/random_ops/test_random_ops.py index 9bf2e5ab77..7800a4cb0c 100644 --- a/test_runner/random_ops/test_random_ops.py +++ b/test_runner/random_ops/test_random_ops.py @@ -125,7 +125,9 @@ class NeonBranch: def create_child_branch(self, parent_timestamp: datetime | None = None) -> NeonBranch | None: return self.project.create_branch(self.id, parent_timestamp) - def create_ro_endpoint(self) -> NeonEndpoint: + def create_ro_endpoint(self) -> NeonEndpoint | None: + if not self.project.check_limit_endpoints(): + return None return NeonEndpoint( self.project, self.neon_api.create_endpoint(self.project_id, self.id, "read_only", {})["endpoint"], @@ -156,11 +158,26 @@ class NeonBranch: return self.updated_at = datetime.fromisoformat(res["branch"]["updated_at"]) parent_id: str = res["branch"]["parent_id"] + # XXX Retry get parent details to work around the issue + # https://databricks.atlassian.net/browse/LKB-279 + target_time = datetime.now() + timedelta(seconds=30) + while datetime.now() < target_time: + try: + parent_def = self.neon_api.get_branch_details(self.project_id, parent_id) + except HTTPError as he: + if he.response.status_code == 404: + log.info("Branch not found, waiting...") + time.sleep(1) + else: + raise HTTPError(he) from he + else: + break + else: + raise RuntimeError(f"Branch {parent_id} not found") + # Creates an object for the parent branch # After the reset operation a new parent branch is created - parent = NeonBranch( - self.project, self.neon_api.get_branch_details(self.project_id, parent_id), True - ) + parent = NeonBranch(self.project, parent_def, True) self.project.branches[parent_id] = parent self.parent = parent parent.children[self.id] = self @@ -173,29 +190,21 @@ class NeonBranch: source_timestamp: str | None = None, preserve_under_name: str | None = None, ) -> dict[str, Any] | None: + if not self.project.check_limit_branches(): + return None endpoints = [ep for ep in self.endpoints.values() if ep.type == "read_only"] # Terminate all the benchmarks running to prevent errors. Errors in benchmark during pgbench are expected for ep in endpoints: ep.terminate_benchmark() self.terminate_benchmark() - try: - res: dict[str, Any] = self.neon_api.restore_branch( - self.project_id, - self.id, - source_branch_id, - source_lsn, - source_timestamp, - preserve_under_name, - ) - except HTTPError as he: - if ( - he.response.status_code == 422 - and he.response.json()["code"] == "BRANCHES_LIMIT_EXCEEDED" - ): - log.info("Branch limit exceeded, skipping") - return None - else: - raise HTTPError(he) from he + res: dict[str, Any] = self.neon_api.restore_branch( + self.project_id, + self.id, + source_branch_id, + source_lsn, + source_timestamp, + preserve_under_name, + ) self.project.wait() self.start_benchmark() for ep in endpoints: @@ -244,26 +253,37 @@ class NeonProject: def delete(self) -> None: self.neon_api.delete_project(self.id) + def check_limit_branches(self) -> bool: + if self.limits["max_branches"] == -1 or len(self.branches) < self.limits["max_branches"]: + return True + log.info("branch limit exceeded (%s/%s)", len(self.branches), self.limits["max_branches"]) + return False + + def check_limit_endpoints(self) -> bool: + if ( + self.limits["max_read_only_endpoints"] == -1 + or self.read_only_endpoints_total < self.limits["max_read_only_endpoints"] + ): + return True + log.info( + "Maximum read only endpoint limit exceeded (%s/%s)", + self.read_only_endpoints_total, + self.limits["max_read_only_endpoints"], + ) + return False + def create_branch( self, parent_id: str | None = None, parent_timestamp: datetime | None = None ) -> NeonBranch | None: self.wait() + if not self.check_limit_branches(): + return None parent_timestamp_str: str | None = None if parent_timestamp: parent_timestamp_str = parent_timestamp.isoformat().replace("+00:00", "Z") - try: - branch_def = self.neon_api.create_branch( + branch_def = self.neon_api.create_branch( self.id, parent_id=parent_id, parent_timestamp=parent_timestamp_str ) - except HTTPError as he: - if ( - he.response.status_code == 422 - and he.response.json()["code"] == "BRANCHES_LIMIT_EXCEEDED" - ): - log.info("Branch limit exceeded, skipping") - return None - else: - raise HTTPError(he) from he new_branch = NeonBranch(self, branch_def) self.wait() return new_branch @@ -401,17 +421,10 @@ def do_action(project: NeonProject, action: str) -> bool: if action == "new_branch" or action == "new_branch_random_time": use_random_time: bool = action == "new_branch_random_time" log.info("Trying to create a new branch %s", "random time" if use_random_time else "") - if 0 <= project.limits["max_branches"] <= len(project.branches): - log.info( - "Maximum branch limit exceeded (%s of %s)", - len(project.branches), - project.limits["max_branches"], - ) - return False parent = project.branches[ random.choice(list(set(project.branches.keys()) - project.reset_branches)) ] - log.info("Parent: %s", parent) + child = parent.create_child_branch() if use_random_time: child = parent.create_child_branch_random_time() else: @@ -429,16 +442,11 @@ def do_action(project: NeonProject, action: str) -> bool: log.info("Leaf branches not found, skipping") return False elif action == "new_ro_endpoint": - if 0 <= project.limits["max_read_only_endpoints"] <= project.read_only_endpoints_total: - log.info( - "Maximum read only endpoint limit exceeded (%s of %s)", - project.read_only_endpoints_total, - project.limits["max_read_only_endpoints"], - ) - return False ep = random.choice( [br for br in project.branches.values() if br.id not in project.reset_branches] ).create_ro_endpoint() + if ep is None: + return False log.info("Created the RO endpoint with id %s branch: %s", ep.id, ep.branch.id) ep.start_benchmark() elif action == "delete_ro_endpoint": diff --git a/test_runner/regress/test_lfc_prewarm.py b/test_runner/regress/test_lfc_prewarm.py index 1fa1ead034..ae36bbda79 100644 --- a/test_runner/regress/test_lfc_prewarm.py +++ b/test_runner/regress/test_lfc_prewarm.py @@ -1,7 +1,6 @@ import random import threading from enum import StrEnum -from time import sleep from typing import Any import pytest @@ -20,28 +19,32 @@ class PrewarmMethod(StrEnum): PREWARM_LABEL = "compute_ctl_lfc_prewarms_total" +PREWARM_ERR_LABEL = "compute_ctl_lfc_prewarm_errors_total" OFFLOAD_LABEL = "compute_ctl_lfc_offloads_total" +OFFLOAD_ERR_LABEL = "compute_ctl_lfc_offload_errors_total" METHOD_VALUES = [e for e in PrewarmMethod] METHOD_IDS = [e.value for e in PrewarmMethod] def check_pinned_entries(cur: Cursor): - # some LFC buffer can be temporary locked by autovacuum or background writer - for _ in range(10): + """ + Wait till none of LFC buffers are pinned + """ + + def none_pinned(): cur.execute("select lfc_value from neon_lfc_stats where lfc_key='file_cache_chunks_pinned'") - n_pinned = cur.fetchall()[0][0] - if n_pinned == 0: - break - sleep(1) - assert n_pinned == 0 + assert cur.fetchall()[0][0] == 0 + + wait_until(none_pinned) def prom_parse(client: EndpointHttpClient) -> dict[str, float]: + labels = PREWARM_LABEL, OFFLOAD_LABEL, PREWARM_ERR_LABEL, OFFLOAD_ERR_LABEL return { - sample.name: sample.value + sample.name: int(sample.value) for family in prom_parse_impl(client.metrics()) for sample in family.samples - if sample.name in (PREWARM_LABEL, OFFLOAD_LABEL) + if sample.name in labels } @@ -54,7 +57,9 @@ def offload_lfc(method: PrewarmMethod, client: EndpointHttpClient, cur: Cursor) assert "error" not in status client.offload_lfc() assert client.prewarm_lfc_status()["status"] == "not_prewarmed" - assert prom_parse(client) == {OFFLOAD_LABEL: 1, PREWARM_LABEL: 0} + parsed = prom_parse(client) + desired = {OFFLOAD_LABEL: 1, PREWARM_LABEL: 0, OFFLOAD_ERR_LABEL: 0, PREWARM_ERR_LABEL: 0} + assert parsed == desired, f"{parsed=} != {desired=}" elif method == PrewarmMethod.POSTGRES: cur.execute("select get_local_cache_state()") return cur.fetchall()[0][0] @@ -81,12 +86,17 @@ def check_prewarmed( assert prom_parse(client)[PREWARM_LABEL] == 1 elif method == PrewarmMethod.COMPUTE_CTL: assert client.prewarm_lfc_status() == desired_status - assert prom_parse(client) == {OFFLOAD_LABEL: 0, PREWARM_LABEL: 1} + desired = {OFFLOAD_LABEL: 0, PREWARM_LABEL: 1, PREWARM_ERR_LABEL: 0, OFFLOAD_ERR_LABEL: 0} + assert prom_parse(client) == desired @pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping") @pytest.mark.parametrize("method", METHOD_VALUES, ids=METHOD_IDS) def test_lfc_prewarm(neon_simple_env: NeonEnv, method: PrewarmMethod): + """ + Test we can offload endpoint's LFC cache to endpoint storage. + Test we can prewarm endpoint with LFC cache loaded from endpoint storage. + """ env = neon_simple_env n_records = 1000000 cfg = [ @@ -140,18 +150,15 @@ def test_lfc_prewarm(neon_simple_env: NeonEnv, method: PrewarmMethod): lfc_used_pages = pg_cur.fetchall()[0][0] log.info(f"Used LFC size: {lfc_used_pages}") pg_cur.execute("select * from get_prewarm_info()") - prewarm_info = pg_cur.fetchall()[0] - log.info(f"Prewarm info: {prewarm_info}") - total, prewarmed, skipped, _ = prewarm_info + total, prewarmed, skipped, _ = pg_cur.fetchall()[0] + log.info(f"Prewarm info: {total=} {prewarmed=} {skipped=}") progress = (prewarmed + skipped) * 100 // total log.info(f"Prewarm progress: {progress}%") - assert lfc_used_pages > 10000 - assert ( - prewarm_info[0] > 0 - and prewarm_info[1] > 0 - and prewarm_info[0] == prewarm_info[1] + prewarm_info[2] - ) + assert total > 0 + assert prewarmed > 0 + assert total == prewarmed + skipped + lfc_cur.execute("select sum(pk) from t") assert lfc_cur.fetchall()[0][0] == n_records * (n_records + 1) / 2 @@ -168,6 +175,9 @@ WORKLOAD_IDS = METHOD_IDS[:-1] @pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping") @pytest.mark.parametrize("method", WORKLOAD_VALUES, ids=WORKLOAD_IDS) def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, method: PrewarmMethod): + """ + Test continiously prewarming endpoint when there is a write-heavy workload going in parallel + """ env = neon_simple_env n_records = 10000 n_threads = 4 @@ -247,5 +257,12 @@ def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, method: PrewarmMet assert total_balance == 0 check_pinned_entries(pg_cur) - if method != PrewarmMethod.POSTGRES: - assert prom_parse(http_client) == {OFFLOAD_LABEL: 1, PREWARM_LABEL: n_prewarms} + if method == PrewarmMethod.POSTGRES: + return + desired = { + OFFLOAD_LABEL: 1, + PREWARM_LABEL: n_prewarms, + OFFLOAD_ERR_LABEL: 0, + PREWARM_ERR_LABEL: 0, + } + assert prom_parse(http_client) == desired diff --git a/test_runner/regress/test_readonly_node.py b/test_runner/regress/test_readonly_node.py index ee934a900d..5612236250 100644 --- a/test_runner/regress/test_readonly_node.py +++ b/test_runner/regress/test_readonly_node.py @@ -201,11 +201,11 @@ def test_readonly_node_gc(neon_env_builder: NeonEnvBuilder): for shard, ps in tenant_get_shards(env, env.initial_tenant): client = ps.http_client() layers_guarded_before_gc = get_layers_protected_by_lease( - client, shard, env.initial_timeline, lease_lsn=lsn + client, shard, env.initial_timeline, lease_lsn=lease_lsn ) gc_result = client.timeline_gc(shard, env.initial_timeline, 0) layers_guarded_after_gc = get_layers_protected_by_lease( - client, shard, env.initial_timeline, lease_lsn=lsn + client, shard, env.initial_timeline, lease_lsn=lease_lsn ) # Note: cannot assert on `layers_removed` here because it could be layers diff --git a/test_runner/regress/test_tenant_size.py b/test_runner/regress/test_tenant_size.py index 190dd914ee..8b291b7cbe 100644 --- a/test_runner/regress/test_tenant_size.py +++ b/test_runner/regress/test_tenant_size.py @@ -740,6 +740,10 @@ def test_lsn_lease_size(neon_env_builder: NeonEnvBuilder, test_output_dir: Path, "pitr_interval": "0s" if zero_gc else "3600s", "gc_period": "0s", "compaction_period": "0s", + # The test exercises leases API, so we need non-zero lease length. + # If this tests ever does GC, we need to accomodate for the initial lease deadline + # after tenant attach, which is also controlled by this variable. + "lsn_lease_length": "600s", } env = neon_env_builder.init_start(initial_tenant_conf=conf) @@ -824,9 +828,7 @@ def insert_with_action( log.info(f"initial size: {initial_size}") with ep.cursor() as cur: - cur.execute( - "CREATE TABLE t0 AS SELECT i::bigint n FROM generate_series(0, 1000000) s(i)" - ) + cur.execute("CREATE TABLE t0 AS SELECT i::bigint n FROM generate_series(0, 10000) s(i)") last_flush_lsn = wait_for_last_flush_lsn(env, ep, tenant, timeline) if action == "lease": @@ -841,15 +843,9 @@ def insert_with_action( raise AssertionError("Invalid action type, only `lease` and `branch`are accepted") with ep.cursor() as cur: - cur.execute( - "CREATE TABLE t1 AS SELECT i::bigint n FROM generate_series(0, 1000000) s(i)" - ) - cur.execute( - "CREATE TABLE t2 AS SELECT i::bigint n FROM generate_series(0, 1000000) s(i)" - ) - cur.execute( - "CREATE TABLE t3 AS SELECT i::bigint n FROM generate_series(0, 1000000) s(i)" - ) + cur.execute("CREATE TABLE t1 AS SELECT i::bigint n FROM generate_series(0, 10000) s(i)") + cur.execute("CREATE TABLE t2 AS SELECT i::bigint n FROM generate_series(0, 10000) s(i)") + cur.execute("CREATE TABLE t3 AS SELECT i::bigint n FROM generate_series(0, 10000) s(i)") last_flush_lsn = wait_for_last_flush_lsn(env, ep, tenant, timeline)