persist: fix multipart blob reads and add blob store benchmarks

.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
 

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` \

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}")

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/&region=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",
+                },
+            },
+        )

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}")

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/&region=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",
+                },
+            },
+        )

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")