Compare commits

..

24 Commits

Author SHA1 Message Date
Arthur Petukhovsky
7c7efe5537 Add walproposer vs vanilla replication test 2022-04-02 01:53:34 +04:00
bojanserafimov
af712798e7 Fix pageserver readme formatting
I put the diagram in a fixed-width block, since it wasn't rendering correctly on github.
2022-04-02 00:36:54 +03:00
Dmitry Ivanov
f5da652388 [proxy] Enable keepalives for all tcp connections (#1448) 2022-03-31 20:44:57 +03:00
Anastasia Lubennikova
8745b022a9 Extend LayerMap dump() function to print also open_layers and frozen_layers.
Add verbose option to chose if we need to print all layer's keys or not.
2022-03-31 17:26:24 +03:00
Arthur Petukhovsky
a40b7cd516 Fix timeouts in test_restarts_under_load (#1436)
* Enable backpressure in test_restarts_under_load

* Remove hacks because #644 is fixed now

* Adjust config in test_restarts_under_load
2022-03-31 17:00:09 +03:00
Konstantin Knizhnik
1aa8fe43cf Fix race condition in image layer (#1440)
* Fix race condition in image layer

refer #1439

* Add explicit drop(inner) in layer load method

* Add explicit drop(inner) in layer load method
2022-03-31 15:47:59 +03:00
Dmitry Rodionov
649f324fe3 make logging in basebackup more consistent 2022-03-30 17:58:51 +03:00
Dmitry Rodionov
8609234204 decrease the log level to debug because it is too noisy 2022-03-30 10:13:38 +03:00
Anton Shyrabokau
5c5629910f Add a test case for reading historic page versions (#1314)
* Add a test case for reading historic page versions

 Test read_page_at_lsn returns correct results when compared to page inspect.
 Validate possiblity of reading pages from dropped relation.
 Ensure funcitons read latest version when null lsn supplied.
 Check that functions do not poison buffer cache with stale page versions.
2022-03-29 22:13:06 -07:00
Kirill Bulatov
277e41f4b7 Show s3 spans in logs and improve the log messages 2022-03-29 19:21:31 +03:00
Arthur Petukhovsky
ce0243bc12 Add metric for last_record_lsn (#1430) 2022-03-29 18:54:24 +03:00
Arseny Sher
ec3bc74165 Add safekeeper information exchange through etcd.
Safekeers now publish to and pull from etcd per-timeline data. Immediate goal is
WAL truncation, for which every safekeeper must know remote_consistent_lsn; the
next would be callmemaybe replacement.

Adds corresponding '--broker' argument to safekeeper and ability to run etcd in
tests.

Adds test checking remote_consistent_lsn is indeed communicated.
2022-03-29 18:16:49 +04:00
Dmitry Rodionov
9594362f74 change python cache version to 2 (fixes python cache in circle CI) 2022-03-29 10:42:30 +03:00
Dmitry Rodionov
eee0f51e0c use cargo-hakari to manage workspace_hack crate
workspace_hack is needed to avoid recompilation when different crates
inside the workspace depend on the same packages but with different
features being enabled. Problem occurs when you build crates separately
one by one. So this is irrelevant to our CI setup because there we build
all binaries at once, but it may be relevant for local development.

this also changes cargo's resolver version to 2
2022-03-29 10:42:04 +03:00
Arthur Petukhovsky
fd78110c2b Add default statement_timeout for tests (#1423) 2022-03-29 09:57:00 +03:00
Anton Shyrabokau
be6a6958e2 CI: rebuild postgres when Makefile changes (#1429) 2022-03-28 18:19:20 -07:00
Kirill Bulatov
0e44887929 Show more S3 logs and less verbove WAL logs 2022-03-29 00:36:06 +03:00
Dhammika Pathirana
1aa57fc262 Fix tone down compact log chatter
Signed-off-by: Dhammika Pathirana <dhammika@gmail.com>
2022-03-28 13:24:13 -07:00
Alexey Kondratov
9a4f0930c0 Turn off S3 for pageserver on staging 2022-03-28 14:14:17 -05:00
Alexey Kondratov
d88f8b4a7e Fix storage deploy condition in ansible playbook 2022-03-28 13:30:40 -05:00
Arthur Petukhovsky
8a901de52a Refactor control file update at safekeeper.
Record global_commit_lsn, have common routine for control file update, add
SafekeeperMemstate.
2022-03-28 21:52:12 +04:00
Alexey Kondratov
a883202495 Enable S3 for pageserver on staging
Follow-up for #1417. Previously we had a problem uploading to S3
due to huge ammount of existing not yet uploaded data. Now we have a
fresh pageserver with LSM storage on staging, so we can try enabling it
once again.
2022-03-28 12:04:40 -05:00
Arseny Sher
780b46ad27 Bump vendor/postgres to fix commit_lsn going backwards. 2022-03-28 20:37:33 +04:00
Arseny Sher
75002adc14 Make shared_buffers large in test_pageserver_catchup.
We intentionally write while pageserver is down, so we shouldn't query it.

Noticed by @petuhovskiy at
https://github.com/zenithdb/postgres/pull/141#issuecomment-1080261700
2022-03-28 20:34:06 +04:00
57 changed files with 1679 additions and 576 deletions

View File

@@ -63,9 +63,10 @@
tags:
- pageserver
# Temporary disabled until LSM storage rewrite lands
# - name: update config
# when: current_version > remote_version or force_deploy
# It seems that currently S3 integration does not play well
# even with fresh pageserver without a burden of old data.
# TODO: turn this back on once the issue is solved.
# - name: update remote storage (s3) config
# lineinfile:
# path: /storage/pageserver/data/pageserver.toml
# line: "{{ item }}"

View File

@@ -34,10 +34,13 @@ jobs:
- checkout
# Grab the postgres git revision to build a cache key.
# Append makefile as it could change the way postgres is built.
# Note this works even though the submodule hasn't been checkout out yet.
- run:
name: Get postgres cache key
command: git rev-parse HEAD:vendor/postgres > /tmp/cache-key-postgres
command: |
git rev-parse HEAD:vendor/postgres > /tmp/cache-key-postgres
cat Makefile >> /tmp/cache-key-postgres
- restore_cache:
name: Restore postgres cache
@@ -78,11 +81,14 @@ jobs:
- checkout
# Grab the postgres git revision to build a cache key.
# Append makefile as it could change the way postgres is built.
# Note this works even though the submodule hasn't been checkout out yet.
- run:
name: Get postgres cache key
command: |
git rev-parse HEAD:vendor/postgres > /tmp/cache-key-postgres
cat Makefile >> /tmp/cache-key-postgres
- restore_cache:
name: Restore postgres cache
@@ -222,12 +228,12 @@ jobs:
- checkout
- restore_cache:
keys:
- v1-python-deps-{{ checksum "poetry.lock" }}
- v2-python-deps-{{ checksum "poetry.lock" }}
- run:
name: Install deps
command: ./scripts/pysync
- save_cache:
key: v1-python-deps-{{ checksum "poetry.lock" }}
key: v2-python-deps-{{ checksum "poetry.lock" }}
paths:
- /home/circleci/.cache/pypoetry/virtualenvs
- run:
@@ -281,12 +287,12 @@ jobs:
- run: git submodule update --init --depth 1
- restore_cache:
keys:
- v1-python-deps-{{ checksum "poetry.lock" }}
- v2-python-deps-{{ checksum "poetry.lock" }}
- run:
name: Install deps
command: ./scripts/pysync
- save_cache:
key: v1-python-deps-{{ checksum "poetry.lock" }}
key: v2-python-deps-{{ checksum "poetry.lock" }}
paths:
- /home/circleci/.cache/pypoetry/virtualenvs
- run:

24
.config/hakari.toml Normal file
View File

@@ -0,0 +1,24 @@
# This file contains settings for `cargo hakari`.
# See https://docs.rs/cargo-hakari/latest/cargo_hakari/config for a full list of options.
hakari-package = "workspace_hack"
# Format for `workspace-hack = ...` lines in other Cargo.tomls. Requires cargo-hakari 0.9.8 or above.
dep-format-version = "2"
# Setting workspace.resolver = "2" in the root Cargo.toml is HIGHLY recommended.
# Hakari works much better with the new feature resolver.
# For more about the new feature resolver, see:
# https://blog.rust-lang.org/2021/03/25/Rust-1.51.0.html#cargos-new-feature-resolver
resolver = "2"
# Add triples corresponding to platforms commonly used by developers here.
# https://doc.rust-lang.org/rustc/platform-support.html
platforms = [
# "x86_64-unknown-linux-gnu",
# "x86_64-apple-darwin",
# "x86_64-pc-windows-msvc",
]
# Write out exact versions rather than a semver range. (Defaults to false.)
# exact-versions = true

368
Cargo.lock generated
View File

@@ -61,18 +61,6 @@ dependencies = [
"backtrace",
]
[[package]]
name = "as-slice"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "45403b49e3954a4b8428a0ac21a4b7afadccf92bfd96273f1a58cd4812496ae0"
dependencies = [
"generic-array 0.12.4",
"generic-array 0.13.3",
"generic-array 0.14.5",
"stable_deref_trait",
]
[[package]]
name = "async-compression"
version = "0.3.12"
@@ -87,6 +75,27 @@ dependencies = [
"zstd-safe",
]
[[package]]
name = "async-stream"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "171374e7e3b2504e0e5236e3b59260560f9fe94bfe9ac39ba5e4e929c5590625"
dependencies = [
"async-stream-impl",
"futures-core",
]
[[package]]
name = "async-stream-impl"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "648ed8c8d2ce5409ccd57453d9d1b214b342a0d69376a6feda1fd6cae3299308"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "async-trait"
version = "0.1.52"
@@ -252,7 +261,7 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4152116fd6e9dadb291ae18fc1ec3575ed6d84c29642d97890f4b4a3417297e4"
dependencies = [
"generic-array 0.14.5",
"generic-array",
]
[[package]]
@@ -419,6 +428,7 @@ dependencies = [
"serde_json",
"tar",
"tokio",
"workspace_hack",
]
[[package]]
@@ -567,7 +577,7 @@ version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bff07008ec701e8028e2ceb8f83f0e4274ee62bd2dbdc4fefff2e9a91824081a"
dependencies = [
"generic-array 0.14.5",
"generic-array",
"subtle",
]
@@ -577,7 +587,7 @@ version = "0.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1d1a86f49236c215f271d40892d5fc950490551400b02ef360692c29815c714"
dependencies = [
"generic-array 0.14.5",
"generic-array",
"subtle",
]
@@ -654,7 +664,7 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3dd60d1080a57a05ab032377049e0591415d2b31afd7028356dbf3cc6dcb066"
dependencies = [
"generic-array 0.14.5",
"generic-array",
]
[[package]]
@@ -714,6 +724,21 @@ dependencies = [
"termcolor",
]
[[package]]
name = "etcd-client"
version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "585de5039d1ecce74773db49ba4e8107e42be7c2cd0b1a9e7fce27181db7b118"
dependencies = [
"http",
"prost",
"tokio",
"tokio-stream",
"tonic",
"tonic-build",
"tower-service",
]
[[package]]
name = "fail"
version = "0.5.0"
@@ -752,6 +777,12 @@ dependencies = [
"winapi",
]
[[package]]
name = "fixedbitset"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "279fb028e20b3c4c320317955b77c5e0c9701f05a1d309905d6fc702cdc5053e"
[[package]]
name = "fnv"
version = "1.0.7"
@@ -867,24 +898,6 @@ dependencies = [
"slab",
]
[[package]]
name = "generic-array"
version = "0.12.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ffdf9f34f1447443d37393cc6c2b8313aebddcd96906caf34e54c68d8e57d7bd"
dependencies = [
"typenum",
]
[[package]]
name = "generic-array"
version = "0.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f797e67af32588215eaaab8327027ee8e71b9dd0b2b26996aedf20c030fce309"
dependencies = [
"typenum",
]
[[package]]
name = "generic-array"
version = "0.14.5"
@@ -903,7 +916,7 @@ checksum = "418d37c8b1d42553c93648be529cb70f920d3baf8ef469b74b9638df426e0b4c"
dependencies = [
"cfg-if",
"libc",
"wasi",
"wasi 0.10.0+wasi-snapshot-preview1",
]
[[package]]
@@ -955,7 +968,7 @@ dependencies = [
"indexmap",
"slab",
"tokio",
"tokio-util",
"tokio-util 0.6.9",
"tracing",
]
@@ -965,15 +978,6 @@ version = "1.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eabb4a44450da02c90444cf74558da904edde8fb4e9035a9a6a4e15445af0bd7"
[[package]]
name = "hash32"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d4041af86e63ac4298ce40e5cca669066e75b6f1aa3390fe2561ffa5e1d9f4cc"
dependencies = [
"byteorder",
]
[[package]]
name = "hashbrown"
version = "0.9.1"
@@ -993,15 +997,12 @@ dependencies = [
]
[[package]]
name = "heapless"
version = "0.6.1"
name = "heck"
version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "634bd4d29cbf24424d0a4bfcbf80c6960129dc24424752a7d1d1390607023422"
checksum = "6d621efb26863f0e9924c6ac577e8275e5e6b77455db64ffa6c65c904e9e132c"
dependencies = [
"as-slice",
"generic-array 0.14.5",
"hash32",
"stable_deref_trait",
"unicode-segmentation",
]
[[package]]
@@ -1125,6 +1126,18 @@ dependencies = [
"tokio-rustls 0.23.2",
]
[[package]]
name = "hyper-timeout"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1"
dependencies = [
"hyper",
"pin-project-lite",
"tokio",
"tokio-io-timeout",
]
[[package]]
name = "ident_case"
version = "1.0.1"
@@ -1358,14 +1371,15 @@ dependencies = [
[[package]]
name = "mio"
version = "0.7.14"
version = "0.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8067b404fe97c70829f082dec8bcf4f71225d7eaea1d8645349cb76fa06205cc"
checksum = "52da4364ffb0e4fe33a9841a98a3f3014fb964045ce4f7a45a398243c8d6b0c9"
dependencies = [
"libc",
"log",
"miow",
"ntapi",
"wasi 0.11.0+wasi-snapshot-preview1",
"winapi",
]
@@ -1378,6 +1392,12 @@ dependencies = [
"winapi",
]
[[package]]
name = "multimap"
version = "0.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a"
[[package]]
name = "nix"
version = "0.23.1"
@@ -1422,17 +1442,6 @@ dependencies = [
"num-traits",
]
[[package]]
name = "num-bigint"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f93ab6289c7b344a8a9f60f88d80aa20032336fe78da341afc91c8a2341fc75f"
dependencies = [
"autocfg",
"num-integer",
"num-traits",
]
[[package]]
name = "num-integer"
version = "0.1.44"
@@ -1533,8 +1542,6 @@ dependencies = [
"lazy_static",
"log",
"nix",
"num-bigint 0.4.3",
"num-traits",
"once_cell",
"postgres 0.19.1 (git+https://github.com/zenithdb/rust-postgres.git?rev=2949d98df52587d562986aad155dd4e889e408b7)",
"postgres-protocol 0.6.1 (git+https://github.com/zenithdb/rust-postgres.git?rev=2949d98df52587d562986aad155dd4e889e408b7)",
@@ -1542,7 +1549,6 @@ dependencies = [
"postgres_ffi",
"rand",
"regex",
"rstar",
"rust-s3",
"scopeguard",
"serde",
@@ -1589,12 +1595,6 @@ dependencies = [
"winapi",
]
[[package]]
name = "pdqselect"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ec91767ecc0a0bbe558ce8c9da33c068066c57ecc8bb8477ef8c1ad3ef77c27"
[[package]]
name = "peeking_take_while"
version = "0.1.2"
@@ -1627,6 +1627,16 @@ version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
[[package]]
name = "petgraph"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a13a2fa9d0b63e5f22328828741e523766fff0ee9e779316902290dff3f824f"
dependencies = [
"fixedbitset",
"indexmap",
]
[[package]]
name = "phf"
version = "0.8.0"
@@ -1846,6 +1856,59 @@ dependencies = [
"thiserror",
]
[[package]]
name = "prost"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "444879275cb4fd84958b1a1d5420d15e6fcf7c235fe47f053c9c2a80aceb6001"
dependencies = [
"bytes",
"prost-derive",
]
[[package]]
name = "prost-build"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "62941722fb675d463659e49c4f3fe1fe792ff24fe5bbaa9c08cd3b98a1c354f5"
dependencies = [
"bytes",
"heck",
"itertools",
"lazy_static",
"log",
"multimap",
"petgraph",
"prost",
"prost-types",
"regex",
"tempfile",
"which",
]
[[package]]
name = "prost-derive"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f9cc1a3263e07e0bf68e96268f37665207b49560d98739662cdfaae215c720fe"
dependencies = [
"anyhow",
"itertools",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "prost-types"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "534b7a0e836e3c482d2693070f982e39e7611da9695d4d1f5a4b186b51faef0a"
dependencies = [
"bytes",
"prost",
]
[[package]]
name = "proxy"
version = "0.1.0"
@@ -1869,11 +1932,13 @@ dependencies = [
"scopeguard",
"serde",
"serde_json",
"socket2",
"thiserror",
"tokio",
"tokio-postgres 0.7.1 (git+https://github.com/zenithdb/rust-postgres.git?rev=2949d98df52587d562986aad155dd4e889e408b7)",
"tokio-postgres-rustls",
"tokio-rustls 0.22.0",
"workspace_hack",
"zenith_metrics",
"zenith_utils",
]
@@ -2048,7 +2113,7 @@ dependencies = [
"serde_urlencoded",
"tokio",
"tokio-rustls 0.23.2",
"tokio-util",
"tokio-util 0.6.9",
"url",
"wasm-bindgen",
"wasm-bindgen-futures",
@@ -2085,18 +2150,6 @@ dependencies = [
"regex",
]
[[package]]
name = "rstar"
version = "0.9.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fc6fc513b8c3853e43a0c3f909ded14ffa82e5170c9c5f6fb175f9c85c8a433"
dependencies = [
"heapless",
"num-traits",
"pdqselect",
"smallvec",
]
[[package]]
name = "rust-ini"
version = "0.17.0"
@@ -2396,7 +2449,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "692ca13de57ce0613a363c8c2f1de925adebc81b04c923ac60c5488bb44abe4b"
dependencies = [
"chrono",
"num-bigint 0.2.6",
"num-bigint",
"num-traits",
]
@@ -2434,12 +2487,6 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d"
[[package]]
name = "stable_deref_trait"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
[[package]]
name = "stringprep"
version = "0.1.2"
@@ -2564,7 +2611,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6db9e6914ab8b1ae1c260a4ae7a49b6c5611b40328a735b21862567685e73255"
dependencies = [
"libc",
"wasi",
"wasi 0.10.0+wasi-snapshot-preview1",
"winapi",
]
@@ -2595,9 +2642,9 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
[[package]]
name = "tokio"
version = "1.16.1"
version = "1.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c27a64b625de6d309e8c57716ba93021dccf1b3b5c97edd6d3dd2d2135afc0a"
checksum = "2af73ac49756f3f7c01172e34a23e5d0216f6c32333757c2c61feb2bbff5a5ee"
dependencies = [
"bytes",
"libc",
@@ -2607,10 +2654,21 @@ dependencies = [
"once_cell",
"pin-project-lite",
"signal-hook-registry",
"socket2",
"tokio-macros",
"winapi",
]
[[package]]
name = "tokio-io-timeout"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "30b74022ada614a1b4834de765f9bb43877f910cc8ce4be40e89042c9223a8bf"
dependencies = [
"pin-project-lite",
"tokio",
]
[[package]]
name = "tokio-macros"
version = "1.7.0"
@@ -2641,7 +2699,7 @@ dependencies = [
"postgres-types 0.2.1 (git+https://github.com/zenithdb/rust-postgres.git?rev=2949d98df52587d562986aad155dd4e889e408b7)",
"socket2",
"tokio",
"tokio-util",
"tokio-util 0.6.9",
]
[[package]]
@@ -2663,7 +2721,7 @@ dependencies = [
"postgres-types 0.2.1 (git+https://github.com/zenithdb/rust-postgres.git?rev=9eb0dbfbeb6a6c1b79099b9f7ae4a8c021877858)",
"socket2",
"tokio",
"tokio-util",
"tokio-util 0.6.9",
]
[[package]]
@@ -2728,6 +2786,20 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-util"
version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "64910e1b9c1901aaf5375561e35b9c057d95ff41a44ede043a03e09279eabaf1"
dependencies = [
"bytes",
"futures-core",
"futures-sink",
"log",
"pin-project-lite",
"tokio",
]
[[package]]
name = "toml"
version = "0.5.8"
@@ -2750,6 +2822,75 @@ dependencies = [
"serde",
]
[[package]]
name = "tonic"
version = "0.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff08f4649d10a70ffa3522ca559031285d8e421d727ac85c60825761818f5d0a"
dependencies = [
"async-stream",
"async-trait",
"base64 0.13.0",
"bytes",
"futures-core",
"futures-util",
"h2",
"http",
"http-body",
"hyper",
"hyper-timeout",
"percent-encoding",
"pin-project",
"prost",
"prost-derive",
"tokio",
"tokio-stream",
"tokio-util 0.6.9",
"tower",
"tower-layer",
"tower-service",
"tracing",
"tracing-futures",
]
[[package]]
name = "tonic-build"
version = "0.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9403f1bafde247186684b230dc6f38b5cd514584e8bec1dd32514be4745fa757"
dependencies = [
"proc-macro2",
"prost-build",
"quote",
"syn",
]
[[package]]
name = "tower"
version = "0.4.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a89fd63ad6adf737582df5db40d286574513c69a11dac5214dc3b5603d6713e"
dependencies = [
"futures-core",
"futures-util",
"indexmap",
"pin-project",
"pin-project-lite",
"rand",
"slab",
"tokio",
"tokio-util 0.7.0",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "tower-layer"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "343bc9466d3fe6b0f960ef45960509f84480bf4fd96f92901afe7ff3df9d3a62"
[[package]]
name = "tower-service"
version = "0.3.1"
@@ -2763,6 +2904,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2d8d93354fe2a8e50d5953f5ae2e47a3fc2ef03292e7ea46e3cc38f549525fb9"
dependencies = [
"cfg-if",
"log",
"pin-project-lite",
"tracing-attributes",
"tracing-core",
@@ -2855,6 +2997,12 @@ dependencies = [
"tinyvec",
]
[[package]]
name = "unicode-segmentation"
version = "1.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7e8820f5d777f6224dc4be3632222971ac30164d4a258d595640799554ebfd99"
[[package]]
name = "unicode-width"
version = "0.1.9"
@@ -2925,6 +3073,7 @@ dependencies = [
"const_format",
"crc32c",
"daemonize",
"etcd-client",
"fs2",
"hex",
"humantime",
@@ -2937,11 +3086,13 @@ dependencies = [
"rust-s3",
"serde",
"serde_json",
"serde_with",
"signal-hook",
"tempfile",
"tokio",
"tokio-postgres 0.7.1 (git+https://github.com/zenithdb/rust-postgres.git?rev=2949d98df52587d562986aad155dd4e889e408b7)",
"tracing",
"url",
"walkdir",
"workspace_hack",
"zenith_metrics",
@@ -2964,6 +3115,12 @@ version = "0.10.0+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a143597ca7c7793eff794def352d41792a93c481eb1042423ff7ff72ba2c31f"
[[package]]
name = "wasi"
version = "0.11.0+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
[[package]]
name = "wasm-bindgen"
version = "0.2.79"
@@ -3130,7 +3287,14 @@ dependencies = [
name = "workspace_hack"
version = "0.1.0"
dependencies = [
"anyhow",
"bytes",
"cc",
"clap 2.34.0",
"either",
"hashbrown 0.11.2",
"libc",
"log",
"memchr",
"num-integer",
"num-traits",
@@ -3138,8 +3302,13 @@ dependencies = [
"quote",
"regex",
"regex-syntax",
"reqwest",
"scopeguard",
"serde",
"syn",
"tokio",
"tracing",
"tracing-core",
]
[[package]]
@@ -3190,6 +3359,7 @@ dependencies = [
"libc",
"once_cell",
"prometheus",
"workspace_hack",
]
[[package]]

View File

@@ -11,6 +11,7 @@ members = [
"zenith_metrics",
"zenith_utils",
]
resolver = "2"
[profile.release]
# This is useful for profiling and, to some extent, debug.

View File

@@ -78,6 +78,11 @@ postgres: postgres-configure \
$(MAKE) -C tmp_install/build/contrib/zenith install
+@echo "Compiling contrib/zenith_test_utils"
$(MAKE) -C tmp_install/build/contrib/zenith_test_utils install
+@echo "Compiling pg_buffercache"
$(MAKE) -C tmp_install/build/contrib/pg_buffercache install
+@echo "Compiling pageinspect"
$(MAKE) -C tmp_install/build/contrib/pageinspect install
.PHONY: postgres-clean
postgres-clean:

View File

@@ -16,4 +16,5 @@ regex = "1"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1"
tar = "0.4"
tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread"] }
tokio = { version = "1.17", features = ["macros", "rt", "rt-multi-thread"] }
workspace_hack = { version = "0.1", path = "../workspace_hack" }

View File

@@ -20,4 +20,4 @@ reqwest = { version = "0.11", default-features = false, features = ["blocking",
pageserver = { path = "../pageserver" }
walkeeper = { path = "../walkeeper" }
zenith_utils = { path = "../zenith_utils" }
workspace_hack = { path = "../workspace_hack" }
workspace_hack = { version = "0.1", path = "../workspace_hack" }

View File

@@ -57,6 +57,10 @@ pub struct LocalEnv {
#[serde(default)]
pub private_key_path: PathBuf,
// A comma separated broker (etcd) endpoints for storage nodes coordination, e.g. 'http://127.0.0.1:2379'.
#[serde(default)]
pub broker_endpoints: Option<String>,
pub pageserver: PageServerConf,
#[serde(default)]

View File

@@ -73,6 +73,8 @@ pub struct SafekeeperNode {
pub http_base_url: String,
pub pageserver: Arc<PageServerNode>,
broker_endpoints: Option<String>,
}
impl SafekeeperNode {
@@ -89,6 +91,7 @@ impl SafekeeperNode {
http_client: Client::new(),
http_base_url: format!("http://127.0.0.1:{}/v1", conf.http_port),
pageserver,
broker_endpoints: env.broker_endpoints.clone(),
}
}
@@ -135,6 +138,9 @@ impl SafekeeperNode {
if !self.conf.sync {
cmd.arg("--no-sync");
}
if let Some(ref ep) = self.broker_endpoints {
cmd.args(&["--broker-endpoints", ep]);
}
if !cmd.status()?.success() {
bail!(

View File

@@ -67,6 +67,8 @@ For more detailed info, see `/walkeeper/README`
`/workspace_hack`:
The workspace_hack crate exists only to pin down some dependencies.
We use [cargo-hakari](https://crates.io/crates/cargo-hakari) for automation.
`/zenith`
Main entry point for the 'zenith' CLI utility.

View File

@@ -17,7 +17,7 @@ lazy_static = "1.4.0"
log = "0.4.14"
clap = "3.0"
daemonize = "0.4.1"
tokio = { version = "1.11", features = ["process", "sync", "macros", "fs", "rt", "io-util", "time"] }
tokio = { version = "1.17", features = ["process", "sync", "macros", "fs", "rt", "io-util", "time"] }
postgres-types = { git = "https://github.com/zenithdb/rust-postgres.git", rev="2949d98df52587d562986aad155dd4e889e408b7" }
postgres-protocol = { git = "https://github.com/zenithdb/rust-postgres.git", rev="2949d98df52587d562986aad155dd4e889e408b7" }
postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="2949d98df52587d562986aad155dd4e889e408b7" }
@@ -44,9 +44,6 @@ nix = "0.23"
once_cell = "1.8.0"
crossbeam-utils = "0.8.5"
fail = "0.5.0"
rstar = "0.9.2"
num-traits = "0.2.14"
num-bigint = "0.4.3"
rust-s3 = { version = "0.28", default-features = false, features = ["no-verify-ssl", "tokio-rustls-tls"] }
async-compression = {version = "0.3", features = ["zstd", "tokio"]}
@@ -54,7 +51,7 @@ async-compression = {version = "0.3", features = ["zstd", "tokio"]}
postgres_ffi = { path = "../postgres_ffi" }
zenith_metrics = { path = "../zenith_metrics" }
zenith_utils = { path = "../zenith_utils" }
workspace_hack = { path = "../workspace_hack" }
workspace_hack = { version = "0.1", path = "../workspace_hack" }
[dev-dependencies]
hex-literal = "0.3"

View File

@@ -13,7 +13,7 @@ keeps track of WAL records which are not synced to S3 yet.
The Page Server consists of multiple threads that operate on a shared
repository of page versions:
```
| WAL
V
+--------------+
@@ -46,7 +46,7 @@ Legend:
---> Data flow
<---
```
Page Service
------------

View File

@@ -65,6 +65,7 @@ impl<'a> Basebackup<'a> {
// prev_lsn to Lsn(0) if we cannot provide the correct value.
let (backup_prev, backup_lsn) = if let Some(req_lsn) = req_lsn {
// Backup was requested at a particular LSN. Wait for it to arrive.
info!("waiting for {}", req_lsn);
timeline.tline.wait_lsn(req_lsn)?;
// If the requested point is the end of the timeline, we can

View File

@@ -25,7 +25,7 @@ fn main() -> Result<()> {
// Basic initialization of things that don't change after startup
virtual_file::init(10);
dump_layerfile_from_path(&path)?;
dump_layerfile_from_path(&path, true)?;
Ok(())
}

View File

@@ -41,7 +41,7 @@ pub mod defaults {
pub const DEFAULT_WAL_REDO_TIMEOUT: &str = "60 s";
pub const DEFAULT_SUPERUSER: &str = "zenith_admin";
pub const DEFAULT_REMOTE_STORAGE_MAX_CONCURRENT_SYNC: usize = 100;
pub const DEFAULT_REMOTE_STORAGE_MAX_CONCURRENT_SYNC: usize = 10;
pub const DEFAULT_REMOTE_STORAGE_MAX_SYNC_ERRORS: u32 = 10;
pub const DEFAULT_PAGE_CACHE_SIZE: usize = 8192;

View File

@@ -48,7 +48,9 @@ use crate::walredo::WalRedoManager;
use crate::CheckpointConfig;
use crate::{ZTenantId, ZTimelineId};
use zenith_metrics::{register_histogram_vec, Histogram, HistogramVec};
use zenith_metrics::{
register_histogram_vec, register_int_gauge_vec, Histogram, HistogramVec, IntGauge, IntGaugeVec,
};
use zenith_utils::crashsafe_dir;
use zenith_utils::lsn::{AtomicLsn, Lsn, RecordLsn};
use zenith_utils::seqwait::SeqWait;
@@ -95,6 +97,15 @@ lazy_static! {
.expect("failed to define a metric");
}
lazy_static! {
static ref LAST_RECORD_LSN: IntGaugeVec = register_int_gauge_vec!(
"pageserver_last_record_lsn",
"Last record LSN grouped by timeline",
&["tenant_id", "timeline_id"]
)
.expect("failed to define a metric");
}
/// Parts of the `.zenith/tenants/<tenantid>/timelines/<timelineid>` directory prefix.
pub const TIMELINES_SEGMENT_NAME: &str = "timelines";
@@ -745,11 +756,12 @@ pub struct LayeredTimeline {
ancestor_timeline: Option<LayeredTimelineEntry>,
ancestor_lsn: Lsn,
// Metrics histograms
// Metrics
reconstruct_time_histo: Histogram,
flush_time_histo: Histogram,
compact_time_histo: Histogram,
create_images_time_histo: Histogram,
last_record_gauge: IntGauge,
/// If `true`, will backup its files that appear after each checkpointing to the remote storage.
upload_layers: AtomicBool,
@@ -982,6 +994,9 @@ impl LayeredTimeline {
&timelineid.to_string(),
])
.unwrap();
let last_record_gauge = LAST_RECORD_LSN
.get_metric_with_label_values(&[&tenantid.to_string(), &timelineid.to_string()])
.unwrap();
LayeredTimeline {
conf,
@@ -1007,6 +1022,7 @@ impl LayeredTimeline {
flush_time_histo,
compact_time_histo,
create_images_time_histo,
last_record_gauge,
upload_layers: AtomicBool::new(upload_layers),
@@ -1325,6 +1341,7 @@ impl LayeredTimeline {
fn finish_write(&self, new_lsn: Lsn) {
assert!(new_lsn.is_aligned());
self.last_record_gauge.set(new_lsn.0 as i64);
self.last_record_lsn.advance(new_lsn);
}
@@ -1594,7 +1611,7 @@ impl LayeredTimeline {
self.compact_level0(target_file_size)?;
timer.stop_and_record();
} else {
info!("Could not compact because no partitioning specified yet");
debug!("Could not compact because no partitioning specified yet");
}
// Call unload() on all frozen layers, to release memory.
@@ -1629,7 +1646,7 @@ impl LayeredTimeline {
let num_deltas = layers.count_deltas(&img_range, &(img_lsn..lsn))?;
info!(
debug!(
"range {}-{}, has {} deltas on this timeline",
img_range.start, img_range.end, num_deltas
);
@@ -1928,7 +1945,7 @@ impl LayeredTimeline {
l.filename().display(),
l.is_incremental(),
);
layers_to_remove.push(Arc::clone(&l));
layers_to_remove.push(Arc::clone(l));
}
// Actually delete the layers from disk and remove them from the map.
@@ -2049,16 +2066,16 @@ impl<'a> TimelineWriter<'_> for LayeredTimelineWriter<'a> {
}
/// Dump contents of a layer file to stdout.
pub fn dump_layerfile_from_path(path: &Path) -> Result<()> {
pub fn dump_layerfile_from_path(path: &Path, verbose: bool) -> Result<()> {
let file = File::open(path)?;
let book = Book::new(file)?;
match book.magic() {
crate::DELTA_FILE_MAGIC => {
DeltaLayer::new_for_path(path, &book)?.dump()?;
DeltaLayer::new_for_path(path, &book)?.dump(verbose)?;
}
crate::IMAGE_FILE_MAGIC => {
ImageLayer::new_for_path(path, &book)?.dump()?;
ImageLayer::new_for_path(path, &book)?.dump(verbose)?;
}
magic => bail!("unrecognized magic identifier: {:?}", magic),
}
@@ -2199,7 +2216,7 @@ pub mod tests {
let mut test_key = Key::from_hex("012222222233333333444444445500000000").unwrap();
let mut blknum = 0;
for _ in 0..50 {
for _ in 0..1000 {
for _ in 0..10000 {
test_key.field6 = blknum;
let writer = tline.writer();
writer.put(

View File

@@ -267,7 +267,7 @@ impl Layer for DeltaLayer {
}
/// debugging function to print out the contents of the layer
fn dump(&self) -> Result<()> {
fn dump(&self, verbose: bool) -> Result<()> {
println!(
"----- delta layer for ten {} tli {} keys {}-{} lsn {}-{} ----",
self.tenantid,
@@ -278,6 +278,10 @@ impl Layer for DeltaLayer {
self.lsn_range.end
);
if !verbose {
return Ok(());
}
let inner = self.load()?;
let path = self.path();

View File

@@ -212,12 +212,16 @@ impl Layer for ImageLayer {
}
/// debugging function to print out the contents of the layer
fn dump(&self) -> Result<()> {
fn dump(&self, verbose: bool) -> Result<()> {
println!(
"----- image layer for ten {} tli {} key {}-{} at {} ----",
self.tenantid, self.timelineid, self.key_range.start, self.key_range.end, self.lsn
);
if !verbose {
return Ok(());
}
let inner = self.load()?;
let mut index_vec: Vec<(&Key, &BlobRef)> = inner.index.iter().collect();
@@ -279,6 +283,7 @@ impl ImageLayer {
// above call to `load_inner`, so it's already been released). And
// while we do that, another thread could unload again, so we have
// to re-check and retry if that happens.
drop(inner);
}
}

View File

@@ -190,7 +190,7 @@ impl Layer for InMemoryLayer {
}
/// debugging function to print out the contents of the layer
fn dump(&self) -> Result<()> {
fn dump(&self, verbose: bool) -> Result<()> {
let inner = self.inner.read().unwrap();
let end_str = inner
@@ -204,6 +204,10 @@ impl Layer for InMemoryLayer {
self.timelineid, self.start_lsn, end_str,
);
if !verbose {
return Ok(());
}
let mut buf = Vec::new();
for (key, vec_map) in inner.index.iter() {
for (lsn, blob_ref) in vec_map.as_slice() {

View File

@@ -16,14 +16,8 @@ use crate::layered_repository::InMemoryLayer;
use crate::repository::Key;
use anyhow::Result;
use lazy_static::lazy_static;
use num_bigint::BigInt;
use num_traits::cast::ToPrimitive;
use num_traits::identities::{One, Zero};
use num_traits::{Bounded, Num, Signed};
use rstar::{RTree, RTreeObject, AABB};
use std::collections::VecDeque;
use std::ops::Range;
use std::ops::{Add, Div, Mul, Neg, Rem, Sub};
use std::sync::Arc;
use tracing::*;
use zenith_metrics::{register_int_gauge, IntGauge};
@@ -57,146 +51,14 @@ pub struct LayerMap {
pub frozen_layers: VecDeque<Arc<InMemoryLayer>>,
/// All the historic layers are kept here
historic_layers: RTree<LayerEnvelope>,
/// L0 layers has key range: (Key::MIN..Key::MAX) and locating them using R-Tree search is very inefficient.
/// So l) layers are also pushed in l0_layers vector.
l0_layers: Vec<Arc<dyn Layer>>,
}
struct LayerEnvelope {
layer: Arc<dyn Layer>,
}
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Debug)]
struct IntKey(i128);
impl Bounded for IntKey {
fn min_value() -> Self {
IntKey(i128::MIN)
}
fn max_value() -> Self {
IntKey(i128::MAX)
}
}
impl Signed for IntKey {
fn is_positive(&self) -> bool {
self.0 > 0
}
fn is_negative(&self) -> bool {
self.0 < 0
}
fn signum(&self) -> Self {
IntKey(self.0.signum())
}
fn abs(&self) -> Self {
IntKey(self.0.abs())
}
fn abs_sub(&self, other: &Self) -> Self {
IntKey(self.0.abs_sub(&other.0))
}
}
impl Neg for IntKey {
type Output = Self;
fn neg(self) -> Self::Output {
IntKey(-self.0)
}
}
impl Rem for IntKey {
type Output = Self;
fn rem(self, rhs: Self) -> Self::Output {
IntKey(self.0 % rhs.0)
}
}
impl Div for IntKey {
type Output = Self;
fn div(self, rhs: Self) -> Self::Output {
IntKey(self.0 / rhs.0)
}
}
impl Add for IntKey {
type Output = Self;
fn add(self, rhs: Self) -> Self::Output {
IntKey(self.0.wrapping_add(rhs.0))
}
}
impl Sub for IntKey {
type Output = Self;
fn sub(self, rhs: Self) -> Self::Output {
IntKey(self.0.wrapping_sub(rhs.0))
}
}
impl Mul for IntKey {
type Output = Self;
fn mul(self, rhs: Self) -> Self::Output {
// Use big integer arithmetic to avoid overflow.
// We have to cacluate sqrt of the result to be able to store result of operation as i128.
// As far as multiplication is used by R-Tree to calculate area and distance, it should not be a problem.
IntKey(
(BigInt::from(self.0) * BigInt::from(rhs.0))
.sqrt()
.to_i128()
.unwrap(),
)
}
}
impl One for IntKey {
fn one() -> Self {
IntKey(1)
}
}
impl Zero for IntKey {
fn zero() -> Self {
IntKey(0)
}
fn is_zero(&self) -> bool {
self.0 == 0
}
}
impl Num for IntKey {
type FromStrRadixErr = <i128 as Num>::FromStrRadixErr;
fn from_str_radix(str: &str, radix: u32) -> Result<Self, Self::FromStrRadixErr> {
Ok(IntKey(i128::from_str_radix(str, radix)?))
}
}
impl PartialEq for LayerEnvelope {
fn eq(&self, other: &Self) -> bool {
// FIXME: ptr_eq might fail to return true for 'dyn'
// references. Clippy complains about this. In practice it
// seems to work, the assertion below would be triggered
// otherwise but this ought to be fixed.
#[allow(clippy::vtable_address_comparisons)]
Arc::ptr_eq(&self.layer, &other.layer)
}
}
impl RTreeObject for LayerEnvelope {
type Envelope = AABB<[IntKey; 2]>;
fn envelope(&self) -> Self::Envelope {
let key_range = self.layer.get_key_range();
let lsn_range = self.layer.get_lsn_range();
AABB::from_corners(
[
IntKey(key_range.start.to_i128()),
IntKey(lsn_range.start.0 as i128),
],
[
IntKey(key_range.end.to_i128() - 1),
IntKey(lsn_range.end.0 as i128 - 1),
], // end is exlusive
)
}
/// TODO: This is a placeholder implementation of a data structure
/// to hold information about all the layer files on disk and in
/// S3. Currently, it's just a vector and all operations perform a
/// linear scan over it. That obviously becomes slow as the
/// number of layers grows. I'm imagining that an R-tree or some
/// other 2D data structure would be the long-term solution here.
historic_layers: Vec<Arc<dyn Layer>>,
}
/// Return value of LayerMap::search
@@ -222,21 +84,19 @@ impl LayerMap {
// Find the latest image layer that covers the given key
let mut latest_img: Option<Arc<dyn Layer>> = None;
let mut latest_img_lsn: Option<Lsn> = None;
let envelope = AABB::from_corners(
[IntKey(key.to_i128()), IntKey(0i128)],
[IntKey(key.to_i128()), IntKey(end_lsn.0 as i128 - 1)],
);
for e in self
.historic_layers
.locate_in_envelope_intersecting(&envelope)
{
let l = &e.layer;
for l in self.historic_layers.iter() {
if l.is_incremental() {
continue;
}
assert!(l.get_key_range().contains(&key));
if !l.get_key_range().contains(&key) {
continue;
}
let img_lsn = l.get_lsn_range().start;
assert!(img_lsn < end_lsn);
if img_lsn >= end_lsn {
// too new
continue;
}
if Lsn(img_lsn.0 + 1) == end_lsn {
// found exact match
return Ok(Some(SearchResult {
@@ -252,16 +112,19 @@ impl LayerMap {
// Search the delta layers
let mut latest_delta: Option<Arc<dyn Layer>> = None;
for e in self
.historic_layers
.locate_in_envelope_intersecting(&envelope)
{
let l = &e.layer;
for l in self.historic_layers.iter() {
if !l.is_incremental() {
continue;
}
assert!(l.get_key_range().contains(&key));
assert!(l.get_lsn_range().start < end_lsn);
if !l.get_key_range().contains(&key) {
continue;
}
if l.get_lsn_range().start >= end_lsn {
// too new
continue;
}
if l.get_lsn_range().end >= end_lsn {
// this layer contains the requested point in the key/lsn space.
// No need to search any further
@@ -319,10 +182,7 @@ impl LayerMap {
/// Insert an on-disk layer
///
pub fn insert_historic(&mut self, layer: Arc<dyn Layer>) {
if layer.get_key_range() == (Key::MIN..Key::MAX) {
self.l0_layers.push(layer.clone());
}
self.historic_layers.insert(LayerEnvelope { layer });
self.historic_layers.push(layer);
NUM_ONDISK_LAYERS.inc();
}
@@ -333,21 +193,17 @@ impl LayerMap {
///
#[allow(dead_code)]
pub fn remove_historic(&mut self, layer: Arc<dyn Layer>) {
if layer.get_key_range() == (Key::MIN..Key::MAX) {
let len_before = self.l0_layers.len();
let len_before = self.historic_layers.len();
// FIXME: ptr_eq might fail to return true for 'dyn'
// references. Clippy complains about this. In practice it
// seems to work, the assertion below would be triggered
// otherwise but this ought to be fixed.
#[allow(clippy::vtable_address_comparisons)]
self.l0_layers.retain(|other| !Arc::ptr_eq(other, &layer));
assert_eq!(self.l0_layers.len(), len_before - 1);
}
assert!(self
.historic_layers
.remove(&LayerEnvelope { layer })
.is_some());
// FIXME: ptr_eq might fail to return true for 'dyn'
// references. Clippy complains about this. In practice it
// seems to work, the assertion below would be triggered
// otherwise but this ought to be fixed.
#[allow(clippy::vtable_address_comparisons)]
self.historic_layers
.retain(|other| !Arc::ptr_eq(other, &layer));
assert_eq!(self.historic_layers.len(), len_before - 1);
NUM_ONDISK_LAYERS.dec();
}
@@ -368,23 +224,13 @@ impl LayerMap {
loop {
let mut made_progress = false;
let envelope = AABB::from_corners(
[IntKey(range_remain.start.to_i128()), IntKey(lsn.0 as i128)],
[
IntKey(range_remain.end.to_i128() - 1),
IntKey(disk_consistent_lsn.0 as i128),
],
);
for e in self
.historic_layers
.locate_in_envelope_intersecting(&envelope)
{
let l = &e.layer;
for l in self.historic_layers.iter() {
if l.is_incremental() {
continue;
}
let img_lsn = l.get_lsn_range().start;
if l.get_key_range().contains(&range_remain.start)
if !l.is_incremental()
&& l.get_key_range().contains(&range_remain.start)
&& img_lsn > lsn
&& img_lsn < disk_consistent_lsn
{
@@ -420,8 +266,8 @@ impl LayerMap {
}
*/
pub fn iter_historic_layers(&self) -> Box<dyn Iterator<Item = Arc<dyn Layer>> + '_> {
Box::new(self.historic_layers.iter().map(|e| e.layer.clone()))
pub fn iter_historic_layers(&self) -> std::slice::Iter<Arc<dyn Layer>> {
self.historic_layers.iter()
}
/// Find the last image layer that covers 'key', ignoring any image layers
@@ -429,22 +275,19 @@ impl LayerMap {
fn find_latest_image(&self, key: Key, lsn: Lsn) -> Option<Arc<dyn Layer>> {
let mut candidate_lsn = Lsn(0);
let mut candidate = None;
let envelope = AABB::from_corners(
[IntKey(key.to_i128()), IntKey(0)],
[IntKey(key.to_i128()), IntKey(lsn.0 as i128)],
);
for e in self
.historic_layers
.locate_in_envelope_intersecting(&envelope)
{
let l = &e.layer;
for l in self.historic_layers.iter() {
if l.is_incremental() {
continue;
}
assert!(l.get_key_range().contains(&key));
if !l.get_key_range().contains(&key) {
continue;
}
let this_lsn = l.get_lsn_range().start;
assert!(this_lsn <= lsn);
if this_lsn > lsn {
continue;
}
if this_lsn < candidate_lsn {
// our previous candidate was better
continue;
@@ -472,16 +315,10 @@ impl LayerMap {
let mut points: Vec<Key>;
points = vec![key_range.start];
let envelope = AABB::from_corners(
[IntKey(key_range.start.to_i128()), IntKey(0)],
[IntKey(key_range.end.to_i128()), IntKey(lsn.0 as i128)],
);
for e in self
.historic_layers
.locate_in_envelope_intersecting(&envelope)
{
let l = &e.layer;
assert!(l.get_lsn_range().start <= lsn);
for l in self.historic_layers.iter() {
if l.get_lsn_range().start > lsn {
continue;
}
let range = l.get_key_range();
if key_range.contains(&range.start) {
points.push(l.get_key_range().start);
@@ -514,29 +351,16 @@ impl LayerMap {
/// given key and LSN range.
pub fn count_deltas(&self, key_range: &Range<Key>, lsn_range: &Range<Lsn>) -> Result<usize> {
let mut result = 0;
if lsn_range.start >= lsn_range.end {
return Ok(0);
}
let envelope = AABB::from_corners(
[
IntKey(key_range.start.to_i128()),
IntKey(lsn_range.start.0 as i128),
],
[
IntKey(key_range.end.to_i128() - 1),
IntKey(lsn_range.end.0 as i128 - 1),
],
);
for e in self
.historic_layers
.locate_in_envelope_intersecting(&envelope)
{
let l = &e.layer;
for l in self.historic_layers.iter() {
if !l.is_incremental() {
continue;
}
assert!(range_overlaps(&l.get_lsn_range(), lsn_range));
assert!(range_overlaps(&l.get_key_range(), key_range));
if !range_overlaps(&l.get_lsn_range(), lsn_range) {
continue;
}
if !range_overlaps(&l.get_key_range(), key_range) {
continue;
}
// We ignore level0 delta layers. Unless the whole keyspace fits
// into one partition
@@ -553,15 +377,37 @@ impl LayerMap {
/// Return all L0 delta layers
pub fn get_level0_deltas(&self) -> Result<Vec<Arc<dyn Layer>>> {
Ok(self.l0_layers.clone())
let mut deltas = Vec::new();
for l in self.historic_layers.iter() {
if !l.is_incremental() {
continue;
}
if l.get_key_range() != (Key::MIN..Key::MAX) {
continue;
}
deltas.push(Arc::clone(l));
}
Ok(deltas)
}
/// debugging function to print out the contents of the layer map
#[allow(unused)]
pub fn dump(&self) -> Result<()> {
pub fn dump(&self, verbose: bool) -> Result<()> {
println!("Begin dump LayerMap");
for e in self.historic_layers.iter() {
e.layer.dump()?;
println!("open_layer:");
if let Some(open_layer) = &self.open_layer {
open_layer.dump(verbose)?;
}
println!("frozen_layers:");
for frozen_layer in self.frozen_layers.iter() {
frozen_layer.dump(verbose)?;
}
println!("historic_layers:");
for layer in self.historic_layers.iter() {
layer.dump(verbose)?;
}
println!("End dump LayerMap");
Ok(())

View File

@@ -143,7 +143,7 @@ pub trait Layer: Send + Sync {
fn delete(&self) -> Result<()>;
/// Dump summary of the contents of the layer to stdout
fn dump(&self) -> Result<()>;
fn dump(&self, verbose: bool) -> Result<()>;
}
// Flag indicating that this version initialize the page

View File

@@ -514,6 +514,7 @@ impl PageServerHandler {
) -> anyhow::Result<()> {
let span = info_span!("basebackup", timeline = %timelineid, tenant = %tenantid, lsn = field::Empty);
let _enter = span.enter();
info!("starting");
// check that the timeline exists
let timeline = tenant_mgr::get_timeline_for_tenant_load(tenantid, timelineid)
@@ -536,7 +537,7 @@ impl PageServerHandler {
basebackup.send_tarball()?;
}
pgb.write_message(&BeMessage::CopyDone)?;
debug!("CopyDone sent!");
info!("done");
Ok(())
}

View File

@@ -321,8 +321,8 @@ pub fn schedule_timeline_checkpoint_upload(
tenant_id, timeline_id
)
} else {
warn!(
"Could not send an upload task for tenant {}, timeline {}: the sync queue is not initialized",
debug!(
"Upload task for tenant {}, timeline {} sent",
tenant_id, timeline_id
)
}
@@ -443,30 +443,38 @@ fn storage_sync_loop<
max_sync_errors: NonZeroU32,
) {
let remote_assets = Arc::new((storage, index.clone()));
info!("Starting remote storage sync loop");
loop {
let index = index.clone();
let loop_step = runtime.block_on(async {
tokio::select! {
new_timeline_states = loop_step(
step = loop_step(
conf,
&mut receiver,
Arc::clone(&remote_assets),
max_concurrent_sync,
max_sync_errors,
)
.instrument(debug_span!("storage_sync_loop_step")) => LoopStep::SyncStatusUpdates(new_timeline_states),
.instrument(info_span!("storage_sync_loop_step")) => step,
_ = thread_mgr::shutdown_watcher() => LoopStep::Shutdown,
}
});
match loop_step {
LoopStep::SyncStatusUpdates(new_timeline_states) => {
// Batch timeline download registration to ensure that the external registration code won't block any running tasks before.
apply_timeline_sync_status_updates(conf, index, new_timeline_states);
debug!("Sync loop step completed");
if new_timeline_states.is_empty() {
debug!("Sync loop step completed, no new timeline states");
} else {
info!(
"Sync loop step completed, {} new timeline state update(s)",
new_timeline_states.len()
);
// Batch timeline download registration to ensure that the external registration code won't block any running tasks before.
apply_timeline_sync_status_updates(conf, index, new_timeline_states);
}
}
LoopStep::Shutdown => {
debug!("Shutdown requested, stopping");
info!("Shutdown requested, stopping");
break;
}
}
@@ -482,7 +490,7 @@ async fn loop_step<
remote_assets: Arc<(S, RemoteIndex)>,
max_concurrent_sync: NonZeroUsize,
max_sync_errors: NonZeroU32,
) -> HashMap<ZTenantId, HashMap<ZTimelineId, TimelineSyncStatusUpdate>> {
) -> LoopStep {
let max_concurrent_sync = max_concurrent_sync.get();
let mut next_tasks = Vec::new();
@@ -490,8 +498,7 @@ async fn loop_step<
if let Some(first_task) = sync_queue::next_task(receiver).await {
next_tasks.push(first_task);
} else {
debug!("Shutdown requested, stopping");
return HashMap::new();
return LoopStep::Shutdown;
};
next_tasks.extend(
sync_queue::next_task_batch(receiver, max_concurrent_sync - 1)
@@ -500,12 +507,17 @@ async fn loop_step<
);
let remaining_queue_length = sync_queue::len();
debug!(
"Processing {} tasks in batch, more tasks left to process: {}",
next_tasks.len(),
remaining_queue_length
);
REMAINING_SYNC_ITEMS.set(remaining_queue_length as i64);
if remaining_queue_length > 0 || !next_tasks.is_empty() {
info!(
"Processing {} tasks in batch, more tasks left to process: {}",
next_tasks.len(),
remaining_queue_length
);
} else {
debug!("No tasks to process");
return LoopStep::SyncStatusUpdates(HashMap::new());
}
let mut task_batch = next_tasks
.into_iter()
@@ -515,8 +527,9 @@ async fn loop_step<
let sync_name = task.kind.sync_name();
let extra_step = match tokio::spawn(
process_task(conf, Arc::clone(&remote_assets), task, max_sync_errors)
.instrument(debug_span!("", sync_id = %sync_id, attempt, sync_name)),
process_task(conf, Arc::clone(&remote_assets), task, max_sync_errors).instrument(
info_span!("process_sync_task", sync_id = %sync_id, attempt, sync_name),
),
)
.await
{
@@ -551,7 +564,7 @@ async fn loop_step<
}
}
new_timeline_states
LoopStep::SyncStatusUpdates(new_timeline_states)
}
async fn process_task<

View File

@@ -29,19 +29,6 @@ pub struct Key {
}
impl Key {
/// 'field2' is used to store tablespaceid for relations and small enum numbers for other relish.
/// As far as Zenith is not supporting tablespace (because of lack of access to local file system),
/// we can assume that only some predefined namespace OIDs are used which can fit in u16
pub fn to_i128(&self) -> i128 {
assert!(self.field2 < 0xFFFFF || self.field2 == 0xFFFFFFFF || self.field2 == 0x22222222);
(((self.field1 & 0xf) as i128) << 120)
| (((self.field2 & 0xFFFF) as i128) << 104)
| ((self.field3 as i128) << 72)
| ((self.field4 as i128) << 40)
| ((self.field5 as i128) << 32)
| self.field6 as i128
}
pub fn next(&self) -> Key {
self.add(1)
}
@@ -596,7 +583,7 @@ mod tests {
use lazy_static::lazy_static;
lazy_static! {
static ref TEST_KEY: Key = Key::from_array(hex!("110000222233333333444444445500000001"));
static ref TEST_KEY: Key = Key::from_array(hex!("112222222233333333444444445500000001"));
}
#[test]
@@ -639,9 +626,9 @@ mod tests {
use std::str::from_utf8;
#[allow(non_snake_case)]
let TEST_KEY_A: Key = Key::from_hex("110000222233333333444444445500000001").unwrap();
let TEST_KEY_A: Key = Key::from_hex("112222222233333333444444445500000001").unwrap();
#[allow(non_snake_case)]
let TEST_KEY_B: Key = Key::from_hex("110000222233333333444444445500000002").unwrap();
let TEST_KEY_B: Key = Key::from_hex("112222222233333333444444445500000002").unwrap();
// Insert a value on the timeline
writer.put(TEST_KEY_A, Lsn(0x20), test_value("foo at 0x20"))?;

View File

@@ -70,7 +70,7 @@ pub fn launch_wal_receiver(
match receivers.get_mut(&(tenantid, timelineid)) {
Some(receiver) => {
info!("wal receiver already running, updating connection string");
debug!("wal receiver already running, updating connection string");
receiver.wal_producer_connstr = wal_producer_connstr.into();
}
None => {

View File

@@ -17,8 +17,8 @@ log = "0.4.14"
memoffset = "0.6.2"
thiserror = "1.0"
serde = { version = "1.0", features = ["derive"] }
workspace_hack = { path = "../workspace_hack" }
zenith_utils = { path = "../zenith_utils" }
workspace_hack = { version = "0.1", path = "../workspace_hack" }
[build-dependencies]
bindgen = "0.59.1"

View File

@@ -22,13 +22,15 @@ rustls = "0.19.1"
scopeguard = "1.1.0"
serde = "1"
serde_json = "1"
socket2 = "0.4.4"
thiserror = "1.0"
tokio = { version = "1.11", features = ["macros"] }
tokio = { version = "1.17", features = ["macros"] }
tokio-postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="2949d98df52587d562986aad155dd4e889e408b7" }
tokio-rustls = "0.22.0"
zenith_utils = { path = "../zenith_utils" }
zenith_metrics = { path = "../zenith_metrics" }
workspace_hack = { version = "0.1", path = "../workspace_hack" }
[dev-dependencies]
tokio-postgres-rustls = "0.8.0"

View File

@@ -41,6 +41,7 @@ impl DatabaseInfo {
let host_port = format!("{}:{}", self.host, self.port);
let socket = TcpStream::connect(host_port).await?;
let socket_addr = socket.peer_addr()?;
socket2::SockRef::from(&socket).set_keepalive(true)?;
Ok((socket_addr, socket))
}

View File

@@ -50,6 +50,10 @@ pub async fn thread_main(
println!("proxy has shut down");
}
// When set for the server socket, the keepalive setting
// will be inherited by all accepted client sockets.
socket2::SockRef::from(&listener).set_keepalive(true)?;
let cancel_map = Arc::new(CancelMap::default());
loop {
let (socket, peer_addr) = listener.accept().await?;
@@ -367,4 +371,24 @@ mod tests {
Ok(())
}
#[tokio::test]
async fn keepalive_is_inherited() -> anyhow::Result<()> {
use tokio::net::{TcpListener, TcpStream};
let listener = TcpListener::bind("127.0.0.1:0").await?;
let port = listener.local_addr()?.port();
socket2::SockRef::from(&listener).set_keepalive(true)?;
let t = tokio::spawn(async move {
let (client, _) = listener.accept().await?;
let keepalive = socket2::SockRef::from(&client).keepalive()?;
anyhow::Ok(keepalive)
});
let _ = TcpStream::connect(("127.0.0.1", port)).await?;
assert!(t.await??, "keepalive should be inherited");
Ok(())
}
}

View File

@@ -4,4 +4,10 @@
# It is intended to be a primary endpoint for all the people who want to
# just setup test environment without going into details of python package management
poetry install --no-root # this installs dev dependencies by default
poetry config --list
if [ -z "${CI}" ]; then
poetry install --no-root --no-interaction --ansi
else
poetry install --no-root
fi

View File

@@ -10,6 +10,8 @@ Prerequisites:
below to run from other directories.
- The zenith git repo, including the postgres submodule
(for some tests, e.g. `pg_regress`)
- Some tests (involving storage nodes coordination) require etcd installed. Follow
[`the guide`](https://etcd.io/docs/v3.5/install/) to obtain it.
### Test Organization

View File

View File

@@ -10,7 +10,9 @@ def test_pageserver_catchup_while_compute_down(zenith_env_builder: ZenithEnvBuil
env = zenith_env_builder.init_start()
env.zenith_cli.create_branch('test_pageserver_catchup_while_compute_down')
pg = env.postgres.create_start('test_pageserver_catchup_while_compute_down')
# Make shared_buffers large to ensure we won't query pageserver while it is down.
pg = env.postgres.create_start('test_pageserver_catchup_while_compute_down',
config_lines=['shared_buffers=512MB'])
pg_conn = pg.connect()
cur = pg_conn.cursor()

View File

@@ -0,0 +1,183 @@
from contextlib import closing
from fixtures.zenith_fixtures import ZenithEnv
from fixtures.log_helper import log
from psycopg2.errors import UndefinedTable
from psycopg2.errors import IoError
pytest_plugins = ("fixtures.zenith_fixtures")
extensions = ["pageinspect", "zenith_test_utils", "pg_buffercache"]
#
# Validation of reading different page versions
#
def test_read_validation(zenith_simple_env: ZenithEnv):
env = zenith_simple_env
env.zenith_cli.create_branch("test_read_validation", "empty")
pg = env.postgres.create_start("test_read_validation")
log.info("postgres is running on 'test_read_validation' branch")
with closing(pg.connect()) as con:
with con.cursor() as c:
for e in extensions:
c.execute("create extension if not exists {};".format(e))
c.execute("create table foo (c int) with (autovacuum_enabled = false)")
c.execute("insert into foo values (1)")
c.execute("select lsn, lower, upper from page_header(get_raw_page('foo', 'main', 0));")
first = c.fetchone()
c.execute("select relfilenode from pg_class where relname = 'foo'")
relfilenode = c.fetchone()[0]
c.execute("insert into foo values (2);")
c.execute("select lsn, lower, upper from page_header(get_raw_page('foo', 'main', 0));")
second = c.fetchone()
assert first != second, "Failed to update page"
log.info("Test table is populated, validating buffer cache")
c.execute(
"select count(*) from pg_buffercache where relfilenode = {}".format(relfilenode))
assert c.fetchone()[0] > 0, "No buffers cached for the test relation"
c.execute(
"select reltablespace, reldatabase, relfilenode from pg_buffercache where relfilenode = {}"
.format(relfilenode))
reln = c.fetchone()
log.info("Clear buffer cache to ensure no stale pages are brought into the cache")
c.execute("select clear_buffer_cache()")
c.execute(
"select count(*) from pg_buffercache where relfilenode = {}".format(relfilenode))
assert c.fetchone()[0] == 0, "Failed to clear buffer cache"
log.info("Cache is clear, reading stale page version")
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn('foo', 'main', 0, '{}'))"
.format(first[0]))
direct_first = c.fetchone()
assert first == direct_first, "Failed fetch page at historic lsn"
c.execute(
"select count(*) from pg_buffercache where relfilenode = {}".format(relfilenode))
assert c.fetchone()[0] == 0, "relation buffers detected after invalidation"
log.info("Cache is clear, reading latest page version without cache")
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn('foo', 'main', 0, NULL))"
)
direct_latest = c.fetchone()
assert second == direct_latest, "Failed fetch page at latest lsn"
c.execute(
"select count(*) from pg_buffercache where relfilenode = {}".format(relfilenode))
assert c.fetchone()[0] == 0, "relation buffers detected after invalidation"
log.info(
"Cache is clear, reading stale page version without cache using relation identifiers"
)
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn( {}, {}, {}, 0, 0, '{}' ))"
.format(reln[0], reln[1], reln[2], first[0]))
direct_first = c.fetchone()
assert first == direct_first, "Failed fetch page at historic lsn using oid"
log.info(
"Cache is clear, reading latest page version without cache using relation identifiers"
)
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn( {}, {}, {}, 0, 0, NULL ))"
.format(reln[0], reln[1], reln[2]))
direct_latest = c.fetchone()
assert second == direct_latest, "Failed fetch page at latest lsn"
c.execute('drop table foo;')
log.info(
"Relation dropped, attempting reading stale page version without cache using relation identifiers"
)
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn( {}, {}, {}, 0, 0, '{}' ))"
.format(reln[0], reln[1], reln[2], first[0]))
direct_first = c.fetchone()
assert first == direct_first, "Failed fetch page at historic lsn using oid"
log.info("Validation page inspect won't allow reading pages of dropped relations")
try:
c.execute("select * from page_header(get_raw_page('foo', 'main', 0));")
assert False, "query should have failed"
except UndefinedTable as e:
log.info("Caught an expected failure: {}".format(e))
def test_read_validation_neg(zenith_simple_env: ZenithEnv):
env = zenith_simple_env
env.zenith_cli.create_branch("test_read_validation_neg", "empty")
pg = env.postgres.create_start("test_read_validation_neg")
log.info("postgres is running on 'test_read_validation_neg' branch")
with closing(pg.connect()) as con:
with con.cursor() as c:
for e in extensions:
c.execute("create extension if not exists {};".format(e))
log.info("read a page of a missing relation")
try:
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn('Unknown', 'main', 0, '0/0'))"
)
assert False, "query should have failed"
except UndefinedTable as e:
log.info("Caught an expected failure: {}".format(e))
c.execute("create table foo (c int) with (autovacuum_enabled = false)")
c.execute("insert into foo values (1)")
log.info("read a page at lsn 0")
try:
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn('foo', 'main', 0, '0/0'))"
)
assert False, "query should have failed"
except IoError as e:
log.info("Caught an expected failure: {}".format(e))
log.info("Pass NULL as an input")
expected = (None, None, None)
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn(NULL, 'main', 0, '0/0'))"
)
assert c.fetchone() == expected, "Expected null output"
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn('foo', NULL, 0, '0/0'))"
)
assert c.fetchone() == expected, "Expected null output"
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn('foo', 'main', NULL, '0/0'))"
)
assert c.fetchone() == expected, "Expected null output"
# This check is currently failing, reading beyond EOF is returning a 0-page
log.info("Read beyond EOF")
c.execute(
"select lsn, lower, upper from page_header(get_raw_page_at_lsn('foo', 'main', 1, NULL))"
)

View File

@@ -12,16 +12,17 @@ from contextlib import closing
from dataclasses import dataclass, field
from multiprocessing import Process, Value
from pathlib import Path
from fixtures.zenith_fixtures import PgBin, Postgres, Safekeeper, ZenithEnv, ZenithEnvBuilder, PortDistributor, SafekeeperPort, zenith_binpath, PgProtocol
from fixtures.utils import lsn_to_hex, mkdir_if_needed, lsn_from_hex
from fixtures.zenith_fixtures import PgBin, Postgres, Safekeeper, SafekeeperHttpClient, ZenithEnv, ZenithEnvBuilder, PortDistributor, SafekeeperPort, zenith_binpath, PgProtocol
from fixtures.utils import etcd_path, lsn_to_hex, mkdir_if_needed, lsn_from_hex
from fixtures.log_helper import log
from typing import List, Optional, Any
from typing import Dict, List, Optional, Any
# basic test, write something in setup with wal acceptors, ensure that commits
# succeed and data is written
def test_normal_work(zenith_env_builder: ZenithEnvBuilder):
zenith_env_builder.num_safekeepers = 3
zenith_env_builder.broker = True
env = zenith_env_builder.init_start()
env.zenith_cli.create_branch('test_wal_acceptors_normal_work')
@@ -326,6 +327,49 @@ def test_race_conditions(zenith_env_builder: ZenithEnvBuilder, stop_value):
proc.join()
# Test that safekeepers push their info to the broker and learn peer status from it
@pytest.mark.skipif(etcd_path() is None, reason="requires etcd which is not present in PATH")
def test_broker(zenith_env_builder: ZenithEnvBuilder):
zenith_env_builder.num_safekeepers = 3
zenith_env_builder.broker = True
zenith_env_builder.enable_local_fs_remote_storage()
env = zenith_env_builder.init_start()
env.zenith_cli.create_branch("test_broker", "main")
pg = env.postgres.create_start('test_broker')
pg.safe_psql("CREATE TABLE t(key int primary key, value text)")
# learn zenith timeline from compute
tenant_id = pg.safe_psql("show zenith.zenith_tenant")[0][0]
timeline_id = pg.safe_psql("show zenith.zenith_timeline")[0][0]
# wait until remote_consistent_lsn gets advanced on all safekeepers
clients = [sk.http_client() for sk in env.safekeepers]
stat_before = [cli.timeline_status(tenant_id, timeline_id) for cli in clients]
log.info(f"statuses is {stat_before}")
pg.safe_psql("INSERT INTO t SELECT generate_series(1,100), 'payload'")
# force checkpoint to advance remote_consistent_lsn
with closing(env.pageserver.connect()) as psconn:
with psconn.cursor() as pscur:
pscur.execute(f"checkpoint {tenant_id} {timeline_id}")
# and wait till remote_consistent_lsn propagates to all safekeepers
started_at = time.time()
while True:
stat_after = [cli.timeline_status(tenant_id, timeline_id) for cli in clients]
if all(
lsn_from_hex(s_after.remote_consistent_lsn) > lsn_from_hex(
s_before.remote_consistent_lsn) for s_after,
s_before in zip(stat_after, stat_before)):
break
elapsed = time.time() - started_at
if elapsed > 20:
raise RuntimeError(
f"timed out waiting {elapsed:.0f}s for remote_consistent_lsn propagation: status before {stat_before}, status current {stat_after}"
)
time.sleep(0.5)
class ProposerPostgres(PgProtocol):
"""Object for running postgres without ZenithEnv"""
def __init__(self,
@@ -352,20 +396,39 @@ class ProposerPostgres(PgProtocol):
""" Path to postgresql.conf """
return os.path.join(self.pgdata_dir, 'postgresql.conf')
def log_path(self) -> str:
""" Path to pg.log """
return os.path.join(self.pg_data_dir_path(), "pg.log")
def create_dir_config(self, wal_acceptors: str):
""" Create dir and config for running --sync-safekeepers """
mkdir_if_needed(self.pg_data_dir_path())
with open(self.config_file_path(), "w") as f:
cfg = [
"synchronous_standby_names = 'walproposer'\n",
"shared_preload_libraries = 'zenith'\n",
"wal_keep_size=10TB\n",
"shared_preload_libraries=zenith\n",
"zenith.page_server_connstring=''\n",
"synchronous_commit=on\n",
"max_wal_senders=10\n",
"wal_log_hints=on\n",
"max_replication_slots=10\n",
"hot_standby=on\n",
"min_wal_size=20GB\n",
"max_wal_size=40GB\n",
"checkpoint_timeout=60min\n",
"log_checkpoints=on\n",
"max_connections=100\n",
"wal_sender_timeout=0\n",
"wal_level=replica\n",
f"zenith.zenith_timeline = '{self.timeline_id.hex}'\n",
f"zenith.zenith_tenant = '{self.tenant_id.hex}'\n",
f"zenith.page_server_connstring = ''\n",
f"wal_acceptors = '{wal_acceptors}'\n",
f"listen_addresses = '{self.listen_addr}'\n",
f"port = '{self.port}'\n",
"synchronous_standby_names = 'walproposer'\n",
"fsync=off\n",
]
f.writelines(cfg)
@@ -397,8 +460,7 @@ class ProposerPostgres(PgProtocol):
def start(self):
""" Start postgres with pg_ctl """
log_path = os.path.join(self.pg_data_dir_path(), "pg.log")
args = ["pg_ctl", "-D", self.pg_data_dir_path(), "-l", log_path, "-w", "start"]
args = ["pg_ctl", "-D", self.pg_data_dir_path(), "-l", self.log_path(), "-w", "start"]
self.pg_bin.run(args)
def stop(self):
@@ -492,6 +554,16 @@ def test_timeline_status(zenith_env_builder: ZenithEnvBuilder):
assert epoch_after_reboot > epoch
@dataclass
class SafekeeperProc:
""" An object representing a running safekeeper daemon. """
proc: 'subprocess.CompletedProcess[bytes]'
port: SafekeeperPort
def http_client(self) -> SafekeeperHttpClient:
return SafekeeperHttpClient(port=self.port.http)
class SafekeeperEnv:
def __init__(self,
repo_dir: Path,
@@ -503,7 +575,7 @@ class SafekeeperEnv:
self.pg_bin = pg_bin
self.num_safekeepers = num_safekeepers
self.bin_safekeeper = os.path.join(str(zenith_binpath), 'safekeeper')
self.safekeepers: Optional[List[subprocess.CompletedProcess[Any]]] = None
self.safekeepers: Optional[List[SafekeeperProc]] = None
self.postgres: Optional[ProposerPostgres] = None
self.tenant_id: Optional[uuid.UUID] = None
self.timeline_id: Optional[uuid.UUID] = None
@@ -527,7 +599,7 @@ class SafekeeperEnv:
return self
def start_safekeeper(self, i):
def start_safekeeper(self, i) -> "SafekeeperProc":
port = SafekeeperPort(
pg=self.port_distributor.get_port(),
http=self.port_distributor.get_port(),
@@ -550,10 +622,12 @@ class SafekeeperEnv:
]
log.info(f'Running command "{" ".join(args)}"')
return subprocess.run(args, check=True)
def get_safekeeper_connstrs(self):
return ','.join([sk_proc.args[2] for sk_proc in self.safekeepers])
return SafekeeperProc(subprocess.run(args, check=True), port)
def get_safekeeper_connstrs(self) -> str:
assert self.safekeepers is not None
return ','.join([sk.proc.args[2] for sk in self.safekeepers])
def create_postgres(self):
pgdata_dir = os.path.join(self.repo_dir, "proposer_pgdata")
@@ -585,8 +659,8 @@ class SafekeeperEnv:
if self.postgres is not None:
self.postgres.stop()
if self.safekeepers is not None:
for sk_proc in self.safekeepers:
self.kill_safekeeper(sk_proc.args[6])
for sk in self.safekeepers:
self.kill_safekeeper(sk.proc.args[6])
def test_safekeeper_without_pageserver(test_output_dir: str,

View File

@@ -1,9 +1,10 @@
import asyncio
import uuid
import asyncpg
import random
import time
from fixtures.zenith_fixtures import ZenithEnvBuilder, Postgres, Safekeeper
from fixtures.zenith_fixtures import ZenithEnv, ZenithEnvBuilder, Postgres, Safekeeper
from fixtures.log_helper import getLogger
from fixtures.utils import lsn_from_hex, lsn_to_hex
from typing import List
@@ -30,10 +31,6 @@ class BankClient(object):
await self.conn.execute('DROP TABLE IF EXISTS bank_log')
await self.conn.execute('CREATE TABLE bank_log(from_uid int, to_uid int, amount int)')
# TODO: Remove when https://github.com/zenithdb/zenith/issues/644 is fixed
await self.conn.execute('ALTER TABLE bank_accs SET (autovacuum_enabled = false)')
await self.conn.execute('ALTER TABLE bank_log SET (autovacuum_enabled = false)')
async def check_invariant(self):
row = await self.conn.fetchrow('SELECT sum(amount) AS sum FROM bank_accs')
assert row['sum'] == self.n_accounts * self.init_amount
@@ -139,12 +136,15 @@ async def wait_for_lsn(safekeeper: Safekeeper,
# On each iteration 1 acceptor is stopped, and 2 others should allow
# background workers execute transactions. In the end, state should remain
# consistent.
async def run_restarts_under_load(pg: Postgres, acceptors: List[Safekeeper], n_workers=10):
async def run_restarts_under_load(env: ZenithEnv,
pg: Postgres,
acceptors: List[Safekeeper],
n_workers=10):
n_accounts = 100
init_amount = 100000
max_transfer = 100
period_time = 10
iterations = 6
period_time = 4
iterations = 10
# Set timeout for this test at 5 minutes. It should be enough for test to complete
# and less than CircleCI's no_output_timeout, taking into account that this timeout
@@ -176,6 +176,11 @@ async def run_restarts_under_load(pg: Postgres, acceptors: List[Safekeeper], n_w
flush_lsn = lsn_to_hex(flush_lsn)
log.info(f'Postgres flush_lsn {flush_lsn}')
pageserver_lsn = env.pageserver.http_client().timeline_detail(
uuid.UUID(tenant_id), uuid.UUID((timeline_id)))["local"]["last_record_lsn"]
sk_ps_lag = lsn_from_hex(flush_lsn) - lsn_from_hex(pageserver_lsn)
log.info(f'Pageserver last_record_lsn={pageserver_lsn} lag={sk_ps_lag / 1024}kb')
# Wait until alive safekeepers catch up with postgres
for idx, safekeeper in enumerate(acceptors):
if idx != victim_idx:
@@ -203,9 +208,8 @@ def test_restarts_under_load(zenith_env_builder: ZenithEnvBuilder):
env = zenith_env_builder.init_start()
env.zenith_cli.create_branch('test_wal_acceptors_restarts_under_load')
pg = env.postgres.create_start('test_wal_acceptors_restarts_under_load')
# Enable backpressure with 1MB maximal lag, because we don't want to block on `wait_for_lsn()` for too long
pg = env.postgres.create_start('test_wal_acceptors_restarts_under_load',
config_lines=['max_replication_write_lag=1MB'])
asyncio.run(run_restarts_under_load(pg, env.safekeepers))
# TODO: Remove when https://github.com/zenithdb/zenith/issues/644 is fixed
pg.stop()
asyncio.run(run_restarts_under_load(env, pg, env.safekeepers))

View File

@@ -1,4 +1,5 @@
import os
import shutil
import subprocess
from typing import Any, List
@@ -76,3 +77,8 @@ def print_gc_result(row):
log.info(
" total: {layers_total}, needed_by_cutoff {layers_needed_by_cutoff}, needed_by_branches: {layers_needed_by_branches}, not_updated: {layers_not_updated}, removed: {layers_removed}"
.format_map(row))
# path to etcd binary or None if not present.
def etcd_path():
return shutil.which("etcd")

View File

@@ -33,7 +33,7 @@ from typing_extensions import Literal
import requests
import backoff # type: ignore
from .utils import (get_self_dir, lsn_from_hex, mkdir_if_needed, subprocess_capture)
from .utils import (etcd_path, get_self_dir, mkdir_if_needed, subprocess_capture, lsn_from_hex)
from fixtures.log_helper import log
"""
This file contains pytest fixtures. A fixture is a test resource that can be
@@ -257,7 +257,8 @@ class PgProtocol:
dbname: Optional[str] = None,
schema: Optional[str] = None,
username: Optional[str] = None,
password: Optional[str] = None) -> str:
password: Optional[str] = None,
statement_timeout_ms: Optional[int] = None) -> str:
"""
Build a libpq connection string for the Postgres instance.
"""
@@ -277,16 +278,23 @@ class PgProtocol:
if schema:
res = f"{res} options='-c search_path={schema}'"
if statement_timeout_ms:
res = f"{res} options='-c statement_timeout={statement_timeout_ms}'"
return res
# autocommit=True here by default because that's what we need most of the time
def connect(self,
*,
autocommit=True,
dbname: Optional[str] = None,
schema: Optional[str] = None,
username: Optional[str] = None,
password: Optional[str] = None) -> PgConnection:
def connect(
self,
*,
autocommit=True,
dbname: Optional[str] = None,
schema: Optional[str] = None,
username: Optional[str] = None,
password: Optional[str] = None,
# individual statement timeout in seconds, 2 minutes should be enough for our tests
statement_timeout: Optional[int] = 120
) -> PgConnection:
"""
Connect to the node.
Returns psycopg2's connection object.
@@ -294,12 +302,12 @@ class PgProtocol:
"""
conn = psycopg2.connect(
self.connstr(
dbname=dbname,
schema=schema,
username=username,
password=password,
))
self.connstr(dbname=dbname,
schema=schema,
username=username,
password=password,
statement_timeout_ms=statement_timeout *
1000 if statement_timeout else None))
# WARNING: this setting affects *all* tests!
conn.autocommit = autocommit
return conn
@@ -425,7 +433,8 @@ class ZenithEnvBuilder:
num_safekeepers: int = 0,
pageserver_auth_enabled: bool = False,
rust_log_override: Optional[str] = None,
default_branch_name=DEFAULT_BRANCH_NAME):
default_branch_name=DEFAULT_BRANCH_NAME,
broker: bool = False):
self.repo_dir = repo_dir
self.rust_log_override = rust_log_override
self.port_distributor = port_distributor
@@ -434,6 +443,7 @@ class ZenithEnvBuilder:
self.num_safekeepers = num_safekeepers
self.pageserver_auth_enabled = pageserver_auth_enabled
self.default_branch_name = default_branch_name
self.broker = broker
self.env: Optional[ZenithEnv] = None
self.s3_mock_server: Optional[MockS3Server] = None
@@ -509,6 +519,8 @@ class ZenithEnvBuilder:
self.env.pageserver.stop(immediate=True)
if self.s3_mock_server:
self.s3_mock_server.kill()
if self.env.broker is not None:
self.env.broker.stop()
class ZenithEnv:
@@ -561,6 +573,16 @@ class ZenithEnv:
default_tenant_id = '{self.initial_tenant.hex}'
""")
self.broker = None
if config.broker:
# keep etcd datadir inside 'repo'
self.broker = Etcd(datadir=os.path.join(self.repo_dir, "etcd"),
port=self.port_distributor.get_port(),
peer_port=self.port_distributor.get_port())
toml += textwrap.dedent(f"""
broker_endpoints = 'http://127.0.0.1:{self.broker.port}'
""")
# Create config for pageserver
pageserver_port = PageserverPort(
pg=self.port_distributor.get_port(),
@@ -603,12 +625,15 @@ class ZenithEnv:
self.zenith_cli.init(toml)
def start(self):
# Start up the page server and all the safekeepers
# Start up the page server, all the safekeepers and the broker
self.pageserver.start()
for safekeeper in self.safekeepers:
safekeeper.start()
if self.broker is not None:
self.broker.start()
def get_safekeeper_connstrs(self) -> str:
""" Get list of safekeeper endpoints suitable for wal_acceptors GUC """
return ','.join([f'localhost:{wa.port.pg}' for wa in self.safekeepers])
@@ -1666,6 +1691,7 @@ class Safekeeper:
class SafekeeperTimelineStatus:
acceptor_epoch: int
flush_lsn: str
remote_consistent_lsn: str
@dataclass
@@ -1674,6 +1700,7 @@ class SafekeeperMetrics:
# As a consequence, values may differ from real original int64s.
flush_lsn_inexact: Dict[Tuple[str, str], int] = field(default_factory=dict)
commit_lsn_inexact: Dict[Tuple[str, str], int] = field(default_factory=dict)
flush_wal_count: Dict[Tuple[str, str], int] = field(default_factory=dict)
class SafekeeperHttpClient(requests.Session):
@@ -1689,7 +1716,8 @@ class SafekeeperHttpClient(requests.Session):
res.raise_for_status()
resj = res.json()
return SafekeeperTimelineStatus(acceptor_epoch=resj['acceptor_state']['epoch'],
flush_lsn=resj['flush_lsn'])
flush_lsn=resj['flush_lsn'],
remote_consistent_lsn=resj['remote_consistent_lsn'])
def get_metrics(self) -> SafekeeperMetrics:
request_result = self.get(f"http://localhost:{self.port}/metrics")
@@ -1707,9 +1735,62 @@ class SafekeeperHttpClient(requests.Session):
all_metrics_text,
re.MULTILINE):
metrics.commit_lsn_inexact[(match.group(1), match.group(2))] = int(match.group(3))
for match in re.finditer(
r'^safekeeper_flush_wal_seconds_count{tenant_id="([0-9a-f]+)",timeline_id="([0-9a-f]+)"} (\S+)$',
all_metrics_text,
re.MULTILINE):
metrics.flush_wal_count[(match.group(1), match.group(2))] = int(match.group(3))
return metrics
@dataclass
class Etcd:
""" An object managing etcd instance """
datadir: str
port: int
peer_port: int
handle: Optional[subprocess.Popen[Any]] = None # handle of running daemon
def check_status(self):
s = requests.Session()
s.mount('http://', requests.adapters.HTTPAdapter(max_retries=1)) # do not retry
s.get(f"http://localhost:{self.port}/health").raise_for_status()
def start(self):
pathlib.Path(self.datadir).mkdir(exist_ok=True)
etcd_full_path = etcd_path()
if etcd_full_path is None:
raise Exception('etcd not found')
with open(os.path.join(self.datadir, "etcd.log"), "wb") as log_file:
args = [
etcd_full_path,
f"--data-dir={self.datadir}",
f"--listen-client-urls=http://localhost:{self.port}",
f"--advertise-client-urls=http://localhost:{self.port}",
f"--listen-peer-urls=http://localhost:{self.peer_port}"
]
self.handle = subprocess.Popen(args, stdout=log_file, stderr=log_file)
# wait for start
started_at = time.time()
while True:
try:
self.check_status()
except Exception as e:
elapsed = time.time() - started_at
if elapsed > 5:
raise RuntimeError(f"timed out waiting {elapsed:.0f}s for etcd start: {e}")
time.sleep(0.5)
else:
break # success
def stop(self):
if self.handle is not None:
self.handle.terminate()
self.handle.wait()
def get_test_output_dir(request: Any) -> str:
""" Compute the working directory for an individual test. """
test_name = request.node.name

View File

@@ -0,0 +1,258 @@
from pathlib import Path
import os
from time import sleep
from typing import Callable, Dict
import uuid
from fixtures.zenith_fixtures import PgBin, PgProtocol, PortDistributor
from fixtures.log_helper import log
from fixtures.benchmark_fixture import MetricReport, ZenithBenchmarker
from batch_others.test_wal_acceptor import ProposerPostgres, SafekeeperEnv
from fixtures.utils import lsn_from_hex, mkdir_if_needed
pytest_plugins = ("fixtures.zenith_fixtures", "fixtures.benchmark_fixture")
def test_walproposer_pgbench(test_output_dir: str,
port_distributor: PortDistributor,
pg_bin: PgBin,
zenbenchmark: ZenithBenchmarker):
# Create the environment in the test-specific output dir
repo_dir = Path(os.path.join(test_output_dir, "repo"))
env = SafekeeperEnv(
repo_dir,
port_distributor,
pg_bin,
num_safekeepers=1,
)
with env:
env.init()
assert env.postgres is not None
def calc_flushes() -> int:
assert env.safekeepers is not None
assert env.tenant_id is not None
assert env.timeline_id is not None
metrics = env.safekeepers[0].http_client().get_metrics()
return int(metrics.flush_wal_count[(env.tenant_id.hex, env.timeline_id.hex)])
pgbench_env = {
"PGHOST": env.postgres.listen_addr,
"PGPORT": str(env.postgres.port),
"PGDATA": env.postgres.pg_data_dir_path(),
"PGUSER": "zenith_admin",
"PGDATABASE": "postgres"
}
run_pgbench(pg_bin, zenbenchmark, pgbench_env, calc_flushes, env.postgres)
def test_vanilla_pgbench(test_output_dir: str,
port_distributor: PortDistributor,
pg_bin: PgBin,
zenbenchmark: ZenithBenchmarker):
# Create the environment in the test-specific output dir
repo_dir = os.path.join(test_output_dir, "repo")
mkdir_if_needed(repo_dir)
pgdata_master = os.path.join(repo_dir, "pgdata_master")
pgdata_replica = os.path.join(repo_dir, "pgdata_replica")
master = ProposerPostgres(pgdata_master,
pg_bin,
uuid.uuid4(),
uuid.uuid4(),
"127.0.0.1",
port_distributor.get_port())
common_config = [
"wal_keep_size=10TB\n",
"shared_preload_libraries=zenith\n",
"zenith.page_server_connstring=''\n",
"synchronous_commit=on\n",
"max_wal_senders=10\n",
"wal_log_hints=on\n",
"max_replication_slots=10\n",
"hot_standby=on\n",
"min_wal_size=20GB\n",
"max_wal_size=40GB\n",
"checkpoint_timeout=60min\n",
"log_checkpoints=on\n",
"max_connections=100\n",
"wal_sender_timeout=0\n",
"wal_level=replica\n",
]
master.initdb()
mkdir_if_needed(master.pg_data_dir_path())
with open(master.config_file_path(), "w") as f:
cfg = [
"fsync=off\n",
f"listen_addresses = '{master.listen_addr}'\n",
f"port = '{master.port}'\n",
"synchronous_standby_names = 'ANY 1 (s1)'\n",
] + common_config
f.writelines(cfg)
with open(os.path.join(pgdata_master, "pg_hba.conf"), "w") as f:
pg_hba = """
host all all 0.0.0.0/0 trust
host all all ::/0 trust
local all all trust
host all all 127.0.0.1/32 trust
host all all ::1/128 trust
host replication all 0.0.0.0/0 trust
host replication all ::/0 trust
# Allow replication connections from localhost, by a user with the
# replication privilege.
local replication all trust
host replication all 127.0.0.1/32 trust
host replication all ::1/128 trust
"""
f.writelines(pg_hba)
master_env = {
"PGHOST": master.listen_addr,
"PGPORT": str(master.port),
"PGDATA": master.pg_data_dir_path(),
"PGUSER": "zenith_admin",
"PGDATABASE": "postgres"
}
master.start()
master.safe_psql("SELECT pg_create_physical_replication_slot('s1');")
pg_bin.run(["pg_basebackup", "-D", pgdata_replica], env=master_env)
replica = ProposerPostgres(pgdata_replica,
pg_bin,
uuid.uuid4(),
uuid.uuid4(),
"127.0.0.1",
port_distributor.get_port())
with open(replica.config_file_path(), "w") as f:
cfg = [
"fsync=on\n",
f"listen_addresses = '{replica.listen_addr}'\n",
f"port = '{replica.port}'\n",
"primary_slot_name = 's1'\n",
f"primary_conninfo = 'application_name=s1 user=zenith_admin host={master.listen_addr} channel_binding=disable port={master.port} sslmode=disable sslcompression=0 sslsni=1 ssl_min_protocol_version=TLSv1.2 gssencmode=disable krbsrvname=postgres target_session_attrs=any'\n",
] + common_config
f.writelines(cfg)
with open(os.path.join(pgdata_replica, "standby.signal"), "w") as f:
pass
replica.start()
def calc_flushes() -> int:
return int(replica.safe_psql("SELECT wal_sync FROM pg_stat_wal")[0][0])
run_pgbench(pg_bin, zenbenchmark, master_env, calc_flushes, master)
replica.stop()
master.stop()
def run_pgbench(pg_bin: PgBin,
zenbenchmark: ZenithBenchmarker,
pgbench_env: Dict[str, str],
calc_flushes: Callable[[], int],
postgres: PgProtocol,
scale=200,
conns_count=32,
pgbench_time=60):
step0_lsn = lsn_from_hex(postgres.safe_psql('select pg_current_wal_insert_lsn()')[0][0])
step0_flush_cnt = calc_flushes()
log.info(f"step0_lsn: {step0_lsn}")
log.info(f"step0_flush_cnt: {step0_flush_cnt}")
cmd = ["pgbench", "-i", "-s", str(scale)]
basepath = pg_bin.run_capture(cmd, pgbench_env)
pgbench_init = basepath + '.stderr'
with open(pgbench_init, 'r') as stdout_f:
stdout = stdout_f.readlines()
stats_str = stdout[-1]
log.info(stats_str)
init_seconds = float(stats_str.split()[2])
zenbenchmark.record("pgbench_init", init_seconds, unit="s", report=MetricReport.LOWER_IS_BETTER)
wal_init_size = get_dir_size(os.path.join(pgbench_env["PGDATA"], 'pg_wal'))
zenbenchmark.record('wal_init_size',
wal_init_size / (1024 * 1024),
'MB',
report=MetricReport.LOWER_IS_BETTER)
step1_lsn = lsn_from_hex(postgres.safe_psql('select pg_current_wal_insert_lsn()')[0][0])
step1_flush_cnt = calc_flushes()
log.info(f"step1_lsn: {step1_lsn}")
log.info(f"step1_flush_cnt: {step1_flush_cnt}")
zenbenchmark.record("init_wal_bytes_per_fsync",
(step1_lsn - step0_lsn) / (step1_flush_cnt - step0_flush_cnt),
unit="b",
report=MetricReport.HIGHER_IS_BETTER)
cmd = [
"pgbench",
"-c",
str(conns_count),
"-N",
"-P",
"1",
"-T",
str(pgbench_time),
]
basepath = pg_bin.run_capture(cmd, pgbench_env)
pgbench_run = basepath + '.stdout'
with open(pgbench_run, 'r') as stdout_f:
stdout = stdout_f.readlines()
for line in stdout:
if "number of transactions actually processed:" in line:
transactions_processed = int(line.split()[-1])
stats_str = stdout[-1]
log.info(stats_str)
pgbench_tps = float(stats_str.split()[2])
step2_lsn = lsn_from_hex(postgres.safe_psql('select pg_current_wal_insert_lsn()')[0][0])
step2_flush_cnt = calc_flushes()
log.info(f"step2_lsn: {step2_lsn}")
log.info(f"step2_flush_cnt: {step2_flush_cnt}")
zenbenchmark.record("tps_pgbench", pgbench_tps, unit="", report=MetricReport.HIGHER_IS_BETTER)
zenbenchmark.record("tx_wal_bytes_per_fsync",
(step2_lsn - step1_lsn) / (step2_flush_cnt - step1_flush_cnt),
unit="b",
report=MetricReport.HIGHER_IS_BETTER)
zenbenchmark.record("txes_per_fsync",
transactions_processed / (step2_flush_cnt - step1_flush_cnt),
unit="",
report=MetricReport.HIGHER_IS_BETTER)
wal_size = get_dir_size(os.path.join(pgbench_env["PGDATA"], 'pg_wal'))
zenbenchmark.record('wal_size',
wal_size / (1024 * 1024),
'MB',
report=MetricReport.LOWER_IS_BETTER)
def get_dir_size(path: str) -> int:
"""Return size in bytes."""
totalbytes = 0
for root, dirs, files in os.walk(path):
for name in files:
totalbytes += os.path.getsize(os.path.join(root, name))
return totalbytes

View File

@@ -15,23 +15,26 @@ tracing = "0.1.27"
clap = "3.0"
daemonize = "0.4.1"
rust-s3 = { version = "0.28", default-features = false, features = ["no-verify-ssl", "tokio-rustls-tls"] }
tokio = { version = "1.11", features = ["macros"] }
tokio = { version = "1.17", features = ["macros"] }
postgres-protocol = { git = "https://github.com/zenithdb/rust-postgres.git", rev="2949d98df52587d562986aad155dd4e889e408b7" }
postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="2949d98df52587d562986aad155dd4e889e408b7" }
anyhow = "1.0"
crc32c = "0.6.0"
humantime = "2.1.0"
walkdir = "2"
url = "2.2.2"
signal-hook = "0.3.10"
serde = { version = "1.0", features = ["derive"] }
serde_with = {version = "1.12.0"}
hex = "0.4.3"
const_format = "0.2.21"
tokio-postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="2949d98df52587d562986aad155dd4e889e408b7" }
etcd-client = "0.8.3"
postgres_ffi = { path = "../postgres_ffi" }
workspace_hack = { path = "../workspace_hack" }
zenith_metrics = { path = "../zenith_metrics" }
zenith_utils = { path = "../zenith_utils" }
workspace_hack = { version = "0.1", path = "../workspace_hack" }
[dev-dependencies]
tempfile = "3.2"

View File

@@ -11,18 +11,19 @@ use std::io::{ErrorKind, Write};
use std::path::{Path, PathBuf};
use std::thread;
use tracing::*;
use url::{ParseError, Url};
use walkeeper::control_file::{self};
use zenith_utils::http::endpoint;
use zenith_utils::zid::ZNodeId;
use zenith_utils::{logging, tcp_listener, GIT_VERSION};
use tokio::sync::mpsc;
use walkeeper::callmemaybe;
use walkeeper::defaults::{DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_PG_LISTEN_ADDR};
use walkeeper::http;
use walkeeper::s3_offload;
use walkeeper::wal_service;
use walkeeper::SafeKeeperConf;
use walkeeper::{broker, callmemaybe};
use zenith_utils::shutdown::exit_now;
use zenith_utils::signals;
@@ -104,6 +105,11 @@ fn main() -> Result<()> {
)
.arg(
Arg::new("id").long("id").takes_value(true).help("safekeeper node id: integer")
).arg(
Arg::new("broker-endpoints")
.long("broker-endpoints")
.takes_value(true)
.help("a comma separated broker (etcd) endpoints for storage nodes coordination, e.g. 'http://127.0.0.1:2379'"),
)
.get_matches();
@@ -154,6 +160,11 @@ fn main() -> Result<()> {
));
}
if let Some(addr) = arg_matches.value_of("broker-endpoints") {
let collected_ep: Result<Vec<Url>, ParseError> = addr.split(',').map(Url::parse).collect();
conf.broker_endpoints = Some(collected_ep?);
}
start_safekeeper(conf, given_id, arg_matches.is_present("init"))
}
@@ -259,11 +270,12 @@ fn start_safekeeper(mut conf: SafeKeeperConf, given_id: Option<ZNodeId>, init: b
threads.push(wal_acceptor_thread);
let conf_cloned = conf.clone();
let callmemaybe_thread = thread::Builder::new()
.name("callmemaybe thread".into())
.spawn(|| {
// thread code
let thread_result = callmemaybe::thread_main(conf, rx);
let thread_result = callmemaybe::thread_main(conf_cloned, rx);
if let Err(e) = thread_result {
error!("callmemaybe thread terminated: {}", e);
}
@@ -271,6 +283,17 @@ fn start_safekeeper(mut conf: SafeKeeperConf, given_id: Option<ZNodeId>, init: b
.unwrap();
threads.push(callmemaybe_thread);
if conf.broker_endpoints.is_some() {
let conf_ = conf.clone();
threads.push(
thread::Builder::new()
.name("broker thread".into())
.spawn(|| {
broker::thread_main(conf_);
})?,
);
}
// TODO: put more thoughts into handling of failed threads
// We probably should restart them.

211
walkeeper/src/broker.rs Normal file
View File

@@ -0,0 +1,211 @@
//! Communication with etcd, providing safekeeper peers and pageserver coordination.
use anyhow::bail;
use anyhow::Context;
use anyhow::Error;
use anyhow::Result;
use etcd_client::Client;
use etcd_client::EventType;
use etcd_client::PutOptions;
use etcd_client::WatchOptions;
use lazy_static::lazy_static;
use regex::Regex;
use serde::{Deserialize, Serialize};
use serde_with::{serde_as, DisplayFromStr};
use std::str::FromStr;
use std::time::Duration;
use tokio::task::JoinHandle;
use tokio::{runtime, time::sleep};
use tracing::*;
use zenith_utils::zid::ZTenantId;
use zenith_utils::zid::ZTimelineId;
use zenith_utils::{
lsn::Lsn,
zid::{ZNodeId, ZTenantTimelineId},
};
use crate::{safekeeper::Term, timeline::GlobalTimelines, SafeKeeperConf};
const RETRY_INTERVAL_MSEC: u64 = 1000;
const PUSH_INTERVAL_MSEC: u64 = 1000;
const LEASE_TTL_SEC: i64 = 5;
// TODO: add global zenith installation ID.
const ZENITH_PREFIX: &str = "zenith";
/// Published data about safekeeper. Fields made optional for easy migrations.
#[serde_as]
#[derive(Deserialize, Serialize)]
pub struct SafekeeperInfo {
/// Term of the last entry.
pub last_log_term: Option<Term>,
/// LSN of the last record.
#[serde_as(as = "Option<DisplayFromStr>")]
pub flush_lsn: Option<Lsn>,
/// Up to which LSN safekeeper regards its WAL as committed.
#[serde_as(as = "Option<DisplayFromStr>")]
pub commit_lsn: Option<Lsn>,
/// LSN up to which safekeeper offloaded WAL to s3.
#[serde_as(as = "Option<DisplayFromStr>")]
pub s3_wal_lsn: Option<Lsn>,
/// LSN of last checkpoint uploaded by pageserver.
#[serde_as(as = "Option<DisplayFromStr>")]
pub remote_consistent_lsn: Option<Lsn>,
#[serde_as(as = "Option<DisplayFromStr>")]
pub peer_horizon_lsn: Option<Lsn>,
}
pub fn thread_main(conf: SafeKeeperConf) {
let runtime = runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let _enter = info_span!("broker").entered();
info!("started, broker endpoints {:?}", conf.broker_endpoints);
runtime.block_on(async {
main_loop(conf).await;
});
}
/// Prefix to timeline related data.
fn timeline_path(zttid: &ZTenantTimelineId) -> String {
format!(
"{}/{}/{}",
ZENITH_PREFIX, zttid.tenant_id, zttid.timeline_id
)
}
/// Key to per timeline per safekeeper data.
fn timeline_safekeeper_path(zttid: &ZTenantTimelineId, sk_id: ZNodeId) -> String {
format!("{}/safekeeper/{}", timeline_path(zttid), sk_id)
}
/// Push once in a while data about all active timelines to the broker.
async fn push_loop(conf: SafeKeeperConf) -> Result<()> {
let mut client = Client::connect(conf.broker_endpoints.as_ref().unwrap(), None).await?;
// Get and maintain lease to automatically delete obsolete data
let lease = client.lease_grant(LEASE_TTL_SEC, None).await?;
let (mut keeper, mut ka_stream) = client.lease_keep_alive(lease.id()).await?;
let push_interval = Duration::from_millis(PUSH_INTERVAL_MSEC);
loop {
// Note: we lock runtime here and in timeline methods as GlobalTimelines
// is under plain mutex. That's ok, all this code is not performance
// sensitive and there is no risk of deadlock as we don't await while
// lock is held.
let active_tlis = GlobalTimelines::get_active_timelines();
for zttid in &active_tlis {
if let Ok(tli) = GlobalTimelines::get(&conf, *zttid, false) {
let sk_info = tli.get_public_info();
let put_opts = PutOptions::new().with_lease(lease.id());
client
.put(
timeline_safekeeper_path(zttid, conf.my_id),
serde_json::to_string(&sk_info)?,
Some(put_opts),
)
.await
.context("failed to push safekeeper info")?;
}
}
// revive the lease
keeper
.keep_alive()
.await
.context("failed to send LeaseKeepAliveRequest")?;
ka_stream
.message()
.await
.context("failed to receive LeaseKeepAliveResponse")?;
sleep(push_interval).await;
}
}
/// Subscribe and fetch all the interesting data from the broker.
async fn pull_loop(conf: SafeKeeperConf) -> Result<()> {
lazy_static! {
static ref TIMELINE_SAFEKEEPER_RE: Regex =
Regex::new(r"^zenith/([[:xdigit:]]+)/([[:xdigit:]]+)/safekeeper/([[:digit:]])$")
.unwrap();
}
let mut client = Client::connect(conf.broker_endpoints.as_ref().unwrap(), None).await?;
loop {
let wo = WatchOptions::new().with_prefix();
// TODO: subscribe only to my timelines
let (_, mut stream) = client.watch(ZENITH_PREFIX, Some(wo)).await?;
while let Some(resp) = stream.message().await? {
if resp.canceled() {
bail!("watch canceled");
}
for event in resp.events() {
if EventType::Put == event.event_type() {
if let Some(kv) = event.kv() {
if let Some(caps) = TIMELINE_SAFEKEEPER_RE.captures(kv.key_str()?) {
let tenant_id = ZTenantId::from_str(caps.get(1).unwrap().as_str())?;
let timeline_id = ZTimelineId::from_str(caps.get(2).unwrap().as_str())?;
let zttid = ZTenantTimelineId::new(tenant_id, timeline_id);
let safekeeper_id = ZNodeId(caps.get(3).unwrap().as_str().parse()?);
let value_str = kv.value_str()?;
match serde_json::from_str::<SafekeeperInfo>(value_str) {
Ok(safekeeper_info) => {
if let Ok(tli) = GlobalTimelines::get(&conf, zttid, false) {
tli.record_safekeeper_info(&safekeeper_info, safekeeper_id)?
}
}
Err(err) => warn!(
"failed to deserialize safekeeper info {}: {}",
value_str, err
),
}
}
}
}
}
}
}
}
async fn main_loop(conf: SafeKeeperConf) {
let mut ticker = tokio::time::interval(Duration::from_millis(RETRY_INTERVAL_MSEC));
let mut push_handle: Option<JoinHandle<Result<(), Error>>> = None;
let mut pull_handle: Option<JoinHandle<Result<(), Error>>> = None;
// Selecting on JoinHandles requires some squats; is there a better way to
// reap tasks individually?
// Handling failures in task itself won't catch panic and in Tokio, task's
// panic doesn't kill the whole executor, so it is better to do reaping
// here.
loop {
tokio::select! {
res = async { push_handle.as_mut().unwrap().await }, if push_handle.is_some() => {
// was it panic or normal error?
let err = match res {
Ok(res_internal) => res_internal.unwrap_err(),
Err(err_outer) => err_outer.into(),
};
warn!("push task failed: {:?}", err);
push_handle = None;
},
res = async { pull_handle.as_mut().unwrap().await }, if pull_handle.is_some() => {
// was it panic or normal error?
let err = match res {
Ok(res_internal) => res_internal.unwrap_err(),
Err(err_outer) => err_outer.into(),
};
warn!("pull task failed: {:?}", err);
pull_handle = None;
},
_ = ticker.tick() => {
if push_handle.is_none() {
push_handle = Some(tokio::spawn(push_loop(conf.clone())));
}
if pull_handle.is_none() {
pull_handle = Some(tokio::spawn(pull_loop(conf.clone())));
}
}
}
}
}

View File

@@ -168,7 +168,14 @@ impl SafekeeperPostgresHandler {
fn handle_identify_system(&mut self, pgb: &mut PostgresBackend) -> Result<()> {
let start_pos = self.timeline.get().get_end_of_wal();
let lsn = start_pos.to_string();
let sysid = self.timeline.get().get_info().server.system_id.to_string();
let sysid = self
.timeline
.get()
.get_state()
.1
.server
.system_id
.to_string();
let lsn_bytes = lsn.as_bytes();
let tli = PG_TLI.to_string();
let tli_bytes = tli.as_bytes();

View File

@@ -86,23 +86,24 @@ async fn timeline_status_handler(request: Request<Body>) -> Result<Response<Body
);
let tli = GlobalTimelines::get(get_conf(&request), zttid, false).map_err(ApiError::from_err)?;
let sk_state = tli.get_info();
let (inmem, state) = tli.get_state();
let flush_lsn = tli.get_end_of_wal();
let acc_state = AcceptorStateStatus {
term: sk_state.acceptor_state.term,
epoch: sk_state.acceptor_state.get_epoch(flush_lsn),
term_history: sk_state.acceptor_state.term_history,
term: state.acceptor_state.term,
epoch: state.acceptor_state.get_epoch(flush_lsn),
term_history: state.acceptor_state.term_history,
};
// Note: we report in memory values which can be lost.
let status = TimelineStatus {
tenant_id: zttid.tenant_id,
timeline_id: zttid.timeline_id,
acceptor_state: acc_state,
commit_lsn: sk_state.commit_lsn,
s3_wal_lsn: sk_state.s3_wal_lsn,
peer_horizon_lsn: sk_state.peer_horizon_lsn,
remote_consistent_lsn: sk_state.remote_consistent_lsn,
commit_lsn: inmem.commit_lsn,
s3_wal_lsn: inmem.s3_wal_lsn,
peer_horizon_lsn: inmem.peer_horizon_lsn,
remote_consistent_lsn: inmem.remote_consistent_lsn,
flush_lsn,
};
Ok(json_response(StatusCode::OK, status)?)

View File

@@ -73,7 +73,7 @@ pub fn handle_json_ctrl(
let inserted_wal = append_logical_message(spg, append_request)?;
let response = AppendResult {
state: spg.timeline.get().get_info(),
state: spg.timeline.get().get_state().1,
inserted_wal,
};
let response_data = serde_json::to_vec(&response)?;
@@ -112,7 +112,7 @@ fn prepare_safekeeper(spg: &mut SafekeeperPostgresHandler) -> Result<()> {
fn send_proposer_elected(spg: &mut SafekeeperPostgresHandler, term: Term, lsn: Lsn) -> Result<()> {
// add new term to existing history
let history = spg.timeline.get().get_info().acceptor_state.term_history;
let history = spg.timeline.get().get_state().1.acceptor_state.term_history;
let history = history.up_to(lsn.checked_sub(1u64).unwrap());
let mut history_entries = history.0;
history_entries.push(TermSwitchEntry { term, lsn });
@@ -142,7 +142,7 @@ fn append_logical_message(
msg: &AppendLogicalMessage,
) -> Result<InsertedWAL> {
let wal_data = encode_logical_message(&msg.lm_prefix, &msg.lm_message);
let sk_state = spg.timeline.get().get_info();
let sk_state = spg.timeline.get().get_state().1;
let begin_lsn = msg.begin_lsn;
let end_lsn = begin_lsn + wal_data.len() as u64;

View File

@@ -1,9 +1,11 @@
//
use std::path::PathBuf;
use std::time::Duration;
use url::Url;
use zenith_utils::zid::{ZNodeId, ZTenantTimelineId};
pub mod broker;
pub mod callmemaybe;
pub mod control_file;
pub mod control_file_upgrade;
@@ -47,6 +49,7 @@ pub struct SafeKeeperConf {
pub ttl: Option<Duration>,
pub recall_period: Duration,
pub my_id: ZNodeId,
pub broker_endpoints: Option<Vec<Url>>,
}
impl SafeKeeperConf {
@@ -71,6 +74,7 @@ impl Default for SafeKeeperConf {
ttl: None,
recall_period: defaults::DEFAULT_RECALL_PERIOD,
my_id: ZNodeId(0),
broker_endpoints: None,
}
}
}

View File

@@ -193,7 +193,7 @@ pub struct SafeKeeperState {
pub peer_horizon_lsn: Lsn,
/// LSN of the oldest known checkpoint made by pageserver and successfully
/// pushed to s3. We don't remove WAL beyond it. Persisted only for
/// informational purposes, we receive it from pageserver.
/// informational purposes, we receive it from pageserver (or broker).
pub remote_consistent_lsn: Lsn,
// Peers and their state as we remember it. Knowing peers themselves is
// fundamental; but state is saved here only for informational purposes and
@@ -202,6 +202,16 @@ pub struct SafeKeeperState {
pub peers: Peers,
}
#[derive(Debug, Clone)]
// In memory safekeeper state. Fields mirror ones in `SafeKeeperState`; values
// are not flushed yet.
pub struct SafekeeperMemState {
pub commit_lsn: Lsn,
pub s3_wal_lsn: Lsn, // TODO: keep only persistent version
pub peer_horizon_lsn: Lsn,
pub remote_consistent_lsn: Lsn,
}
impl SafeKeeperState {
pub fn new(zttid: &ZTenantTimelineId, peers: Vec<ZNodeId>) -> SafeKeeperState {
SafeKeeperState {
@@ -470,14 +480,12 @@ struct SafeKeeperMetrics {
}
impl SafeKeeperMetrics {
fn new(tenant_id: ZTenantId, timeline_id: ZTimelineId, commit_lsn: Lsn) -> Self {
fn new(tenant_id: ZTenantId, timeline_id: ZTimelineId) -> Self {
let tenant_id = tenant_id.to_string();
let timeline_id = timeline_id.to_string();
let m = Self {
Self {
commit_lsn: COMMIT_LSN_GAUGE.with_label_values(&[&tenant_id, &timeline_id]),
};
m.commit_lsn.set(u64::from(commit_lsn) as f64);
m
}
}
}
@@ -487,10 +495,14 @@ pub struct SafeKeeper<CTRL: control_file::Storage, WAL: wal_storage::Storage> {
// Cached metrics so we don't have to recompute labels on each update.
metrics: SafeKeeperMetrics,
/// not-yet-flushed pairs of same named fields in s.*
pub commit_lsn: Lsn,
pub peer_horizon_lsn: Lsn,
pub s: SafeKeeperState, // persistent part
/// Maximum commit_lsn between all nodes, can be ahead of local flush_lsn.
pub global_commit_lsn: Lsn,
/// LSN since the proposer safekeeper currently talking to appends WAL;
/// determines epoch switch point.
epoch_start_lsn: Lsn,
pub inmem: SafekeeperMemState, // in memory part
pub s: SafeKeeperState, // persistent part
pub control_store: CTRL,
pub wal_store: WAL,
@@ -513,9 +525,15 @@ where
}
SafeKeeper {
metrics: SafeKeeperMetrics::new(state.tenant_id, ztli, state.commit_lsn),
commit_lsn: state.commit_lsn,
peer_horizon_lsn: state.peer_horizon_lsn,
metrics: SafeKeeperMetrics::new(state.tenant_id, ztli),
global_commit_lsn: state.commit_lsn,
epoch_start_lsn: Lsn(0),
inmem: SafekeeperMemState {
commit_lsn: state.commit_lsn,
s3_wal_lsn: state.s3_wal_lsn,
peer_horizon_lsn: state.peer_horizon_lsn,
remote_consistent_lsn: state.remote_consistent_lsn,
},
s: state,
control_store,
wal_store,
@@ -530,8 +548,7 @@ where
.up_to(self.wal_store.flush_lsn())
}
#[cfg(test)]
fn get_epoch(&self) -> Term {
pub fn get_epoch(&self) -> Term {
self.s.acceptor_state.get_epoch(self.wal_store.flush_lsn())
}
@@ -602,9 +619,6 @@ where
// pass wal_seg_size to read WAL and find flush_lsn
self.wal_store.init_storage(&self.s)?;
// update tenant_id/timeline_id in metrics
self.metrics = SafeKeeperMetrics::new(msg.tenant_id, msg.ztli, self.commit_lsn);
info!(
"processed greeting from proposer {:?}, sending term {:?}",
msg.proposer_id, self.s.acceptor_state.term
@@ -684,12 +698,49 @@ where
Ok(None)
}
/// Advance commit_lsn taking into account what we have locally
pub fn update_commit_lsn(&mut self) -> Result<()> {
let commit_lsn = min(self.global_commit_lsn, self.wal_store.flush_lsn());
assert!(commit_lsn >= self.inmem.commit_lsn);
self.inmem.commit_lsn = commit_lsn;
self.metrics.commit_lsn.set(self.inmem.commit_lsn.0 as f64);
// If new commit_lsn reached epoch switch, force sync of control
// file: walproposer in sync mode is very interested when this
// happens. Note: this is for sync-safekeepers mode only, as
// otherwise commit_lsn might jump over epoch_start_lsn.
// Also note that commit_lsn can reach epoch_start_lsn earlier
// that we receive new epoch_start_lsn, and we still need to sync
// control file in this case.
if commit_lsn == self.epoch_start_lsn && self.s.commit_lsn != commit_lsn {
self.persist_control_file()?;
}
// We got our first commit_lsn, which means we should sync
// everything to disk, to initialize the state.
if self.s.commit_lsn == Lsn(0) && commit_lsn > Lsn(0) {
self.wal_store.flush_wal()?;
self.persist_control_file()?;
}
Ok(())
}
/// Persist in-memory state to the disk.
fn persist_control_file(&mut self) -> Result<()> {
self.s.commit_lsn = self.inmem.commit_lsn;
self.s.peer_horizon_lsn = self.inmem.peer_horizon_lsn;
self.control_store.persist(&self.s)
}
/// Handle request to append WAL.
#[allow(clippy::comparison_chain)]
fn handle_append_request(
&mut self,
msg: &AppendRequest,
mut require_flush: bool,
require_flush: bool,
) -> Result<Option<AcceptorProposerMessage>> {
if self.s.acceptor_state.term < msg.h.term {
bail!("got AppendRequest before ProposerElected");
@@ -701,25 +752,22 @@ where
return Ok(Some(AcceptorProposerMessage::AppendResponse(resp)));
}
// After ProposerElected, which performs truncation, we should get only
// indeed append requests (but flush_lsn is advanced only on record
// boundary, so might be less).
assert!(self.wal_store.flush_lsn() <= msg.h.begin_lsn);
// Now we know that we are in the same term as the proposer,
// processing the message.
self.epoch_start_lsn = msg.h.epoch_start_lsn;
// TODO: don't update state without persisting to disk
self.s.proposer_uuid = msg.h.proposer_uuid;
let mut sync_control_file = false;
// do the job
if !msg.wal_data.is_empty() {
self.wal_store.write_wal(msg.h.begin_lsn, &msg.wal_data)?;
// If this was the first record we ever receieved, initialize
// If this was the first record we ever received, initialize
// commit_lsn to help find_end_of_wal skip the hole in the
// beginning.
if self.s.commit_lsn == Lsn(0) {
self.s.commit_lsn = msg.h.begin_lsn;
sync_control_file = true;
require_flush = true;
if self.global_commit_lsn == Lsn(0) {
self.global_commit_lsn = msg.h.begin_lsn;
}
}
@@ -728,35 +776,22 @@ where
self.wal_store.flush_wal()?;
}
// Advance commit_lsn taking into account what we have locally.
// commit_lsn can be 0, being unknown to new walproposer while he hasn't
// collected majority of its epoch acks yet, ignore it in this case.
// Update global_commit_lsn, verifying that it cannot decrease.
if msg.h.commit_lsn != Lsn(0) {
let commit_lsn = min(msg.h.commit_lsn, self.wal_store.flush_lsn());
// If new commit_lsn reached epoch switch, force sync of control
// file: walproposer in sync mode is very interested when this
// happens. Note: this is for sync-safekeepers mode only, as
// otherwise commit_lsn might jump over epoch_start_lsn.
sync_control_file |= commit_lsn == msg.h.epoch_start_lsn;
self.commit_lsn = commit_lsn;
self.metrics
.commit_lsn
.set(u64::from(self.commit_lsn) as f64);
assert!(msg.h.commit_lsn >= self.global_commit_lsn);
self.global_commit_lsn = msg.h.commit_lsn;
}
self.peer_horizon_lsn = msg.h.truncate_lsn;
self.inmem.peer_horizon_lsn = msg.h.truncate_lsn;
self.update_commit_lsn()?;
// Update truncate and commit LSN in control file.
// To avoid negative impact on performance of extra fsync, do it only
// when truncate_lsn delta exceeds WAL segment size.
sync_control_file |=
self.s.peer_horizon_lsn + (self.s.server.wal_seg_size as u64) < self.peer_horizon_lsn;
if sync_control_file {
self.s.commit_lsn = self.commit_lsn;
self.s.peer_horizon_lsn = self.peer_horizon_lsn;
}
if sync_control_file {
self.control_store.persist(&self.s)?;
if self.s.peer_horizon_lsn + (self.s.server.wal_seg_size as u64)
< self.inmem.peer_horizon_lsn
{
self.persist_control_file()?;
}
trace!(
@@ -780,6 +815,10 @@ where
/// Flush WAL to disk. Return AppendResponse with latest LSNs.
fn handle_flush(&mut self) -> Result<Option<AcceptorProposerMessage>> {
self.wal_store.flush_wal()?;
// commit_lsn can be updated because we have new flushed data locally.
self.update_commit_lsn()?;
Ok(Some(AcceptorProposerMessage::AppendResponse(
self.append_response(),
)))

View File

@@ -230,7 +230,7 @@ impl ReplicationConn {
let mut wal_seg_size: usize;
loop {
wal_seg_size = spg.timeline.get().get_info().server.wal_seg_size as usize;
wal_seg_size = spg.timeline.get().get_state().1.server.wal_seg_size as usize;
if wal_seg_size == 0 {
error!("Cannot start replication before connecting to wal_proposer");
sleep(Duration::from_secs(1));

View File

@@ -17,12 +17,14 @@ use tracing::*;
use zenith_utils::lsn::Lsn;
use zenith_utils::zid::{ZNodeId, ZTenantTimelineId};
use crate::broker::SafekeeperInfo;
use crate::callmemaybe::{CallmeEvent, SubscriptionStateKey};
use crate::control_file;
use crate::control_file::Storage as cf_storage;
use crate::safekeeper::{
AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, SafeKeeperState,
SafekeeperMemState,
};
use crate::send_wal::HotStandbyFeedback;
use crate::wal_storage;
@@ -340,7 +342,7 @@ impl Timeline {
let replica_state = shared_state.replicas[replica_id].unwrap();
let deactivate = shared_state.notified_commit_lsn == Lsn(0) || // no data at all yet
(replica_state.last_received_lsn != Lsn::MAX && // Lsn::MAX means that we don't know the latest LSN yet.
replica_state.last_received_lsn >= shared_state.sk.commit_lsn);
replica_state.last_received_lsn >= shared_state.sk.inmem.commit_lsn);
if deactivate {
shared_state.deactivate(&self.zttid, callmemaybe_tx)?;
return Ok(true);
@@ -349,6 +351,11 @@ impl Timeline {
Ok(false)
}
fn is_active(&self) -> bool {
let shared_state = self.mutex.lock().unwrap();
shared_state.active
}
/// Timed wait for an LSN to be committed.
///
/// Returns the last committed LSN, which will be at least
@@ -394,7 +401,7 @@ impl Timeline {
rmsg = shared_state.sk.process_msg(msg)?;
// locally available commit lsn. flush_lsn can be smaller than
// commit_lsn if we are catching up safekeeper.
commit_lsn = shared_state.sk.commit_lsn;
commit_lsn = shared_state.sk.inmem.commit_lsn;
// if this is AppendResponse, fill in proper hot standby feedback and disk consistent lsn
if let Some(AcceptorProposerMessage::AppendResponse(ref mut resp)) = rmsg {
@@ -410,8 +417,61 @@ impl Timeline {
Ok(rmsg)
}
pub fn get_info(&self) -> SafeKeeperState {
self.mutex.lock().unwrap().sk.s.clone()
pub fn get_state(&self) -> (SafekeeperMemState, SafeKeeperState) {
let shared_state = self.mutex.lock().unwrap();
(shared_state.sk.inmem.clone(), shared_state.sk.s.clone())
}
/// Prepare public safekeeper info for reporting.
pub fn get_public_info(&self) -> SafekeeperInfo {
let shared_state = self.mutex.lock().unwrap();
SafekeeperInfo {
last_log_term: Some(shared_state.sk.get_epoch()),
flush_lsn: Some(shared_state.sk.wal_store.flush_lsn()),
// note: this value is not flushed to control file yet and can be lost
commit_lsn: Some(shared_state.sk.inmem.commit_lsn),
s3_wal_lsn: Some(shared_state.sk.inmem.s3_wal_lsn),
// TODO: rework feedbacks to avoid max here
remote_consistent_lsn: Some(max(
shared_state.get_replicas_state().remote_consistent_lsn,
shared_state.sk.inmem.remote_consistent_lsn,
)),
peer_horizon_lsn: Some(shared_state.sk.inmem.peer_horizon_lsn),
}
}
/// Update timeline state with peer safekeeper data.
pub fn record_safekeeper_info(&self, sk_info: &SafekeeperInfo, _sk_id: ZNodeId) -> Result<()> {
let mut shared_state = self.mutex.lock().unwrap();
// Note: the check is too restrictive, generally we can update local
// commit_lsn if our history matches (is part of) history of advanced
// commit_lsn provider.
if let (Some(commit_lsn), Some(last_log_term)) = (sk_info.commit_lsn, sk_info.last_log_term)
{
if last_log_term == shared_state.sk.get_epoch() {
shared_state.sk.global_commit_lsn =
max(commit_lsn, shared_state.sk.global_commit_lsn);
shared_state.sk.update_commit_lsn()?;
let local_commit_lsn = min(commit_lsn, shared_state.sk.wal_store.flush_lsn());
shared_state.sk.inmem.commit_lsn =
max(local_commit_lsn, shared_state.sk.inmem.commit_lsn);
}
}
if let Some(s3_wal_lsn) = sk_info.s3_wal_lsn {
shared_state.sk.inmem.s3_wal_lsn = max(s3_wal_lsn, shared_state.sk.inmem.s3_wal_lsn);
}
if let Some(remote_consistent_lsn) = sk_info.remote_consistent_lsn {
shared_state.sk.inmem.remote_consistent_lsn = max(
remote_consistent_lsn,
shared_state.sk.inmem.remote_consistent_lsn,
);
}
if let Some(peer_horizon_lsn) = sk_info.peer_horizon_lsn {
shared_state.sk.inmem.peer_horizon_lsn =
max(peer_horizon_lsn, shared_state.sk.inmem.peer_horizon_lsn);
}
// TODO: sync control file
Ok(())
}
pub fn add_replica(&self, state: ReplicaState) -> usize {
@@ -495,7 +555,7 @@ impl GlobalTimelines {
}
/// Get a timeline with control file loaded from the global TIMELINES map.
/// If control file doesn't exist, bails out.
/// If control file doesn't exist and create=false, bails out.
pub fn get(
conf: &SafeKeeperConf,
zttid: ZTenantTimelineId,
@@ -537,4 +597,14 @@ impl GlobalTimelines {
}
}
}
/// Get ZTenantTimelineIDs of all active timelines.
pub fn get_active_timelines() -> Vec<ZTenantTimelineId> {
let timelines = TIMELINES.lock().unwrap();
timelines
.iter()
.filter(|&(_, tli)| tli.is_active())
.map(|(zttid, _)| *zttid)
.collect()
}
}

View File

@@ -1,22 +1,50 @@
# This file is generated by `cargo hakari`.
# To regenerate, run:
# cargo hakari generate
[package]
name = "workspace_hack"
version = "0.1.0"
edition = "2021"
description = "workspace-hack package, managed by hakari"
# You can choose to publish this crate: see https://docs.rs/cargo-hakari/latest/cargo_hakari/publishing.
publish = false
[target.'cfg(all())'.dependencies]
libc = { version = "0.2", features = ["default", "extra_traits", "std"] }
memchr = { version = "2", features = ["default", "std", "use_std"] }
# The parts of the file between the BEGIN HAKARI SECTION and END HAKARI SECTION comments
# are managed by hakari.
### BEGIN HAKARI SECTION
[dependencies]
anyhow = { version = "1", features = ["backtrace", "std"] }
bytes = { version = "1", features = ["serde", "std"] }
clap = { version = "2", features = ["ansi_term", "atty", "color", "strsim", "suggestions", "vec_map"] }
either = { version = "1", features = ["use_std"] }
hashbrown = { version = "0.11", features = ["ahash", "inline-more", "raw"] }
libc = { version = "0.2", features = ["extra_traits", "std"] }
log = { version = "0.4", default-features = false, features = ["serde", "std"] }
memchr = { version = "2", features = ["std", "use_std"] }
num-integer = { version = "0.1", default-features = false, features = ["std"] }
num-traits = { version = "0.2", default-features = false, features = ["std"] }
regex = { version = "1", features = ["aho-corasick", "default", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
regex-syntax = { version = "0.6", features = ["default", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
serde = { version = "1", features = ["default", "derive", "serde_derive", "std"] }
num-traits = { version = "0.2", features = ["std"] }
regex = { version = "1", features = ["aho-corasick", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
regex-syntax = { version = "0.6", features = ["unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
reqwest = { version = "0.11", default-features = false, features = ["__rustls", "__tls", "blocking", "hyper-rustls", "json", "rustls", "rustls-pemfile", "rustls-tls", "rustls-tls-webpki-roots", "serde_json", "stream", "tokio-rustls", "tokio-util", "webpki-roots"] }
scopeguard = { version = "1", features = ["use_std"] }
serde = { version = "1", features = ["alloc", "derive", "serde_derive", "std"] }
tokio = { version = "1", features = ["bytes", "fs", "io-util", "libc", "macros", "memchr", "mio", "net", "num_cpus", "once_cell", "process", "rt", "rt-multi-thread", "signal-hook-registry", "sync", "time", "tokio-macros"] }
tracing = { version = "0.1", features = ["attributes", "std", "tracing-attributes"] }
tracing-core = { version = "0.1", features = ["lazy_static", "std"] }
[target.'cfg(all())'.build-dependencies]
libc = { version = "0.2", features = ["default", "extra_traits", "std"] }
memchr = { version = "2", features = ["default", "std", "use_std"] }
proc-macro2 = { version = "1", features = ["default", "proc-macro"] }
quote = { version = "1", features = ["default", "proc-macro"] }
regex = { version = "1", features = ["aho-corasick", "default", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
regex-syntax = { version = "0.6", features = ["default", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
syn = { version = "1", features = ["clone-impls", "default", "derive", "full", "parsing", "printing", "proc-macro", "quote", "visit", "visit-mut"] }
[build-dependencies]
cc = { version = "1", default-features = false, features = ["jobserver", "parallel"] }
clap = { version = "2", features = ["ansi_term", "atty", "color", "strsim", "suggestions", "vec_map"] }
either = { version = "1", features = ["use_std"] }
libc = { version = "0.2", features = ["extra_traits", "std"] }
log = { version = "0.4", default-features = false, features = ["serde", "std"] }
memchr = { version = "2", features = ["std", "use_std"] }
proc-macro2 = { version = "1", features = ["proc-macro"] }
quote = { version = "1", features = ["proc-macro"] }
regex = { version = "1", features = ["aho-corasick", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
regex-syntax = { version = "0.6", features = ["unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
serde = { version = "1", features = ["alloc", "derive", "serde_derive", "std"] }
syn = { version = "1", features = ["clone-impls", "derive", "extra-traits", "full", "parsing", "printing", "proc-macro", "quote", "visit", "visit-mut"] }
### END HAKARI SECTION

View File

@@ -1,23 +1 @@
//! This crate contains no code.
//!
//! The workspace_hack crate exists only to pin down some dependencies,
//! so that those dependencies always build with the same features,
//! under a few different cases that can be problematic:
//! - Running `cargo check` or `cargo build` from a crate sub-directory
//! instead of the workspace root.
//! - Running `cargo install`, which can only be done per-crate
//!
//! The dependency lists in Cargo.toml were automatically generated by
//! a tool called
//! [Hakari](https://github.com/facebookincubator/cargo-guppy/tree/main/tools/hakari).
//!
//! Hakari doesn't have a CLI yet; in the meantime the example code in
//! their `README` file is enough to regenerate the dependencies.
//! Hakari's output was pasted into Cargo.toml, except for the
//! following manual edits:
//! - `winapi` dependency was removed. This is probably just due to the
//! fact that Hakari's target analysis is incomplete.
//!
//! There isn't any penalty to this data falling out of date; it just
//! means that under the conditions above Cargo will rebuild more
//! packages than strictly necessary.
// This is a stub lib.rs.

View File

@@ -15,4 +15,4 @@ control_plane = { path = "../control_plane" }
walkeeper = { path = "../walkeeper" }
postgres_ffi = { path = "../postgres_ffi" }
zenith_utils = { path = "../zenith_utils" }
workspace_hack = { path = "../workspace_hack" }
workspace_hack = { version = "0.1", path = "../workspace_hack" }

View File

@@ -8,3 +8,4 @@ prometheus = {version = "0.13", default_features=false} # removes protobuf depen
libc = "0.2"
lazy_static = "1.4"
once_cell = "1.8.0"
workspace_hack = { version = "0.1", path = "../workspace_hack" }

View File

@@ -16,7 +16,7 @@ routerify = "3"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1"
thiserror = "1.0"
tokio = { version = "1.11", features = ["macros"]}
tokio = { version = "1.17", features = ["macros"]}
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
nix = "0.23.0"
@@ -30,7 +30,7 @@ git-version = "0.3.5"
serde_with = "1.12.0"
zenith_metrics = { path = "../zenith_metrics" }
workspace_hack = { path = "../workspace_hack" }
workspace_hack = { version = "0.1", path = "../workspace_hack" }
[dev-dependencies]
byteorder = "1.4.3"

View File

@@ -160,7 +160,7 @@ pub fn serve_thread_main<S>(
where
S: Future<Output = ()> + Send + Sync,
{
info!("Starting a http endpoint at {}", listener.local_addr()?);
info!("Starting an HTTP endpoint at {}", listener.local_addr()?);
// Create a Service from the router above to handle incoming requests.
let service = RouterService::new(router_builder.build().map_err(|err| anyhow!(err))?).unwrap();