WIP: Azurite testing

MaterializeInc · Dec 30, 2024 · 2b03352 · 2b03352
1 parent 2222c8c
commit 2b03352
Show file tree

Hide file tree

Showing 24 changed files with 736 additions and 160 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/ci/nightly/pipeline.template.yml b/ci/nightly/pipeline.template.yml
@@ -338,6 +338,18 @@ steps:
               args: [--persistent-user-tables]
         skip: "Persistence tests disabled"
 
+      - id: azurite-testdrive
+        label: "testdrive with Azurite blob store"
+        depends_on: build-aarch64
+        timeout_in_minutes: 180
+        agents:
+          queue: hetzner-aarch64-8cpu-16gb
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: testdrive
+              args: [--azurite]
+
+
   - group: Limits
     key: limits-group
     steps:

diff --git a/misc/python/materialize/checks/mzcompose_actions.py b/misc/python/materialize/checks/mzcompose_actions.py
@@ -72,7 +72,7 @@ def execute(self, e: Executor) -> None:
             name=self.mz_service,
             image=image,
             external_metadata_store=True,
-            external_minio=True,
+            external_blob_store=True,
             environment_extra=self.environment_extra,
             system_parameter_defaults=self.system_parameter_defaults,
             additional_system_parameter_defaults=self.additional_system_parameter_defaults,

