serhiy-bzhezytskyy commented on code in PR #108: URL: https://github.com/apache/solr-orbit/pull/108#discussion_r4291354292
########## solrorbit/tools/launchr/errors.py: ########## @@ -0,0 +1,50 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Exception hierarchy for launchr. + +Every failure launchr raises deliberately is a :class:`LaunchrError`. The +orchestrator treats these as expected outcomes worth reporting; anything else +escaping a plugin is a bug and is logged with a traceback. +""" + + +class LaunchrError(Exception): + """Base class for every error launchr raises deliberately.""" + + +class ConfigError(LaunchrError): + """The configuration is invalid, incomplete, or self-contradictory.""" + + +class ProvisioningError(LaunchrError): + """A provisioning strategy could not create or tear down infrastructure.""" + + +class DeployError(LaunchrError): + """A Solr install method could not bring Solr up, or Solr came up wrong.""" Review Comment: A failed deploy leaves the orchestrator without a deployment to collect logs from (see the comment on `docker.py`), so the error carries what the install method managed to read. ```suggestion class DeployError(LaunchrError): """A Solr install method could not bring Solr up, or Solr came up wrong. ``logs`` carries whatever the install method could read before it cleaned up, because a failed deploy leaves the orchestrator with no deployment to ask. """ def __init__(self, message: str, *, logs: dict[str, bytes] | None = None): super().__init__(message) self.logs = logs or {} ``` ########## solrorbit/tools/launchr/install/docker.py: ########## @@ -0,0 +1,143 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Run Solr in a Docker container on the local machine. + +This wraps :class:`solrorbit.builder.solr_provisioner.SolrDockerLauncher`, the +same launcher Solr Orbit's own ``docker`` pipeline uses, rather than growing a +second container-management path. It is single-container with embedded +ZooKeeper in SolrCloud mode, and it already knows that ``-c`` is required below +Solr 10 and rejected from 10 onwards. + +The cluster-config is applied as container environment: ``SOLR_HEAP``, +``GC_TUNE`` and ``SOLR_OPTS``. +""" + +import logging +import subprocess +from typing import ClassVar + +from solrorbit.builder.solr_provisioner import SolrDockerLauncher +from solrorbit.tools.launchr import translate +from solrorbit.tools.launchr.errors import DeployError +from solrorbit.tools.launchr.models import ( + ProvisionedInfra, + ResolvedClusterConfig, + SolrDeployment, +) + +logger = logging.getLogger(__name__) + +#: Container name prefix. Distinct from Solr Orbit's own ``solr-orbit`` +#: container so a launchr run and a hand-run ``--pipeline=docker`` do not fight +#: over the same name. +CONTAINER_PREFIX = "launchr-solr" + + +def resolve_image(version: str, *, registry: str, nightly_registry: str) -> str: + """Return the full image reference for a Solr version. + + Mirrors ``SolrDockerLauncher``'s rule so the image recorded in the report is + the image that was actually pulled. + """ + repository = nightly_registry if "SNAPSHOT" in version.upper() else registry + return f"{repository}:{version}" + + +class DockerInstall: + """Starts one Solr container per cell, and removes it afterwards.""" + + API_VERSION: ClassVar[int] = 1 + name: ClassVar[str] = "docker" + + def __init__(self, *, registry: str = "solr", nightly_registry: str = "apache/solr-nightly"): + self.registry = registry + self.nightly_registry = nightly_registry + + def deploy(self, infra: ProvisionedInfra, version: str, image: str | None, + cluster_config: ResolvedClusterConfig) -> SolrDeployment: + endpoints = infra.endpoints + if len(endpoints) != 1: + raise DeployError( + f"The 'docker' install method runs a single container, but the profile " + f"provisioned {len(endpoints)} Solr endpoints." + ) + endpoint = endpoints[0] + resolved_image = image or resolve_image( + version, registry=self.registry, nightly_registry=self.nightly_registry + ) + container = f"{CONTAINER_PREFIX}-{endpoint.port}" + + # SolrDockerLauncher reads heap/GC/opts off an object with a `.variables` + # dict; it needs nothing else from a real ClusterConfigInstance. + variables = translate.to_orbit_variables(cluster_config.env) + launcher = SolrDockerLauncher( + port=endpoint.port, + startup_timeout=infra.profile.startup_timeout, + container_name=container, + cluster_config=_Variables(variables), + image_repository=self.registry, + nightly_image_repository=self.nightly_registry, + ) + + logger.info("Starting Solr %s as container %s from %s", version, container, resolved_image) + try: + launcher.start(version_tag=version, mode="cloud") + except Exception as exc: + # Keep the container around only long enough to grab its logs; the + # orchestrator calls collect_logs() then undeploy() on failure. + raise DeployError(f"Could not start Solr {version} in Docker: {exc}") from exc Review Comment: When `launcher.start()` raises (a `_wait_for_ready()` timeout, say), `deploy()` raises before it returns a `SolrDeployment`. In `_run_cell` `deployment` is then still `None`, so neither `collect_logs()` nor `undeploy()` runs, and the comment is not what happens: - the cell stores no Solr log, in exactly the case where one is wanted; - the `launchr-solr-<port>` container keeps running after the last cell (the next cell's `docker rm -f` in `SolrDockerLauncher.start` hides it only when another cell follows on that port). So read the log and remove the container here, and hand the log over on the error: ```suggestion except Exception as exc: # No SolrDeployment exists yet, so the orchestrator can neither collect # logs nor undeploy: read the log and remove the container here. logs = self._container_logs(container) subprocess.run(["docker", "rm", "-f", container], capture_output=True, text=True) raise DeployError(f"Could not start Solr {version} in Docker: {exc}", logs=logs) from exc ``` ########## solrorbit/tools/launchr/install/docker.py: ########## @@ -0,0 +1,143 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Run Solr in a Docker container on the local machine. + +This wraps :class:`solrorbit.builder.solr_provisioner.SolrDockerLauncher`, the +same launcher Solr Orbit's own ``docker`` pipeline uses, rather than growing a +second container-management path. It is single-container with embedded +ZooKeeper in SolrCloud mode, and it already knows that ``-c`` is required below +Solr 10 and rejected from 10 onwards. + +The cluster-config is applied as container environment: ``SOLR_HEAP``, +``GC_TUNE`` and ``SOLR_OPTS``. +""" + +import logging +import subprocess +from typing import ClassVar + +from solrorbit.builder.solr_provisioner import SolrDockerLauncher +from solrorbit.tools.launchr import translate +from solrorbit.tools.launchr.errors import DeployError +from solrorbit.tools.launchr.models import ( + ProvisionedInfra, + ResolvedClusterConfig, + SolrDeployment, +) + +logger = logging.getLogger(__name__) + +#: Container name prefix. Distinct from Solr Orbit's own ``solr-orbit`` +#: container so a launchr run and a hand-run ``--pipeline=docker`` do not fight +#: over the same name. +CONTAINER_PREFIX = "launchr-solr" + + +def resolve_image(version: str, *, registry: str, nightly_registry: str) -> str: + """Return the full image reference for a Solr version. + + Mirrors ``SolrDockerLauncher``'s rule so the image recorded in the report is + the image that was actually pulled. + """ + repository = nightly_registry if "SNAPSHOT" in version.upper() else registry + return f"{repository}:{version}" + + +class DockerInstall: + """Starts one Solr container per cell, and removes it afterwards.""" + + API_VERSION: ClassVar[int] = 1 + name: ClassVar[str] = "docker" + + def __init__(self, *, registry: str = "solr", nightly_registry: str = "apache/solr-nightly"): + self.registry = registry + self.nightly_registry = nightly_registry + + def deploy(self, infra: ProvisionedInfra, version: str, image: str | None, + cluster_config: ResolvedClusterConfig) -> SolrDeployment: + endpoints = infra.endpoints + if len(endpoints) != 1: + raise DeployError( + f"The 'docker' install method runs a single container, but the profile " + f"provisioned {len(endpoints)} Solr endpoints." + ) + endpoint = endpoints[0] + resolved_image = image or resolve_image( + version, registry=self.registry, nightly_registry=self.nightly_registry + ) + container = f"{CONTAINER_PREFIX}-{endpoint.port}" + + # SolrDockerLauncher reads heap/GC/opts off an object with a `.variables` + # dict; it needs nothing else from a real ClusterConfigInstance. + variables = translate.to_orbit_variables(cluster_config.env) + launcher = SolrDockerLauncher( + port=endpoint.port, + startup_timeout=infra.profile.startup_timeout, + container_name=container, + cluster_config=_Variables(variables), + image_repository=self.registry, + nightly_image_repository=self.nightly_registry, + ) + + logger.info("Starting Solr %s as container %s from %s", version, container, resolved_image) + try: + launcher.start(version_tag=version, mode="cloud") + except Exception as exc: + # Keep the container around only long enough to grab its logs; the + # orchestrator calls collect_logs() then undeploy() on failure. + raise DeployError(f"Could not start Solr {version} in Docker: {exc}") from exc + + return SolrDeployment( + infra=infra, + version=version, + image=resolved_image, + endpoints=[endpoint], + cluster_config=cluster_config, + handles={"launcher": launcher, "container": container}, + metadata={"container_pid": launcher.pid}, + ) + + def undeploy(self, deployment: SolrDeployment) -> None: + launcher = deployment.handles.get("launcher") + container = deployment.handles.get("container") + if launcher is not None: + launcher.stop() + elif container: + subprocess.run(["docker", "rm", "-f", container], capture_output=True, text=True) + # The container ran without a data volume, so its index died with it and + # the next cell starts from empty. + + def collect_logs(self, deployment: SolrDeployment) -> dict[str, bytes]: + container = deployment.handles.get("container") + if not container: + return {} + try: Review Comment: The log reading moves into a helper so `deploy()` can use it: ```suggestion container = deployment.handles.get("container") return self._container_logs(container) if container else {} @staticmethod def _container_logs(container: str) -> dict[str, bytes]: try: ``` ########## solrorbit/tools/launchr/orchestrator.py: ########## @@ -0,0 +1,406 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""The lifecycle loop. + +For each provisioning group: provision once, then for each cell deploy Solr, +health-check it, resolve where its version sits on the timeline, run Solr Orbit, +enrich the report, store it, undeploy. Tear the group down when its cells are +done. + +Two invariants this module exists to guarantee: + +**Failure isolation.** A cell that fails produces a ``failed`` report and does +not stop the next cell. A group whose provisioning fails fails only its own +cells. + +**Nothing leaks.** ``undeploy`` and ``teardown`` run in ``finally`` blocks and +never raise, so a crash cannot leave a container running or an instance +billing. +""" + +import logging +import shutil +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from solrorbit.tools.launchr import enrich as enrich_module +from solrorbit.tools.launchr import health, orbit, retry, summary, translate +from solrorbit.tools.launchr.config import LaunchrConfig +from solrorbit.tools.launchr.errors import BenchmarkError, LaunchrError Review Comment: ```suggestion from solrorbit.tools.launchr.errors import BenchmarkError, DeployError, LaunchrError ``` ########## solrorbit/tools/launchr/config.py: ########## @@ -0,0 +1,483 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Load, merge and validate launchr's configuration. + +Precedence is ``CLI flag > environment variable > launchr.yaml > built-in +default``, resolved once here and never re-read. Every value remembers where it +came from, which is the only practical way to debug a precedence problem in CI +-- see ``launchr validate-config``. +""" + +import logging +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Mapping + +import yaml + +from solrorbit.tools.launchr.errors import ConfigError Review Comment: ```suggestion from solrorbit.tools.launchr import translate from solrorbit.tools.launchr.errors import ConfigError ``` ########## solrorbit/tools/launchr/orchestrator.py: ########## @@ -0,0 +1,406 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""The lifecycle loop. + +For each provisioning group: provision once, then for each cell deploy Solr, +health-check it, resolve where its version sits on the timeline, run Solr Orbit, +enrich the report, store it, undeploy. Tear the group down when its cells are +done. + +Two invariants this module exists to guarantee: + +**Failure isolation.** A cell that fails produces a ``failed`` report and does +not stop the next cell. A group whose provisioning fails fails only its own +cells. + +**Nothing leaks.** ``undeploy`` and ``teardown`` run in ``finally`` blocks and +never raise, so a crash cannot leave a container running or an instance +billing. +""" + +import logging +import shutil +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from solrorbit.tools.launchr import enrich as enrich_module +from solrorbit.tools.launchr import health, orbit, retry, summary, translate +from solrorbit.tools.launchr.config import LaunchrConfig +from solrorbit.tools.launchr.errors import BenchmarkError, LaunchrError +from solrorbit.tools.launchr.install import get_install_method +from solrorbit.tools.launchr.matrix import ProfileGroup, plan +from solrorbit.tools.launchr.models import ( + MatrixCell, + OrbitInvocation, + ProvisionedInfra, + RunOutcome, + SolrDeployment, + new_run_id, + new_session_id, +) +from solrorbit.tools.launchr.provisioning import get_strategy +from solrorbit.tools.launchr.storage import get_backend +from solrorbit.tools.launchr.versions import VersionCache, VersionTimeline + +logger = logging.getLogger(__name__) + + +class Session: + """One ``launchr run``.""" + + def __init__(self, config: LaunchrConfig, *, session_id: str | None = None, + version_timestamp: datetime | None = None, branch: str | None = None): + self.config = config + self.session_id = session_id or new_session_id() + self.started_at = datetime.now(timezone.utc) + self.storage = get_backend(config.storage_backend, config.storage_path) + self.timeline = VersionTimeline( + VersionCache(config.version_cache_path), offline=config.offline + ) + self.override_timestamp = version_timestamp + self.override_branch = branch + self.outcomes: list[RunOutcome] = [] + self._orbit_version, self._orbit_revision = _orbit_identity() + + # -- the loop --------------------------------------------------------- + + def run(self) -> int: + """Execute the whole matrix. Returns a process exit code.""" + groups = plan(self.config) + total = sum(len(g.cells) for g in groups) + logger.info("Session %s: %d cell(s) in %d group(s); storage %s", + self.session_id, total, len(groups), self.storage.location) + + interrupted = False + index = 0 + try: + for group in groups: + index = self._run_group(group, start_index=index, total=total) + except KeyboardInterrupt: + interrupted = True + logger.warning("Session %s interrupted; infrastructure has been released", + self.session_id) + + finished_at = datetime.now(timezone.utc) + self._write_summary(groups, finished_at) + + if interrupted: + return 130 + + failed = [o for o in self.outcomes if not o.completed] + if failed: + logger.error("Session %s finished with %d/%d cell(s) failed", + self.session_id, len(failed), len(self.outcomes)) + for outcome in failed: + logger.error(" failed: %s -- %s", outcome.cell.describe(), outcome.error) + return 1 + logger.info("Session %s finished: %d/%d cell(s) completed", + self.session_id, len(self.outcomes), len(self.outcomes)) + return 0 + + def _run_group(self, group: ProfileGroup, *, start_index: int, total: int) -> int: + strategy = get_strategy(group.profile.provisioning) + index = start_index + infra: ProvisionedInfra | None = None + + try: + infra = retry.with_retries( + lambda: strategy.provision(group.profile, nodes=group.nodes), + what=f"provisioning {group.describe()}", + ) + except Exception as exc: # noqa: BLE001 - a group's cells all fail together + logger.error("Could not provision %s: %s", group.describe(), exc) + for cell in group.cells: + index += 1 + self._record_unprovisioned(cell, exc) + return index + + try: + for cell in group.cells: + index += 1 + logger.info("cell %d/%d %s", index, total, cell.describe()) + self.outcomes.append(self._run_cell(cell, infra, strategy)) + finally: + try: + strategy.teardown(infra) + except Exception as exc: # noqa: BLE001 - teardown must never raise + logger.error("Teardown of %s failed: %s", group.describe(), exc) + return index + + # -- one cell --------------------------------------------------------- + + def _run_cell(self, cell: MatrixCell, infra: ProvisionedInfra, strategy) -> RunOutcome: + outcome = RunOutcome( + cell=cell, + run_id=new_run_id(), + session_id=self.session_id, + status=RunOutcome.FAILED, + started_at=datetime.now(timezone.utc), + ) + cluster_config = translate.resolve(self.config.cluster_config(cell.cluster_config)) + install = get_install_method( + infra.install_method, + registry=self.config.registry, + nightly_registry=self.config.nightly_registry, + ) + + workdir = Path(tempfile.mkdtemp(prefix=f"launchr-{outcome.run_id}-")) + deployment: SolrDeployment | None = None + report: dict | None = None + argv: list[str] | None = None + system_info = None + version_info = None + logs: dict[str, bytes] = {} + + try: + deployment = retry.with_retries( + lambda: install.deploy(infra, cell.solr_version, None, cluster_config), + what=f"deploying Solr {cell.solr_version}", + ) + system_info = health.probe( + deployment.endpoints[0], timeout=float(infra.profile.startup_timeout) + ) + deployment.system_info = system_info + self._check_version(cell, system_info, managed=infra.install_method != "none") + + version_info = self.timeline.resolve( + cell.solr_version, system_info, + override_timestamp=self.override_timestamp, + override_branch=self.override_branch, + ) + + workload_params = self._workload_params(cell, cluster_config) + test_run_id = _new_test_run_id() + outcome.test_run_id = test_run_id + argv = orbit.build_argv( + cell, deployment, + test_run_id=test_run_id, + session_id=self.session_id, + run_id=outcome.run_id, + workload_params=workload_params, + workload_repository=str(self.config.benchmark.get("workload_repository", "default")), + test_mode=bool(self.config.benchmark.get("test_mode", False)), + user_tags=dict(self.config.benchmark.get("user_tags", {}) or {}), + ) + result = strategy.run_on_driver(infra, OrbitInvocation( + argv=argv, test_run_id=test_run_id, log_path=workdir / "solr-orbit.log", + )) + if not result.ok: + raise BenchmarkError( + f"Solr Orbit exited {result.returncode} for {cell.describe()}. " + f"Its output is in the stored logs/ directory." + ) + report = _parse_report(result.report) + outcome.status = RunOutcome.COMPLETED + + except KeyboardInterrupt: + # Let an interrupt unwind: the finally block below undeploys Solr and + # _run_group's releases the infrastructure on the way out. + outcome.error = "interrupted" + outcome.finished_at = datetime.now(timezone.utc) + shutil.rmtree(workdir, ignore_errors=True) + raise + except Exception as exc: # noqa: BLE001 - one cell's failure is data, not a crash + outcome.error = f"{type(exc).__name__}: {exc}" + if isinstance(exc, LaunchrError): + logger.error("Cell %s failed: %s", cell.describe(), exc) + else: + logger.exception("Cell %s failed unexpectedly", cell.describe()) + if deployment is not None: + try: + logs = install.collect_logs(deployment) + except Exception as log_exc: # noqa: BLE001 + logger.warning("Could not collect Solr logs: %s", log_exc) + + finally: + if deployment is not None: + try: + install.undeploy(deployment) + except Exception as exc: # noqa: BLE001 - undeploy must never raise + logger.error("Undeploying Solr for %s failed: %s", cell.describe(), exc) + + outcome.finished_at = datetime.now(timezone.utc) + try: + self._store(cell, outcome, report, cluster_config, deployment, system_info, + version_info, argv, workdir, logs) + except Exception as exc: # noqa: BLE001 + logger.error("Could not store the report for %s: %s", cell.describe(), exc) + outcome.status = RunOutcome.FAILED + outcome.error = outcome.error or f"storage failed: {exc}" + finally: + shutil.rmtree(workdir, ignore_errors=True) + return outcome + + # -- storage ---------------------------------------------------------- + + def _cell_prefix(self, cell: MatrixCell, outcome: RunOutcome) -> str: + date = self.started_at.strftime("%Y-%m-%d") + return f"runs/{cell.workload}/{date}/{outcome.run_id}" + + def _store(self, cell: MatrixCell, outcome: RunOutcome, report: dict | None, + cluster_config, deployment, system_info, version_info, + argv: list[str] | None, workdir: Path, logs: dict[str, bytes]) -> None: + prefix = self._cell_prefix(cell, outcome) + block = enrich_module.build_block( + cell=cell, + outcome=outcome, + cluster_config=cluster_config, + workload_params=self._workload_params(cell, cluster_config), + workload_repository=str(self.config.benchmark.get("workload_repository", "default")), + benchmark=self.config.benchmark, + argv=argv, + deployment=deployment, + system_info=system_info, + version_info=version_info, + infra_metadata=deployment.infra.metadata if deployment else None, + workload_revision=enrich_module.workload_revision_of(report), + user_labels=dict(self.config.benchmark.get("user_tags", {}) or {}), + orbit_version=self._orbit_version, + orbit_revision=self._orbit_revision, + ) + enriched = enrich_module.enrich(report, block, cell=cell, test_run_id=outcome.test_run_id) + self.storage.write(f"{prefix}/results.json", _json_bytes(enriched)) + outcome.result_path = f"{prefix}/results.json" + + if outcome.completed: + return + + # Logs are stored for failures only: the thing you need, exactly when + # you need it, without unbounded growth on the happy path. + orbit_log = workdir / "solr-orbit.log" + if orbit_log.exists(): + self.storage.write(f"{prefix}/logs/solr-orbit.log", orbit_log.read_bytes()) + for name, data in logs.items(): + self.storage.write(f"{prefix}/logs/{name}", data) + + def _write_summary(self, groups: list[ProfileGroup], finished_at: datetime) -> None: + document = summary.build( + self.session_id, self.started_at, finished_at, groups, self.outcomes, + launchr_version=self._orbit_version, + ) + path = summary.summary_path(self.session_id, self.started_at) + try: + retry.with_retries( + lambda: self.storage.write(path, summary.render(document)), + what="writing the session summary", + backoff=retry.STORAGE_BACKOFF, + ) + logger.info("Session summary: %s", path) + except Exception as exc: # noqa: BLE001 + logger.error("Could not write the session summary: %s", exc) + + # -- helpers ---------------------------------------------------------- + + def _workload_params(self, cell: MatrixCell, cluster_config) -> dict[str, Any]: + """Workload params for a cell: the cluster-config's, then the workload's own.""" + params: dict[str, Any] = dict(cluster_config.workload_params) + for workload in self.config.workloads: + if workload.name == cell.workload: + params.update(workload.workload_params) + break + return params + + def _check_version(self, cell: MatrixCell, system_info, *, managed: bool) -> None: + """Warn, or fail, when the cluster is not running the version we asked for. + + For a cluster launchr deployed this is a real error -- the report would + claim a version that did not run. For an externally managed one it is a + configuration mistake worth failing on too, because the data point would + be mislabelled either way. + """ + reported = (system_info.solr_version or "").strip() + wanted = cell.solr_version.strip() + if not reported or reported == wanted: + return + # A SNAPSHOT image reports its base version, e.g. 10.0.0-SNAPSHOT -> 10.0.0. + if wanted.upper().startswith(reported.upper()): Review Comment: `"9.11.0".startswith("9.1")` is true, so a cluster reporting `9.1` is accepted for a `9.11.0` cell and the report carries the wrong version label. Only the SNAPSHOT case needs the exception: ```suggestion if wanted.upper() == f"{reported}-SNAPSHOT".upper(): ``` ########## solrorbit/tools/launchr/config.py: ########## @@ -0,0 +1,483 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Load, merge and validate launchr's configuration. + +Precedence is ``CLI flag > environment variable > launchr.yaml > built-in +default``, resolved once here and never re-read. Every value remembers where it +came from, which is the only practical way to debug a precedence problem in CI +-- see ``launchr validate-config``. +""" + +import logging +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Mapping + +import yaml + +from solrorbit.tools.launchr.errors import ConfigError +from solrorbit.tools.launchr.models import ( + DEFAULT_SOLR_PORT, + ClusterConfigSpec, + ProfileSpec, + WorkloadSpec, +) + +logger = logging.getLogger(__name__) + +DEFAULT_CONFIG_FILE = "launchr.yaml" + +def default_storage_path(env: Mapping[str, str] | None = None) -> str: + """Where runs land when nothing says otherwise. + + Under Solr Orbit's home, beside the caches and installs it already keeps + there, rather than in whatever directory launchr was invoked from. A result + store is the history of a machine's benchmarking, not an artifact of one + shell session: defaulting it to the working directory means running from + somewhere else silently starts an empty history, and running from a + repository checkout litters it. + + ``BENCHMARK_HOME`` moves it, the same variable Solr Orbit already honours + for everything else it stores. + """ + environ = os.environ if env is None else env + home = environ.get("BENCHMARK_HOME") or os.path.expanduser("~") + return os.path.join(home, ".solr-orbit", "benchmark-results") + + +#: Built-in defaults. Chosen so that ``launchr run --versions X --workloads Y`` +#: works on a machine with nothing but Docker: local provisioning, a container +#: per cell, results under Solr Orbit's home. +BUILT_IN: dict[str, Any] = { + "solr_versions": [], + "registry": "docker.io/library/solr", + "nightly_registry": "docker.io/apache/solr-nightly", + # Empty means "decide when the config is loaded", so BENCHMARK_HOME is read + # at the moment it matters rather than frozen at import. + "storage": {"backend": "local", "path": ""}, + "provisioning_profiles": { + "local-docker": { + "provisioning": "local", + "install_method": "docker", + "nodes": 1, + "port": DEFAULT_SOLR_PORT, + } + }, + "cluster_configs": { + "default-1s1r": {"heap": "4g", "gc": "G1GC", "shards": 1, "replicas": 1} + }, + "defaults": {"provisioning": ["local-docker"], "cluster_config": ["default-1s1r"]}, + "workloads": [], + "benchmark": { + "iterations": 1, + "aggregate": False, + "test_mode": False, + "workload_repository": "default", + "user_tags": {}, + }, + "versions": {"cache": "~/.launchr/version-cache.json", "offline": False}, +} + +#: Setting path -> environment variable. Only settings with a CLI flag get one, +#: so the three ways of saying the same thing stay in step. +ENV_VARS: dict[str, str] = { + "solr_versions": "SOLR_VERSIONS", + "workloads": "LAUNCHR_WORKLOADS", + "defaults.provisioning": "LAUNCHR_PROVISIONING", + "defaults.cluster_config": "LAUNCHR_CLUSTER_CONFIG", + "storage.backend": "LAUNCHR_STORAGE_BACKEND", + "storage.path": "LAUNCHR_STORAGE_PATH", + "registry": "LAUNCHR_REGISTRY", +} + +#: Keys whose value is a comma-separated list on the command line and in the +#: environment, but a real list in YAML. +LIST_SETTINGS = {"solr_versions", "workloads", "defaults.provisioning", "defaults.cluster_config"} + +#: Config keys that must never hold a secret. launchr reads credentials from the +#: ambient chain (environment, cloud profile, instance role) and nowhere else. +_CREDENTIAL_HINTS = ("password", "secret", "token", "access_key", "secret_key", "credential", "api_key") + + +@dataclass +class Provenance: + """Where each resolved setting came from.""" + + sources: dict[str, str] = field(default_factory=dict) + + def record(self, key: str, source: str) -> None: + self.sources[key] = source + + def of(self, key: str) -> str: + return self.sources.get(key, "built-in default") + + +@dataclass +class LaunchrConfig: + """A fully resolved configuration.""" + + data: dict[str, Any] + provenance: Provenance + config_path: Path | None = None + + # -- scalars ---------------------------------------------------------- + + @property + def registry(self) -> str: + return str(self.data["registry"]) + + @property + def nightly_registry(self) -> str: + return str(self.data["nightly_registry"]) + + @property + def solr_versions(self) -> list[str]: + return [str(v) for v in self.data["solr_versions"]] + + @property + def storage_backend(self) -> str: + return str(self.data["storage"]["backend"]) + + @property + def storage_path(self) -> str: + return str(self.data["storage"]["path"]) + + @property + def benchmark(self) -> Mapping[str, Any]: + return self.data["benchmark"] + + @property + def version_cache_path(self) -> str | None: + return self.data.get("versions", {}).get("cache") + + @property + def offline(self) -> bool: + return bool(self.data.get("versions", {}).get("offline", False)) + + # -- structured sections --------------------------------------------- + + def profile(self, name: str) -> ProfileSpec: + """Return one provisioning profile, validated.""" + profiles = self.data["provisioning_profiles"] + if name not in profiles: + known = ", ".join(sorted(profiles)) or "(none defined)" + raise ConfigError(f"Unknown provisioning profile {name!r}. Defined: {known}") + raw = dict(profiles[name] or {}) + provisioning = raw.pop("provisioning", None) + install_method = raw.pop("install_method", None) + if not provisioning: + raise ConfigError(f"Provisioning profile {name!r} does not set 'provisioning'") + if not install_method: + raise ConfigError(f"Provisioning profile {name!r} does not set 'install_method'") + options = dict(raw.pop("options", {}) or {}) + spec = ProfileSpec( + name=name, + provisioning=str(provisioning), + install_method=str(install_method), + nodes=int(raw.pop("nodes", 1)), + port=int(raw.pop("port", DEFAULT_SOLR_PORT)), + solr_url=raw.pop("solr_url", None), + startup_timeout=int(raw.pop("startup_timeout", 300)), + options=options, + ) + if raw: + # Unknown keys are nearly always typos, and a silently ignored + # instance_type is exactly the kind of thing that wastes a night. + raise ConfigError( + f"Provisioning profile {name!r} has unknown keys: {', '.join(sorted(raw))}. " + f"Put plugin-specific settings under 'options:'." + ) + return spec + + @property + def profile_names(self) -> list[str]: + return sorted(self.data["provisioning_profiles"]) + + def cluster_config(self, name: str) -> ClusterConfigSpec: + """Return one cluster-config, validated.""" + configs = self.data["cluster_configs"] + if name not in configs: + known = ", ".join(sorted(configs)) or "(none defined)" + raise ConfigError(f"Unknown cluster config {name!r}. Defined: {known}") + raw = dict(configs[name] or {}) + spec = ClusterConfigSpec( + name=name, + heap=raw.pop("heap", None), + gc=raw.pop("gc", None), + solr_opts=raw.pop("solr_opts", None), + shards=raw.pop("shards", None), + replicas=raw.pop("replicas", None), + ) + if raw: + raise ConfigError( + f"Cluster config {name!r} has unknown keys: {', '.join(sorted(raw))}" + ) + return spec + + @property + def cluster_config_names(self) -> list[str]: + return sorted(self.data["cluster_configs"]) + + @property + def workloads(self) -> list[WorkloadSpec]: + """The workloads to run, as specs.""" + specs: list[WorkloadSpec] = [] + for entry in self.data["workloads"]: + if isinstance(entry, str): + specs.append(WorkloadSpec(name=entry)) + continue + raw = dict(entry or {}) + name = raw.pop("name", None) + if not name: + raise ConfigError(f"Workload entry {entry!r} has no 'name'") + procedures = raw.pop("test_procedures", None) + if procedures in (None, []): + tests: tuple[str | None, ...] = (None,) + elif isinstance(procedures, str): + tests = (procedures,) + else: + tests = tuple(str(p) for p in procedures) + specs.append(WorkloadSpec( + name=str(name), + test_procedures=tests, + workload_params=dict(raw.pop("workload_params", {}) or {}), + )) + if raw: + raise ConfigError( + f"Workload {name!r} has unknown keys: {', '.join(sorted(raw))}" + ) + return specs + + def default_profiles(self) -> list[str]: + return [str(p) for p in self.data["defaults"]["provisioning"]] + + def default_cluster_configs(self) -> list[str]: + return [str(c) for c in self.data["defaults"]["cluster_config"]] + + # -- reporting -------------------------------------------------------- + + def describe(self) -> str: + """Render the resolved configuration with the provenance of each value.""" + lines = [f"config file: {self.config_path or '(none; built-in defaults)'}", ""] + for key in sorted(_flatten(self.data)): + value = _get(self.data, key) + lines.append(f"{key} = {value!r}".ljust(60) + f" <- {self.provenance.of(key)}") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Loading +# --------------------------------------------------------------------------- + + +def load(config_path: str | os.PathLike | None = None, *, env: Mapping[str, str] | None = None, + overrides: Mapping[str, Any] | None = None) -> LaunchrConfig: + """Resolve the configuration from defaults, a YAML file, the environment and CLI overrides. + + :param config_path: explicit config file. When omitted, ``./launchr.yaml`` + is used if it exists; otherwise built-in defaults apply. + :param env: environment mapping, for testing. Defaults to ``os.environ``. + :param overrides: CLI overrides, keyed by dotted setting path. ``None`` + values are ignored, so an unset flag does not clobber the file. + """ + environ = os.environ if env is None else env + data = _deep_copy(BUILT_IN) + provenance = Provenance() + + resolved_path = _locate(config_path) + if resolved_path is not None: + file_data = _read_yaml(resolved_path) + _merge(data, file_data, provenance, origin=str(resolved_path)) + + for key, var in ENV_VARS.items(): + raw = environ.get(var) + if raw is None or raw == "": + continue + value = _coerce(key, raw) + _set(data, key, value) + provenance.record(key, f"${var}") + + for key, value in (overrides or {}).items(): + if value is None: + continue + _set(data, key, value) + provenance.record(key, "CLI flag") + + if not str(data["storage"].get("path") or "").strip(): + data["storage"]["path"] = default_storage_path(environ) + provenance.record("storage.path", "default") + + config = LaunchrConfig(data=data, provenance=provenance, config_path=resolved_path) + validate(config) + return config + + +def apply_solr_port(config: LaunchrConfig, port: int) -> None: + """Override the Solr port on every provisioning profile. + + A convenience for the common case of one local profile. Assuming 8983 is a + bad default for a Solr developer, who very likely has their own Solr there, + and the provisioning pipelines refuse to start when the port is busy. + """ + for name, profile in config.data["provisioning_profiles"].items(): + if profile is None: + continue + profile["port"] = int(port) + config.provenance.record(f"provisioning_profiles.{name}.port", "CLI flag") + + +def _locate(config_path: str | os.PathLike | None) -> Path | None: + if config_path: + path = Path(os.path.expanduser(str(config_path))) + if not path.exists(): + raise ConfigError(f"Config file not found: {path}") + return path + default = Path(DEFAULT_CONFIG_FILE) + return default if default.exists() else None + + +def _read_yaml(path: Path) -> dict: + try: + loaded = yaml.safe_load(path.read_text(encoding="utf-8")) + except yaml.YAMLError as exc: + raise ConfigError(f"Could not parse {path}: {exc}") from exc + if loaded is None: + return {} + if not isinstance(loaded, dict): + raise ConfigError(f"{path} must contain a YAML mapping at the top level") + unknown = set(loaded) - set(BUILT_IN) + if unknown: + raise ConfigError( + f"{path} has unknown top-level keys: {', '.join(sorted(unknown))}. " + f"Known: {', '.join(sorted(BUILT_IN))}" + ) + return loaded + + +def validate(config: LaunchrConfig) -> None: + """Check the configuration for problems worth catching before anything runs.""" + _reject_credentials(config.data) + + # Resolving each section is the validation: the properties raise on bad input. + for name in config.profile_names: + config.profile(name) + for name in config.cluster_config_names: + config.cluster_config(name) Review Comment: `translate.resolve()` is what rejects an unknown `gc` or a bad shard count, and today it runs in `_run_cell` before the `try`: `gc: nonexistent` raises `ConfigError` out of `Session.run()` after the infrastructure is provisioned, with no failed-cell report and no later cells. Resolving here makes `validate-config` and `--dry-run` catch it, which is what `test_bad_gc_is_caught_at_load_time` says it checks (it calls `translate.resolve` after a `load` that does not raise, so it would pass either way; it should assert that the `load` raises). ```suggestion # Resolving is what rejects an unknown gc or a bad shard count; left to # run time it would abort the session from inside a cell. translate.resolve(config.cluster_config(name)) ``` ########## solrorbit/tools/launchr/orchestrator.py: ########## @@ -0,0 +1,406 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""The lifecycle loop. + +For each provisioning group: provision once, then for each cell deploy Solr, +health-check it, resolve where its version sits on the timeline, run Solr Orbit, +enrich the report, store it, undeploy. Tear the group down when its cells are +done. + +Two invariants this module exists to guarantee: + +**Failure isolation.** A cell that fails produces a ``failed`` report and does +not stop the next cell. A group whose provisioning fails fails only its own +cells. + +**Nothing leaks.** ``undeploy`` and ``teardown`` run in ``finally`` blocks and +never raise, so a crash cannot leave a container running or an instance +billing. +""" + +import logging +import shutil +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from solrorbit.tools.launchr import enrich as enrich_module +from solrorbit.tools.launchr import health, orbit, retry, summary, translate +from solrorbit.tools.launchr.config import LaunchrConfig +from solrorbit.tools.launchr.errors import BenchmarkError, LaunchrError +from solrorbit.tools.launchr.install import get_install_method +from solrorbit.tools.launchr.matrix import ProfileGroup, plan +from solrorbit.tools.launchr.models import ( + MatrixCell, + OrbitInvocation, + ProvisionedInfra, + RunOutcome, + SolrDeployment, + new_run_id, + new_session_id, +) +from solrorbit.tools.launchr.provisioning import get_strategy +from solrorbit.tools.launchr.storage import get_backend +from solrorbit.tools.launchr.versions import VersionCache, VersionTimeline + +logger = logging.getLogger(__name__) + + +class Session: + """One ``launchr run``.""" + + def __init__(self, config: LaunchrConfig, *, session_id: str | None = None, + version_timestamp: datetime | None = None, branch: str | None = None): + self.config = config + self.session_id = session_id or new_session_id() + self.started_at = datetime.now(timezone.utc) + self.storage = get_backend(config.storage_backend, config.storage_path) + self.timeline = VersionTimeline( + VersionCache(config.version_cache_path), offline=config.offline + ) + self.override_timestamp = version_timestamp + self.override_branch = branch + self.outcomes: list[RunOutcome] = [] + self._orbit_version, self._orbit_revision = _orbit_identity() + + # -- the loop --------------------------------------------------------- + + def run(self) -> int: + """Execute the whole matrix. Returns a process exit code.""" + groups = plan(self.config) + total = sum(len(g.cells) for g in groups) + logger.info("Session %s: %d cell(s) in %d group(s); storage %s", + self.session_id, total, len(groups), self.storage.location) + + interrupted = False + index = 0 + try: + for group in groups: + index = self._run_group(group, start_index=index, total=total) + except KeyboardInterrupt: + interrupted = True + logger.warning("Session %s interrupted; infrastructure has been released", + self.session_id) + + finished_at = datetime.now(timezone.utc) + self._write_summary(groups, finished_at) + + if interrupted: + return 130 + + failed = [o for o in self.outcomes if not o.completed] + if failed: + logger.error("Session %s finished with %d/%d cell(s) failed", + self.session_id, len(failed), len(self.outcomes)) + for outcome in failed: + logger.error(" failed: %s -- %s", outcome.cell.describe(), outcome.error) + return 1 + logger.info("Session %s finished: %d/%d cell(s) completed", + self.session_id, len(self.outcomes), len(self.outcomes)) + return 0 + + def _run_group(self, group: ProfileGroup, *, start_index: int, total: int) -> int: + strategy = get_strategy(group.profile.provisioning) + index = start_index + infra: ProvisionedInfra | None = None + + try: + infra = retry.with_retries( + lambda: strategy.provision(group.profile, nodes=group.nodes), + what=f"provisioning {group.describe()}", + ) + except Exception as exc: # noqa: BLE001 - a group's cells all fail together + logger.error("Could not provision %s: %s", group.describe(), exc) + for cell in group.cells: + index += 1 + self._record_unprovisioned(cell, exc) + return index + + try: + for cell in group.cells: + index += 1 + logger.info("cell %d/%d %s", index, total, cell.describe()) + self.outcomes.append(self._run_cell(cell, infra, strategy)) + finally: + try: + strategy.teardown(infra) + except Exception as exc: # noqa: BLE001 - teardown must never raise + logger.error("Teardown of %s failed: %s", group.describe(), exc) + return index + + # -- one cell --------------------------------------------------------- + + def _run_cell(self, cell: MatrixCell, infra: ProvisionedInfra, strategy) -> RunOutcome: + outcome = RunOutcome( + cell=cell, + run_id=new_run_id(), + session_id=self.session_id, + status=RunOutcome.FAILED, + started_at=datetime.now(timezone.utc), + ) + cluster_config = translate.resolve(self.config.cluster_config(cell.cluster_config)) + install = get_install_method( + infra.install_method, + registry=self.config.registry, + nightly_registry=self.config.nightly_registry, + ) + + workdir = Path(tempfile.mkdtemp(prefix=f"launchr-{outcome.run_id}-")) + deployment: SolrDeployment | None = None + report: dict | None = None + argv: list[str] | None = None + system_info = None + version_info = None + logs: dict[str, bytes] = {} + + try: + deployment = retry.with_retries( + lambda: install.deploy(infra, cell.solr_version, None, cluster_config), + what=f"deploying Solr {cell.solr_version}", + ) + system_info = health.probe( + deployment.endpoints[0], timeout=float(infra.profile.startup_timeout) + ) + deployment.system_info = system_info + self._check_version(cell, system_info, managed=infra.install_method != "none") + + version_info = self.timeline.resolve( + cell.solr_version, system_info, + override_timestamp=self.override_timestamp, + override_branch=self.override_branch, + ) + + workload_params = self._workload_params(cell, cluster_config) + test_run_id = _new_test_run_id() + outcome.test_run_id = test_run_id + argv = orbit.build_argv( + cell, deployment, + test_run_id=test_run_id, + session_id=self.session_id, + run_id=outcome.run_id, + workload_params=workload_params, + workload_repository=str(self.config.benchmark.get("workload_repository", "default")), + test_mode=bool(self.config.benchmark.get("test_mode", False)), + user_tags=dict(self.config.benchmark.get("user_tags", {}) or {}), + ) + result = strategy.run_on_driver(infra, OrbitInvocation( + argv=argv, test_run_id=test_run_id, log_path=workdir / "solr-orbit.log", + )) + if not result.ok: + raise BenchmarkError( + f"Solr Orbit exited {result.returncode} for {cell.describe()}. " + f"Its output is in the stored logs/ directory." + ) + report = _parse_report(result.report) + outcome.status = RunOutcome.COMPLETED + + except KeyboardInterrupt: + # Let an interrupt unwind: the finally block below undeploys Solr and + # _run_group's releases the infrastructure on the way out. + outcome.error = "interrupted" + outcome.finished_at = datetime.now(timezone.utc) + shutil.rmtree(workdir, ignore_errors=True) + raise + except Exception as exc: # noqa: BLE001 - one cell's failure is data, not a crash + outcome.error = f"{type(exc).__name__}: {exc}" Review Comment: Store the logs a failed deploy carried (there is no `deployment` to collect from): ```suggestion outcome.error = f"{type(exc).__name__}: {exc}" if deployment is None and isinstance(exc, DeployError): logs = exc.logs ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
