diff --git a/misc/python/materialize/checks/executors.py b/misc/python/materialize/checks/executors.py index 52b8d91b8dc6e..ae1a0c41f586b 100644 --- a/misc/python/materialize/checks/executors.py +++ b/misc/python/materialize/checks/executors.py @@ -26,10 +26,13 @@ class Executor: # scenarios which are still in development and not available a few versions # back already. current_mz_version: MzVersion - # All the system settings we have already set in previous Mz versions. No - # need to set them again in a future version since they should be - # persisted. - system_settings: set[str] = set() + + def __init__(self) -> None: + # All the system settings we have already set in previous Mz versions. + # No need to set them again in a future version since they should be + # persisted. Per instance so that a fresh environment (and its fresh + # executor) starts with a clean slate. + self.system_settings: set[str] = set() def testdrive( self, input: str, caller: Traceback | None = None, mz_service: str | None = None @@ -55,6 +58,7 @@ def join(self, handle: Any) -> None: class MzcomposeExecutor(Executor): def __init__(self, composition: Composition) -> None: + super().__init__() self.composition = composition def mzcompose_composition(self) -> Composition: @@ -85,7 +89,7 @@ def run_pyaction( class MzcomposeExecutorParallel(MzcomposeExecutor): def __init__(self, composition: Composition) -> None: - self.composition = composition + super().__init__(composition) self.exception: BaseException | None = None def testdrive( @@ -148,6 +152,7 @@ def join(self, handle: Any) -> None: class CloudtestExecutor(Executor): def __init__(self, application: MaterializeApplication, version: MzVersion) -> None: + super().__init__() self.application = application self.seed = random.getrandbits(32) self.current_mz_version = version diff --git a/misc/python/materialize/checks/mzcompose_actions.py b/misc/python/materialize/checks/mzcompose_actions.py index 6d0a7fbe066f5..e9a37a1bcd4ec 100644 --- a/misc/python/materialize/checks/mzcompose_actions.py +++ b/misc/python/materialize/checks/mzcompose_actions.py @@ -100,6 +100,13 @@ def execute(self, e: Executor) -> None: self.tag or MzVersion.parse_cargo(), config_name ) + # Scenario-level parameters come from the command line + # (--system-param) and win over action-specific ones. + additional_system_parameter_defaults = { + **self.additional_system_parameter_defaults, + **self.scenario.additional_system_parameter_defaults, + } + mz = Materialized( name=self.mz_service, image=image, @@ -111,7 +118,7 @@ def execute(self, e: Executor) -> None: blob_store_is_azure=self.scenario.features.azurite_enabled(), environment_extra=self.environment_extra, system_parameter_defaults=self.system_parameter_defaults, - additional_system_parameter_defaults=self.additional_system_parameter_defaults, + additional_system_parameter_defaults=additional_system_parameter_defaults, system_parameter_version=self.system_parameter_version, soft_assertions=self.soft_assertions, sanity_restart=False, @@ -121,7 +128,7 @@ def execute(self, e: Executor) -> None: restart=self.restart, force_migrations=self.force_migrations, publish=self.publish, - default_replication_factor=2, + default_replication_factor=self.scenario.default_replication_factor, support_external_clusterd=True, builtin_system_cluster_replication_factor=self.builtin_system_cluster_replication_factor, builtin_probe_cluster_replication_factor=self.builtin_probe_cluster_replication_factor, @@ -214,10 +221,13 @@ def execute(self, e: Executor) -> None: """) self.handle = e.testdrive(input=input, mz_service=self.mz_service) - e.system_settings.update(system_settings) + self.applied_settings = system_settings def join(self, e: Executor) -> None: e.join(self.handle) + # Only record the settings as applied once the testdrive fragment that + # applies them has actually succeeded. + e.system_settings.update(self.applied_settings) class SetupSqlServerTesting(MzcomposeAction): @@ -303,10 +313,23 @@ def execute(self, e: Executor) -> None: port=6877, user="mz_system", ) + + # Drop all existing replicas so that quickstart runs solely on + # clusterd_compute_1. A leftover orchestrated replica would keep + # serving queries and mask failures of the unorchestrated one. + replica_names = [row[0] for row in c.sql_query(""" + SELECT mz_cluster_replicas.name + FROM mz_cluster_replicas + JOIN mz_clusters ON mz_cluster_replicas.cluster_id = mz_clusters.id + WHERE mz_clusters.name = 'quickstart' + """)] + drop_replicas = "\n".join( + f"DROP CLUSTER REPLICA quickstart.{name};" for name in replica_names + ) c.sql( - """ + f""" ALTER CLUSTER quickstart SET (MANAGED = false); - DROP CLUSTER REPLICA quickstart.r1; + {drop_replicas} CREATE CLUSTER REPLICA quickstart.r1 STORAGECTL ADDRESSES ['clusterd_compute_1:2100'], STORAGE ADDRESSES ['clusterd_compute_1:2103'], diff --git a/misc/python/materialize/checks/scenarios.py b/misc/python/materialize/checks/scenarios.py index dcdc47b74317d..bbaaafac24eed 100644 --- a/misc/python/materialize/checks/scenarios.py +++ b/misc/python/materialize/checks/scenarios.py @@ -9,6 +9,7 @@ from dataclasses import dataclass from random import Random +from typing import Any from materialize.checks.actions import Action, Initialize, Manipulate, Validate from materialize.checks.checks import Check @@ -53,10 +54,20 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + additional_system_parameter_defaults: dict[str, str] | None = None, + default_replication_factor: int = 2, ) -> None: - self._checks = checks + # Copy the list since checks() shuffles it in place. + self._checks = list(checks) self.executor = executor self.features = features + # Applied by StartMz on top of any action-specific parameters so that + # --system-param and --default-replication-factor reach every Mz + # instance started by a scenario. + self.additional_system_parameter_defaults = ( + additional_system_parameter_defaults or {} + ) + self.default_replication_factor = default_replication_factor self.rng = None if seed is None else Random(seed) self._base_version = MzVersion.parse_cargo() @@ -306,8 +317,9 @@ def __init__( features: Features, seed: str | None, change_entries: list[SystemVarChangeEntry], + **kwargs: Any, ): - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) self.change_entries = change_entries def actions(self) -> list[Action]: diff --git a/misc/python/materialize/checks/scenarios_backup_restore.py b/misc/python/materialize/checks/scenarios_backup_restore.py index fce7983dca3d0..e43843f551ce3 100644 --- a/misc/python/materialize/checks/scenarios_backup_restore.py +++ b/misc/python/materialize/checks/scenarios_backup_restore.py @@ -26,6 +26,9 @@ class BackupAndRestoreAfterManipulate(Scenario): """ def actions(self) -> list[Action]: + # Restart Mz through StartMz instead of Restore(restart_mz=True) so + # that the restarted instance uses the same service definition as the + # one it was originally started with. return [ StartMz(self), Initialize(self), @@ -33,7 +36,8 @@ def actions(self) -> list[Action]: Manipulate(self, phase=2), Backup(), KillMz(), - Restore(), + Restore(restart_mz=False), + StartMz(self), Validate(self), ] @@ -48,7 +52,8 @@ def actions(self) -> list[Action]: Manipulate(self, phase=1), Backup(), KillMz(), - Restore(), + Restore(restart_mz=False), + StartMz(self), Manipulate(self, phase=2), Validate(self), ] @@ -86,18 +91,22 @@ def actions(self) -> list[Action]: Initialize(self), Backup(), KillMz(), - Restore(), + Restore(restart_mz=False), + StartMz(self), Manipulate(self, phase=1), Backup(), KillMz(), - Restore(), + Restore(restart_mz=False), + StartMz(self), Manipulate(self, phase=2), Backup(), KillMz(), - Restore(), + Restore(restart_mz=False), + StartMz(self), Validate(self), Backup(), KillMz(), - Restore(), + Restore(restart_mz=False), + StartMz(self), Validate(self), ] diff --git a/misc/python/materialize/checks/scenarios_upgrade.py b/misc/python/materialize/checks/scenarios_upgrade.py index 80545ee221cc3..06742cf3451ac 100644 --- a/misc/python/materialize/checks/scenarios_upgrade.py +++ b/misc/python/materialize/checks/scenarios_upgrade.py @@ -9,6 +9,7 @@ from dataclasses import dataclass +from typing import Any from materialize.checks.actions import Action, Initialize, Manipulate, Sleep, Validate from materialize.checks.checks import Check @@ -100,11 +101,12 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self._self_managed_versions = get_self_managed_versions( max_version=MzVersion.parse_cargo() ) - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def base_version(self) -> MzVersion: return self._self_managed_versions[-1] @@ -146,11 +148,12 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self._self_managed_versions = get_self_managed_versions( max_version=MzVersion.parse_cargo() ) - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def base_version(self) -> MzVersion: return self._self_managed_versions[-2] @@ -273,9 +276,10 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self.minor_versions = get_minor_versions() - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def base_version(self) -> MzVersion: return self.minor_versions[3] @@ -325,15 +329,6 @@ def actions(self) -> list[Action]: class UpgradeV80Migration(Scenario): """Test upgrade v26.17.1 (catalog 80) -> v26.18.0 (catalog 81, buggy v80_to_v81) -> v26.24.0 (catalog 83, partial fix in v82_to_v83) -> current (catalog 84, full fix in v83_to_v84)""" - def __init__( - self, - checks: list[type[Check]], - executor: Executor, - features: Features, - seed: str | None = None, - ): - super().__init__(checks, executor, features, seed) - def base_version(self) -> MzVersion: return MzVersion.parse_mz("v26.17.1") @@ -677,9 +672,10 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self.self_managed_versions = get_compatible_upgrade_from_versions() - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def base_version(self) -> MzVersion: return self.self_managed_versions[0] @@ -733,9 +729,10 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self.self_managed_versions = get_compatible_upgrade_from_versions() - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def base_version(self) -> MzVersion: return self.self_managed_versions[0] @@ -799,9 +796,10 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self.self_managed_versions = get_compatible_upgrade_from_versions() - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def _generate_random_upgrade_path( self, @@ -913,9 +911,10 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self.self_managed_versions = get_compatible_upgrade_from_versions() - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def base_version(self) -> MzVersion: return self.self_managed_versions[0] diff --git a/misc/python/materialize/checks/scenarios_zero_downtime.py b/misc/python/materialize/checks/scenarios_zero_downtime.py index a6d254d3e130f..d284c44668690 100644 --- a/misc/python/materialize/checks/scenarios_zero_downtime.py +++ b/misc/python/materialize/checks/scenarios_zero_downtime.py @@ -8,6 +8,8 @@ # by the Apache License, Version 2.0. +from typing import Any + from materialize.checks.actions import ( Action, BumpVersion, @@ -293,9 +295,10 @@ def __init__( executor: Executor, features: Features, seed: str | None = None, + **kwargs: Any, ): self.minor_versions = get_minor_versions() - super().__init__(checks, executor, features, seed) + super().__init__(checks, executor, features, seed, **kwargs) def base_version(self) -> MzVersion: return self.minor_versions[3] diff --git a/test/cloudtest/test_upgrade.py b/test/cloudtest/test_upgrade.py index 60e40b3705c23..ba3bf79903bd1 100644 --- a/test/cloudtest/test_upgrade.py +++ b/test/cloudtest/test_upgrade.py @@ -93,6 +93,9 @@ def test_upgrade(aws_region: str | None, log_filter: str | None, dev: bool) -> N AlterConnectionHost, } ) + # Sort so that parallel jobs, each of which computes its own shard, agree + # on the shard split. Set iteration order differs between processes. + checks.sort(key=lambda ch: ch.__name__) checks = buildkite.shard_list(checks, lambda ch: ch.__name__) if buildkite.get_parallelism_index() != 0 or buildkite.get_parallelism_count() != 1: print( diff --git a/test/platform-checks/mzcompose.py b/test/platform-checks/mzcompose.py index 59f309ea7e644..3cc6a788ec91c 100644 --- a/test/platform-checks/mzcompose.py +++ b/test/platform-checks/mzcompose.py @@ -268,7 +268,9 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: scenarios = [globals()[args.scenario]] else: base_scenarios = {SystemVarChange} - scenarios = all_subclasses(Scenario) - base_scenarios + scenarios = sorted( + all_subclasses(Scenario) - base_scenarios, key=lambda s: s.__name__ + ) if args.check: all_checks = {check.__name__: check for check in all_subclasses(Check)} @@ -287,9 +289,12 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: additional_system_parameter_defaults = {} for val in args.system_param or []: - x = val[0].split("=", maxsplit=1) - assert len(x) == 2, f"--system-param '{val}' should be the format =" - additional_system_parameter_defaults[x[0]] = x[1] + for param in val: + x = param.split("=", maxsplit=1) + assert ( + len(x) == 2 + ), f"--system-param '{param}' should be the format =" + additional_system_parameter_defaults[x[0]] = x[1] with c.override( *create_mzs( @@ -300,7 +305,24 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: external_metadata_store=args.external_metadata_store, ) ): - executor = MzcomposeExecutor(composition=c) + executor_class = ( + MzcomposeExecutorParallel + if args.execution_mode is ExecutionMode.PARALLEL + else MzcomposeExecutor + ) + + ran_scenario = False + + def reset_environment() -> None: + # Scenarios (and checks in oneatatime mode) expect a fresh + # environment, so wipe all state, including the metadata and blob + # stores, before every run but the first. + nonlocal ran_scenario + if ran_scenario: + c.down(destroy_volumes=True, sanity_restart_mz=False) + ran_scenario = True + setup(c, args.external_blob_store) + for scenario_class in scenarios: assert issubclass( scenario_class, Scenario @@ -308,22 +330,17 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: print(f"Testing scenario {scenario_class}...") - executor_class = ( - MzcomposeExecutorParallel - if args.execution_mode is ExecutionMode.PARALLEL - else MzcomposeExecutor - ) - executor = executor_class(composition=c) - execution_mode = args.execution_mode if execution_mode in [ExecutionMode.SEQUENTIAL, ExecutionMode.PARALLEL]: - setup(c, args.external_blob_store) + reset_environment() scenario = scenario_class( checks=checks, - executor=executor, + executor=executor_class(composition=c), features=features, seed=args.seed, + additional_system_parameter_defaults=additional_system_parameter_defaults, + default_replication_factor=args.default_replication_factor, ) scenario.run() elif execution_mode is ExecutionMode.ONEATATIME: @@ -334,12 +351,14 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: c.override_current_testcase_name( f"Check '{check}' with scenario '{scenario_class}'" ) - setup(c, args.external_blob_store) + reset_environment() scenario = scenario_class( checks=[check], - executor=executor, + executor=executor_class(composition=c), features=features, seed=args.seed, + additional_system_parameter_defaults=additional_system_parameter_defaults, + default_replication_factor=args.default_replication_factor, ) scenario.run() else: