diff --git a/.buildkite/image_build/image_build.sh b/.buildkite/image_build/image_build.sh index 52f120534b..b81cd4ec0c 100755 --- a/.buildkite/image_build/image_build.sh +++ b/.buildkite/image_build/image_build.sh @@ -160,6 +160,42 @@ print_bake_config() { (cd "$(dirname "${BAKE_CONFIG_FILE}")" && buildkite-agent artifact upload "$(basename "${BAKE_CONFIG_FILE}")") } +record_buildkit_trace() ( + local metadata_file="$1" + local helper=".buildkite/scripts/ci-otel/ci_otel.py" + local trace_file="${BUILD_TMP_DIR}/buildkit-trace.json" + local build_ref record_id + + if [[ "${BUILDKITE:-}" != "true" || ! -f "${helper}" ]] || + ! command -v timeout >/dev/null 2>&1 || + ! command -v python3 >/dev/null 2>&1; then + return 0 + fi + build_ref="$(timeout 3s python3 "${helper}" build-ref "${metadata_file}" "${TARGET}" 2>/dev/null)" || { + echo "vLLM CI OTel: BuildKit metadata unavailable; trace skipped" >&2 + return 0 + } + record_id="${build_ref##*/}" + if [[ -z "${record_id}" ]]; then + return 0 + fi + + echo "--- :mag: Recording BuildKit trace" + if ! timeout 20s docker buildx history trace "${record_id}" >"${trace_file}"; then + echo "vLLM CI OTel: BuildKit history trace unavailable; upload skipped" >&2 + return 0 + fi + + export CI_INFRA_BUILDX_REF="${build_ref}" + export CI_INFRA_OTEL_SPOOL_DIR="${BUILD_TMP_DIR}/otel-spans" + timeout 5s python3 "${helper}" record-buildkit "${trace_file}" || { + echo "vLLM CI OTel: BuildKit trace conversion skipped" >&2 + return 0 + } + timeout 5s python3 "${helper}" flush || + echo "vLLM CI OTel: BuildKit trace upload skipped" >&2 +) + ################################# # Main Script # ################################# @@ -269,7 +305,18 @@ export PARENT_COMMIT print_bake_config echo "--- :docker: Building ${TARGET}" -docker --debug buildx bake -f "${VLLM_BAKE_FILE_PATH}" -f "${CI_HCL_PATH}" --progress plain "${TARGET}" +BUILD_TMP_DIR="$(mktemp -d)" +trap 'rm -rf -- "${BUILD_TMP_DIR}"' EXIT +BUILD_METADATA_FILE="${BUILD_TMP_DIR}/build-metadata.json" +BUILD_STATUS=0 +docker --debug buildx bake -f "${VLLM_BAKE_FILE_PATH}" -f "${CI_HCL_PATH}" \ + --progress plain --metadata-file "${BUILD_METADATA_FILE}" "${TARGET}" || BUILD_STATUS=$? + +record_buildkit_trace "${BUILD_METADATA_FILE}" || true + +if [[ "${BUILD_STATUS}" -ne 0 ]]; then + exit "${BUILD_STATUS}" +fi echo "--- :white_check_mark: Build complete" diff --git a/.buildkite/scripts/ci-otel/ci_otel.py b/.buildkite/scripts/ci-otel/ci_otel.py new file mode 100644 index 0000000000..0d25594bec --- /dev/null +++ b/.buildkite/scripts/ci-otel/ci_otel.py @@ -0,0 +1,650 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +"""Dependency-free OTLP exporter and pytest timing plugin for CI.""" + +from __future__ import annotations + +import hashlib +import json +import os +import secrets +import struct +import sys +import time +import urllib.request +from collections.abc import Iterable +from dataclasses import asdict, dataclass +from pathlib import Path + +ENDPOINT = os.getenv("CI_INFRA_OTEL_ENDPOINT", "https://ci.vllm.ai/api/otel/v1/traces") +AUDIENCE = os.getenv("CI_INFRA_OTEL_AUDIENCE", "https://ci.vllm.ai/api/otel") +MAX_BATCH_SIZE = 2_000 +TEST_SPOOL_BATCH_SIZE = 100 + + +def _parse_dockerfile_step(name: str) -> tuple[str, int, int, str] | None: + """Parse "[stage index/total] label" without regex.""" + if not name.startswith("["): + return None + close = name.find("]") + if close == -1: + return None + header = name[1:close] + label = name[close + 1 :].lstrip() + if not label: + return None + parts = header.rsplit(None, 1) + if len(parts) != 2: + return None + stage, step = parts + if "/" not in step: + return None + index_str, total_str = step.split("/", 1) + if not index_str.isdigit() or not total_str.isdigit(): + return None + return stage, int(index_str), int(total_str), label + + +@dataclass(frozen=True) +class Span: + trace_id: str + span_id: str + parent_span_id: str | None + name: str + start_ns: int + end_ns: int + attributes: dict[str, str | int | bool] + status_code: int = 1 + + +def _varint(value: int) -> bytes: + if value < 0: + value += 1 << 64 + encoded = bytearray() + while value > 0x7F: + encoded.append((value & 0x7F) | 0x80) + value >>= 7 + encoded.append(value) + return bytes(encoded) + + +def _field(number: int, wire_type: int) -> bytes: + return _varint((number << 3) | wire_type) + + +def _bytes_field(number: int, value: bytes) -> bytes: + return _field(number, 2) + _varint(len(value)) + value + + +def _string_field(number: int, value: str) -> bytes: + return _bytes_field(number, value.encode()) + + +def _varint_field(number: int, value: int) -> bytes: + return _field(number, 0) + _varint(value) + + +def _fixed64_field(number: int, value: int) -> bytes: + return _field(number, 1) + struct.pack(" bytes: + if isinstance(value, bool): + encoded_value = _varint_field(2, int(value)) + elif isinstance(value, int): + encoded_value = _varint_field(3, value) + else: + encoded_value = _string_field(1, value) + return _string_field(1, name) + _bytes_field(2, encoded_value) + + +def _resource_attributes() -> dict[str, str | int | bool]: + build_number = os.getenv("BUILDKITE_BUILD_NUMBER", "0") + attributes: dict[str, str | int | bool] = { + "service.name": "vllm-ci", + "buildkite.organization.slug": os.getenv("BUILDKITE_ORGANIZATION_SLUG", ""), + "buildkite.pipeline.slug": os.getenv("BUILDKITE_PIPELINE_SLUG", ""), + "buildkite.build.id": os.getenv("BUILDKITE_BUILD_ID", ""), + "buildkite.build.number": int(build_number) if build_number.isdigit() else 0, + "buildkite.build.branch": os.getenv("BUILDKITE_BRANCH", ""), + "buildkite.build.commit": os.getenv("BUILDKITE_COMMIT", ""), + "buildkite.job.id": os.getenv("BUILDKITE_JOB_ID", ""), + "buildkite.job.label": os.getenv("BUILDKITE_LABEL", ""), + "buildkite.agent.queue": os.getenv("BUILDKITE_AGENT_META_DATA_QUEUE", ""), + } + return {name: value for name, value in attributes.items() if value != ""} + + +def _encode_span(span: Span) -> bytes: + encoded = b"".join( + ( + _bytes_field(1, bytes.fromhex(span.trace_id)), + _bytes_field(2, bytes.fromhex(span.span_id)), + _bytes_field(4, bytes.fromhex(span.parent_span_id)) + if span.parent_span_id + else b"", + _string_field(5, span.name), + _varint_field(6, 1), + _fixed64_field(7, span.start_ns), + _fixed64_field(8, span.end_ns), + ) + ) + attributes = dict(span.attributes) + if os.getenv("BUILDKITE_BUILD_URL") and os.getenv("BUILDKITE_JOB_ID"): + attributes["buildkite.job.web_url"] = ( + f"{os.environ['BUILDKITE_BUILD_URL']}#{os.environ['BUILDKITE_JOB_ID']}" + ) + for name, value in attributes.items(): + encoded += _bytes_field(9, _attribute(name, value)) + return encoded + _bytes_field(15, _varint_field(3, span.status_code)) + + +def encode_request(spans: Iterable[Span]) -> bytes: + resource = b"".join( + _bytes_field(1, _attribute(name, value)) + for name, value in _resource_attributes().items() + ) + scope_spans = _bytes_field(1, _string_field(1, "vllm.ci")) + for span in spans: + scope_spans += _bytes_field(2, _encode_span(span)) + return _bytes_field(1, _bytes_field(1, resource) + _bytes_field(2, scope_spans)) + + +def new_context() -> tuple[str, str, str | None]: + parts = os.getenv("TRACEPARENT", "").lower().split("-") + valid = ( + len(parts) == 4 + and [len(part) for part in parts] == [2, 32, 16, 2] + and all(all(char in "0123456789abcdef" for char in part) for part in parts) + and parts[1] != "0" * 32 + and parts[2] != "0" * 16 + ) + if valid: + return parts[1], secrets.token_hex(8), parts[2] + build_id = os.getenv("BUILDKITE_BUILD_ID", "local") + trace_id = hashlib.sha256(build_id.encode()).hexdigest()[:32] + return trace_id, secrets.token_hex(8), None + + +def _spool_dir() -> Path | None: + value = os.getenv("CI_INFRA_OTEL_SPOOL_DIR") + return Path(value) if value else None + + +def record_spans(spans: Iterable[Span]) -> bool: + try: + spool_dir = _spool_dir() + records = [json.dumps(asdict(span), separators=(",", ":")) for span in spans] + if spool_dir is None or not records: + return False + spool_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + with (spool_dir / f"spans-{os.getpid()}.jsonl").open( + "a", encoding="utf-8" + ) as output: + output.write("\n".join(records) + "\n") + return True + except Exception as error: + print(f"CI timing spool skipped: {error}", file=sys.stderr) + return False + + +def _read_spool() -> tuple[list[Span], list[Path]]: + spool_dir = _spool_dir() + if spool_dir is None or not spool_dir.is_dir(): + return [], [] + spans: list[Span] = [] + loaded_files: list[Path] = [] + for spool_file in sorted(spool_dir.glob("spans-*.jsonl")): + try: + spans.extend( + Span(**json.loads(record)) + for record in spool_file.read_text(encoding="utf-8").splitlines() + if record + ) + loaded_files.append(spool_file) + except Exception as error: + print( + f"CI timing spool ignored {spool_file.name}: {error}", file=sys.stderr + ) + return spans, loaded_files + + +def load_spans() -> list[Span]: + return _read_spool()[0] + + +def buildx_ref(metadata_path: str, target: str) -> str: + """Read the exact Buildx history reference emitted by bake.""" + metadata = json.loads(Path(metadata_path).read_text(encoding="utf-8")) + target_metadata = metadata.get(target, metadata) + reference = target_metadata.get("buildx.build.ref") + if not isinstance(reference, str) or not reference: + raise ValueError(f"Buildx metadata has no history reference for {target}") + return reference + + +def _buildkit_tags(span: dict) -> dict[str, str | int | bool]: + tags: dict[str, str | int | bool] = {} + for tag in span.get("tags", []): + key = tag.get("key") + value = tag.get("value") + if isinstance(key, str) and isinstance(value, (str, int, bool)): + tags[key] = value + return tags + + +def _buildkit_parent_ids(span: dict) -> list[str]: + return [ + reference["spanID"] + for reference in span.get("references", []) + if reference.get("refType") == "CHILD_OF" + and isinstance(reference.get("spanID"), str) + ] + + +def _buildkit_span_id(trace_id: str, value: str) -> str: + return hashlib.sha256(f"{trace_id}:{value}".encode()).hexdigest()[:16] + + +def buildkit_spans(payload: dict) -> list[Span]: + """Normalize Buildx's Jaeger JSON into dashboard-oriented OTLP spans.""" + trace_id, _, _ = new_context() + traces = payload.get("data", []) + native_trace_id = ( + traces[0].get("traceID", "buildkit") + if traces and isinstance(traces[0], dict) + else "buildkit" + ) + span_scope = ( + os.getenv("CI_INFRA_BUILDX_REF") + or os.getenv("BUILDKITE_JOB_ID") + or str(native_trace_id) + ) + steps: list[tuple[str, int, int, str, dict, int, int, int]] = [] + + for trace in traces: + raw_spans = [ + span + for span in trace.get("spans", []) + if isinstance(span, dict) and isinstance(span.get("spanID"), str) + ] + spans_by_id = {span["spanID"]: span for span in raw_spans} + children: dict[str, list[dict]] = {} + for span in raw_spans: + for parent_id in _buildkit_parent_ids(span): + if parent_id in spans_by_id: + children.setdefault(parent_id, []).append(span) + + def envelope( + span: dict, child_spans: dict[str, list[dict]] = children + ) -> tuple[int, int]: + start_us = int(span["startTime"]) + end_us = start_us + int(span["duration"]) + pending = list(child_spans.get(span["spanID"], [])) + seen: set[str] = set() + while pending: + child = pending.pop() + child_id = child["spanID"] + if child_id in seen: + continue + seen.add(child_id) + child_start = int(child["startTime"]) + start_us = min(start_us, child_start) + end_us = max(end_us, child_start + int(child["duration"])) + pending.extend(child_spans.get(child_id, [])) + return start_us * 1_000, end_us * 1_000 + + selected: dict[str, tuple[dict, int, int]] = {} + for span in raw_spans: + name = span.get("operationName") + if not isinstance(name, str) or name.startswith("cache request: "): + continue + step = _parse_dockerfile_step(name) + is_internal = name.startswith("[internal] ") + is_export = name.startswith(("exporting to ", "export to ")) + if not (step or is_internal or is_export): + continue + try: + start_ns, end_ns = envelope(span) + except (KeyError, TypeError, ValueError): + continue + vertex = _buildkit_tags(span).get("vertex") + key = str(vertex) if vertex else f"{name}:{span['spanID']}" + previous = selected.get(key) + if previous is None or end_ns - start_ns > previous[2] - previous[1]: + selected[key] = (span, start_ns, end_ns) + + for span, start_ns, end_ns in selected.values(): + name = span["operationName"] + step = _parse_dockerfile_step(name) + tags = _buildkit_tags(span) + failed = ( + tags.get("error") is True or tags.get("otel.status_code") == "ERROR" + ) + if step: + stage, step_index, step_total, label = step + else: + stage = "BuildKit internals" + label = name.removeprefix("[internal] ") + step_index = step_total = 0 + steps.append( + ( + stage, + step_index, + step_total, + label, + tags, + start_ns, + end_ns, + 2 if failed else 1, + ) + ) + + output: list[Span] = [] + build_ref = os.getenv("CI_INFRA_BUILDX_REF", "") + for stage in dict.fromkeys( + step[0] for step in steps if step[0] != "BuildKit internals" + ): + stage_steps = sorted( + (step for step in steps if step[0] == stage), key=lambda step: step[5] + ) + stage_id = _buildkit_span_id(trace_id, f"{span_scope}:stage:{stage}") + stage_start = min(step[5] for step in stage_steps) + stage_end = max(step[6] for step in stage_steps) + stage_failed = any(step[7] == 2 for step in stage_steps) + stage_attributes: dict[str, str | int | bool] = { + "ci.span.kind": "docker-stage", + "docker.stage": stage, + "docker.step.count": len(stage_steps), + } + if build_ref: + stage_attributes["docker.build.ref"] = build_ref + output.append( + Span( + trace_id=trace_id, + span_id=stage_id, + parent_span_id=None, + name="docker.stage", + start_ns=stage_start, + end_ns=stage_end, + attributes=stage_attributes, + status_code=2 if stage_failed else 1, + ) + ) + for ( + stage_name, + index, + total, + label, + tags, + start_ns, + end_ns, + status, + ) in stage_steps: + vertex = str(tags.get("vertex", "")) + attributes: dict[str, str | int | bool] = { + "ci.span.kind": "docker-instruction", + "docker.stage": stage_name, + "docker.step.label": label, + } + if index: + attributes["docker.step.index"] = index + attributes["docker.step.total"] = total + attributes["docker.instruction"] = label.split(None, 1)[0] + if vertex: + attributes["docker.vertex"] = vertex + output.append( + Span( + trace_id=trace_id, + span_id=_buildkit_span_id( + trace_id, + f"{span_scope}:step:{stage_name}:{index}:{vertex}:{label}", + ), + parent_span_id=stage_id, + name="docker.instruction", + start_ns=start_ns, + end_ns=end_ns, + attributes=attributes, + status_code=status, + ) + ) + for stage_name, index, total, label, tags, start_ns, end_ns, status in ( + step for step in steps if step[0] == "BuildKit internals" + ): + vertex = str(tags.get("vertex", "")) + attributes: dict[str, str | int | bool] = { + "ci.span.kind": "docker-internal", + "docker.stage": stage_name, + "docker.step.label": label, + } + if vertex: + attributes["docker.vertex"] = vertex + output.append( + Span( + trace_id=trace_id, + span_id=_buildkit_span_id( + trace_id, f"{span_scope}:internal:{vertex}:{label}" + ), + parent_span_id=None, + name="docker.internal", + start_ns=start_ns, + end_ns=end_ns, + attributes=attributes, + status_code=status, + ) + ) + return output + + +def record_buildkit_trace(trace_path: str) -> bool: + payload = json.loads(Path(trace_path).read_text(encoding="utf-8")) + return record_spans(buildkit_spans(payload)) + + +def _remaining_seconds(deadline: float) -> float: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("CI timing upload deadline exceeded") + return remaining + + +def _oidc_token(deadline: float) -> str: + access_token = os.getenv("BUILDKITE_AGENT_ACCESS_TOKEN", "") + job_id = os.getenv("BUILDKITE_JOB_ID", "") + if not access_token or not job_id: + raise RuntimeError("Buildkite job credentials are unavailable") + endpoint = os.getenv( + "BUILDKITE_AGENT_ENDPOINT", "https://agent.buildkite.com/v3" + ).rstrip("/") + request = urllib.request.Request( + f"{endpoint}/jobs/{job_id}/oidc/tokens", + data=json.dumps({"audience": AUDIENCE, "lifetime": 300}).encode(), + method="POST", + headers={ + "Authorization": f"Token {access_token}", + "Content-Type": "application/json", + "User-Agent": "vllm-ci-otel/1", + }, + ) + with urllib.request.urlopen( + request, timeout=_remaining_seconds(deadline) + ) as response: + token = json.loads(response.read()).get("token") + if not isinstance(token, str) or not token: + raise RuntimeError("Buildkite OIDC response did not contain a token") + return token + + +def export_spans(spans: list[Span], timeout_seconds: float = 3.0) -> bool: + if not spans or os.getenv("BUILDKITE") != "true": + return False + try: + deadline = time.monotonic() + max(timeout_seconds, 0.1) + token = _oidc_token(deadline) + for offset in range(0, len(spans), MAX_BATCH_SIZE): + request = urllib.request.Request( + ENDPOINT, + data=encode_request(spans[offset : offset + MAX_BATCH_SIZE]), + method="POST", + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/x-protobuf", + "User-Agent": "vllm-ci-otel/1", + }, + ) + with urllib.request.urlopen( + request, timeout=_remaining_seconds(deadline) + ) as response: + if response.status != 200: + raise RuntimeError(f"OTLP endpoint returned {response.status}") + return True + except Exception as error: + print(f"CI timing upload skipped: {error}", file=sys.stderr) + return False + + +def flush_spans(timeout_seconds: float = 3.0) -> bool: + spans, spool_files = _read_spool() + if not spool_files: + return True + if not spans or not export_spans(spans, timeout_seconds): + return False + for spool_file in spool_files: + try: + spool_file.unlink() + except FileNotFoundError: + pass + except Exception as error: + print(f"CI timing spool cleanup skipped: {error}", file=sys.stderr) + return True + + +_test_runs: dict[str, tuple[int, str]] = {} +_test_spans: list[Span] = [] + + +def _spool_test_spans() -> None: + if not _test_spans: + return + record_spans(_test_spans) + _test_spans.clear() + + +def _test_span(nodeid: str, start_ns: int, outcome: str) -> Span: + return Span( + trace_id=os.environ["CI_INFRA_TRACE_ID"], + span_id=secrets.token_hex(8), + parent_span_id=os.environ["CI_INFRA_COMMAND_SPAN_ID"], + name="pytest.test", + start_ns=start_ns, + end_ns=time.time_ns(), + attributes={ + "ci.span.kind": "test", + "test.nodeid": nodeid, + "test.file": nodeid.split("::", 1)[0], + "test.outcome": outcome, + }, + status_code=2 if outcome == "failed" else 1, + ) + + +def pytest_runtest_logstart(nodeid: str, location: tuple[str, int | None, str]): + try: + if os.getenv("CI_INFRA_TRACE_ID") and os.getenv("CI_INFRA_COMMAND_SPAN_ID"): + _test_runs[nodeid] = (time.time_ns(), "unknown") + except Exception: + pass + + +def pytest_runtest_logreport(report): + try: + if report.nodeid not in _test_runs: + return + start_ns, outcome = _test_runs[report.nodeid] + if report.failed: + outcome = "failed" + elif report.skipped and outcome != "failed": + outcome = "skipped" + elif report.when == "call" and outcome == "unknown": + outcome = "passed" + _test_runs[report.nodeid] = (start_ns, outcome) + except Exception: + pass + + +def pytest_runtest_logfinish(nodeid: str, location: tuple[str, int | None, str]): + try: + run = _test_runs.pop(nodeid, None) + if run: + _test_spans.append(_test_span(nodeid, *run)) + if len(_test_spans) >= TEST_SPOOL_BATCH_SIZE: + _spool_test_spans() + except Exception: + pass + + +def pytest_sessionfinish(session, exitstatus: int): + try: + for nodeid, run in list(_test_runs.items()): + _test_spans.append(_test_span(nodeid, *run)) + _test_runs.clear() + _spool_test_spans() + except Exception: + pass + + +def _record_command(arguments: list[str]) -> None: + trace_id, span_id, parent_span_id, start_ns, index, exit_code, label = arguments + exit_status = int(exit_code) + record_spans( + [ + Span( + trace_id=trace_id, + span_id=span_id, + parent_span_id=None if parent_span_id == "-" else parent_span_id, + name="ci.command", + start_ns=int(start_ns), + end_ns=time.time_ns(), + attributes={ + "ci.span.kind": "command", + "ci.command.index": int(index), + "ci.command.label": label, + "process.exit.code": exit_status, + }, + status_code=1 if exit_status == 0 else 2, + ) + ] + ) + + +def main(arguments: list[str] | None = None) -> int: + arguments = sys.argv[1:] if arguments is None else arguments + try: + if arguments == ["new-context"]: + trace_id, span_id, parent_span_id = new_context() + print(trace_id, span_id, parent_span_id or "-", time.time_ns()) + return 0 + if arguments == ["flush"]: + return 0 if flush_spans() else 1 + if len(arguments) == 3 and arguments[0] == "build-ref": + print(buildx_ref(arguments[1], arguments[2])) + return 0 + if len(arguments) == 2 and arguments[0] == "record-buildkit": + return 0 if record_buildkit_trace(arguments[1]) else 1 + if len(arguments) == 8 and arguments[0] == "record-command": + _record_command(arguments[1:]) + return 0 + print( + "usage: ci_otel.py {new-context|flush|build-ref ...|" + "record-buildkit ...|record-command ...}", + file=sys.stderr, + ) + return 2 + except Exception as error: + print(f"CI timing helper failed: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.buildkite/scripts/ci-otel/ci_otel.sh b/.buildkite/scripts/ci-otel/ci_otel.sh new file mode 100755 index 0000000000..feec16fa65 --- /dev/null +++ b/.buildkite/scripts/ci-otel/ci_otel.sh @@ -0,0 +1,133 @@ +#!/bin/sh + +[ "${_CI_INFRA_OTEL_INITIALIZED:-0}" = "1" ] && return 0 + +# No-op fallbacks keep every generated wrapper safe when setup is unavailable. +ci_otel_start() { :; } +ci_otel_finish() { :; } +ci_otel_run() { shift 2; env "$@"; return $?; } + +if [ -z "${CI_INFRA_OTEL_DIR:-}" ]; then + echo "vLLM CI OTel: helper directory is unset; tracing disabled" >&2 || : + return 0 +fi +export CI_INFRA_OTEL_DIR + +_CI_INFRA_OTEL_OWNS_RUNTIME=0 +if [ -z "${CI_INFRA_OTEL_RUNTIME_DIR:-}" ]; then + CI_INFRA_OTEL_RUNTIME_DIR="$(mktemp -d 2>/dev/null || :)" + [ -n "${CI_INFRA_OTEL_RUNTIME_DIR}" ] && _CI_INFRA_OTEL_OWNS_RUNTIME=1 +fi +CI_INFRA_OTEL_SPOOL_DIR="${CI_INFRA_OTEL_SPOOL_DIR:-${CI_INFRA_OTEL_RUNTIME_DIR}/spans}" +export CI_INFRA_OTEL_RUNTIME_DIR CI_INFRA_OTEL_SPOOL_DIR + +_CI_INFRA_OTEL_PYTHON="$(command -v python3 2>/dev/null || :)" +if [ -z "${CI_INFRA_OTEL_RUNTIME_DIR}" ] || + [ -z "${_CI_INFRA_OTEL_PYTHON}" ] || + [ ! -f "${CI_INFRA_OTEL_DIR}/ci_otel.py" ] || + ! mkdir -p "${CI_INFRA_OTEL_SPOOL_DIR}" || + ! PYTHONPATH="${CI_INFRA_OTEL_DIR}${PYTHONPATH:+:${PYTHONPATH}}" \ + "${_CI_INFRA_OTEL_PYTHON}" -c "import ci_otel" >/dev/null 2>&1; then + echo "vLLM CI OTel: tracing unavailable; test command will run normally" >&2 || : + if [ "${_CI_INFRA_OTEL_OWNS_RUNTIME}" = "1" ]; then + rm -rf -- "${CI_INFRA_OTEL_RUNTIME_DIR}" || : + fi + return 0 +fi + +_CI_INFRA_OTEL_SHIM_DIR="${CI_INFRA_OTEL_RUNTIME_DIR}/bin" +if mkdir -p "${_CI_INFRA_OTEL_SHIM_DIR}"; then + _CI_INFRA_OTEL_SHIM="${_CI_INFRA_OTEL_SHIM_DIR}/pytest" + if [ ! -e "${_CI_INFRA_OTEL_SHIM}" ] && [ ! -L "${_CI_INFRA_OTEL_SHIM}" ]; then + ln -s "${CI_INFRA_OTEL_DIR}/ci_pytest.sh" "${_CI_INFRA_OTEL_SHIM}" || : + fi + if [ -L "${_CI_INFRA_OTEL_SHIM}" ] && + [ "$(readlink "${_CI_INFRA_OTEL_SHIM}" 2>/dev/null || :)" = \ + "${CI_INFRA_OTEL_DIR}/ci_pytest.sh" ]; then + case ":${CI_INFRA_OTEL_SHIM_PATHS:-}:" in + *":${_CI_INFRA_OTEL_SHIM_DIR}:"*) ;; + *) + CI_INFRA_OTEL_SHIM_PATHS="${CI_INFRA_OTEL_SHIM_PATHS:+${CI_INFRA_OTEL_SHIM_PATHS}:}${_CI_INFRA_OTEL_SHIM_DIR}" + PATH="${_CI_INFRA_OTEL_SHIM_DIR}:${PATH}" + ;; + esac + export CI_INFRA_OTEL_SHIM_PATHS PATH + hash -r 2>/dev/null || : + fi +fi + +_ci_otel_python() { + if command -v timeout >/dev/null 2>&1; then + timeout 2s "${_CI_INFRA_OTEL_PYTHON}" "$@" + else + "${_CI_INFRA_OTEL_PYTHON}" "$@" + fi +} + +ci_otel_start() { + _CI_INFRA_OTEL_COMMAND_INDEX="$1" + _CI_INFRA_OTEL_COMMAND_LABEL="${2:-command ${_CI_INFRA_OTEL_COMMAND_INDEX}}" + _CI_INFRA_OTEL_CONTEXT="$(_ci_otel_python "${CI_INFRA_OTEL_DIR}/ci_otel.py" new-context)" || + return 0 + # The helper prints four whitespace-free fields. + # shellcheck disable=SC2086 + set -- ${_CI_INFRA_OTEL_CONTEXT} + [ "$#" -eq 4 ] || return 0 + + _CI_INFRA_OTEL_TRACE_ID="$1" + _CI_INFRA_OTEL_COMMAND_SPAN_ID="$2" + _CI_INFRA_OTEL_PARENT_SPAN_ID="$3" + _CI_INFRA_OTEL_START_NS="$4" + CI_INFRA_TRACE_ID="${_CI_INFRA_OTEL_TRACE_ID}" + CI_INFRA_COMMAND_SPAN_ID="${_CI_INFRA_OTEL_COMMAND_SPAN_ID}" + export CI_INFRA_TRACE_ID CI_INFRA_COMMAND_SPAN_ID + _CI_INFRA_OTEL_ACTIVE=1 +} + +ci_otel_finish() { + _CI_INFRA_OTEL_COMMAND_STATUS="${1:-0}" + [ "${_CI_INFRA_OTEL_ACTIVE:-0}" = "1" ] || return 0 + _CI_INFRA_OTEL_ACTIVE=0 + _ci_otel_python "${CI_INFRA_OTEL_DIR}/ci_otel.py" record-command \ + "${_CI_INFRA_OTEL_TRACE_ID}" "${_CI_INFRA_OTEL_COMMAND_SPAN_ID}" \ + "${_CI_INFRA_OTEL_PARENT_SPAN_ID}" "${_CI_INFRA_OTEL_START_NS}" \ + "${_CI_INFRA_OTEL_COMMAND_INDEX}" "${_CI_INFRA_OTEL_COMMAND_STATUS}" \ + "${_CI_INFRA_OTEL_COMMAND_LABEL}" || : + CI_INFRA_TRACE_ID="" + CI_INFRA_COMMAND_SPAN_ID="" + export CI_INFRA_TRACE_ID CI_INFRA_COMMAND_SPAN_ID +} + +# Run a simple command with tracing. Only for commands that do not modify +# shell state (export, cd, etc.) — those need the explicit start/finish pair. +# Assignment-prefixed arguments (VAR=value) are routed through env so the +# shell does not try to execute the assignment as a program. +ci_otel_run() { + _CI_INFRA_OTEL_RUN_INDEX="$1" + _CI_INFRA_OTEL_RUN_LABEL="$2" + shift 2 + ci_otel_start "${_CI_INFRA_OTEL_RUN_INDEX}" "${_CI_INFRA_OTEL_RUN_LABEL}" || : + env "$@" + _CI_INFRA_OTEL_RUN_STATUS=$? + ci_otel_finish "${_CI_INFRA_OTEL_RUN_STATUS}" || : + return "${_CI_INFRA_OTEL_RUN_STATUS}" +} + +_ci_otel_on_exit() { + _CI_INFRA_OTEL_EXIT_STATUS=$? + trap - 0 + ci_otel_finish "${_CI_INFRA_OTEL_EXIT_STATUS}" || : + if command -v timeout >/dev/null 2>&1; then + timeout 4s "${_CI_INFRA_OTEL_PYTHON}" "${CI_INFRA_OTEL_DIR}/ci_otel.py" flush || : + else + "${_CI_INFRA_OTEL_PYTHON}" "${CI_INFRA_OTEL_DIR}/ci_otel.py" flush || : + fi + if [ "${_CI_INFRA_OTEL_OWNS_RUNTIME}" = "1" ]; then + rm -rf -- "${CI_INFRA_OTEL_RUNTIME_DIR}" || : + fi + exit "${_CI_INFRA_OTEL_EXIT_STATUS}" +} + +trap _ci_otel_on_exit 0 +_CI_INFRA_OTEL_INITIALIZED=1 +: diff --git a/.buildkite/scripts/ci-otel/ci_pytest.sh b/.buildkite/scripts/ci-otel/ci_pytest.sh new file mode 100755 index 0000000000..9ae19190bb --- /dev/null +++ b/.buildkite/scripts/ci-otel/ci_pytest.sh @@ -0,0 +1,42 @@ +#!/bin/sh + +real_pytest="" +old_ifs="${IFS}" +IFS=: +for path_dir in ${PATH}; do + path_dir="${path_dir:-.}" + case ":${CI_INFRA_OTEL_SHIM_PATHS:-}:" in + *":${path_dir}:"*) continue ;; + esac + if [ -x "${path_dir}/pytest" ]; then + real_pytest="${path_dir}/pytest" + break + fi +done +IFS="${old_ifs}" + +if [ -z "${real_pytest}" ]; then + echo "vLLM CI OTel: pytest executable not found" >&2 || : + exit 127 +fi + +helper_dir="${CI_INFRA_OTEL_DIR:-}" +otel_pythonpath="${helper_dir}${PYTHONPATH:+:${PYTHONPATH}}" +otel_available() { + if command -v timeout >/dev/null 2>&1; then + timeout 2s env PYTHONPATH="${otel_pythonpath}" python3 -c "import ci_otel" \ + >/dev/null 2>&1 + else + PYTHONPATH="${otel_pythonpath}" python3 -c "import ci_otel" >/dev/null 2>&1 + fi +} + +if [ -n "${helper_dir}" ] && + [ -n "${CI_INFRA_TRACE_ID:-}" ] && + [ -n "${CI_INFRA_COMMAND_SPAN_ID:-}" ] && + otel_available; then + PYTHONPATH="${otel_pythonpath}" exec "${real_pytest}" -p ci_otel "$@" +fi + +echo "vLLM CI OTel: pytest tracing skipped; running pytest normally" >&2 || : +exec "${real_pytest}" "$@" diff --git a/.buildkite/scripts/ci-otel/tests/test_ci_otel.py b/.buildkite/scripts/ci-otel/tests/test_ci_otel.py new file mode 100644 index 0000000000..ff279de953 --- /dev/null +++ b/.buildkite/scripts/ci-otel/tests/test_ci_otel.py @@ -0,0 +1,788 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from __future__ import annotations + +import json +import os +import shlex +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +SCRIPTS_DIR = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(SCRIPTS_DIR)) + +import ci_otel # noqa: E402 +from ci_otel import Span # noqa: E402 + + +def _span(**attributes) -> Span: + return Span( + trace_id="01" * 16, + span_id="02" * 8, + parent_span_id=None, + name="ci.command", + start_ns=100, + end_ns=200, + attributes=attributes, + ) + + +def _quoted(value: str) -> str: + return shlex.quote(value) + + +def test_context_continues_traceparent(monkeypatch): + monkeypatch.setenv( + "TRACEPARENT", + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + ) + + trace_id, span_id, parent_span_id = ci_otel.new_context() + + assert trace_id == "4bf92f3577b34da6a3ce929d0e0e4736" + assert len(span_id) == 16 + assert parent_span_id == "00f067aa0ba902b7" + + +def test_otlp_payload_contains_dashboard_identity(monkeypatch): + monkeypatch.setenv("BUILDKITE_ORGANIZATION_SLUG", "vllm") + monkeypatch.setenv("BUILDKITE_PIPELINE_SLUG", "ci") + monkeypatch.setenv("BUILDKITE_BUILD_ID", "build-id") + monkeypatch.setenv("BUILDKITE_BUILD_NUMBER", "42") + monkeypatch.setenv("BUILDKITE_JOB_ID", "job-id") + + payload = ci_otel.encode_request([_span(**{"ci.span.kind": "command"})]) + + for value in (b"ci.command", b"ci.span.kind", b"buildkite.job.id", b"job-id"): + assert value in payload + + +def test_otlp_payload_accepts_negative_integer_attributes(): + payload = ci_otel.encode_request([_span(**{"process.exit.code": -1})]) + + assert b"process.exit.code" in payload + + +def test_spool_round_trip_is_local(monkeypatch, tmp_path): + monkeypatch.setenv("CI_INFRA_OTEL_SPOOL_DIR", str(tmp_path)) + monkeypatch.setattr( + ci_otel, + "_oidc_token", + lambda deadline: (_ for _ in ()).throw(AssertionError("unexpected upload")), + ) + span = _span(**{"ci.command.index": 1}) + + assert ci_otel.record_spans([span]) is True + assert ci_otel.load_spans() == [span] + + +def test_buildx_ref_supports_bake_metadata(tmp_path): + metadata = tmp_path / "metadata.json" + metadata.write_text( + json.dumps({"test-ci": {"buildx.build.ref": "builder/node/record-id"}}) + ) + + assert ci_otel.buildx_ref(str(metadata), "test-ci") == "builder/node/record-id" + + +def test_buildkit_trace_groups_stages_and_includes_descendant_time(monkeypatch): + monkeypatch.setenv("BUILDKITE_BUILD_ID", "build-id") + payload = { + "data": [ + { + "spans": [ + { + "spanID": "01" * 8, + "operationName": "[base 1/2] FROM alpine:3.21", + "startTime": 100, + "duration": 10, + "tags": [{"key": "vertex", "value": "sha256:from"}], + }, + { + "spanID": "02" * 8, + "operationName": "registry pull", + "references": [{"refType": "CHILD_OF", "spanID": "01" * 8}], + "startTime": 105, + "duration": 95, + }, + { + "spanID": "03" * 8, + "operationName": "[base 2/2] RUN uv pip install vllm", + "startTime": 210, + "duration": 40, + "tags": [{"key": "vertex", "value": "sha256:run"}], + }, + { + "spanID": "04" * 8, + "operationName": "[test 1/1] COPY --from=base /opt /opt", + "startTime": 260, + "duration": 20, + "tags": [{"key": "vertex", "value": "sha256:copy"}], + }, + { + "spanID": "05" * 8, + "operationName": "[internal] load build context", + "startTime": 90, + "duration": 30, + "tags": [{"key": "vertex", "value": "sha256:context"}], + }, + { + "spanID": "06" * 8, + "operationName": ( + "cache request: [base 2/2] RUN uv pip install vllm" + ), + "startTime": 205, + "duration": 2, + "tags": [{"key": "vertex", "value": "sha256:run"}], + }, + ] + } + ] + } + + spans = ci_otel.buildkit_spans(payload) + stages = { + span.attributes.get("docker.stage"): span + for span in spans + if span.name == "docker.stage" + } + instructions = [span for span in spans if span.name == "docker.instruction"] + + assert set(stages) == {"base", "test"} + assert stages["base"].start_ns == 100_000 + assert stages["base"].end_ns == 250_000 + assert len(instructions) == 3 + assert len([span for span in spans if span.name == "docker.internal"]) == 1 + from_span = next( + span for span in instructions if span.attributes["docker.instruction"] == "FROM" + ) + assert from_span.end_ns == 200_000 + assert from_span.parent_span_id == stages["base"].span_id + assert all(span.trace_id == stages["base"].trace_id for span in spans) + + +def test_record_buildkit_trace_spools_normalized_spans(monkeypatch, tmp_path): + trace_file = tmp_path / "trace.json" + trace_file.write_text( + json.dumps( + { + "data": [ + { + "spans": [ + { + "spanID": "01" * 8, + "operationName": "[base 1/1] RUN true", + "startTime": 100, + "duration": 20, + } + ] + } + ] + } + ) + ) + monkeypatch.setenv("CI_INFRA_OTEL_SPOOL_DIR", str(tmp_path / "spool")) + + assert ci_otel.record_buildkit_trace(str(trace_file)) is True + assert {span.attributes["ci.span.kind"] for span in ci_otel.load_spans()} == { + "docker-stage", + "docker-instruction", + } + + +def test_export_uses_buildkite_oidc_and_otlp(monkeypatch): + for name, value in { + "BUILDKITE": "true", + "BUILDKITE_AGENT_ACCESS_TOKEN": "agent-token", + "BUILDKITE_AGENT_ENDPOINT": "https://agent.example/v3", + "BUILDKITE_JOB_ID": "job-id", + }.items(): + monkeypatch.setenv(name, value) + requests = [] + + class Response: + status = 200 + + def __init__(self, body=b""): + self.body = body + + def read(self): + return self.body + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def open_request(request, timeout): + requests.append((request, timeout)) + if request.full_url.endswith("/oidc/tokens"): + return Response(b'{"token":"oidc-token"}') + return Response() + + monkeypatch.setattr(ci_otel.urllib.request, "urlopen", open_request) + + assert ci_otel.export_spans([_span()], timeout_seconds=0.5) is True + oidc_request, upload_request = (request for request, _ in requests) + assert oidc_request.full_url.endswith("/jobs/job-id/oidc/tokens") + assert oidc_request.get_header("Authorization") == "Token agent-token" + assert json.loads(oidc_request.data)["audience"] == ci_otel.AUDIENCE + assert upload_request.get_header("Authorization") == "Bearer oidc-token" + assert upload_request.get_header("Content-type") == "application/x-protobuf" + assert all(0 < timeout <= 0.5 for _, timeout in requests) + + +def test_export_failure_is_soft_and_bounded(monkeypatch): + monkeypatch.setenv("BUILDKITE", "true") + monkeypatch.setattr( + ci_otel, + "_oidc_token", + lambda deadline: (_ for _ in ()).throw(TimeoutError("unavailable")), + ) + + started = time.monotonic() + assert ci_otel.export_spans([_span()], timeout_seconds=0.1) is False + assert time.monotonic() - started < 0.5 + + +def test_export_batches_with_one_oidc_token(monkeypatch): + monkeypatch.setenv("BUILDKITE", "true") + monkeypatch.setattr(ci_otel, "MAX_BATCH_SIZE", 2) + token_calls = [] + requests = [] + + class Response: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def token(deadline): + token_calls.append(deadline) + return "token" + + def open_request(request, timeout): + requests.append(request) + return Response() + + monkeypatch.setattr(ci_otel, "_oidc_token", token) + monkeypatch.setattr(ci_otel.urllib.request, "urlopen", open_request) + + assert ci_otel.export_spans([_span() for _ in range(5)]) is True + assert len(token_calls) == 1 + assert len(requests) == 3 + + +def test_successful_flush_removes_spool_and_does_not_duplicate(monkeypatch, tmp_path): + monkeypatch.setenv("CI_INFRA_OTEL_SPOOL_DIR", str(tmp_path)) + uploaded = [] + assert ci_otel.record_spans([_span()]) is True + monkeypatch.setattr( + ci_otel, "export_spans", lambda spans, timeout: uploaded.extend(spans) or True + ) + + assert ci_otel.flush_spans() is True + assert len(uploaded) == 1 + assert list(tmp_path.glob("spans-*.jsonl")) == [] + assert ci_otel.flush_spans() is True + assert len(uploaded) == 1 + + +def test_failed_flush_preserves_spool(monkeypatch, tmp_path): + monkeypatch.setenv("CI_INFRA_OTEL_SPOOL_DIR", str(tmp_path)) + assert ci_otel.record_spans([_span()]) is True + monkeypatch.setattr(ci_otel, "export_spans", lambda spans, timeout: False) + + assert ci_otel.flush_spans() is False + assert len(list(tmp_path.glob("spans-*.jsonl"))) == 1 + + +def test_cli_rejects_bad_arguments_without_traceback(capsys): + assert ci_otel.main(["bogus"]) == 2 + assert ci_otel.main(["record-command", "too", "short"]) == 2 + assert ( + ci_otel.main(["record-command", "t", "s", "-", "not-an-int", "1", "0", "label"]) + == 1 + ) + + stderr = capsys.readouterr().err + assert "usage:" in stderr + assert "CI timing helper failed:" in stderr + assert "Traceback" not in stderr + + +def test_pytest_spans_are_spooled_incrementally(monkeypatch): + recorded = [] + monkeypatch.setenv("CI_INFRA_TRACE_ID", "01" * 16) + monkeypatch.setenv("CI_INFRA_COMMAND_SPAN_ID", "02" * 8) + monkeypatch.setattr(ci_otel, "TEST_SPOOL_BATCH_SIZE", 1) + monkeypatch.setattr( + ci_otel, "record_spans", lambda spans: recorded.extend(spans) or True + ) + ci_otel._test_runs.clear() + ci_otel._test_spans.clear() + ci_otel._test_runs["test_sample.py::test_one"] = (time.time_ns(), "passed") + + ci_otel.pytest_runtest_logfinish("test_sample.py::test_one", ("", 0, "")) + + assert len(recorded) == 1 + assert ci_otel._test_spans == [] + + +def test_shell_wrapper_records_commands_without_changing_shell_state(tmp_path): + first = f"ci_otel_start 1 {_quoted('export VALUE=ready')}" + second = f"ci_otel_start 2 {_quoted('check VALUE')}" + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; {first}; ' + f"export VALUE=ready; ci_otel_finish 0; {second}; " + 'test "$VALUE" = ready; ci_otel_finish 0' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_SPOOL_DIR": str(tmp_path), + }, + ) + + assert result.returncode == 0, result.stderr + records = "".join(path.read_text() for path in tmp_path.glob("spans-*.jsonl")) + assert "export VALUE=ready" in records + assert "check VALUE" in records + + +def test_shell_wrapper_preserves_failure_status(tmp_path): + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f"ci_otel_start 1 {_quoted('false')}; false; status=$?; " + 'ci_otel_finish "$status"; exit "$status"' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_SPOOL_DIR": str(tmp_path), + }, + ) + + assert result.returncode == 1 + + +def test_ci_otel_run_records_command_and_preserves_status(tmp_path): + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f"ci_otel_run 1 {_quoted('true')} true; " + f"ci_otel_run 2 {_quoted('false')} false; " + 'echo "status=$?"' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_SPOOL_DIR": str(tmp_path), + }, + ) + + assert result.returncode == 0, result.stderr + assert "status=1" in result.stdout + records = "".join(path.read_text() for path in tmp_path.glob("spans-*.jsonl")) + assert '"ci.command.index":1' in records + assert '"ci.command.index":2' in records + assert '"process.exit.code":0' in records + assert '"process.exit.code":1' in records + + +def test_ci_otel_run_fail_open_when_tracing_unavailable(tmp_path): + shell = ( + f'CI_INFRA_OTEL_DIR="{tmp_path / "missing"}"; export CI_INFRA_OTEL_DIR; ' + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f"ci_otel_run 1 {_quoted('echo ran')} echo ran" + ) + + result = subprocess.run( + ["/bin/sh", "-e", "-c", shell], + check=False, + capture_output=True, + text=True, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "ran" + + +def test_ci_otel_run_handles_assignment_prefixed_commands(tmp_path): + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f"ci_otel_run 1 {_quoted('MY_VAR=hello env')} " + "MY_VAR=hello env" + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_SPOOL_DIR": str(tmp_path), + }, + ) + + assert result.returncode == 0, result.stderr + assert "MY_VAR=hello" in result.stdout + + +def test_missing_helpers_do_not_block_the_test_command(tmp_path): + output = tmp_path / "ran" + shell = ( + f'CI_INFRA_OTEL_DIR="{tmp_path / "missing"}"; export CI_INFRA_OTEL_DIR; ' + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; printf ran > "{output}"' + ) + + result = subprocess.run( + ["/bin/sh", "-e", "-c", shell], + check=False, + capture_output=True, + text=True, + ) + + assert result.returncode == 0, result.stderr + assert output.read_text() == "ran" + + +def test_unset_helper_dir_does_not_block_the_test_command(): + shell = f'unset CI_INFRA_OTEL_DIR; . "{SCRIPTS_DIR / "ci_otel.sh"}"; echo ran' + + result = subprocess.run( + ["/bin/bash", "-e", "-c", shell], + check=False, + capture_output=True, + text=True, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "ran" + + +def test_helper_dir_assignment_is_exported(): + shell = ( + "unset CI_INFRA_OTEL_DIR; " + f'CI_INFRA_OTEL_DIR="{SCRIPTS_DIR}" . "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f'/bin/sh -c \'test "$CI_INFRA_OTEL_DIR" = "{SCRIPTS_DIR}"\'' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], check=False, capture_output=True, text=True + ) + + assert result.returncode == 0, result.stderr + + +@pytest.mark.parametrize("failure", ["new-context", "record-command"]) +def test_helper_failure_cannot_change_job_status(tmp_path, failure): + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_python = fake_bin / "python3" + fake_python.write_text( + "#!/bin/sh\n" + '[ "$1" = "-c" ] && exit 0\n' + '[ "$2" = "new-context" ] && {\n' + ' [ "$CI_OTEL_TEST_FAILURE" = "new-context" ] && exit 99\n' + ' echo "01010101010101010101010101010101 0202020202020202 - 1"\n' + " exit 0\n" + "}\n" + '[ "$2" = "record-command" ] && exit 99\n' + "exit 99\n" + ) + fake_python.chmod(0o755) + shell = ( + f'PATH="{fake_bin}:$PATH"; . "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f"ci_otel_start 1 {_quoted('true')}; ci_otel_finish 0; echo ran" + ) + + result = subprocess.run( + ["/bin/sh", "-e", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_RUNTIME_DIR": str(tmp_path / "runtime"), + "CI_OTEL_TEST_FAILURE": failure, + }, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "ran" + + +def test_flush_failure_preserves_job_status(tmp_path): + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_python = fake_bin / "python3" + fake_python.write_text("#!/bin/sh\nexit 99\n") + fake_python.chmod(0o755) + + def run(status: int): + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f'PATH="{fake_bin}"; export PATH; exit {status}' + ) + return subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={**os.environ, "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR)}, + ) + + assert run(0).returncode == 0 + assert run(7).returncode == 7 + + +def test_shell_wrapper_uses_private_trace_ids(tmp_path): + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f"ci_otel_start 1 {_quoted('true')}; " + 'expected="$CI_INFRA_TRACE_ID"; ' + "CI_INFRA_TRACE_ID=corrupted; CI_INFRA_COMMAND_SPAN_ID=corrupted; " + 'ci_otel_finish 0; printf "%s" "$expected"' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_SPOOL_DIR": str(tmp_path), + }, + ) + + assert result.returncode == 0, result.stderr + record = json.loads(next(tmp_path.glob("spans-*.jsonl")).read_text()) + assert record["trace_id"] == result.stdout + assert record["span_id"] != "corrupted" + + +def test_shell_setup_is_idempotent(tmp_path): + runtime = tmp_path / "runtime" + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + 'printf "%s\n%s" "$PATH" "$CI_INFRA_OTEL_SHIM_PATHS"' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_RUNTIME_DIR": str(runtime), + }, + ) + + assert result.returncode == 0, result.stderr + path, shim_paths = result.stdout.splitlines() + assert path.split(":").count(str(runtime / "bin")) == 1 + assert shim_paths.split(":").count(str(runtime / "bin")) == 1 + + +def test_helper_owned_runtime_is_removed_on_exit(tmp_path): + runtime_file = tmp_path / "runtime" + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f'printf "%s" "$CI_INFRA_OTEL_RUNTIME_DIR" > "{runtime_file}"' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={**os.environ, "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR)}, + ) + + assert result.returncode == 0, result.stderr + assert not Path(runtime_file.read_text()).exists() + + +def test_caller_owned_runtime_is_preserved(tmp_path): + runtime = tmp_path / "runtime" + shell = f'. "{SCRIPTS_DIR / "ci_otel.sh"}"' + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_RUNTIME_DIR": str(runtime), + }, + ) + + assert result.returncode == 0, result.stderr + assert runtime.exists() + + +def test_pytest_shim_recovers_from_stale_recorded_path(tmp_path): + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + real_pytest = bin_dir / "pytest" + real_pytest.write_text("#!/bin/sh\nprintf 'CURRENT-PYTEST'") + real_pytest.chmod(0o755) + + result = subprocess.run( + [str(SCRIPTS_DIR / "ci_pytest.sh"), "--version"], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "PATH": str(bin_dir), + "CI_INFRA_OTEL_REAL_PYTEST": "/missing/pytest", + }, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout == "CURRENT-PYTEST" + + +def test_path_change_after_source_uses_new_pytest(tmp_path): + old_bin = tmp_path / "old" + new_bin = tmp_path / "new" + runtime = tmp_path / "runtime" + old_bin.mkdir() + new_bin.mkdir() + for directory, output in ((old_bin, "OLD"), (new_bin, "NEW")): + executable = directory / "pytest" + executable.write_text(f"#!/bin/sh\nprintf '{output}'") + executable.chmod(0o755) + shell = ( + f'PATH="{old_bin}:$PATH"; . "{SCRIPTS_DIR / "ci_otel.sh"}"; ' + f'PATH="{new_bin}:$PATH"; hash -r; pytest' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_RUNTIME_DIR": str(runtime), + }, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout == "NEW" + + +def test_pytest_shim_records_distinct_test_intervals_with_pythonpath_override( + tmp_path, +): + test_file = tmp_path / "test_sample.py" + test_file.write_text( + "import time\n" + "def test_first(): time.sleep(0.02)\n" + "def test_second(): time.sleep(0.02)\n" + ) + workspace = tmp_path / "workspace" + workspace.mkdir() + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + pytest_executable = bin_dir / "pytest" + pytest_executable.write_text( + f'#!/bin/sh\nexec {shlex.quote(sys.executable)} -m pytest "$@"\n' + ) + pytest_executable.chmod(0o755) + shell = ( + f'. "{SCRIPTS_DIR / "ci_otel.sh"}"\n' + f"ci_otel_start 1 {_quoted('pytest tests')}\n" + f"PYTHONPATH={shlex.quote(str(workspace))} " + f"pytest -q {shlex.quote(str(test_file))}\n" + "status=$?\n" + 'ci_otel_finish "$status"\n' + 'exit "$status"' + ) + + result = subprocess.run( + ["/bin/sh", "-c", shell], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "CI_INFRA_OTEL_DIR": str(SCRIPTS_DIR), + "CI_INFRA_OTEL_SPOOL_DIR": str(tmp_path / "spans"), + "PATH": f"{bin_dir}:{os.environ['PATH']}", + "PYTEST_ADDOPTS": "", + }, + ) + + assert result.returncode == 0, result.stderr + records = [ + json.loads(line) + for path in (tmp_path / "spans").glob("spans-*.jsonl") + for line in path.read_text().splitlines() + ] + tests = [record for record in records if record["name"] == "pytest.test"] + assert len(tests) == 2 + assert tests[0]["start_ns"] < tests[1]["start_ns"] + assert tests[0]["end_ns"] <= tests[1]["start_ns"] + + +def test_pytest_plugin_failure_cannot_fail_pytest(tmp_path): + test_file = tmp_path / "test_sample.py" + test_file.write_text("def test_sample(): assert True\n") + + result = subprocess.run( + [sys.executable, "-m", "pytest", "-q", "-p", "ci_otel", str(test_file)], + check=False, + capture_output=True, + text=True, + env={ + **os.environ, + "PYTHONPATH": str(SCRIPTS_DIR), + "CI_INFRA_TRACE_ID": "01" * 16, + "CI_INFRA_COMMAND_SPAN_ID": "02" * 8, + "CI_INFRA_OTEL_SPOOL_DIR": "/dev/null/spans", + }, + ) + + assert result.returncode == 0, result.stderr + assert "1 passed" in result.stdout