misc.python.materialize.zippy.mz_actions

  1# Copyright Materialize, Inc. and contributors. All rights reserved.
  2#
  3# Use of this software is governed by the Business Source License
  4# included in the LICENSE file at the root of this repository.
  5#
  6# As of the Change Date specified in that file, in accordance with
  7# the Business Source License, use of this software will be governed
  8# by the Apache License, Version 2.0.
  9
 10
 11from materialize.mzcompose.composition import Composition
 12from materialize.mzcompose.services.materialized import (
 13    LEADER_STATUS_HEALTHCHECK,
 14    DeploymentStatus,
 15    Materialized,
 16)
 17from materialize.zippy.balancerd_actions import restart_balancerd
 18from materialize.zippy.balancerd_capabilities import BalancerdIsRunning
 19from materialize.zippy.blob_store_capabilities import BlobStoreIsRunning
 20from materialize.zippy.crdb_capabilities import CockroachIsRunning
 21from materialize.zippy.framework import (
 22    Action,
 23    Capability,
 24    Mz0dtDeployBaseAction,
 25    State,
 26)
 27from materialize.zippy.mz_capabilities import MzIsRunning
 28from materialize.zippy.table_capabilities import TableExists
 29from materialize.zippy.view_capabilities import ViewExists
 30
 31
 32class MzStart(Action):
 33    """Starts a Mz instance (all components are running in the same container)."""
 34
 35    @classmethod
 36    def requires(cls) -> set[type[Capability]]:
 37        return {CockroachIsRunning, BlobStoreIsRunning}
 38
 39    @classmethod
 40    def incompatible_with(cls) -> set[type[Capability]]:
 41        return {MzIsRunning}
 42
 43    def run(self, c: Composition, state: State) -> None:
 44        print(
 45            f"Starting Mz with additional_system_parameter_defaults = {state.additional_system_parameter_defaults}"
 46        )
 47
 48        with c.override(
 49            Materialized(
 50                name=state.mz_service,
 51                external_blob_store=True,
 52                blob_store_is_azure=c.blob_store() == "azurite",
 53                external_metadata_store=True,
 54                deploy_generation=state.deploy_generation,
 55                system_parameter_defaults=state.system_parameter_defaults,
 56                sanity_restart=False,
 57                restart="on-failure",
 58                additional_system_parameter_defaults=state.additional_system_parameter_defaults,
 59                metadata_store="cockroach",
 60                default_replication_factor=2,
 61            )
 62        ):
 63            c.up(state.mz_service)
 64
 65        for config_param in [
 66            "max_tables",
 67            "max_sources",
 68            "max_objects_per_schema",
 69            "max_materialized_views",
 70            "max_sinks",
 71        ]:
 72            c.sql(
 73                f"ALTER SYSTEM SET {config_param} TO 1000",
 74                user="mz_system",
 75                port=6877,
 76                print_statement=False,
 77                service=state.mz_service,
 78            )
 79
 80        c.sql(
 81            """
 82            ALTER CLUSTER quickstart SET (MANAGED = false);
 83            """,
 84            user="mz_system",
 85            port=6877,
 86            service=state.mz_service,
 87        )
 88
 89        # Make sure all eligible LIMIT queries use the PeekPersist optimization
 90        c.sql(
 91            "ALTER SYSTEM SET persist_fast_path_limit = 1000000000",
 92            user="mz_system",
 93            port=6877,
 94            service=state.mz_service,
 95        )
 96
 97    def provides(self) -> list[Capability]:
 98        return [MzIsRunning()]
 99
100
101class MzStop(Action):
102    """Stops the entire Mz instance (all components are running in the same container)."""
103
104    @classmethod
105    def requires(cls) -> set[type[Capability]]:
106        # Technically speaking, we do not need balancerd to be up in order to kill Mz
107        # However, without this protection we frequently end up in a situation where
108        # both are down and Zippy enters a prolonged period of restarting one or the
109        # other and no other useful work can be performed in the meantime.
110        return {MzIsRunning, BalancerdIsRunning}
111
112    def run(self, c: Composition, state: State) -> None:
113        c.kill(state.mz_service)
114
115    def withholds(self) -> set[type[Capability]]:
116        return {MzIsRunning}
117
118
119class MzRestart(Action):
120    """Restarts the entire Mz instance (all components are running in the same container)."""
121
122    @classmethod
123    def requires(cls) -> set[type[Capability]]:
124        return {MzIsRunning}
125
126    def run(self, c: Composition, state: State) -> None:
127        with c.override(
128            Materialized(
129                name=state.mz_service,
130                external_blob_store=True,
131                blob_store_is_azure=c.blob_store() == "azurite",
132                external_metadata_store=True,
133                deploy_generation=state.deploy_generation,
134                system_parameter_defaults=state.system_parameter_defaults,
135                sanity_restart=False,
136                restart="on-failure",
137                additional_system_parameter_defaults=state.additional_system_parameter_defaults,
138                metadata_store="cockroach",
139                default_replication_factor=2,
140            )
141        ):
142            c.kill(state.mz_service)
143            c.up(state.mz_service)
144
145
146class Mz0dtDeploy(Mz0dtDeployBaseAction):
147    """Switches Mz to a new deployment using 0dt."""
148
149    @classmethod
150    def requires(cls) -> set[type[Capability]]:
151        return {MzIsRunning}
152
153    def run(self, c: Composition, state: State) -> None:
154        state.deploy_generation += 1
155
156        state.mz_service = (
157            "materialized" if state.deploy_generation % 2 == 0 else "materialized2"
158        )
159
160        print(f"Deploying generation {state.deploy_generation} on {state.mz_service}")
161
162        with c.override(
163            Materialized(
164                name=state.mz_service,
165                external_blob_store=True,
166                blob_store_is_azure=c.blob_store() == "azurite",
167                external_metadata_store=True,
168                deploy_generation=state.deploy_generation,
169                system_parameter_defaults=state.system_parameter_defaults,
170                sanity_restart=False,
171                restart="on-failure",
172                healthcheck=LEADER_STATUS_HEALTHCHECK,
173                additional_system_parameter_defaults=state.additional_system_parameter_defaults,
174                metadata_store="cockroach",
175                default_replication_factor=2,
176            ),
177        ):
178            c.up(state.mz_service, detach=True)
179            c.await_mz_deployment_status(
180                DeploymentStatus.READY_TO_PROMOTE, state.mz_service
181            )
182            c.promote_mz(state.mz_service)
183            c.await_mz_deployment_status(DeploymentStatus.IS_LEADER, state.mz_service)
184            c.stop(
185                (
186                    "materialized2"
187                    if state.mz_service == "materialized"
188                    else "materialized"
189                ),
190                wait=True,
191            )
192
193        # Balancerd's resolver is fixed at startup and still points at the
194        # previous generation. Re-point it at the new leader.
195        if c.is_running("balancerd"):
196            restart_balancerd(c, state)
197
198
199class KillClusterd(Action):
200    """Kills the clusterd processes in the environmentd container. The process orchestrator will restart them."""
201
202    @classmethod
203    def requires(cls) -> list[set[type[Capability]]]:
204        # Only kill once a dataflow-bearing object exists, so the kill has
205        # something to disrupt.
206        return [{MzIsRunning, ViewExists}, {MzIsRunning, TableExists}]
207
208    def run(self, c: Composition, state: State) -> None:
209        # Depending on the workload, clusterd may not be running, hence the || true
210        c.exec(state.mz_service, "bash", "-c", "kill -9 `pidof clusterd` || true")