Copilot commented on code in PR #108: URL: https://github.com/apache/solr-orbit/pull/108#discussion_r4289307489
########## 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: If `launcher.start()` raises after `docker run` has created the container (for example, when `_wait_for_ready()` times out), `deploy()` raises before returning a `SolrDeployment`. The orchestrator therefore has nothing to pass to `undeploy()` and leaves the `launchr-solr-*` container running, so the next cell can hit the same port and the claimed cleanup guarantee is broken. Stop the local launcher in this exception path (while preserving the original startup error). ########## solrorbit/tools/launchr/orbit.py: ########## @@ -0,0 +1,431 @@ +# 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. +"""Build and run the Solr Orbit command line. + +Every launchr cell, on every kind of infrastructure, runs the same shape of +command:: + + python -m solrorbit.benchmark run --pipeline=benchmark-only --target-hosts=... + +Solr Orbit is invoked through ``sys.executable -m`` rather than a ``solr-orbit`` +binary on ``PATH``, so launchr always drives the interpreter and the Solr Orbit +version it was itself installed with. + +Two things are deliberately absent from the argv: + +``--cluster-config`` + Solr Orbit does not apply a cluster-config under ``benchmark-only`` -- it + only records the names in the report. launchr applies heap/GC/opts to the + Solr process itself and records what it applied. + +``--distribution-version`` + Solr Orbit derives it from the live cluster, which is more trustworthy than + anything launchr could assert. +""" + +import json +import logging +import os +import signal +import subprocess +import sys +import time +from pathlib import Path + +from solrorbit import paths +from solrorbit.tools.launchr.errors import BenchmarkError, ConfigError +from solrorbit.tools.launchr.models import ( + MatrixCell, + OrbitInvocation, + OrbitResult, + SolrDeployment, +) + +logger = logging.getLogger(__name__) + +ORBIT_MODULE = "solrorbit.benchmark" +PIPELINE = "benchmark-only" + +#: Report filenames Solr Orbit writes, by whether the run was aggregated. +TEST_RUN_REPORT = "test_run.json" +AGGREGATED_REPORT = "aggregated_test_run.json" + + +# --------------------------------------------------------------------------- +# Where Solr Orbit puts its reports +# --------------------------------------------------------------------------- + + +def orbit_root() -> Path: + """Return Solr Orbit's ``node.root.dir``. + + Read from ``~/.solr-orbit/benchmark.ini`` (honouring ``BENCHMARK_HOME``) so + launchr follows the same configuration the benchmark itself does. Falls back + to the documented default when no config file exists yet, which is the case + on a machine that has never run Solr Orbit. + """ + # Imported lazily: solrorbit.config pulls in a good deal of the package. + from solrorbit import config as orbit_config + + config_file = orbit_config.ConfigFile() + if config_file.present: + try: + parser = config_file.load() + root = parser.get("node", "root.dir", fallback=None) + if root: + return Path(os.path.expanduser(root)) + except Exception as exc: # noqa: BLE001 - a malformed ini is not fatal here + logger.warning("Could not read node.root.dir from %s: %s", config_file.location, exc) + return Path(paths.benchmark_confdir()) / "benchmarks" + + +def report_path(test_run_id: str, *, aggregated: bool = False, root: Path | None = None) -> Path: + """Return the path Solr Orbit writes a given run's report to.""" + base = root if root is not None else orbit_root() + if aggregated: + return base / "aggregated_results" / test_run_id / AGGREGATED_REPORT + return base / "test-runs" / test_run_id / TEST_RUN_REPORT + + +# --------------------------------------------------------------------------- +# Building the argv +# --------------------------------------------------------------------------- + + +def _validate_tag(key: str, value: str) -> None: + """Reject user tags Solr Orbit's parser cannot round-trip. + + ``extract_user_tags_from_string`` splits on ``,`` and then on ``:``, so a + key or value containing either is silently mangled or raises deep inside + Solr Orbit. Failing here names the offending tag. + """ + for field, text in (("key", key), ("value", value)): + if "," in text or ":" in text: + raise ConfigError( + f"User tag {field} {text!r} contains ',' or ':', which Solr Orbit's " + f"--user-tag parser uses as separators. Choose a different {field}." + ) + + +def render_user_tags(tags: dict[str, object]) -> str: + """Render a tag dict as Solr Orbit's ``--user-tag`` format.""" + parts = [] + for key, value in tags.items(): + text = "" if value is None else str(value) + _validate_tag(str(key), text) + parts.append(f"{key}:{text}") + return ",".join(parts) + + +def build_argv(cell: MatrixCell, deployment: SolrDeployment, *, test_run_id: str, + session_id: str, run_id: str, workload_params: dict, + workload_repository: str = "default", test_mode: bool = False, + user_tags: dict | None = None, + include_tasks: str | None = None, exclude_tasks: str | None = None) -> list[str]: + """Build the full Solr Orbit command line for one cell.""" + tags = { + "launchr-session": session_id, + "launchr-run": run_id, + "launchr-cluster-config": cell.cluster_config, + "launchr-profile": cell.profile, + } + tags.update(user_tags or {}) + + argv = [ + sys.executable, "-m", ORBIT_MODULE, "run", + f"--pipeline={PIPELINE}", + f"--target-hosts={deployment.target_hosts}", + f"--client-options={deployment.client_options}", + f"--workload={cell.workload}", + f"--workload-repository={workload_repository}", + f"--test-run-id={test_run_id}", + f"--user-tag={render_user_tags(tags)}", + "--test-iterations=1", + "--aggregate=false", Review Comment: This hard-codes `--aggregate=false` regardless of `benchmark.aggregate`, while `enrich.build_block()` records the configured value as `launchr.benchmark.aggregated`. A config with `aggregate: true` is therefore either run differently from its report or produces an aggregate report that this code never locates (`OrbitResult` and `report_path` are fixed to the non-aggregated path). Reject unsupported aggregation during config validation or thread the setting through the invocation and report lookup. ########## 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)) Review Comment: Cluster-config translation happens before `_run_cell` enters its exception handler. An invalid GC name or shard/replica value is therefore raised out of the cell loop after infrastructure has been provisioned, without a failed-cell report and without allowing later cells/groups to run. Validate these resolved configs during config loading (or move this setup into the guarded failure path) so a bad cell cannot abort the session this way. ########## solrorbit/tools/launchr/storage/local.py: ########## @@ -0,0 +1,89 @@ +# 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. +"""Local filesystem storage. + +The development backend, and the reference implementation of the layout every +other backend reproduces. Writes are atomic within a directory: a temp file +followed by ``os.replace``, so a crashed run never leaves a half-written report +for reportr to trip over. +""" + +import logging +import os +from pathlib import Path +from typing import ClassVar + +from solrorbit.tools.launchr.errors import StorageError + +logger = logging.getLogger(__name__) + + +class LocalStorageBackend: + """Stores reports under a local directory.""" + + API_VERSION: ClassVar[int] = 1 + name: ClassVar[str] = "local" + + def __init__(self, path: str | os.PathLike): + self.root = Path(os.path.expanduser(str(path))).resolve() + + def __repr__(self) -> str: + return f"LocalStorageBackend({str(self.root)!r})" + + @property + def location(self) -> str: + """A human-readable description of where this backend writes.""" + return str(self.root) + + def _resolve(self, path: str) -> Path: + """Map a relative storage path to a filesystem path, refusing escapes.""" + if not path or path.startswith("/"): + raise StorageError(f"Storage paths must be relative, got {path!r}") + target = (self.root / path).resolve() + if target != self.root and self.root not in target.parents: + raise StorageError(f"Storage path {path!r} escapes the storage root {self.root}") + return target + + def write(self, path: str, data: bytes, *, overwrite: bool = False) -> None: + target = self._resolve(path) + if target.exists() and not overwrite: + raise StorageError( + f"Refusing to overwrite {path!r}: stored reports are immutable. " + f"A re-run gets a new run id." + ) + target.parent.mkdir(parents=True, exist_ok=True) + tmp = target.with_name(target.name + ".tmp") + tmp.write_bytes(data) + os.replace(tmp, target) + logger.debug("Wrote %d bytes to %s", len(data), target) + + def read(self, path: str) -> bytes: + target = self._resolve(path) + try: + return target.read_bytes() + except OSError as exc: + raise StorageError(f"Could not read {path!r} from {self.root}: {exc}") from exc + + def list(self, prefix: str) -> list[str]: + base = self.root / prefix if prefix else self.root Review Comment: `list()` does not use the backend's path-safety check, so a caller can pass an absolute or `..` prefix and make it walk outside the storage root; in many cases this then raises an unhandled `ValueError` while converting paths. Resolve and validate the prefix through `_resolve()` before recursing, just as `write()`, `read()`, and `exists()` do. ########## 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) + config.workloads Review Comment: `validate()` does not actually validate the translated cluster config: `LaunchrConfig.cluster_config()` only constructs a spec, so an unknown `gc` is accepted by `validate-config` and `run` later raises from `_run_cell` before its per-cell `try` block. That aborts the whole session without a failed-cell report, contrary to the failure-isolation contract. Resolve each cluster config during validation (or move this setup into the isolated cell path). ########## solrorbit/tools/launchr/cli.py: ########## @@ -0,0 +1,322 @@ +# 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 ``launchr`` command line. + +Four subcommands: + +``run`` + Walk the matrix. ``--dry-run`` prints the plan and the exact Solr Orbit + command lines without touching anything -- it is offline and instant, which + is what makes it usable as a review tool. +``validate-config`` + Print the resolved configuration with the provenance of every value. +``list`` + Inventory of profiles, cluster configs and plugins. +``version`` + What launchr this is. +""" + +import argparse +import logging +import signal +import sys +from datetime import datetime + +from solrorbit.tools.launchr import __version__, matrix, orbit, translate +from solrorbit.tools.launchr import config as config_module +from solrorbit.tools.launchr import install as install_module +from solrorbit.tools.launchr import provisioning as provisioning_module +from solrorbit.tools.launchr import storage as storage_module +from solrorbit.tools.launchr.errors import LaunchrError +from solrorbit.tools.launchr.models import ( + NodeHandle, + ProvisionedInfra, + SolrDeployment, + SolrEndpoint, + new_run_id, + new_session_id, +) + +PROGRAM = "launchr" + +logger = logging.getLogger(__name__) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog=PROGRAM, + description="Orchestrate Apache Solr Orbit benchmarks over a matrix of versions, " + "workloads, hardware profiles and cluster configs.", + ) + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") + parser.add_argument("-v", "--verbose", action="store_true", help="Log at DEBUG level.") + parser.add_argument("-q", "--quiet", action="store_true", help="Log warnings and errors only.") + + sub = parser.add_subparsers(dest="subcommand", metavar="<command>") + + run = sub.add_parser("run", help="Run the benchmark matrix") + _add_config_args(run) + run.add_argument("--versions", help="Comma-separated Solr versions, e.g. 9.0.0,9.7.0,9.11.0.") + run.add_argument("--workloads", help="Comma-separated workload names.") + run.add_argument("--provisioning", help="Comma-separated provisioning profile names.") + run.add_argument("--cluster-config", help="Comma-separated cluster config names.") + run.add_argument("--storage", dest="storage_backend", + help=f"Storage backend ({', '.join(storage_module.available())}).") + run.add_argument("--storage-path", help="Storage root: a directory for the local backend.") + run.add_argument("--registry", help="Docker repository for released versions.") + run.add_argument("--solr-port", type=int, metavar="PORT", + help="Port for Solr to listen on, overriding every provisioning profile. " + "Useful when the default is taken by a Solr of your own.") + run.add_argument("--test-mode", action="store_true", default=None, + help="Truncate every corpus to 1000 documents. Minutes instead of hours; " + "for smoke tests, never for publishable results.") + run.add_argument("--branch", help="Override the inferred git branch for SNAPSHOT versions.") + run.add_argument("--version-timestamp", metavar="ISO8601", + help="Override the x-axis timestamp for every version in this run.") + run.add_argument("--dry-run", action="store_true", + help="Print the plan and the exact solr-orbit command lines; run nothing.") + + validate = sub.add_parser("validate-config", help="Print the resolved configuration") + _add_config_args(validate) + + listing = sub.add_parser("list", help="List configured profiles, cluster configs or plugins") + listing.add_argument("what", nargs="?", default="all", + choices=["all", "profiles", "cluster-configs", "plugins"]) + _add_config_args(listing) + + sub.add_parser("version", help="Print the launchr version") + return parser + + +def _add_config_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("-c", "--config", metavar="FILE", + help=f"Config file (default: ./{config_module.DEFAULT_CONFIG_FILE} if present).") + + +def _overrides(args: argparse.Namespace) -> dict: + """Map CLI arguments onto dotted config paths. + + ``None`` means "not given", and is dropped by the config loader so an unset + flag never clobbers the file. + """ + def split(value): + return [p.strip() for p in value.split(",") if p.strip()] if value else None + + mapping = { + "solr_versions": split(getattr(args, "versions", None)), + "workloads": split(getattr(args, "workloads", None)), + "defaults.provisioning": split(getattr(args, "provisioning", None)), + "defaults.cluster_config": split(getattr(args, "cluster_config", None)), + "storage.backend": getattr(args, "storage_backend", None), + "storage.path": getattr(args, "storage_path", None), + "registry": getattr(args, "registry", None), + "benchmark.test_mode": getattr(args, "test_mode", None), + } + return {k: v for k, v in mapping.items() if v is not None} + + +def _configure_logging(args: argparse.Namespace) -> None: + level = logging.INFO + if getattr(args, "verbose", False): + level = logging.DEBUG + elif getattr(args, "quiet", False): + level = logging.WARNING + logging.basicConfig(level=level, format="%(asctime)s %(levelname)-7s %(message)s", + datefmt="%H:%M:%S", stream=sys.stderr) + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def cmd_run(args: argparse.Namespace) -> int: + config = config_module.load(args.config, overrides=_overrides(args)) + if args.solr_port: + config_module.apply_solr_port(config, args.solr_port) + if args.dry_run: + return _dry_run(config) + + from solrorbit.tools.launchr import orchestrator + + timestamp = None + if args.version_timestamp: + try: + timestamp = datetime.fromisoformat(args.version_timestamp) + except ValueError: + raise LaunchrError( + f"--version-timestamp {args.version_timestamp!r} is not an ISO-8601 timestamp" + ) from None + return orchestrator.run(config, version_timestamp=timestamp, branch=args.branch) + + +def _dry_run(config: config_module.LaunchrConfig) -> int: + """Print the plan and the argv for each cell, touching nothing. + + Deliberately offline and instant: this is how a reviewer checks a matrix + expansion and a cluster-config translation without burning benchmark time. + Anything that needs the network or a workload checkout belongs in a separate + command. + """ + groups = matrix.plan(config) + print(matrix.describe_plan(groups)) + + session_id = new_session_id() + print("Solr Orbit invocations:\n") + for group in groups: + for cell in group.cells: + cluster_config = translate.resolve(config.cluster_config(cell.cluster_config)) + params = dict(cluster_config.workload_params) + for workload in config.workloads: + if workload.name == cell.workload: + params.update(workload.workload_params) + break + endpoint = (SolrEndpoint.from_url(group.profile.solr_url.split(",")[0]) + if group.profile.solr_url + else SolrEndpoint(port=group.profile.port)) + deployment = SolrDeployment( + infra=ProvisionedInfra( + profile=group.profile, strategy=group.profile.provisioning, + install_method=group.profile.install_method, + nodes=[NodeHandle(name="solr-1", endpoint=endpoint)], + ), + version=cell.solr_version, image=None, endpoints=[endpoint], + cluster_config=cluster_config, + ) Review Comment: For a preconfigured profile with comma-separated solr_url values, dry-run keeps only the first URL. Actual provisioning returns all configured endpoints and build_argv targets all of them, so the promised exact command shown by --dry-run is incomplete and can hide the real multi-node target set. Build the dry-run deployment from every URL, not just split(",")[0]. ########## 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()): + return Review Comment: This accepts any textual prefix as a SNAPSHOT match: a cluster reporting `9.1` is accepted for a requested `9.11.0` because `"9.11.0".startswith("9.1")`. The run can therefore benchmark the wrong Solr version under the requested label. Restrict this exception to an exact requested `*-SNAPSHOT` whose base version equals `reported`. ########## 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)) Review Comment: The main report write bypasses the retry policy even though storage writes are documented as retryable. A transient backend failure here is caught by the outer handler, marks an otherwise successful benchmark as failed, and leaves no `results.json` or `result_path` to inspect. Wrap this write in the configured storage retry policy (and apply the same treatment to failure-log writes). ########## solrorbit/tools/launchr/enrich.py: ########## @@ -0,0 +1,158 @@ +# 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. +"""Add launchr's metadata to a Solr Orbit report. + +Solr Orbit's ``test_run.json`` is copied **verbatim** and a single ``launchr`` +key is added. Nothing in Solr Orbit's own key space is touched, so +``TestRun.from_dict()`` still parses a stored report and ``solr-orbit compare`` +still consumes it. + +The ``launchr`` block carries what a chart needs and a benchmark cannot know +about itself: where the version sits on the timeline, what hardware ran it, and +what configuration was actually applied. +""" + +from datetime import datetime, timezone +from typing import Any, Mapping + +from solrorbit.tools.launchr.models import ( + MatrixCell, + ResolvedClusterConfig, + RunOutcome, + SolrDeployment, + SystemInfo, + VersionInfo, +) + +#: Bumped when the ``launchr`` block changes incompatibly. reportr refuses a +#: report whose schema_version it does not understand rather than mis-reading it. +SCHEMA_VERSION = 1 + + +def _iso(value: datetime | None) -> str | None: + if value is None: + return None + stamped = value if value.tzinfo else value.replace(tzinfo=timezone.utc) + return stamped.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def build_block(*, cell: MatrixCell, outcome: RunOutcome, cluster_config: ResolvedClusterConfig, + workload_params: Mapping[str, Any], workload_repository: str, + benchmark: Mapping[str, Any], argv: list[str] | None, + deployment: SolrDeployment | None = None, + system_info: SystemInfo | None = None, + version_info: VersionInfo | None = None, + infra_metadata: Mapping[str, Any] | None = None, + workload_revision: str | None = None, + user_labels: Mapping[str, Any] | None = None, + orbit_version: str | None = None, + orbit_revision: str | None = None) -> dict: + """Build the ``launchr`` block for one cell.""" + solr: dict[str, Any] = {"version": cell.solr_version} + if deployment is not None: + solr["image"] = deployment.image + solr["endpoints"] = [str(e) for e in deployment.endpoints] + if system_info is not None: + solr["impl_version"] = system_info.solr_impl_version or None + solr["commit"] = system_info.commit + solr["lucene_version"] = system_info.lucene_version + solr["mode"] = system_info.mode + solr["jvm"] = {"vendor": system_info.jvm_vendor, "version": system_info.jvm_version} + if version_info is not None: + solr["version_timestamp"] = _iso(version_info.timestamp) + solr["version_timestamp_source"] = version_info.source + solr["version_timestamp_precision"] = version_info.precision + solr["version_kind"] = version_info.kind + solr["branch"] = version_info.branch + solr["branch_source"] = version_info.branch_source + + infra: dict[str, Any] = { + "provisioning_profile": cell.profile, + "install_method": deployment.infra.install_method if deployment else None, + "provisioning": deployment.infra.strategy if deployment else None, + } + if infra_metadata: + infra.update(infra_metadata) + + block: dict[str, Any] = { + "schema_version": SCHEMA_VERSION, + "run_id": outcome.run_id, + "session_id": outcome.session_id, + "cell_id": cell.cell_id, + "status": outcome.status, + "started_at": _iso(outcome.started_at), + "finished_at": _iso(outcome.finished_at), + "solr": solr, + "infra": infra, + "cluster_config": { + "name": cluster_config.name, + "heap": cluster_config.spec.heap if cluster_config.spec else None, + "gc": cluster_config.spec.gc if cluster_config.spec else None, + "shards": cluster_config.spec.shards if cluster_config.spec else None, + "replicas": cluster_config.spec.replicas if cluster_config.spec else None, + "solr_opts": cluster_config.spec.solr_opts if cluster_config.spec else None, + "applied_env": dict(cluster_config.env), Review Comment: `install_method: none` cannot apply any of these environment settings, but this line still serializes the requested values as `cluster_config.applied_env`. Reports for preconfigured clusters therefore claim that heap/GC/opts were applied when they were not, contradicting the field's meaning and the metadata contract. Emit an empty applied environment for externally managed (or not-yet-deployed) cells while retaining the requested values in the other fields. -- 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]
