mirror of
https://github.com/XRPLF/rippled.git
synced 2026-08-21 22:30:57 +00:00
fix(telemetry): make the log-trace correlation checks meaningful
Both checks selected on {job="xrpld"}. Loki's OTLP ingestion promotes
service.name to the label `service_name` and keeps a `job` attribute as
structured metadata, which a stream selector cannot match, so the selector
returned zero streams whatever had been ingested. The collector config and
TESTING.md already say to select on `service_name`.
Invert the cross-reference. Picking an arbitrary trace from Tempo and
expecting it in Loki fails even when correlation works, because a log line
carries a trace_id only when emitted inside a sampled span and most spans
log nothing at `warning` level. Start from a logged trace_id instead and
resolve it in Tempo, which is the invariant worth asserting, and try every
id found so one unexported trace does not fail the check.
Bound the log queries in time. Nothing here set start/end, so every query
relied on Loki's one-hour default and returned nothing when re-run later to
investigate a result.
This commit is contained in:
Pratik Mankawde
@@ -33,6 +33,7 @@ import asyncio
|
||||
import fnmatch
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
@@ -41,6 +42,11 @@ from typing import Any
|
||||
|
||||
import aiohttp
|
||||
|
||||
# Loki's default query window is the last hour. A validation run finishes in
|
||||
# minutes, but bounding the range explicitly keeps the query reproducible when
|
||||
# someone re-runs it later to investigate a result.
|
||||
LOG_QUERY_WINDOW_SECONDS = 4 * 60 * 60
|
||||
|
||||
logger = logging.getLogger("validate_telemetry")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -167,6 +173,19 @@ class ValidationReport:
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _log_query_window() -> dict[str, str]:
|
||||
"""Loki query_range bounds covering a validation run.
|
||||
|
||||
Returns:
|
||||
start/end parameters in nanoseconds since the epoch.
|
||||
"""
|
||||
now = time.time()
|
||||
return {
|
||||
"start": str(int((now - LOG_QUERY_WINDOW_SECONDS) * 1_000_000_000)),
|
||||
"end": str(int(now * 1_000_000_000)),
|
||||
}
|
||||
|
||||
|
||||
async def _tempo_search(
|
||||
session: aiohttp.ClientSession,
|
||||
tempo_url: str,
|
||||
@@ -849,9 +868,13 @@ async def validate_log_trace_correlation(
|
||||
# Check 1: Any logs with trace_id exist.
|
||||
try:
|
||||
params = {
|
||||
"query": '{job="xrpld"} |= "trace_id="',
|
||||
# Loki's OTLP ingestion promotes service.name to the label
|
||||
# `service_name`. A `job` attribute is structured metadata, which a
|
||||
# stream selector cannot match — see otel-collector-config.yaml.
|
||||
"query": '{service_name="xrpld"} |= "trace_id="',
|
||||
"limit": 5,
|
||||
"direction": "backward",
|
||||
**_log_query_window(),
|
||||
}
|
||||
async with session.get(
|
||||
f"{loki_url}/loki/api/v1/query_range", params=params
|
||||
@@ -882,56 +905,79 @@ async def validate_log_trace_correlation(
|
||||
)
|
||||
)
|
||||
|
||||
# Check 2: Cross-reference a trace_id from Tempo to Loki.
|
||||
# Check 2: Cross-reference a trace_id from a log line back to Tempo.
|
||||
#
|
||||
# Driven from the log side on purpose. A trace_id only reaches a log line
|
||||
# when that line is emitted inside a sampled span, and at `warning` level
|
||||
# most spans produce no log output at all — so picking an arbitrary trace
|
||||
# from Tempo and expecting it in Loki fails even when correlation works.
|
||||
# Starting from a logged trace_id tests the invariant that matters: an id
|
||||
# written to a log must resolve to a trace that was actually exported.
|
||||
try:
|
||||
# Get a recent trace from Tempo.
|
||||
traces = await _tempo_search(
|
||||
session,
|
||||
tempo_url,
|
||||
'{resource.service.name="xrpld"}',
|
||||
limit=1,
|
||||
)
|
||||
loki_params = {
|
||||
"query": '{service_name="xrpld"} |= "trace_id="',
|
||||
"limit": 5,
|
||||
"direction": "backward",
|
||||
**_log_query_window(),
|
||||
}
|
||||
async with session.get(
|
||||
f"{loki_url}/loki/api/v1/query_range", params=loki_params
|
||||
) as resp:
|
||||
data = await resp.json()
|
||||
streams = data.get("data", {}).get("result", [])
|
||||
|
||||
if traces:
|
||||
trace_id = traces[0].get("traceID", "")
|
||||
if trace_id:
|
||||
# Search Loki for this trace_id.
|
||||
loki_params = {
|
||||
"query": f'{{job="xrpld"}} |= "{trace_id}"',
|
||||
"limit": 5,
|
||||
"direction": "backward",
|
||||
}
|
||||
async with session.get(
|
||||
f"{loki_url}/loki/api/v1/query_range",
|
||||
params=loki_params,
|
||||
) as loki_resp:
|
||||
loki_data = await loki_resp.json()
|
||||
loki_streams = loki_data.get("data", {}).get("result", [])
|
||||
loki_count = sum(len(s.get("values", [])) for s in loki_streams)
|
||||
report.add(
|
||||
CheckResult(
|
||||
name="log.trace_id_cross_reference",
|
||||
category="log",
|
||||
passed=loki_count > 0,
|
||||
message=(
|
||||
f"trace_id {trace_id[:16]}... found in "
|
||||
f"{loki_count} Loki entries"
|
||||
if loki_count > 0
|
||||
else f"trace_id {trace_id[:16]}... not found " "in Loki"
|
||||
),
|
||||
details={
|
||||
"trace_id": trace_id,
|
||||
"loki_count": loki_count,
|
||||
},
|
||||
)
|
||||
)
|
||||
else:
|
||||
logged_ids = [
|
||||
match.group(1)
|
||||
for stream in streams
|
||||
for _, line in stream.get("values", [])
|
||||
if (match := re.search(r"trace_id=([0-9a-f]{32})", line))
|
||||
]
|
||||
|
||||
if not logged_ids:
|
||||
report.add(
|
||||
CheckResult(
|
||||
name="log.trace_id_cross_reference",
|
||||
category="log",
|
||||
passed=False,
|
||||
message="No traces in Tempo to cross-reference",
|
||||
message=(
|
||||
"No logged trace_id to cross-reference. Log lines carry one only "
|
||||
"when emitted inside a sampled span; raise the log level or widen "
|
||||
"the workload if this persists."
|
||||
),
|
||||
)
|
||||
)
|
||||
else:
|
||||
# Try every id found, not just the first: one unexported trace
|
||||
# should not fail the check while correlation demonstrably works.
|
||||
resolved: str | None = None
|
||||
span_count = 0
|
||||
unique_ids = list(dict.fromkeys(logged_ids))
|
||||
for candidate in unique_ids:
|
||||
try:
|
||||
spans = await _tempo_get_trace(session, tempo_url, candidate)
|
||||
except Exception: # noqa: BLE001 - a 404 is "not found", not an error
|
||||
continue
|
||||
if spans:
|
||||
resolved, span_count = candidate, len(spans)
|
||||
break
|
||||
|
||||
report.add(
|
||||
CheckResult(
|
||||
name="log.trace_id_cross_reference",
|
||||
category="log",
|
||||
passed=resolved is not None,
|
||||
message=(
|
||||
f"logged trace_id {resolved[:16]}... resolves to "
|
||||
f"{span_count} spans in Tempo"
|
||||
if resolved
|
||||
else f"none of {len(unique_ids)} logged trace_id(s) resolve in "
|
||||
"Tempo; the spans they name were not exported"
|
||||
),
|
||||
details={
|
||||
"trace_id": resolved,
|
||||
"span_count": span_count,
|
||||
"candidates": len(unique_ids),
|
||||
},
|
||||
)
|
||||
)
|
||||
except Exception as exc:
|
||||
|
||||
Reference in New Issue
Block a user