[CI] Add repository-local OTel tracing helpers (#52851)

Signed-off-by: khluu <khluu000@gmail.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Kevin H. Luu
2026-09-01 02:36:46 -07:00
committed by GitHub
co-authored by Codex
parent 8f03625b3d
commit 504bb8b0c3
5 changed files with 1661 additions and 1 deletions
+48 -1
View File
@@ -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"
+650
View File
@@ -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("<Q", value)
def _attribute(name: str, value: str | int | bool) -> 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())
+133
View File
@@ -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
:
+42
View File
@@ -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}" "$@"
@@ -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