From ba186767f70905f8b14fabe5387dbc1b8b1bdf0b Mon Sep 17 00:00:00 2001 From: justin Date: Fri, 11 Sep 2026 15:38:16 -0500 Subject: [PATCH] persist: fix multipart blob reads and add blob store benchmarks `S3Blob::get` reads a multipart-uploaded object part by part, using the `x-amz-mp-parts-count` header from its first request to learn how many parts to fetch. A store that does not return that header is indistinguishable from one holding a single-part object, so persist fetched only the first part and handed the decoder a truncated blob, which panicked as `Invalid Parquet file. Corrupt footer`. rustfs is such a store, and every cold read of a blob above the 8 MiB multipart threshold, which is every large compaction output, crashed environmentd. The blob cache masked it for freshly written blobs. `get` now also reads the object's total size from `Content-Range`, which every store returns on a part request, fetches the remainder by byte range when the part count is missing, and checks the reassembled length against the total. Any store that returns a short object now fails the read, which persist retries, instead of corrupting data downstream. The existing path, where the header is present, is unchanged. Adds two ways to measure a blob store: * `persistcli bench blob` drives one store through persist's own blob client at a given object size and concurrency, writing, listing, reading and deleting, and reports throughput, latency percentiles and the retries persist's retry loop needed. Reads verify object length. * `test/blob-store-benchmark` sweeps that over object sizes, concurrency levels and stored volumes for several stores, samples each store container's CPU and memory, and writes a CSV. Also teaches the existing compositions about more blob stores. garage and rustfs join minio and azurite as `external_blob_store` targets for `Materialized` and `Testdrive`, resolved through a new `blob_store` module. The parallel benchmark takes `--blob-store` and `--other-blob-store` in place of `--azurite`, so two stores can be compared directly, and prints persist's blob operation counts and latencies after each scenario. Its new `BlobStoreReadsWrites` scenario disables the blob cache so that reads reach the store. garage ships without a shell, so it gets an mzbuild image that adds one for the cluster setup a fresh node needs. Tests: unit tests for the `Content-Range` parser and the byte-range chunking in `src/persist/src/s3.rs`. Release note: Materialize now reads multipart objects correctly from S3-compatible blob stores that do not return the `x-amz-mp-parts-count` response header. Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 1 + .../lint-main/checks/check-mzcompose-files.sh | 2 + .../mzcompose/services/blob_store.py | 35 ++ .../materialize/mzcompose/services/garage.py | 62 +++ .../mzcompose/services/materialized.py | 18 +- .../materialize/mzcompose/services/rustfs.py | 107 ++++++ .../mzcompose/services/testdrive.py | 20 +- .../parallel_benchmark/scenarios.py | 94 +++++ src/persist-client/src/cli/bench.rs | 288 +++++++++++++- src/persist/src/s3.rs | 158 ++++++-- test/blob-store-benchmark/mzcompose | 14 + test/blob-store-benchmark/mzcompose.py | 362 ++++++++++++++++++ test/garage/Dockerfile | 29 ++ test/garage/entrypoint.sh | 51 +++ test/garage/garage.toml | 51 +++ test/garage/mzbuild.yml | 10 + test/parallel-benchmark/mzcompose.py | 172 ++++++++- 17 files changed, 1410 insertions(+), 64 deletions(-) create mode 100644 misc/python/materialize/mzcompose/services/blob_store.py create mode 100644 misc/python/materialize/mzcompose/services/garage.py create mode 100644 misc/python/materialize/mzcompose/services/rustfs.py create mode 100755 test/blob-store-benchmark/mzcompose create mode 100644 test/blob-store-benchmark/mzcompose.py create mode 100644 test/garage/Dockerfile create mode 100755 test/garage/entrypoint.sh create mode 100644 test/garage/garage.toml create mode 100644 test/garage/mzbuild.yml diff --git a/.gitignore b/.gitignore index ac1a2b870ee48..2414df635e2c4 100644 --- a/.gitignore +++ b/.gitignore @@ -42,6 +42,7 @@ test/scalability/results/**/*.png misc/mcp-materialize/dist misc/wasm/target parallel-benchmark.db +blob-store-benchmark*.csv license_key /trufflehog.log diff --git a/ci/test/lint-main/checks/check-mzcompose-files.sh b/ci/test/lint-main/checks/check-mzcompose-files.sh index 9045e3d8934c2..9b090044bfb61 100755 --- a/ci/test/lint-main/checks/check-mzcompose-files.sh +++ b/ci/test/lint-main/checks/check-mzcompose-files.sh @@ -22,6 +22,7 @@ check_all_files_referenced_in_ci() { COMPOSITIONS=$(find . -name mzcompose.py \ -not -wholename "./misc/python/materialize/cli/mzcompose.py" `# Only glue code, no workflows` \ -not -wholename "./misc/monitoring/mzcompose.py" `# Only run manually` \ + -not -wholename "./test/blob-store-benchmark/mzcompose.py" `# Only run manually` \ -not -wholename "./test/canary-environment/mzcompose.py" `# Only run manually` \ -not -wholename "./test/console/mzcompose.py" `# Only run manually` \ -not -wholename "./test/mzcompose_examples/mzcompose.py" `# Example only` \ @@ -45,6 +46,7 @@ check_default_workflow_references_others() { while IFS= read -r file; do MZCOMPOSE_TEST_FILES+=("$file") done < <(find ./test -name "mzcompose.py" \ + -not -wholename "./test/blob-store-benchmark/mzcompose.py" `# Only run manually` \ -not -wholename "./test/canary-environment/mzcompose.py" `# Only run manually` \ -not -wholename "./test/ssh-connection/mzcompose.py" `# Handled differently` \ -not -wholename "./test/scalability/mzcompose.py" `# Other workflows are for manual usage` \ diff --git a/misc/python/materialize/mzcompose/services/blob_store.py b/misc/python/materialize/mzcompose/services/blob_store.py new file mode 100644 index 0000000000000..59421b61bac02 --- /dev/null +++ b/misc/python/materialize/mzcompose/services/blob_store.py @@ -0,0 +1,35 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +"""The blob stores persist can run against in a composition. + +Each is a service of the same name that `Materialized` and `Testdrive` accept +as `external_blob_store`. +""" + +from materialize.mzcompose.services.azurite import azure_blob_uri +from materialize.mzcompose.services.garage import garage_blob_uri +from materialize.mzcompose.services.minio import minio_blob_uri +from materialize.mzcompose.services.rustfs import rustfs_blob_uri + +BLOB_STORES = ["minio", "azurite", "garage", "rustfs"] + + +def blob_store_uri(blob_store: str) -> str: + """The persist blob URL for the named blob store service.""" + match blob_store: + case "minio": + return minio_blob_uri() + case "azurite": + return azure_blob_uri() + case "garage": + return garage_blob_uri() + case "rustfs": + return rustfs_blob_uri() + raise ValueError(f"unknown blob store {blob_store!r}, expected one of {BLOB_STORES}") diff --git a/misc/python/materialize/mzcompose/services/garage.py b/misc/python/materialize/mzcompose/services/garage.py new file mode 100644 index 0000000000000..9370add9f29f3 --- /dev/null +++ b/misc/python/materialize/mzcompose/services/garage.py @@ -0,0 +1,62 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + + +from materialize.mzcompose.service import Service + +# Garage only accepts keys shaped like the ones it generates: `GK` followed by +# 12 hex bytes, with a 32 hex byte secret. The entrypoint imports this pair so +# the blob URL can be a constant. +GARAGE_ACCESS_KEY_ID = "GK0123456789abcdef01234567" +GARAGE_SECRET_ACCESS_KEY = "0123456789abcdef" * 4 + + +def garage_blob_uri(address: str = "garage") -> str: + return f"s3://{GARAGE_ACCESS_KEY_ID}:{GARAGE_SECRET_ACCESS_KEY}@persist/persist?endpoint=http://{address}:3900/®ion=garage" + + +class Garage(Service): + """Single-node garage, an S3-compatible blob store, at `garage:3900`. + + The container is healthy only once the buckets exist, since garage creates + them after the server is up. Start it before, or wait for it separately + from, anything that depends on it with `service_started`. + """ + + def __init__( + self, + name: str = "garage", + setup_materialize: bool = False, + additional_buckets: list[str] = [], + ports: list[int | str] = [3900], + allow_host_ports: bool = False, + ) -> None: + buckets = (["persist"] if setup_materialize else []) + additional_buckets + super().__init__( + name=name, + config={ + "mzbuild": "garage", + "ports": ports, + "allow_host_ports": allow_host_ports, + "environment": [ + f"GARAGE_ACCESS_KEY_ID={GARAGE_ACCESS_KEY_ID}", + f"GARAGE_SECRET_ACCESS_KEY={GARAGE_SECRET_ACCESS_KEY}", + f"GARAGE_BUCKETS={' '.join(buckets)}", + ], + "healthcheck": { + "test": [ + "CMD-SHELL", + "test -f /var/lib/garage/meta/ready && garage status >/dev/null", + ], + "timeout": "5s", + "interval": "1s", + "start_period": "30s", + }, + }, + ) diff --git a/misc/python/materialize/mzcompose/services/materialized.py b/misc/python/materialize/mzcompose/services/materialized.py index 36477be460f4c..e559f12f9fa63 100644 --- a/misc/python/materialize/mzcompose/services/materialized.py +++ b/misc/python/materialize/mzcompose/services/materialized.py @@ -31,7 +31,7 @@ ServiceConfig, ServiceDependency, ) -from materialize.mzcompose.services.azurite import azure_blob_uri +from materialize.mzcompose.services.blob_store import blob_store_uri from materialize.mzcompose.services.listener_config import ( resolve_listeners_config_path, ) @@ -40,7 +40,6 @@ METADATA_STORE, metadata_store_companions, ) -from materialize.mzcompose.services.minio import minio_blob_uri class MaterializeEmulator(Service): @@ -250,14 +249,15 @@ def __init__( command += [f"--environment-id={environment_id}"] if external_blob_store: - blob_store = "azurite" if blob_store_is_azure else "minio" - depends_graph[blob_store] = {"condition": "service_started"} - address = blob_store if external_blob_store == True else external_blob_store - persist_blob_url = ( - azure_blob_uri(address) - if blob_store_is_azure - else minio_blob_uri(address) + # A string names the blob store service (see `BLOB_STORES`), `True` + # picks minio, or azurite with `blob_store_is_azure`. + blob_store = ( + external_blob_store + if isinstance(external_blob_store, str) + else ("azurite" if blob_store_is_azure else "minio") ) + depends_graph[blob_store] = {"condition": "service_started"} + persist_blob_url = blob_store_uri(blob_store) if persist_blob_url: command.append(f"--persist-blob-url={persist_blob_url}") diff --git a/misc/python/materialize/mzcompose/services/rustfs.py b/misc/python/materialize/mzcompose/services/rustfs.py new file mode 100644 index 0000000000000..d4a072cc2c855 --- /dev/null +++ b/misc/python/materialize/mzcompose/services/rustfs.py @@ -0,0 +1,107 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + + +from materialize.mzcompose.service import Service + +RUSTFS_VERSION = "1.0.0-rc.5" + + +def rustfs_blob_uri(address: str = "rustfs") -> str: + return f"s3://minioadmin:minioadmin@persist/persist?endpoint=http://{address}:9000/®ion=rustfs" + + +class RustFs(Service): + """Single-node rustfs, an S3-compatible blob store, at `rustfs:9000`. + + The container is healthy only once the buckets exist, since they are + created through the S3 API after the server is up. Start it before, or wait + for it separately from, anything that depends on it with `service_started`. + + One volume, which is zero-parity erasure coding, matching the single drive + and `EC:0` the minio image runs with. rustfs refuses several volumes that + share a device, so more drives would need more devices, not just more + directories. + + NOTE: rustfs 1.0.0-rc.5 answers `GetObject` with `partNumber` without the + `x-amz-mp-parts-count` header, and its `HeadObject` ignores `partNumber`. + Persist reads multipart-uploaded blobs part by part and falls back to byte + ranges when the count is missing (see `S3Blob::get`), so reads work but + cost the store one extra round trip per part beyond the first. + """ + + def __init__( + self, + name: str = "rustfs", + image: str = f"rustfs/rustfs:{RUSTFS_VERSION}", + setup_materialize: bool = False, + additional_buckets: list[str] = [], + ports: list[int | str] = [9000], + allow_host_ports: bool = False, + ) -> None: + buckets = (["persist"] if setup_materialize else []) + additional_buckets + # rustfs keeps objects in an erasure-coded layout, so buckets cannot be + # pre-created as directories the way the minio image does it. Creation + # is retried until the bucket is visible: rustfs answers its health + # endpoint before it accepts bucket operations, and the PUT fails when + # the bucket survived a restart. + s3 = "curl -s -o /dev/null --aws-sigv4 aws:amz:rustfs:s3 --user minioadmin:minioadmin" + create_buckets = "".join( + f"until {s3} -f -I http://127.0.0.1:9000/{bucket}; do " + f"{s3} -X PUT http://127.0.0.1:9000/{bucket}; sleep 0.5; done; " + for bucket in buckets + ) + # `$$` keeps docker compose from interpolating the shell variables. + command = ( + "rustfs & pid=$$!; trap 'kill $$pid' TERM INT; " + "until curl -sf -o /dev/null http://127.0.0.1:9000/health; do sleep 0.2; done; " + f"{create_buckets}touch /tmp/rustfs-ready; " + "wait $$pid" + ) + super().__init__( + name=name, + config={ + "image": image, + "entrypoint": ["sh", "-c"], + "command": [command], + "ports": ports, + "allow_host_ports": allow_host_ports, + "environment": [ + "RUSTFS_ACCESS_KEY=minioadmin", + "RUSTFS_SECRET_KEY=minioadmin", + "RUSTFS_CONSOLE_ENABLE=false", + # Speed over durability, like the minio image's patched-out + # fdatasync and garage's fsync-off default. The new-bucket + # tier would otherwise override the process-wide mode. + "RUSTFS_DURABILITY_MODE=none", + "RUSTFS_NEW_BUCKET_DURABILITY_MODE=inherit", + # Background work no composition needs on a store that is + # thrown away at the end of the run. minio runs with + # MINIO_HEAL_DISABLE=on for the same reason. + # + # NOTE: in 1.0.0-rc.5 the two switches leave the startup + # logs unchanged, so they may gate less than their names + # suggest. The scanner preset is set as well, since that + # one documents what it controls (sleep factor, maximum + # sleep, cycle interval). + "RUSTFS_HEAL_ENABLED=false", + "RUSTFS_SCANNER_ENABLED=false", + "RUSTFS_SCANNER_SPEED=slowest", + ], + "healthcheck": { + "test": [ + "CMD-SHELL", + "test -f /tmp/rustfs-ready && curl -sf -o /dev/null http://127.0.0.1:9000/health", + ], + "timeout": "5s", + "interval": "1s", + "start_period": "30s", + }, + }, + ) diff --git a/misc/python/materialize/mzcompose/services/testdrive.py b/misc/python/materialize/mzcompose/services/testdrive.py index ecc344f04c23b..87a375d32ccb6 100644 --- a/misc/python/materialize/mzcompose/services/testdrive.py +++ b/misc/python/materialize/mzcompose/services/testdrive.py @@ -22,13 +22,12 @@ Service, ServiceConfig, ) -from materialize.mzcompose.services.azurite import azure_blob_uri +from materialize.mzcompose.services.blob_store import blob_store_uri from materialize.mzcompose.services.metadata_store import ( EXTERNAL_METADATA_STORE_ADDRESS, METADATA_STORE, metadata_store_companions, ) -from materialize.mzcompose.services.minio import minio_blob_uri SANITIZER_TIMEOUT_FACTOR = 10 @@ -79,7 +78,7 @@ def __init__( no_consistency_checks: bool = False, check_statement_logging: bool = False, external_metadata_store: str | bool = EXTERNAL_METADATA_STORE_ADDRESS, - external_blob_store: bool = False, + external_blob_store: str | bool = False, blob_store_is_azure: bool = False, fivetran_destination: bool = False, fivetran_destination_url: str = "http://fivetran-destination:6874", @@ -220,16 +219,13 @@ def __init__( if set_persist_urls: if external_blob_store: - blob_store = "azurite" if blob_store_is_azure else "minio" - address = ( - blob_store if external_blob_store == True else external_blob_store - ) - persist_blob_url = ( - azure_blob_uri(address) - if blob_store_is_azure - else minio_blob_uri(address) + # Same meaning as for `Materialized`. + blob_store = ( + external_blob_store + if isinstance(external_blob_store, str) + else ("azurite" if blob_store_is_azure else "minio") ) - entrypoint.append(f"--persist-blob-url={persist_blob_url}") + entrypoint.append(f"--persist-blob-url={blob_store_uri(blob_store)}") else: entrypoint.append("--persist-blob-url=file:///mzdata/persist/blob") diff --git a/misc/python/materialize/parallel_benchmark/scenarios.py b/misc/python/materialize/parallel_benchmark/scenarios.py index 7da94760d90a3..7b8ca9e29fba7 100644 --- a/misc/python/materialize/parallel_benchmark/scenarios.py +++ b/misc/python/materialize/parallel_benchmark/scenarios.py @@ -10,6 +10,7 @@ import queue import time from copy import deepcopy +from textwrap import dedent import psycopg @@ -1587,3 +1588,96 @@ def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]): conn_pool_size=100, conn_pool_setup=["SET TRANSACTION_ISOLATION TO 'SERIALIZABLE'"], ) + + +class BlobStoreReadsWrites(Scenario): + r"""Measures the persist round trips to the blob store behind a batch write + and a snapshot read, with what would otherwise hide the store switched off. + + Each INSERT carries far more than `persist_inline_writes_single_max_bytes`, + so persist writes its batch to the blob store rather than inlining it into + consensus, and the measured latency includes that PUT. The blob cache is + disabled for the run, so every read of `blob_read` fetches its parts from + the store instead of from memory, where the default 128 MiB would hold the + whole table after the first read. The cache takes the larger of its static + limit and its per-thread scale, and compositions turn the scaling on, so + both are switched off. + + Reads and writes go to separate tables so the read latency stays stationary + over the load phase instead of growing with the rows written. The read + aggregates a column rather than counting rows: a bare `count(*)` projects + away every column, and persist then answers it from part metadata without + fetching a single part. + + Neither loop is blob-only: the write also runs its generating query on the + cluster and ships the rows through the coordinator, and the read decodes a + million rows. The `blob_get` and `blob_set` latencies the harness prints + from persist's metrics after the scenario separate the store's share from + Materialize's own work. + + To compare blob stores rather than Materialize versions: + + bin/mzcompose --find parallel-benchmark run default \ + --scenario BlobStoreReadsWrites \ + --blob-store garage --other-blob-store minio + """ + + def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]): + mz = conn_infos["materialized"] + setup = dedent(""" + $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} + ALTER SYSTEM SET persist_blob_cache_mem_limit_bytes = 0 + ALTER SYSTEM SET persist_blob_cache_scale_with_threads = false + + > DROP TABLE IF EXISTS blob_read CASCADE + > DROP TABLE IF EXISTS blob_write CASCADE + > CREATE TABLE blob_read (a int, b text) + > CREATE TABLE blob_write (a int, b text) + """) + # Hex digests do not compress, so each row costs real bytes in the + # store. Ten batches rather than one so the table starts out as several + # parts, the way a table written over time looks. + setup += "".join( + "> INSERT INTO blob_read SELECT n, md5(n::text) FROM generate_series(1, 100000) AS n\n" + for _ in range(10) + ) + setup += dedent(""" + > SELECT count(*) FROM blob_read + 1000000 + """) + self.init( + [ + TdPhase(setup), + LoadPhase( + duration=120, + actions=[ + OpenLoop( + action=ReuseConnQuery( + "INSERT INTO blob_write SELECT n, md5(n::text) FROM generate_series(1, 100000) AS n", + mz, + strict_serializable=False, + ), + dist=Periodic(per_second=1), + ), + ClosedLoop( + action=ReuseConnQuery( + "SELECT sum(a) FROM blob_read", + mz, + strict_serializable=False, + ), + ), + ], + ), + # Nothing resets the services between scenarios when the + # benchmark runs against an existing environment, so undo what + # this one changed. + TdPhase(""" + > DROP TABLE IF EXISTS blob_read CASCADE + > DROP TABLE IF EXISTS blob_write CASCADE + + $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} + ALTER SYSTEM RESET persist_blob_cache_mem_limit_bytes + ALTER SYSTEM RESET persist_blob_cache_scale_with_threads + """), + ], + ) diff --git a/src/persist-client/src/cli/bench.rs b/src/persist-client/src/cli/bench.rs index 3bea661ab7418..ce566bfa51e33 100644 --- a/src/persist-client/src/cli/bench.rs +++ b/src/persist-client/src/cli/bench.rs @@ -11,13 +11,24 @@ use futures_util::stream::StreamExt; use futures_util::{TryStreamExt, stream}; +use std::future::Future; use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; +use anyhow::anyhow; +use bytes::Bytes; +use mz_ore::cast::{CastFrom, CastLossy}; +use mz_ore::metrics::MetricsRegistry; +use mz_ore::now::SYSTEM_TIME; +use mz_ore::url::SensitiveUrl; use mz_persist::indexed::encoding::BlobTraceBatchPart; +use mz_persist::location::ExternalError; -use crate::cli::args::StateArgs; +use crate::cfg::PersistConfig; +use crate::cli::args::{READ_ALL_BUILD_INFO, StateArgs, make_blob}; +use crate::internal::machine::retry_external; use crate::internal::state::BatchPart; +use crate::metrics::Metrics; /// Commands for read-only inspection of persist state #[derive(Debug, clap::Args)] @@ -32,6 +43,9 @@ pub(crate) enum Command { /// Fetch the blobs in a shard as quickly as possible, repeated some number /// of times. S3Fetch(S3FetchArgs), + /// Write, list, read and delete objects of one size at a fixed + /// concurrency, through persist's own blob store client. + Blob(BlobArgs), } /// Fetch the blobs in a shard as quickly as possible, repeated some number of @@ -52,6 +66,7 @@ pub struct S3FetchArgs { pub async fn run(command: BenchArgs) -> Result<(), anyhow::Error> { match command.command { Command::S3Fetch(args) => bench_s3(&args).await?, + Command::Blob(args) => bench_blob(&args).await?, } Ok(()) @@ -125,3 +140,272 @@ async fn bench_s3(args: &S3FetchArgs) -> Result<(), anyhow::Error> { Ok(()) } + +/// Drives a blob store through persist's own blob client and +/// reports what each operation cost, as one CSV row per operation. +/// +/// Writes `count` objects of `size_bytes` under `prefix`, lists `list_prefix`, +/// reads the objects back, and deletes them, each with `concurrency` +/// operations in flight. Writes, reads and deletes are retried the way persist +/// retries them, so a store that throttles shows up as latency and in the +/// `retries` column rather than as a failed run. The reads check that every +/// object comes back at its written size, so a store that hands back partial +/// objects does fail the run. +#[derive(Debug, Clone, clap::Parser)] +pub struct BlobArgs { + /// Blob store to benchmark, in the form `--persist-blob-url` takes. + #[clap(long, env = "BLOB_URI")] + blob_uri: SensitiveUrl, + + /// Key prefix the objects are written under. + #[clap(long, default_value = "bench")] + prefix: String, + + /// Key prefix to list. Defaults to `prefix`. A parent of it measures a + /// listing that spans objects kept from earlier runs. + #[clap(long)] + list_prefix: Option, + + /// Size of each object in bytes. + #[clap(long)] + size_bytes: usize, + + /// Number of objects. + #[clap(long)] + count: usize, + + /// Operations in flight at once. + #[clap(long, default_value_t = 1)] + concurrency: usize, + + /// How long to read objects picked at random, in seconds. With 0, each + /// object is read once instead. + #[clap(long, default_value_t = 0)] + read_secs: u64, + + /// Skip the read phase. + #[clap(long)] + skip_read: bool, + + /// Skip the list phase. + #[clap(long)] + skip_list: bool, + + /// Leave the objects in place instead of deleting them, so that later + /// runs see a fuller store. + #[clap(long)] + keep: bool, + + /// Seed for the object contents and the read order. + #[clap(long, default_value_t = 0)] + seed: u64, + + /// Omit the CSV header row. + #[clap(long)] + no_header: bool, +} + +async fn bench_blob(args: &BlobArgs) -> Result<(), anyhow::Error> { + if args.count == 0 { + return Err(anyhow!("--count must be positive")); + } + let cfg = PersistConfig::new_default_configs(&READ_ALL_BUILD_INFO, SYSTEM_TIME.clone()); + let metrics = Arc::new(Metrics::new(&cfg, &MetricsRegistry::new())); + let blob = make_blob(&cfg, &args.blob_uri, true, Arc::clone(&metrics)).await?; + let concurrency = args.concurrency.max(1); + // The same retry loops persist's own writers, readers and garbage + // collector run their blob operations under. + let set_retries = &metrics.retries.external.batch_set; + let get_retries = &metrics.retries.external.fetch_batch_get; + let delete_retries = &*metrics.retries.external.batch_delete; + + let mut rng = SplitMix64(args.seed); + // Incompressible, like the parquet parts persist writes, so that a store + // that compresses cannot shrink the work. + let payload = random_bytes(&mut rng, args.size_bytes); + let keys: Vec = (0..args.count) + .map(|i| format!("{}/{i:08}", args.prefix)) + .collect(); + + if !args.no_header { + println!( + "op,size_bytes,concurrency,ops,bytes,elapsed_secs,ops_per_sec,mib_per_sec,p50_ms,p90_ms,p99_ms,max_ms,retries" + ); + } + + let start = Instant::now(); + let retries_before = set_retries.retries.get(); + let latencies = run_ops(concurrency, keys.iter(), |key| { + let blob = Arc::clone(&blob); + let payload = payload.clone(); + async move { + retry_external(set_retries, || blob.set(key, payload.clone())).await; + Ok(()) + } + }) + .await?; + report( + args, + "set", + &latencies, + start.elapsed(), + args.size_bytes, + set_retries.retries.get() - retries_before, + ); + + if !args.skip_list { + let list_prefix = args.list_prefix.as_deref().unwrap_or(&args.prefix); + let start = Instant::now(); + let mut listed = 0; + blob.list_keys_and_metadata(list_prefix, &mut |_| listed += 1) + .await?; + let elapsed = start.elapsed(); + report(args, "list", &vec![elapsed; listed], elapsed, 0, 0); + } + + if !args.skip_read { + let deadline = Instant::now() + Duration::from_secs(args.read_secs); + let read_keys: Box> = if args.read_secs == 0 { + Box::new(keys.iter()) + } else { + Box::new(std::iter::from_fn(|| { + (Instant::now() < deadline) + .then(|| &keys[usize::cast_from(rng.next() % u64::cast_from(keys.len()))]) + })) + }; + let size = args.size_bytes; + let start = Instant::now(); + let retries_before = get_retries.retries.get(); + let latencies = run_ops(concurrency, read_keys, |key| { + let blob = Arc::clone(&blob); + async move { + // Only the store call is retried. A short or missing object + // is the store lying, not a transient, and ends the run. + match retry_external(get_retries, || blob.get(key)).await { + Some(value) if value.len() == size => Ok(()), + Some(value) => Err(anyhow!( + "{key}: read {} bytes of a {size} byte object", + value.len() + ) + .into()), + None => Err(anyhow!("{key}: missing").into()), + } + } + }) + .await?; + report( + args, + "get", + &latencies, + start.elapsed(), + args.size_bytes, + get_retries.retries.get() - retries_before, + ); + } + + if !args.keep { + let start = Instant::now(); + let retries_before = delete_retries.retries.get(); + let latencies = run_ops(concurrency, keys.iter(), |key| { + let blob = Arc::clone(&blob); + async move { + retry_external(delete_retries, || blob.delete(key)).await; + Ok(()) + } + }) + .await?; + report( + args, + "delete", + &latencies, + start.elapsed(), + 0, + delete_retries.retries.get() - retries_before, + ); + } + + Ok(()) +} + +/// Runs `op` over `items` with `concurrency` in flight, returning each +/// operation's latency. +async fn run_ops( + concurrency: usize, + items: impl IntoIterator, + op: F, +) -> Result, ExternalError> +where + F: Fn(K) -> Fut, + Fut: Future>, +{ + stream::iter(items) + .map(|item| { + let start = Instant::now(); + let fut = op(item); + async move { + fut.await?; + Ok(start.elapsed()) + } + }) + .buffer_unordered(concurrency) + .try_collect() + .await +} + +fn report( + args: &BlobArgs, + op: &str, + latencies: &[Duration], + elapsed: Duration, + bytes_each: usize, + retries: u64, +) { + let mut sorted = latencies.to_vec(); + sorted.sort(); + let ops = sorted.len(); + let ms = |d: Duration| d.as_secs_f64() * 1000.0; + // Nearest-rank percentiles over the sorted latencies. + let pct = |q: usize| { + if ops == 0 { + f64::NAN + } else { + ms(sorted[(ops - 1) * q / 100]) + } + }; + let bytes = ops * bytes_each; + let secs = elapsed.as_secs_f64(); + println!( + "{op},{},{},{ops},{bytes},{secs:.3},{:.1},{:.1},{:.2},{:.2},{:.2},{:.2},{retries}", + args.size_bytes, + args.concurrency, + f64::cast_lossy(ops) / secs, + f64::cast_lossy(bytes) / secs / (1024.0 * 1024.0), + pct(50), + pct(90), + pct(99), + sorted.last().map_or(f64::NAN, |d| ms(*d)), + ); +} + +/// SplitMix64, so that the payload and read order are reproducible without +/// pulling in a random number generator. +struct SplitMix64(u64); + +impl SplitMix64 { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } +} + +fn random_bytes(rng: &mut SplitMix64, len: usize) -> Bytes { + let mut buf = Vec::with_capacity(len + 8); + while buf.len() < len { + buf.extend_from_slice(&rng.next().to_le_bytes()); + } + buf.truncate(len); + Bytes::from(buf) +} diff --git a/src/persist/src/s3.rs b/src/persist/src/s3.rs index bf32439330ab8..26b7653870388 100644 --- a/src/persist/src/s3.rs +++ b/src/persist/src/s3.rs @@ -24,6 +24,7 @@ use aws_credential_types::Credentials; use aws_sdk_s3::Client as S3Client; use aws_sdk_s3::config::{AsyncSleep, Sleep}; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; +use aws_sdk_s3::operation::get_object::GetObjectOutput; use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart}; use aws_types::region::Region; @@ -356,6 +357,14 @@ impl Blob for S3Blob { // the headers before the full data body has completed. This gives us // the number of parts. We can then proceed to fetch the body of the // first request concurrently with the rest of the parts of the object. + // + // Not every S3-compatible store reports the part count. Its response + // to the first request is then indistinguishable from a single-part + // object's, except that `Content-Range` still carries the object's + // total size. The remainder is fetched by byte range in that case, and + // the reassembled length is checked against the total either way, so a + // store that misbehaves fails the get instead of handing the decoder a + // truncated blob. // For each header and body that we fetch, we track the fastest, and // any large deviations from it. @@ -385,56 +394,72 @@ impl Blob for S3Blob { } }; - // Get the remaining number of parts - let num_parts = match first_part.parts_count() { - // For a non-multipart upload, parts_count will be None. The rest of the code works - // perfectly well if we just pretend this was a multipart upload of 1 part. - None => 1, - // For any positive value greater than 0, just return it. - Some(parts @ 1..) => parts, + // The object's total size comes from `Content-Range`, which s3 returns + // for any request that names a part. + let total_len = first_part + .content_range() + .and_then(parse_content_range_total); + let first_len = first_part + .content_length() + .and_then(|len| u64::try_from(len).ok()); + + let remaining: Vec = match first_part.parts_count() { + // For a non-multipart upload, parts_count will be None, and the + // first request already returned the whole object. A store that + // does not report the count looks the same, except that the first + // part falls short of the total. Its remainder is fetched by byte + // range in chunks of the first part's size, which is the size the + // upload used for every part but the last. + None => match (first_len, total_len) { + (Some(first_len), Some(total_len)) if first_len < total_len => { + remaining_ranges(first_len, total_len) + .into_iter() + .map(PartRequest::Range) + .collect() + } + _ => Vec::new(), + }, + Some(parts @ 1..) => (2..=parts).map(PartRequest::Number).collect(), // A non-positive value is invalid. Some(bad) => { assert!(bad <= 0); return Err(anyhow!("unexpected number of s3 object parts: {}", bad).into()); } }; + let num_requests = remaining.len() + 1; trace!( - "s3 download first header took {:?} ({num_parts} parts)", + "s3 download first header took {:?} ({num_requests} requests)", start_overall.elapsed(), ); let mut body_futures = FuturesOrdered::new(); - let mut first_part = Some(first_part); + let mut requests = vec![PartRequest::First(first_part)]; + requests.extend(remaining); - // Fetch the headers of the rest of the parts. (Starting at part 2 because we already - // did part 1.) - for part_num in 1..=num_parts { + for request in requests { // Clone a handle to our MinElapsed trackers so we can give one to // each download task. let min_header_elapsed = Arc::clone(&min_header_elapsed); let min_body_elapsed = Arc::clone(&min_body_elapsed); let get_invalid_resp = self.metrics.get_invalid_resp.clone(); - let first_part = first_part.take(); let path = &path; let request_future = async move { - // Fetch the headers of the rest of the parts. (Using the existing headers - // for part 1. - let mut object = match first_part { - Some(first_part) => { - assert_eq!(part_num, 1, "only the first part should be prefetched"); - first_part - } - None => { - assert_ne!(part_num, 1, "first part should be prefetched"); - // Request our headers. + let mut object = match request { + // Fetched above, together with the headers that shaped + // the remaining requests. + PartRequest::First(object) => object, + other => { let header_start = Instant::now(); - let object = self - .client - .get_object() - .bucket(&self.bucket) - .key(path) - .part_number(part_num) + let req = self.client.get_object().bucket(&self.bucket).key(path); + let req = match other { + PartRequest::Number(part_num) => req.part_number(part_num), + PartRequest::Range(range) => { + req.range(format!("bytes={}-{}", range.start, range.end - 1)) + } + PartRequest::First(_) => unreachable!("handled above"), + }; + let object = req .send() .await .inspect_err(|err| self.update_error_metrics("GetObject", err)) @@ -444,7 +469,6 @@ impl Blob for S3Blob { object } }; - // Request the body. let body_start = Instant::now(); @@ -503,10 +527,24 @@ impl Blob for S3Blob { segments.append(&mut part_body); } + // A store that reports neither the part count nor a range it honors + // hands back less than the object. Fail the get here rather than let + // the decoder find out. + if let Some(total_len) = total_len { + let fetched_len: u64 = segments.iter().map(|s| u64::cast_from(s.len())).sum(); + if fetched_len != total_len { + self.metrics.get_invalid_resp.inc(); + return Err(anyhow!( + "s3 GetObject {path} returned {fetched_len} bytes of a {total_len} byte object" + ) + .into()); + } + } + debug!( - "s3 GetObject took {:?} ({} parts)", + "s3 GetObject took {:?} ({} requests)", start_overall.elapsed(), - num_parts + num_requests ); Ok(Some(SegmentedBytes::from(segments))) } @@ -915,6 +953,35 @@ impl S3Blob { } } +/// One request of a multi-request `get`. +enum PartRequest { + /// The first part, already fetched to learn the object's shape. + First(GetObjectOutput), + /// A part by the number it was uploaded with. + Number(i32), + /// A byte range, for stores that do not report the part count. + Range(Range), +} + +/// The total size in a `Content-Range` header (`bytes 0-8388607/9711660`), if +/// the header states one. +fn parse_content_range_total(content_range: &str) -> Option { + content_range.rsplit_once('/')?.1.trim().parse().ok() +} + +/// Byte ranges covering `[first_len, total_len)` in chunks of `first_len`. +fn remaining_ranges(first_len: u64, total_len: u64) -> Vec> { + let chunk_len = if first_len == 0 { total_len } else { first_len }; + let mut ranges = Vec::new(); + let mut start = first_len; + while start < total_len { + let end = cmp::min(start + chunk_len, total_len); + ranges.push(start..end); + start = end; + } + ranges +} + #[derive(Clone, Debug)] struct MultipartConfig { multipart_threshold: usize, @@ -1084,6 +1151,33 @@ mod tests { use super::*; + #[mz_ore::test] + fn content_range_total() { + assert_eq!( + parse_content_range_total("bytes 0-8388607/9711660"), + Some(9711660) + ); + assert_eq!(parse_content_range_total("bytes */42"), Some(42)); + assert_eq!(parse_content_range_total("bytes 0-1/*"), None); + assert_eq!(parse_content_range_total("garbage"), None); + } + + #[mz_ore::test] + fn ranges_for_unreported_parts() { + let mib = 1024 * 1024; + assert_eq!(remaining_ranges(8 * mib, 9711660), vec![8 * mib..9711660]); + assert_eq!( + remaining_ranges(8 * mib, 3 * 8 * mib + 5), + vec![ + 8 * mib..16 * mib, + 16 * mib..24 * mib, + 24 * mib..24 * mib + 5 + ] + ); + assert_eq!(remaining_ranges(8 * mib, 8 * mib), Vec::>::new()); + assert_eq!(remaining_ranges(0, 10), vec![0..10]); + } + #[mz_ore::test(tokio::test(flavor = "multi_thread"))] #[cfg_attr(coverage, ignore)] // https://github.com/MaterializeInc/database-issues/issues/5586 #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `TLS_method` on OS `linux` diff --git a/test/blob-store-benchmark/mzcompose b/test/blob-store-benchmark/mzcompose new file mode 100755 index 0000000000000..1f866645dabc8 --- /dev/null +++ b/test/blob-store-benchmark/mzcompose @@ -0,0 +1,14 @@ +#!/usr/bin/env bash + +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. +# +# mzcompose — runs Docker Compose with Materialize customizations. + +exec "$(dirname "$0")"/../../bin/pyactivate -m materialize.cli.mzcompose "$@" diff --git a/test/blob-store-benchmark/mzcompose.py b/test/blob-store-benchmark/mzcompose.py new file mode 100644 index 0000000000000..0d7570e5e0208 --- /dev/null +++ b/test/blob-store-benchmark/mzcompose.py @@ -0,0 +1,362 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +""" +Measures the blob stores persist can run against by driving each through +persist's own blob client (`persistcli bench blob`) over a matrix of object +sizes, concurrency levels and stored volumes, and compares throughput, latency +and the store's CPU and memory use side by side. +""" + +import argparse +import csv +import re +import subprocess +import threading +import time +from collections import defaultdict + +from materialize import MZ_ROOT +from materialize.mzcompose.composition import Composition, WorkflowArgumentParser +from materialize.mzcompose.services.azurite import Azurite +from materialize.mzcompose.services.blob_store import BLOB_STORES, blob_store_uri +from materialize.mzcompose.services.garage import Garage +from materialize.mzcompose.services.minio import Minio +from materialize.mzcompose.services.persistcli import Persistcli +from materialize.mzcompose.services.rustfs import RustFs +from materialize.mzcompose.test_result import ( + FailedTestExecutionError, + TestFailureDetails, +) + +SERVICES = [ + Minio(setup_materialize=True), + Garage(setup_materialize=True), + RustFs(setup_materialize=True), + Azurite(), + Persistcli(), +] + +# Columns `persistcli bench blob` prints, in order. +PERSISTCLI_FIELDS = [ + "op", + "size_bytes", + "concurrency", + "ops", + "bytes", + "elapsed_secs", + "ops_per_sec", + "mib_per_sec", + "p50_ms", + "p90_ms", + "p99_ms", + "max_ms", + "retries", +] +CSV_FIELDS = ["store", "fill_bytes"] + PERSISTCLI_FIELDS + ["cpu_pct_max", "mem_bytes_max"] + +UNITS = {"": 1, "k": 1024, "kib": 1024, "m": 1024**2, "mib": 1024**2, "g": 1024**3, "gib": 1024**3} + + +def parse_bytes(text: str) -> int: + match = re.fullmatch(r"\s*(\d+)\s*([a-zA-Z]*)\s*", text) + if not match or match.group(2).lower() not in UNITS: + raise argparse.ArgumentTypeError(f"not a size: {text!r}") + return int(match.group(1)) * UNITS[match.group(2).lower()] + + +def format_bytes(n: float) -> str: + for unit in ["B", "KiB", "MiB", "GiB"]: + if n < 1024 or unit == "GiB": + return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}" + n /= 1024 + raise AssertionError("unreachable") + + +class ContainerStats(threading.Thread): + """Samples `docker stats` for one container until stopped, keeping the + peak CPU percentage and memory use. + + `docker stats --no-stream` blocks for about a second per sample to compute + the CPU rate, so this is a thread rather than a poll inside the cell loop. + """ + + def __init__(self, container: str): + super().__init__(daemon=True) + self.container = container + self.cpu_pct_max = 0.0 + self.mem_bytes_max = 0 + self._stop = threading.Event() + + def run(self) -> None: + while not self._stop.is_set(): + try: + out = subprocess.check_output( + [ + "docker", + "stats", + "--no-stream", + "--format", + "{{.CPUPerc}}\t{{.MemUsage}}", + self.container, + ], + text=True, + stderr=subprocess.DEVNULL, + ) + except subprocess.CalledProcessError: + time.sleep(1) + continue + cpu, mem = out.strip().split("\t") + self.cpu_pct_max = max(self.cpu_pct_max, float(cpu.rstrip("%"))) + self.mem_bytes_max = max(self.mem_bytes_max, parse_docker_mem(mem.split("/")[0])) + + def stop(self) -> None: + self._stop.set() + self.join() + + +def parse_docker_mem(text: str) -> int: + """`docker stats` prints memory like `1.234GiB` or `512MiB`.""" + match = re.fullmatch(r"\s*([\d.]+)\s*([A-Za-z]+)\s*", text) + assert match, f"unexpected docker stats memory: {text!r}" + return int(float(match.group(1)) * UNITS[match.group(2).lower()]) + + +def bench( + c: Composition, + store: str, + args: argparse.Namespace, + prefix: str, + size: int, + count: int, + concurrency: int, + fill_bytes: int, + read_secs: int, + keep: bool, + skip_list: bool, +) -> list[dict[str, str]]: + """Runs one `persistcli bench blob` invocation and returns its rows. + + A negative `read_secs` skips the read phase. + """ + stats = ContainerStats(f"{c.project_name}-{store}-1") + stats.start() + try: + output = c.run( + "persistcli", + "bench", + "blob", + f"--blob-uri={blob_store_uri(store)}", + f"--prefix={prefix}", + # Everything under the store's prefix, so that objects kept from + # the fill phase count towards the listing. + "--list-prefix=", + f"--size-bytes={size}", + f"--count={count}", + f"--concurrency={concurrency}", + *([f"--read-secs={read_secs}"] if read_secs >= 0 else ["--skip-read"]), + *(["--keep"] if keep else []), + *(["--skip-list"] if skip_list else []), + "--no-header", + capture=True, + rm=True, + ).stdout + finally: + stats.stop() + rows = [] + for record in csv.DictReader(output.splitlines(), fieldnames=PERSISTCLI_FIELDS): + record["store"] = store + record["fill_bytes"] = str(fill_bytes) + record["cpu_pct_max"] = f"{stats.cpu_pct_max:.1f}" + record["mem_bytes_max"] = str(stats.mem_bytes_max) + rows.append(record) + return rows + + +def objects_per_cell(args: argparse.Namespace, size: int, concurrency: int) -> int: + """How many objects a cell writes: enough bytes to be representative and + enough objects to keep `concurrency` busy, within the caps.""" + count = max(concurrency, args.bytes_per_cell // size) + count = min(count, args.max_objects_per_cell, max(1, args.max_bytes_per_cell // size)) + return count + + +def print_report(rows: list[dict[str, str]]) -> None: + stores = list(dict.fromkeys(row["store"] for row in rows)) + by_cell: dict[tuple[str, str, str], dict[tuple[str, str], dict[str, str]]] = defaultdict(dict) + for row in rows: + by_cell[(row["fill_bytes"], row["op"], row["size_bytes"])][ + (row["store"], row["concurrency"]) + ] = row + for (fill_bytes, op, size_bytes), cells in by_cell.items(): + title = f"{op} {format_bytes(int(size_bytes))} objects" + if int(fill_bytes): + title += f", {format_bytes(int(fill_bytes))} already stored" + print(f"\n=== {title}") + header = f"{'CONC':>5} {'STORE':<8} {'OPS/S':>9} {'MiB/S':>8} {'P50 ms':>9} {'P90 ms':>9} {'P99 ms':>9} {'MAX ms':>9} {'RETRIES':>7} {'CPU%':>7} {'MEM':>9}" + print(header) + concurrencies = sorted({int(conc) for (_, conc) in cells}, key=int) + for concurrency in concurrencies: + for store in stores: + row = cells.get((store, str(concurrency))) + if row is None: + continue + print( + f"{concurrency:>5} {store:<8} {float(row['ops_per_sec']):>9.1f} {float(row['mib_per_sec']):>8.1f} " + f"{float(row['p50_ms']):>9.2f} {float(row['p90_ms']):>9.2f} {float(row['p99_ms']):>9.2f} {float(row['max_ms']):>9.2f} " + f"{int(row['retries']):>7} {float(row['cpu_pct_max']):>7.1f} {format_bytes(int(row['mem_bytes_max'])):>9}" + ) + + +def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: + parser.add_argument( + "--blob-store", + action="append", + choices=BLOB_STORES, + help="Blob stores to benchmark (default: minio, garage, rustfs)", + ) + parser.add_argument( + "--size", + action="append", + type=parse_bytes, + help="Object sizes, like 4KiB or 8MiB (default: 4KiB 64KiB 1MiB 8MiB 64MiB)", + ) + parser.add_argument( + "--concurrency", + action="append", + type=int, + help="Operations in flight (default: 1 8 32 128)", + ) + parser.add_argument( + "--fill", + action="append", + type=parse_bytes, + help="Stored volumes to run the matrix at, ascending (default: 0). Fill objects are 8MiB and stay for the rest of the store's run.", + ) + parser.add_argument( + "--bytes-per-cell", + type=parse_bytes, + default=256 * 1024**2, + help="Bytes each cell writes, before the caps (default: 256MiB)", + ) + parser.add_argument( + "--max-objects-per-cell", + type=int, + default=4096, + help="Cap on objects per cell, which bounds the small-object cells (default: 4096)", + ) + parser.add_argument( + "--max-bytes-per-cell", + type=parse_bytes, + default=2 * 1024**3, + help="Cap on bytes per cell, which bounds the large-object cells (default: 2GiB)", + ) + parser.add_argument( + "--read-secs", + type=int, + default=10, + help="Seconds of random reads per cell (default: 10)", + ) + parser.add_argument( + "--csv", + default="blob-store-benchmark.csv", + help="Where to write every row, relative to the repository root", + ) + args = parser.parse_args() + + stores = args.blob_store or ["minio", "garage", "rustfs"] + sizes = args.size or [4 * 1024, 64 * 1024, 1024**2, 8 * 1024**2, 64 * 1024**2] + concurrencies = args.concurrency or [1, 8, 32, 128] + fills = sorted(set(args.fill or [0])) + fill_object_bytes = 8 * 1024**2 + + # Written as cells complete, so a run that dies keeps what it measured. + csv_path = MZ_ROOT / args.csv + csv_file = open(csv_path, "w", newline="") + writer = csv.DictWriter(csv_file, fieldnames=CSV_FIELDS) + writer.writeheader() + rows: list[dict[str, str]] = [] + failures: list[str] = [] + for store in stores: + print(f"+++ Benchmarking {store}") + c.up(store) + filled = 0 + for fill_bytes in fills: + if fill_bytes > filled: + print(f"--- Filling {store} to {format_bytes(fill_bytes)}") + try: + bench( + c, + store, + args, + prefix=f"fill/{fill_bytes}", + size=fill_object_bytes, + count=(fill_bytes - filled) // fill_object_bytes, + concurrency=32, + fill_bytes=fill_bytes, + read_secs=-1, + keep=True, + skip_list=True, + ) + except Exception as e: + # The cells at this fill level would measure a store + # holding less than they claim, so skip the rest of them. + failures.append(f"{store}: fill to {format_bytes(fill_bytes)}: {e}") + print(f"Fill failed, skipping the remaining fill levels for {store}: {e}") + break + filled = fill_bytes + for size in sizes: + for concurrency in concurrencies: + count = objects_per_cell(args, size, concurrency) + cell = f"{store}: {count} x {format_bytes(size)} at concurrency {concurrency}, {format_bytes(filled)} stored" + print(f"--- {cell}") + try: + cell_rows = bench( + c, + store, + args, + prefix=f"cell/{fill_bytes}/{size}/{concurrency}", + size=size, + count=count, + concurrency=concurrency, + fill_bytes=fill_bytes, + read_secs=args.read_secs, + keep=False, + skip_list=False, + ) + except Exception as e: + # One failing cell should not cost the rest of the + # matrix. The objects it wrote stay behind, which is + # noise for later list cells of this store only. + failures.append(f"{cell}: {e}") + print(f"Cell failed, continuing: {e}") + continue + rows.extend(cell_rows) + writer.writerows(cell_rows) + csv_file.flush() + for row in cell_rows: + print( + f" {row['op']:<7} {float(row['ops_per_sec']):>9.1f} ops/s {float(row['mib_per_sec']):>8.1f} MiB/s " + f"p50 {float(row['p50_ms']):>8.2f} ms p99 {float(row['p99_ms']):>8.2f} ms retries {row['retries']}" + ) + c.kill(store) + c.rm(store, destroy_volumes=True) + csv_file.close() + + print(f"+++ Results ({csv_path})") + print_report(rows) + if failures: + print("+++ Failed cells") + for failure in failures: + print(f" {failure}") + raise FailedTestExecutionError( + errors=[ + TestFailureDetails(message=failure, details=None) for failure in failures + ] + ) diff --git a/test/garage/Dockerfile b/test/garage/Dockerfile new file mode 100644 index 0000000000000..397408ff7999b --- /dev/null +++ b/test/garage/Dockerfile @@ -0,0 +1,29 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# The upstream image is built FROM scratch, with no shell, so it cannot run +# the cluster setup that a fresh garage node needs before it serves S3 +# requests. Take its statically linked binary and give it a shell. +FROM dxflrs/garage:v2.4.1 AS garage + +MZFROM debian-base + +COPY --from=garage /garage /usr/local/bin/garage +COPY garage.toml /etc/garage.toml +COPY entrypoint.sh /usr/local/bin/garage-entrypoint +RUN useradd -m garage \ + && mkdir -p /var/lib/garage/meta /var/lib/garage/data \ + && chown -R garage:garage /var/lib/garage + +# Per-connection INFO lines from the RPC layer would otherwise log every +# healthcheck. +ENV RUST_LOG=garage=warn + +USER garage +ENTRYPOINT ["garage-entrypoint"] diff --git a/test/garage/entrypoint.sh b/test/garage/entrypoint.sh new file mode 100755 index 0000000000000..048b339ca0010 --- /dev/null +++ b/test/garage/entrypoint.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash + +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. +# +# Starts a single-node garage and performs the cluster setup a fresh node +# needs before it accepts S3 requests: a layout, the S3 key from +# GARAGE_ACCESS_KEY_ID/GARAGE_SECRET_ACCESS_KEY, and the buckets named in +# GARAGE_BUCKETS. Garage only accepts these over RPC from a running node, so +# they cannot be baked into the image. The ready marker tells the healthcheck +# that setup has completed, and skips it when the container restarts with its +# metadata intact. + +set -euo pipefail + +: "${GARAGE_ACCESS_KEY_ID:?}" +: "${GARAGE_SECRET_ACCESS_KEY:?}" +: "${GARAGE_BUCKETS:=}" + +ready=/var/lib/garage/meta/ready + +garage server & +pid=$! +trap 'kill "$pid"' TERM INT + +until garage status >/dev/null 2>&1; do + sleep 0.2 +done + +if [[ ! -f "$ready" ]]; then + # `node id` prints `@`, and `layout assign` wants the id. + node_id=$(garage node id -q | cut -d@ -f1) + # The capacity only weights data placement across nodes, so any value + # works for one node. + garage layout assign -z dc1 -c 1T "$node_id" + garage layout apply --version 1 + garage key import --yes -n persist "$GARAGE_ACCESS_KEY_ID" "$GARAGE_SECRET_ACCESS_KEY" + for bucket in $GARAGE_BUCKETS; do + garage bucket create "$bucket" + garage bucket allow --read --write --owner "$bucket" --key persist + done + touch "$ready" +fi + +wait "$pid" diff --git a/test/garage/garage.toml b/test/garage/garage.toml new file mode 100644 index 0000000000000..78a3f192b49af --- /dev/null +++ b/test/garage/garage.toml @@ -0,0 +1,51 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# Single-node garage for compositions. Tuned like the minio image, for speed +# over durability: fsync is off (the default), stored blocks are not compressed +# (persist blobs arrive compressed already), and the periodic scrub is off, as +# minio runs with MINIO_HEAL_DISABLE=on. +# +# The block knobs below matter because garage stores every object as a chain of +# `block_size` files. Persist writes blobs from a few KiB up to its ~128MiB +# batch target, so the defaults (1MiB blocks, and read concurrency sized as +# backpressure for spinning disks) turn one large-blob get into dozens of file +# opens contending for 16 slots. + +metadata_dir = "/var/lib/garage/meta" +data_dir = "/var/lib/garage/data" +db_engine = "lmdb" +replication_factor = 1 +compression_level = "none" +disable_scrub = true + +# Upstream's recommendation for storing large files, against a 1MiB default. +# NB: garage cannot change this on a store that already holds data, which is +# fine here because compositions always start from an empty volume. +block_size = "10M" +# Default 16, a backpressure mechanism for HDD read speed. Compositions run on +# local SSD, where the cap only serializes concurrent readers. +block_max_concurrent_reads = 256 +# Default 3, with 10-30 recommended for NVMe. +block_max_concurrent_writes_per_request = 30 + +rpc_bind_addr = "[::]:3901" +rpc_public_addr = "127.0.0.1:3901" +# Only ever used within this container. +rpc_secret = "8b1f4c2a9d6e3f70a5b2c4d8e1f6a3b9c7d2e5f8a1b4c7d0e3f6a9b2c5d8e1f4" + +[s3_api] +# The region must match the one in the blob URL, garage rejects requests +# signed for any other. +s3_region = "garage" +api_bind_addr = "[::]:3900" +root_domain = ".s3.garage.localhost" + +[admin] +api_bind_addr = "[::]:3903" diff --git a/test/garage/mzbuild.yml b/test/garage/mzbuild.yml new file mode 100644 index 0000000000000..e70a393754a65 --- /dev/null +++ b/test/garage/mzbuild.yml @@ -0,0 +1,10 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +name: garage diff --git a/test/parallel-benchmark/mzcompose.py b/test/parallel-benchmark/mzcompose.py index 1690cb5e53144..3765185162f85 100644 --- a/test/parallel-benchmark/mzcompose.py +++ b/test/parallel-benchmark/mzcompose.py @@ -14,12 +14,17 @@ import argparse import gc +import json +import math import os +import re import time +from collections import defaultdict from pathlib import Path import matplotlib.pyplot as plt import numpy +import requests from matplotlib.markers import MarkerStyle from materialize import MZ_ROOT, buildkite @@ -36,7 +41,9 @@ ) from materialize.mzcompose.services.azurite import Azurite from materialize.mzcompose.services.balancerd import Balancerd +from materialize.mzcompose.services.blob_store import BLOB_STORES from materialize.mzcompose.services.cockroach import Cockroach +from materialize.mzcompose.services.garage import Garage from materialize.mzcompose.services.kafka import Kafka as KafkaService from materialize.mzcompose.services.kgen import Kgen as KgenService from materialize.mzcompose.services.materialized import Materialized @@ -45,6 +52,7 @@ from materialize.mzcompose.services.mz import Mz from materialize.mzcompose.services.postgres import Postgres from materialize.mzcompose.services.redpanda import Redpanda +from materialize.mzcompose.services.rustfs import RustFs from materialize.mzcompose.services.schema_registry import SchemaRegistry from materialize.mzcompose.services.testdrive import Testdrive from materialize.mzcompose.test_result import ( @@ -104,6 +112,8 @@ def known_regression(scenario: str, other_tag: str | None) -> bool: Cockroach(setup_materialize=True, in_memory=True), Minio(setup_materialize=True), Azurite(), + Garage(setup_materialize=True), + RustFs(setup_materialize=True), KgenService(), Postgres(), MySql(), @@ -291,6 +301,125 @@ def upload_plots( print(f"Saving plots to {plot_paths}") +PERSIST_BLOB_OPS = ["blob_get", "blob_set", "blob_delete", "blob_list_keys"] + +METRIC_LINE = re.compile( + r"^(?P[A-Za-z_:][A-Za-z0-9_:]*)(?:\{(?P[^}]*)\})?\s+(?P\S+)" +) +METRIC_LABEL = re.compile(r'(\w+)="((?:[^"\\]|\\.)*)"') + + +def persist_metrics_texts(c: Composition) -> list[str]: + """The prometheus exposition of each process in the materialized container. + + environmentd is reached through its published port. The clusterds are found + through the service discovery files the process orchestrator writes, whose + targets are TCP proxies that only listen inside the container. + """ + texts = [ + requests.get( + f"http://127.0.0.1:{c.port('materialized', 6878)}/metrics", timeout=30 + ).text + ] + files = c.exec( + "materialized", + "sh", + "-c", + "ls /mzdata/prometheus/*.json 2>/dev/null || true", + capture=True, + silent=True, + ).stdout.split() + for file in files: + static_configs = json.loads( + c.exec("materialized", "cat", file, capture=True, silent=True).stdout + ) + for static_config in static_configs: + if static_config["labels"].get("mz_orchestrator_port") != "internal-http": + continue + for target in static_config["targets"]: + texts.append( + c.exec( + "materialized", + "curl", + "-sf", + f"http://{target}/metrics", + capture=True, + silent=True, + ).stdout + ) + return texts + + +def histogram_quantile(q: float, buckets: dict[float, float]) -> float: + """`histogram_quantile` over cumulative buckets keyed by upper bound.""" + total = buckets.get(math.inf, 0.0) + if total == 0: + return math.nan + rank = q * total + prev_le, prev_count = 0.0, 0.0 + for le in sorted(buckets): + count = buckets[le] + if count >= rank: + if le == math.inf: + return prev_le + return prev_le + (le - prev_le) * (rank - prev_count) / (count - prev_count) + prev_le, prev_count = le, count + return prev_le + + +def report_persist_blob_ops(c: Composition) -> None: + """Prints what persist asked of the blob store, summed over environmentd + and its clusterds, since the container started. + + The measured queries only reach the store through persist, so this is the + share of their latency the store itself accounts for, which the query + statistics alone cannot separate from Materialize's own work. + """ + succeeded: dict[str, float] = defaultdict(float) + failed: dict[str, float] = defaultdict(float) + byte_count: dict[str, float] = defaultdict(float) + seconds: dict[str, float] = defaultdict(float) + buckets: dict[str, dict[float, float]] = defaultdict(lambda: defaultdict(float)) + for text in persist_metrics_texts(c): + for line in text.splitlines(): + match = METRIC_LINE.match(line) + if not match or not match["name"].startswith("mz_persist_external_"): + continue + labels = dict(METRIC_LABEL.findall(match["labels"] or "")) + op = labels.get("op") + if op not in PERSIST_BLOB_OPS: + continue + value = float(match["value"]) + match match["name"]: + case "mz_persist_external_succeeded_count": + succeeded[op] += value + case "mz_persist_external_failed_count": + failed[op] += value + case "mz_persist_external_bytes_count": + byte_count[op] += value + case "mz_persist_external_seconds": + seconds[op] += value + case "mz_persist_external_op_latency_bucket": + buckets[op][float(labels["le"])] += value + + print("Persist blob store operations (all processes, since startup):") + print( + f" {'OP':<15} {'COUNT':>8} {'FAILED':>7} {'MiB':>9} {'MEAN ms':>9} {'P50 ms':>9} {'P99 ms':>9}" + ) + for op in PERSIST_BLOB_OPS: + count = succeeded[op] + failed[op] + mean = seconds[op] / count * 1000 if count else math.nan + # Only gets and sets record a latency histogram. + if op in buckets: + p50 = f"{histogram_quantile(0.5, buckets[op]) * 1000:9.2f}" + p99 = f"{histogram_quantile(0.99, buckets[op]) * 1000:9.2f}" + else: + p50 = p99 = f"{'':>9}" + print( + f" {op:<15} {int(count):>8} {int(failed[op]):>7} {byte_count[op] / 2**20:9.1f} {mean:9.2f} {p50} {p99}" + ) + + def report( mz_string: str, scenario: Scenario, @@ -384,6 +513,7 @@ def run_once( service_names: list[str], tag: str | None, params: str | None, + blob_store: str, args, suffix: str, sqlite_store: bool, @@ -457,8 +587,7 @@ def run_once( default_size=args.size, soft_assertions=False, external_metadata_store=True, - external_blob_store=True, - blob_store_is_azure=args.azurite, + external_blob_store=blob_store, sanity_restart=False, additional_system_parameter_defaults=additional_system_parameter_defaults, metadata_store="cockroach", @@ -467,8 +596,7 @@ def run_once( no_reset=True, seed=1, metadata_store="cockroach", - external_blob_store=True, - blob_store_is_azure=args.azurite, + external_blob_store=blob_store, ), ] target = None @@ -492,6 +620,10 @@ def run_once( mz_string = f"{mz_version} ({target.host})" else: print("~~~ Starting up services") + # On its own first: garage and rustfs create their buckets + # after the server is up, and Materialized only waits for the + # store's container to have started. + c.up(blob_store) c.up(*service_names, Service("testdrive", idle=True)) c.verify_build_profile() @@ -568,12 +700,19 @@ def run_once( failures.extend(new_failures) stats[scenario] = new_stats state.measurements.close() + if not target: + try: + report_persist_blob_ops(c) + except Exception as e: + # Diagnostics only, and this runs on the failure path + # too, where Materialize may be gone. + print(f"Could not collect persist blob store metrics: {e}") if not target: print( "~~~ Resetting services to prevent interference between scenarios" ) - services = service_names + ["cockroach", "testdrive", "minio"] + services = service_names + ["cockroach", "testdrive", blob_store] c.kill(*services) c.rm(*services, destroy_volumes=True) c.rm_volumes("mzdata") @@ -825,7 +964,17 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: help="Store results in SQLite instead of in memory", ) parser.add_argument( - "--azurite", action="store_true", help="Use Azurite as blob store instead of S3" + "--blob-store", + choices=BLOB_STORES, + default="minio", + help="Blob store to run persist against", + ) + + parser.add_argument( + "--other-blob-store", + choices=BLOB_STORES, + default=None, + help="Blob store for the 'OTHER' Mz instance, to compare blob stores rather than (or as well as) Materialize versions. Defaults to --blob-store.", ) parser.add_argument( @@ -883,6 +1032,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: service_names, tag=None, params=args.this_params, + blob_store=args.blob_store, args=args, suffix=f"this_run{run_number}", sqlite_store=args.sqlite_store, @@ -904,10 +1054,13 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: if f.test_class_name_override not in retried_scenario_names ] + this_failures - if args.other_tag: + if args.other_tag or args.other_blob_store: assert not args.mz_url, "Can't set both --mz-url and --other-tag" - tag = resolve_tag(args.other_tag) - print(f"--- Running against other tag for comparison: {tag}") + tag = resolve_tag(args.other_tag) if args.other_tag else None + other_blob_store = args.other_blob_store or args.blob_store + print( + f"--- Running against other configuration for comparison: tag {tag or 'same'}, blob store {other_blob_store}" + ) guarantees_orig = args.guarantees args.guarantees = False other_stats, other_failures = run_once( @@ -916,6 +1069,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: service_names, tag=tag, params=args.other_params, + blob_store=other_blob_store, args=args, suffix=f"other_run{run_number}", sqlite_store=args.sqlite_store,