From 2ff0a4ae828811051984e2f7521f4436e85aacad Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Thu, 16 Jan 2025 13:07:55 +0100 Subject: [PATCH] extract the l0stack generator into a reusable python module --- test_runner/deep_layers_with_delta.py | 118 +++++------------- test_runner/fixtures/common_types.py | 4 + test_runner/fixtures/neon_fixtures.py | 4 + .../pageserver/makelayers/__init__.py | 0 .../fixtures/pageserver/makelayers/l0stack.py | 104 +++++++++++++++ ...er_max_throughput_getpage_at_latest_lsn.py | 16 +-- 6 files changed, 149 insertions(+), 97 deletions(-) create mode 100644 test_runner/fixtures/pageserver/makelayers/__init__.py create mode 100644 test_runner/fixtures/pageserver/makelayers/l0stack.py diff --git a/test_runner/deep_layers_with_delta.py b/test_runner/deep_layers_with_delta.py index da2d6bcc35..d67fc9ca37 100644 --- a/test_runner/deep_layers_with_delta.py +++ b/test_runner/deep_layers_with_delta.py @@ -15,33 +15,28 @@ Measure impact of that. Some commands from shell history used in combination with this script: - NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=futures-unordered ./target/release/neon_local pageserver restart - NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=serial ./target/release/neon_local pageserver restart - NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=parallel ./target/release/neon_local pageserver restart - -cat >> .neon/pageserver_1/pageserver.toml <<"EOF" -# customizations -virtual_file_io_mode = "direct" -page_service_pipelining={mode="pipelined", max_batch_size=32, execution="concurrent-futures"} +``` + ./target/release/neon_local stop + rm -rf .neon + ./target/release/neon_local init + cat >> .neon/pageserver_1/pageserver.toml <<"EOF" + # customizations + virtual_file_io_mode = "direct" + page_service_pipelining={mode="pipelined", max_batch_size=32, execution="concurrent-futures"} EOF + ./target/release/neon_local start -starting from scrach: - psql 'postgresql://localhost:1235/storage_controller' -c 'DELETE FROM tenant_shards' - NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=... ./target/release/neon_local pageserver restart - ./target/debug/neon_local endpoint stop foo - rm -rf .neon/endpoints/foo - ./target/debug/neon_local endpoint create foo - echo 'full_page_writes=off' >> .neon/endpoints/foo/postgresql.conf - ./target/debug/neon_local endpoint start foo - non-package-mode-py3.10christian@neon-hetzner-dev-christian:[~/src/neon/test_runner]: poetry run python3 deep_layers_with_delta.py - Need 6400 pages, 38400 rows - 0 - modified rows 6400 - 1 - modified rows 6400 - ... - 19 - modified rows 6400 + psql 'postgresql://localhost:1235/storage_controller' -c 'DELETE FROM tenant_shards' + NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=futures-unordered ./target/release/neon_local pageserver restart + sleep 2 + ./target/release/neon_local tenant create --set-default + ./target/debug/neon_local endpoint stop foo + rm -rf .neon/endpoints/foo + ./target/debug/neon_local endpoint create foo + echo 'full_page_writes=off' >> .neon/endpoints/foo/postgresql.conf + ./target/debug/neon_local endpoint start foo + pushd test_runner; poetry run python3 deep_layers_with_delta.py; popd +``` Layer files created: @@ -77,82 +72,25 @@ Now use pagebench to measure perf, e.g. """ - -from pathlib import Path -import time -from fixtures.common_types import Lsn, TenantId, TenantShardId, TimelineId +import psycopg2 +from fixtures.common_types import TenantShardId, TimelineId from fixtures.pageserver.http import PageserverHttpClient +from fixtures.pageserver.makelayers.l0stack import L0StackShape, make_l0_stack_standalone ps_http = PageserverHttpClient(port=9898, is_testing_enabled_or_skip=lambda: None) vps_http = PageserverHttpClient(port=1234, is_testing_enabled_or_skip=lambda: None) tenants = ps_http.tenant_list() assert len(tenants) == 1 -tenant_id = tenants[0]["id"] +tenant_shard_id = TenantShardId.parse(tenants[0]["id"]) -timlines = ps_http.timeline_list(tenant_id) +timlines = ps_http.timeline_list(tenant_shard_id) assert len(timlines) == 1 -timeline_id = timlines[0]["timeline_id"] - -config = { - "gc_period": "0s", # disable periodic gc - "checkpoint_timeout": "10 years", - "compaction_period": "1h", # doesn't matter, but 0 value will kill walredo every 10s - "compaction_threshold": 100000, # we just want L0s - "compaction_target_size": 134217728, - "checkpoint_distance": 268435456, - "image_creation_threshold": 100000, # we just want L0s -} - -vps_http.set_tenant_config(tenant_id, config) +timeline_id = TimelineId(timlines[0]["timeline_id"]) connstr = "postgresql://cloud_admin@localhost:55432/postgres" - -import psycopg2 - -def last_record_lsn( - pageserver_http_client: PageserverHttpClient, - tenant: TenantId | TenantShardId, - timeline: TimelineId, -) -> Lsn: - detail = pageserver_http_client.timeline_detail(tenant, timeline) - - lsn_str = detail["last_record_lsn"] - assert isinstance(lsn_str, str) - return Lsn(lsn_str) - conn = psycopg2.connect(connstr) -conn.autocommit = True -cur = conn.cursor() -# each tuple is 23 (header) + 100 bytes = 123 bytes -# page header si 24 bytes -# 8k page size -# (8k-24bytes) / 123 bytes = 63 tuples per page -# set fillfactor to 10 to have 6 tuples per page -cur.execute("DROP TABLE IF EXISTS data") -cur.execute("CREATE TABLE data(id bigint, row char(92)) with (fillfactor=10)") -desired_size = 50 * 1024 * 1024 # 50MiB -need_pages = desired_size // 8192 -need_rows = need_pages * 6 -print(f"Need {need_pages} pages, {need_rows} rows") -cur.execute(f"INSERT INTO data SELECT i,'row'||i FROM generate_series(1, {need_rows}) as i") - -# every iteration updates one tuple in each page -delta_stack_height = 20 -for i in range( 0, delta_stack_height): - print(i) - cur.execute(f"UPDATE data set row = row||',u' where id % 6 = {i%6}") - print("modified rows", cur.rowcount) - cur.execute("SELECT pg_current_wal_flush_lsn()") - flush_lsn = Lsn(cur.fetchall()[0][0]) - - while True: - last_record = last_record_lsn(ps_http, tenant_id, timeline_id) - if last_record >= flush_lsn: - break - time.sleep(0.1) - - ps_http.timeline_checkpoint(tenant_id, timeline_id) - ps_http.timeline_compact(tenant_id, timeline_id) +shape = L0StackShape(logical_table_size_mib=50, delta_stack_height=20) +make_l0_stack_standalone(vps_http, ps_http, tenant_shard_id, timeline_id, conn, shape) diff --git a/test_runner/fixtures/common_types.py b/test_runner/fixtures/common_types.py index 6c22b31e00..c82c7578d1 100644 --- a/test_runner/fixtures/common_types.py +++ b/test_runner/fixtures/common_types.py @@ -208,6 +208,10 @@ class ShardIndex: shard_count=int(input[2:4], 16), ) + @property + def is_sharded(self) -> bool: + return self.shard_count != 0 + class TenantShardId: def __init__(self, tenant_id: TenantId, shard_number: int, shard_count: int): diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index 7d32449612..3dea732199 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -313,6 +313,10 @@ class PgProtocol: """ return self.safe_psql(query, log_query=log_query)[0][0] + def show_timeline_id(self) -> TimelineId: + """SHOW neon.timeline_id""" + return TimelineId(cast("str", self.safe_psql("show neon.timeline_id")[0][0])) + class PageserverWalReceiverProtocol(StrEnum): VANILLA = "vanilla" diff --git a/test_runner/fixtures/pageserver/makelayers/__init__.py b/test_runner/fixtures/pageserver/makelayers/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/test_runner/fixtures/pageserver/makelayers/l0stack.py b/test_runner/fixtures/pageserver/makelayers/l0stack.py new file mode 100644 index 0000000000..33267e3bb2 --- /dev/null +++ b/test_runner/fixtures/pageserver/makelayers/l0stack.py @@ -0,0 +1,104 @@ +from dataclasses import dataclass + +from psycopg2.extensions import connection as PgConnection + +from fixtures.common_types import Lsn, ShardIndex, TenantShardId, TimelineId +from fixtures.log_helper import log +from fixtures.neon_fixtures import Endpoint +from fixtures.pageserver.http import PageserverHttpClient +from fixtures.pageserver.utils import wait_for_last_record_lsn + + +@dataclass +class L0StackShape: + logical_table_size_mib: int = 50 + delta_stack_height: int = 20 + + +def make_l0_stack(endpoint: Endpoint, shape: L0StackShape): + env = endpoint.env + + # TDOO: wait for storcon to finish any reonciles before jumping to action here? + shards = env.storage_controller.tenant_describe(endpoint.tenant_id) + + shard_idx = ShardIndex.parse(shards[0]["shard_id"]) + assert shard_idx.shard_count == 1, "see above" + tenant_shard_id = TenantShardId( + endpoint.tenant_id, shard_idx.shard_number, shard_idx.shard_count + ) + + ps = env.get_pageserver(shards[0]["node_id"]) + + timeline_id = endpoint.show_timeline_id() + + vps_http = env.storage_controller.pageserver_api() + ps_http = ps.http_client() + endpoint_conn = endpoint.connect() + make_l0_stack_standalone(vps_http, ps_http, tenant_shard_id, timeline_id, endpoint_conn, shape) + + +def make_l0_stack_standalone( + vps_http: PageserverHttpClient, + ps_http: PageserverHttpClient, + tenant_shard_id: TenantShardId, + timeline_id: TimelineId, + endpoint_conn: PgConnection, + shape: L0StackShape, +): + """ + Creates stack of L0 deltas each of which should have 1 Value::Delta per page in `data`. + + Usable from scripts (ad-hoc manual testing in neon_local) and pytest code alike. + """ + + assert ( + not tenant_shard_id.shard_index.is_sharded + ), "the current implementation only supports unsharded tenants" + + tenant_id = tenant_shard_id.tenant_id + conn = endpoint_conn + desired_size = shape.logical_table_size_mib * 1024 * 1024 + + config = { + "gc_period": "0s", # disable periodic gc + "checkpoint_timeout": "10 years", + "compaction_period": "1h", # doesn't matter, but 0 value will kill walredo every 10s + "compaction_threshold": 100000, # we just want L0s + "compaction_target_size": 134217728, + "checkpoint_distance": 268435456, + "image_creation_threshold": 100000, # we just want L0s + } + + vps_http.set_tenant_config(tenant_id, config) + + conn.autocommit = True + cur = conn.cursor() + + # each tuple is 23 (header) + 100 bytes = 123 bytes + # page header si 24 bytes + # 8k page size + # (8k-24bytes) / 123 bytes = 63 tuples per page + # set fillfactor to 10 to have 6 tuples per page + cur.execute("DROP TABLE IF EXISTS data") + cur.execute("CREATE TABLE data(id bigint, row char(92)) with (fillfactor=10)") + need_pages = desired_size // 8192 + need_rows = need_pages * 6 + log.info(f"Need {need_pages} pages, {need_rows} rows") + cur.execute(f"INSERT INTO data SELECT i,'row'||i FROM generate_series(1, {need_rows}) as i") + + def settle_and_flush(): + cur.execute("SELECT pg_current_wal_flush_lsn()") + flush_lsn = Lsn(cur.fetchall()[0][0]) + wait_for_last_record_lsn(ps_http, tenant_shard_id, timeline_id, flush_lsn) + ps_http.timeline_checkpoint(tenant_id, timeline_id) + ps_http.timeline_compact(tenant_id, timeline_id) + + # create an L0 for the initial data we just inserted + settle_and_flush() + # every iteration updates one tuple in each page + delta_stack_height = shape.delta_stack_height + for i in range(0, delta_stack_height): + log.info(i) + cur.execute(f"UPDATE data set row = row||',u' where id % 6 = {i%6}") + log.info("modified rows", cur.rowcount) + settle_and_flush() 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 24d8384520..1e1734fd82 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 @@ -1,11 +1,11 @@ from __future__ import annotations -from dataclasses import dataclass import dataclasses import json -from pathlib import Path import sys import time +from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING import pytest @@ -238,7 +238,6 @@ def setup_tenant_template(env: NeonEnv, pg_bin: PgBin, scale: int): return (template_tenant, template_timeline, config) - def run_pagebench_benchmark( env: NeonEnv, pg_bin: PgBin, record, duration_secs: int, n_clients: int ): @@ -254,7 +253,7 @@ def run_pagebench_benchmark( pageserver_cpu_seconds_total: float pageserver_layers_visited_per_vectored_read_global_buckets: dict[int, int] - def __sub__(self, other: "Metrics") -> "Metrics": + def __sub__(self, other: Metrics) -> Metrics: return Metrics( time=self.time - other.time, pageserver_cpu_seconds_total=self.pageserver_cpu_seconds_total @@ -265,7 +264,7 @@ def run_pagebench_benchmark( }, ) - def normalize(self, by) -> "Metrics": + def normalize(self, by) -> Metrics: return Metrics( time=self.time / by, pageserver_cpu_seconds_total=self.pageserver_cpu_seconds_total / by, @@ -277,6 +276,7 @@ def run_pagebench_benchmark( def get_metrics() -> Metrics: pageserver_metrics = ps_http.get_metrics() + def parse_le_label(s): try: return int(s) @@ -285,6 +285,7 @@ def run_pagebench_benchmark( return sys.maxsize else: raise + return Metrics( time=time.time(), pageserver_cpu_seconds_total=pageserver_metrics.query_one( @@ -328,7 +329,6 @@ def run_pagebench_benchmark( metrics = metrics_after - metrics_before - # # Record the results # @@ -368,4 +368,6 @@ def run_pagebench_benchmark( report=MetricReport.TEST_PARAM, ) else: - record(f"counters.{metric}", metric_value=value, unit="", report=MetricReport.TEST_PARAM) + record( + f"counters.{metric}", metric_value=value, unit="", report=MetricReport.TEST_PARAM + )