Class Clusterd

misc/python/materialize/mzcompose/services/clusterd.py:25–117

class Clusterd(Service):
    def __init__(
        self,
        name: str = "clusterd",
        image: str | None = None,
        environment_id: str | None = None,
        environment_extra: list[str] = [],
        memory: str | None = None,
        cpu: str | None = None,
        options: list[str] = [],
        restart: str = "no",
        stop_grace_period: str = "120s",
        scratch_directory: str = "/scratch",
        volumes: list[str] = [],
        workers: int = 1,
        process_names: list[str] = [],
        mz_service: str = "materialized",
    ) -> None:
        environment = [
            "CLUSTERD_LOG_FILTER",
            f"CLUSTERD_GRPC_HOST={name}",
            # For old Mz versions
            "CLUSTERD_USE_CTP=true",
            "MZ_SOFT_ASSERTIONS=1",
            "MZ_EAT_MY_DATA=1",
            # Defaults that were previously set by the clusterd entrypoint.sh.
            "CLUSTERD_STORAGE_CONTROLLER_LISTEN_ADDR=0.0.0.0:2100",
            "CLUSTERD_COMPUTE_CONTROLLER_LISTEN_ADDR=0.0.0.0:2101",
            "CLUSTERD_INTERNAL_HTTP_LISTEN_ADDR=0.0.0.0:6878",
            "CLUSTERD_SECRETS_READER=local-file",
            "CLUSTERD_SECRETS_READER_LOCAL_FILE_DIR=/mzdata/secrets",
            "LD_PRELOAD=libeatmydata.so",
            f"CLUSTERD_PERSIST_PUBSUB_URL=http://{mz_service}:6879",
            *environment_extra,
        ]

        if not environment_id:
            environment_id = DEFAULT_MZ_ENVIRONMENT_ID

        environment += [f"CLUSTERD_ENVIRONMENT_ID={environment_id}"]

        process_names = process_names if process_names else [name]
        process_index = process_names.index(name)
        compute_timely_config = timely_config(
            process_names, 2102, workers, DEFAULT_COMPUTE_EXERT_PROPORTIONALITY
        )
        storage_timely_config = timely_config(
            process_names, 2103, workers, DEFAULT_STORAGE_EXERT_PROPORTIONALITY
        )

        environment += [
            f"CLUSTERD_PROCESS={process_index}",
            f"CLUSTERD_COMPUTE_TIMELY_CONFIG={compute_timely_config}",
            f"CLUSTERD_STORAGE_TIMELY_CONFIG={storage_timely_config}",
        ]

        options = ["clusterd", f"--scratch-directory={scratch_directory}", *options]