diff --git a/misc/python/materialize/data_ingest/transaction_def.py b/misc/python/materialize/data_ingest/transaction_def.py
@@ -82,7 +82,7 @@ def generate(self, fields: list[Field]) -> Iterator[Transaction | None]:
                 Materialized(
                     name=self.workload.mz_service,
                     ports=ports,
-                    external_minio=True,
+                    external_blob_store=True,
                     external_metadata_store=True,
                     system_parameter_defaults=get_default_system_parameters(
                         zero_downtime=True
@@ -129,7 +129,7 @@ def generate(self, fields: list[Field]) -> Iterator[Transaction | None]:
                 Materialized(
                     name=self.workload.mz_service,
                     ports=ports,
-                    external_minio=True,
+                    external_blob_store=True,
                     external_metadata_store=True,
                     system_parameter_defaults=get_default_system_parameters(
                         zero_downtime=True

diff --git a/misc/python/materialize/mzcompose/services/azure.py b/misc/python/materialize/mzcompose/services/azure.py
@@ -7,14 +7,12 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0.
 
-import os
 
-from materialize import MZ_ROOT
-from materialize.mzcompose import loader
-from materialize.mzcompose.service import (
-    Service,
-    ServiceHealthcheck
-)
+from materialize.mzcompose.service import Service, ServiceHealthcheck
+
+
+def azure_blob_uri(address: str = "azurite") -> str:
+    return f"http://devstoreaccount1.{address}:10000/container"
 
 
 class Azurite(Service):
@@ -23,18 +21,27 @@ class Azurite(Service):
     def __init__(
         self,
         name: str = "azurite",
-        aliases: list[str] = ["azurite"],
+        aliases: list[str] = ["azurite", "devstoreaccount1.azurite"],
         image: str | None = None,
         command: list[str] | None = None,
         in_memory: bool = False,
         healthcheck: ServiceHealthcheck | None = None,
         stop_grace_period: str = "120s",
     ):
         if image is None:
-            image = f"mcr.microsoft.com/azure-storage/azurite:{self.DEFAULT_AZURITE_TAG}"
+            image = (
+                f"mcr.microsoft.com/azure-storage/azurite:{self.DEFAULT_AZURITE_TAG}"
+            )
 
         if command is None:
-            command = ["azurite-blob", "--blobHost", "0.0.0.0:10000", "--blobPort", "10000"]
+            command = [
+                "azurite-blob",
+                "--blobHost",
+                "0.0.0.0",
+                "--blobPort",
+                "10000",
+                "--disableProductStyleUrl",
+            ]
 
         if in_memory:
             command.append("--inMemoryPersistence")

diff --git a/misc/python/materialize/mzcompose/services/materialized.py b/misc/python/materialize/mzcompose/services/materialized.py
@@ -27,6 +27,7 @@
     ServiceConfig,
     ServiceDependency,
 )
+from materialize.mzcompose.services.azure import azure_blob_uri
 from materialize.mzcompose.services.minio import minio_blob_uri
 from materialize.mzcompose.services.postgres import METADATA_STORE
 
@@ -49,7 +50,8 @@ def __init__(
         environment_id: str | None = None,
         propagate_crashes: bool = True,
         external_metadata_store: str | bool = False,
-        external_minio: str | bool = False,
+        external_blob_store: str | bool = False,
+        blob_store_is_azure: bool = False,
         unsafe_mode: bool = True,
         restart: str | None = None,
         use_default_volumes: bool = True,
@@ -149,10 +151,15 @@ def __init__(
             environment_id = DEFAULT_MZ_ENVIRONMENT_ID
         command += [f"--environment-id={environment_id}"]
 
-        if external_minio:
-            depends_graph["minio"] = {"condition": "service_healthy"}
-            address = "minio" if external_minio == True else external_minio
-            persist_blob_url = minio_blob_uri(address)
+        if external_blob_store:
+            blob_store = "azurite" if blob_store_is_azure else "minio"
+            depends_graph[blob_store] = {"condition": "service_healthy"}
+            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)
+            )
 
         if persist_blob_url:
             command.append(f"--persist-blob-url={persist_blob_url}")

diff --git a/misc/python/materialize/mzcompose/services/testdrive.py b/misc/python/materialize/mzcompose/services/testdrive.py
@@ -52,7 +52,7 @@ def __init__(
         aws_secret_access_key: str | None = "minioadmin",
         no_consistency_checks: bool = False,
         external_metadata_store: bool = False,
-        external_minio: bool = False,
+        external_blob_store: bool = False,
         fivetran_destination: bool = False,
         fivetran_destination_url: str = "http://fivetran-destination:6874",
         fivetran_destination_files_path: str = "/share/tmp",
@@ -157,7 +157,7 @@ def __init__(
                 f"--fivetran-destination-files-path={fivetran_destination_files_path}"
             )
 
-        if external_minio:
+        if external_blob_store:
             depends_graph["minio"] = {"condition": "service_healthy"}
             persist_blob_url = "s3://minioadmin:minioadmin@persist/persist?endpoint=http://minio:9000/&region=minio"
             entrypoint.append(f"--persist-blob-url={persist_blob_url}")

diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py
@@ -1614,7 +1614,7 @@ def run(self, exe: Executor) -> bool:
         with self.composition.override(
             Materialized(
                 restart="on-failure",
-                external_minio="toxiproxy",
+                external_blob_store="toxiproxy",
                 external_metadata_store="toxiproxy",
                 ports=["6975:6875", "6976:6876", "6977:6877"],
                 sanity_restart=self.sanity_restart,
@@ -1655,7 +1655,7 @@ def run(self, exe: Executor) -> bool:
         with self.composition.override(
             Materialized(
                 name=mz_service,
-                external_minio="toxiproxy",
+                external_blob_store="toxiproxy",
                 external_metadata_store="toxiproxy",
                 ports=ports,
                 sanity_restart=self.sanity_restart,

diff --git a/misc/python/materialize/zippy/backup_and_restore_actions.py b/misc/python/materialize/zippy/backup_and_restore_actions.py
@@ -28,7 +28,7 @@ def run(self, c: Composition, state: State) -> None:
         with c.override(
             Materialized(
                 name=state.mz_service,
-                external_minio=True,
+                external_blob_store=True,
                 external_metadata_store=True,
                 deploy_generation=state.deploy_generation,
                 system_parameter_defaults=state.system_parameter_defaults,

diff --git a/misc/python/materialize/zippy/mz_actions.py b/misc/python/materialize/zippy/mz_actions.py
@@ -81,7 +81,7 @@ def run(self, c: Composition, state: State) -> None:
         with c.override(
             Materialized(
                 name=state.mz_service,
-                external_minio=True,
+                external_blob_store=True,
                 external_metadata_store=True,
                 deploy_generation=state.deploy_generation,
                 system_parameter_defaults=state.system_parameter_defaults,
@@ -158,7 +158,7 @@ def run(self, c: Composition, state: State) -> None:
         with c.override(
             Materialized(
                 name=state.mz_service,
-                external_minio=True,
+                external_blob_store=True,
                 external_metadata_store=True,
                 deploy_generation=state.deploy_generation,
                 system_parameter_defaults=state.system_parameter_defaults,
@@ -190,7 +190,7 @@ def run(self, c: Composition, state: State) -> None:
         with c.override(
             Materialized(
                 name=state.mz_service,
-                external_minio=True,
+                external_blob_store=True,
                 external_metadata_store=True,
                 deploy_generation=state.deploy_generation,
                 system_parameter_defaults=state.system_parameter_defaults,

diff --git a/src/persist/src/abs.rs b/src/persist/src/abs.rs
@@ -16,7 +16,7 @@ use anyhow::anyhow;
 use async_trait::async_trait;
 use azure_core::StatusCode;
 use azure_identity::create_default_credential;
-use azure_storage::{prelude::*, EMULATOR_ACCOUNT};
+use azure_storage::{prelude::*, CloudLocation, EMULATOR_ACCOUNT};
 use azure_storage_blobs::prelude::*;
 use bytes::Bytes;
 use futures_util::StreamExt;
@@ -63,9 +63,15 @@ impl ABSBlobConfig {
     ) -> Result<Self, Error> {
         let client = if account == EMULATOR_ACCOUNT {
             info!("Connecting to Azure emulator");
-            ClientBuilder::emulator()
-                .blob_service_client()
-                .container_client(container)
+            ClientBuilder::with_location(
+                CloudLocation::Emulator {
+                    address: url.domain().expect("domain for Azure emulator").to_string(),
+                    port: url.port().expect("port for Azure emulator"),
+                },
+                StorageCredentials::emulator(),
+            )
+            .blob_service_client()
+            .container_client(container)
         } else {
             let sas_credentials = match url.query() {
                 Some(query) => Some(StorageCredentials::sas_token(query)),
@@ -167,7 +173,9 @@ impl ABSBlob {
             // TODO: we could move this logic into the test harness.
             // it's currently here because it's surprisingly annoying to
             // create the container out-of-band
-            let _ = config.client.create().await;
+            if let Err(e) = config.client.create().await {
+                warn!("Failed to create container: {e}");
+            }
         }
 
         let ret = ABSBlob {

diff --git a/test/backup-restore-postgres/mzcompose.py b/test/backup-restore-postgres/mzcompose.py
@@ -25,7 +25,7 @@
     Minio(setup_materialize=True),
     Mc(),
     Materialized(
-        external_minio=True,
+        external_blob_store=True,
         external_metadata_store=True,
         sanity_restart=False,
         metadata_store="postgres-metadata",

diff --git a/test/backup-restore/mzcompose.py b/test/backup-restore/mzcompose.py
@@ -25,7 +25,7 @@
     Minio(setup_materialize=True),
     Mc(),
     Materialized(
-        external_minio=True,
+        external_blob_store=True,
         external_metadata_store=True,
         sanity_restart=False,
         metadata_store="cockroach",

diff --git a/test/cluster/mzcompose.py b/test/cluster/mzcompose.py
@@ -3218,7 +3218,7 @@ def workflow_test_incident_70(c: Composition) -> None:
     with c.override(
         Materialized(
             external_metadata_store=True,
-            external_minio=True,
+            external_blob_store=True,
             sanity_restart=False,
         ),
         Minio(setup_materialize=True),

diff --git a/test/data-ingest/mzcompose.py b/test/data-ingest/mzcompose.py
@@ -56,7 +56,7 @@
     # Fixed port so that we keep the same port after restarting Mz in disruptions
     Materialized(
         ports=["16875:6875"],
-        external_minio=True,
+        external_blob_store=True,
         external_metadata_store=True,
         system_parameter_defaults=get_default_system_parameters(zero_downtime=True),
         additional_system_parameter_defaults={"enable_table_keys": "true"},
@@ -65,7 +65,7 @@
     Materialized(
         name="materialized2",
         ports=["26875:6875"],
-        external_minio=True,
+        external_blob_store=True,
         external_metadata_store=True,
         system_parameter_defaults=get_default_system_parameters(zero_downtime=True),
         additional_system_parameter_defaults={"enable_table_keys": "true"},

diff --git a/test/feature-benchmark/mzcompose.py b/test/feature-benchmark/mzcompose.py
@@ -318,7 +318,7 @@ def create_mz_service(
         additional_system_parameter_defaults=additional_system_parameter_defaults,
         external_metadata_store=True,
         metadata_store="cockroach",
-        external_minio=True,
+        external_blob_store=True,
         sanity_restart=False,
     )
 

diff --git a/test/mzcompose_examples/mzcompose.py b/test/mzcompose_examples/mzcompose.py
@@ -73,7 +73,7 @@ def workflow_mz_with_options(c: Composition) -> None:
 
 
 def workflow_minio(c: Composition) -> None:
-    mz = Materialized(external_minio=True)
+    mz = Materialized(external_blob_store=True)
 
     with c.override(mz):
         c.up("materialized")
diff --git a/test/parallel-benchmark/mzcompose.py b/test/parallel-benchmark/mzcompose.py
@@ -438,7 +438,7 @@ def run_once(
                 default_size=args.size,
                 soft_assertions=False,
                 external_metadata_store=True,
-                external_minio=True,
+                external_blob_store=True,
                 sanity_restart=False,
                 additional_system_parameter_defaults=ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS
                 | {"max_connections": "100000"},

diff --git a/test/parallel-workload/mzcompose.py b/test/parallel-workload/mzcompose.py
@@ -83,7 +83,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
 
     with c.override(
         Materialized(
-            external_minio="toxiproxy",
+            external_blob_store="toxiproxy",
             external_metadata_store="toxiproxy",
             ports=["6975:6875", "6976:6876", "6977:6877"],
             sanity_restart=sanity_restart,

diff --git a/test/platform-checks/mzcompose.py b/test/platform-checks/mzcompose.py
@@ -54,7 +54,7 @@ def create_mzs(
         Materialized(
             name=mz_name,
             external_metadata_store=True,
-            external_minio=True,
+            external_blob_store=True,
             sanity_restart=False,
             volumes_extra=["secrets:/share/secrets"],
             metadata_store="cockroach",

diff --git a/test/testdrive-old-kafka-src-syntax/mzcompose.py b/test/testdrive-old-kafka-src-syntax/mzcompose.py
@@ -37,9 +37,9 @@
     Postgres(),
     MySql(),
     Minio(setup_materialize=True, additional_directories=["copytos3"]),
-    Materialized(external_minio=True),
+    Materialized(external_blob_store=True),
     FivetranDestination(volumes_extra=["tmp:/share/tmp"]),
-    Testdrive(external_minio=True),
+    Testdrive(external_blob_store=True),
 ]
 
 
@@ -133,7 +133,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
 
     materialized = Materialized(
         default_size=args.default_size,
-        external_minio=True,
+        external_blob_store=True,
         additional_system_parameter_defaults=additional_system_parameter_defaults,
     )
 
@@ -144,7 +144,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
         validate_catalog_store=True,
         default_timeout=args.default_timeout,
         volumes_extra=["mzdata:/mzdata"],
-        external_minio=True,
+        external_blob_store=True,
         fivetran_destination=True,
         fivetran_destination_files_path="/share/tmp",
         entrypoint_extra=[