mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
Chris
8c7b3996cb
* test(scanner): validate OTLP histogram measurement windows * feat(scanner): extract exact integer counter windows * test(ecstore): drain guarded delete fixture rename tails * test(ilm): retain transition timeout evidence
222 lines
10 KiB
Python
222 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Extract histogram and integer counter windows without filling missing data."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
MAX_SNAPSHOT_BYTES = 64 * 1024 * 1024
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def unique_object(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
require(key not in result, "duplicate JSON field")
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def uint(value, name):
|
|
# OTLP JSON encodes uint64 as strings; accepting floats loses nanoseconds.
|
|
if isinstance(value, str) and value.isascii() and value.isdecimal():
|
|
value = int(value)
|
|
require(type(value) is int and 0 <= value < 2**64, f"invalid {name}")
|
|
return value
|
|
|
|
|
|
def attributes(value):
|
|
require(isinstance(value, list), "attributes must be an array")
|
|
result = {}
|
|
for entry in value:
|
|
require(isinstance(entry, dict), "invalid attribute")
|
|
key = entry.get("key")
|
|
require(isinstance(key, str) and key and key not in result, "duplicate or missing attribute key")
|
|
require(isinstance(entry.get("value"), dict) and entry["value"], "missing attribute value")
|
|
result[key] = entry["value"]
|
|
return result
|
|
|
|
|
|
def identity(resource, scope, metric, point, resource_schema, scope_schema):
|
|
require(uint(resource.get("droppedAttributesCount", 0), "resource dropped attributes") == 0,
|
|
"resource identity has dropped attributes")
|
|
require(uint(scope.get("droppedAttributesCount", 0), "scope dropped attributes") == 0,
|
|
"scope identity has dropped attributes")
|
|
return {
|
|
"resource_schema": resource_schema,
|
|
"resource": attributes(resource.get("attributes", [])),
|
|
"scope_schema": scope_schema,
|
|
"scope": {"name": scope.get("name", ""), "version": scope.get("version", ""),
|
|
"attributes": attributes(scope.get("attributes", []))},
|
|
"name": metric["name"],
|
|
"unit": metric.get("unit", ""),
|
|
"attributes": attributes(point.get("attributes", [])),
|
|
}
|
|
|
|
|
|
def point_times(point):
|
|
require(uint(point.get("flags", 0), "point flags") == 0, "point has no recorded value or unknown flags")
|
|
start = uint(point.get("startTimeUnixNano"), "startTimeUnixNano")
|
|
end = uint(point.get("timeUnixNano"), "timeUnixNano")
|
|
require(0 < start <= end, "missing or reversed point timestamps")
|
|
return {"start": start, "end": end}
|
|
|
|
|
|
def counter_point(point):
|
|
require(set(point) & {"asInt", "asDouble"} == {"asInt"}, "counter must have one integer value")
|
|
value = uint(point.get("asInt"), "counter value")
|
|
require(value < 2**63, "counter exceeds OTLP signed int64 range")
|
|
return {**point_times(point), "kind": "sum", "value": value}
|
|
|
|
|
|
def histogram_point(point):
|
|
times = point_times(point)
|
|
bounds = point.get("explicitBounds", [])
|
|
buckets = point.get("bucketCounts", [])
|
|
require(isinstance(bounds, list) and isinstance(buckets, list), "invalid histogram buckets")
|
|
require(all(type(v) in (float, int) and math.isfinite(v) for v in bounds), "non-finite histogram boundary")
|
|
require(all(a < b for a, b in zip(bounds, bounds[1:])), "unordered histogram boundaries")
|
|
require(len(buckets) == len(bounds) + 1, "missing or inconsistent histogram buckets")
|
|
buckets = [uint(value, "bucket count") for value in buckets]
|
|
count = uint(point.get("count", 0), "histogram count")
|
|
require(sum(buckets) == count, "histogram count does not match buckets")
|
|
return {**times, "kind": "histogram", "count": count, "bounds": bounds, "buckets": buckets}
|
|
|
|
|
|
def snapshot(document, names):
|
|
require(isinstance(document, dict), "snapshot must be an OTLP export object")
|
|
series = {}
|
|
seen = set()
|
|
for resource_metrics in document.get("resourceMetrics", []):
|
|
resource = resource_metrics.get("resource", {})
|
|
for scope_metrics in resource_metrics.get("scopeMetrics", []):
|
|
scope = scope_metrics.get("scope", {})
|
|
for metric in scope_metrics.get("metrics", []):
|
|
if metric.get("name") not in names:
|
|
continue
|
|
kinds = set(metric) & {"histogram", "exponentialHistogram", "sum", "gauge", "summary"}
|
|
require(kinds in ({"histogram"}, {"sum"}), "ambiguous or unsupported metric type")
|
|
kind = next(iter(kinds))
|
|
data = metric[kind]
|
|
require(isinstance(data, dict), f"{metric['name']} has invalid metric data")
|
|
require(data.get("aggregationTemporality") in (2, "AGGREGATION_TEMPORALITY_CUMULATIVE"),
|
|
f"{metric['name']} is not cumulative")
|
|
if kind == "sum":
|
|
require(data.get("isMonotonic") is True, "counter sum must be monotonic")
|
|
for point in data.get("dataPoints", []):
|
|
key = identity(resource, scope, metric, point,
|
|
resource_metrics.get("schemaUrl", ""), scope_metrics.get("schemaUrl", ""))
|
|
encoded = json.dumps(key, sort_keys=True, allow_nan=False)
|
|
require(encoded not in series, "duplicate metric series in snapshot")
|
|
series[encoded] = (key, histogram_point(point) if kind == "histogram" else counter_point(point))
|
|
seen.add(metric["name"])
|
|
require(seen == set(names), "snapshot is missing requested metric observations")
|
|
return series
|
|
|
|
|
|
def quantile_bounds(buckets, bounds, numerator):
|
|
count = sum(buckets)
|
|
if count == 0:
|
|
return None
|
|
rank = (count * numerator + 99) // 100
|
|
cumulative = 0
|
|
for index, bucket in enumerate(buckets):
|
|
cumulative += bucket
|
|
if cumulative >= rank:
|
|
# The bucket only bounds the quantile; interpolation invents precision.
|
|
return {"lower_exclusive": bounds[index - 1] if index else None,
|
|
"upper_inclusive": bounds[index] if index < len(bounds) else None}
|
|
raise ValueError("quantile rank exceeds histogram count")
|
|
|
|
|
|
def window(before, after, names, expected_start=None, expected_end=None):
|
|
require(names and len(names) == len(set(names)), "metric names must be nonempty and unique")
|
|
require((expected_start is None) == (expected_end is None), "both expected window boundaries are required")
|
|
if expected_start is not None:
|
|
expected_start = uint(expected_start, "expected start")
|
|
expected_end = uint(expected_end, "expected end")
|
|
require(0 < expected_start < expected_end, "invalid expected window")
|
|
first, last = snapshot(before, names), snapshot(after, names)
|
|
require(first.keys() == last.keys(), "metric series identity changed or is missing at a boundary")
|
|
rows = []
|
|
for key in sorted(first):
|
|
descriptor, a = first[key]
|
|
_, b = last[key]
|
|
require(a["kind"] == b["kind"], "metric type changed within measurement window")
|
|
require(a["start"] == b["start"], "metric reset within measurement window")
|
|
require(a["end"] < b["end"], "duplicate or reversed snapshot time")
|
|
if expected_start is not None:
|
|
require((a["end"], b["end"]) == (expected_start, expected_end),
|
|
"metric does not match the requested measurement window")
|
|
row = {**descriptor, "kind": a["kind"], "window_start_ns": a["end"], "window_end_ns": b["end"]}
|
|
if a["kind"] == "sum":
|
|
delta = b["value"] - a["value"]
|
|
require(delta >= 0, "integer counter decreased")
|
|
rows.append({**row, "delta": delta})
|
|
continue
|
|
require(a["bounds"] == b["bounds"], "histogram boundaries changed")
|
|
buckets = [end - start for start, end in zip(a["buckets"], b["buckets"])]
|
|
require(all(value >= 0 for value in buckets), "histogram bucket counter decreased")
|
|
count = b["count"] - a["count"]
|
|
require(count == sum(buckets), "window count does not match buckets")
|
|
rows.append({**row,
|
|
"count": count, "explicit_bounds": a["bounds"], "bucket_counts": buckets,
|
|
"p95": quantile_bounds(buckets, a["bounds"], 95),
|
|
"p99": quantile_bounds(buckets, a["bounds"], 99)})
|
|
return {"window_checked": expected_start is not None, "series": rows}
|
|
|
|
|
|
def load(path, input_format):
|
|
require(path.stat().st_size <= MAX_SNAPSHOT_BYTES, "oversized OTLP snapshot")
|
|
raw = path.read_bytes()
|
|
if input_format == "protobuf":
|
|
try:
|
|
from google.protobuf.json_format import MessageToDict
|
|
from google.protobuf.message import DecodeError
|
|
from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ExportMetricsServiceRequest
|
|
except ImportError as error:
|
|
raise ValueError("protobuf input requires opentelemetry-proto in the active Python environment") from error
|
|
request = ExportMetricsServiceRequest()
|
|
try:
|
|
request.ParseFromString(raw)
|
|
except DecodeError as error:
|
|
raise ValueError("malformed OTLP protobuf") from error
|
|
document = MessageToDict(request)
|
|
else:
|
|
document = json.loads(raw, object_pairs_hook=unique_object)
|
|
return document, hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--before", type=Path, required=True)
|
|
parser.add_argument("--after", type=Path, required=True)
|
|
parser.add_argument("--metric", action="append", required=True)
|
|
parser.add_argument("--format", choices=("json", "protobuf"), default="json")
|
|
parser.add_argument("--start-ns", type=int)
|
|
parser.add_argument("--end-ns", type=int)
|
|
parser.add_argument("--out", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
before, before_sha = load(args.before, args.format)
|
|
after, after_sha = load(args.after, args.format)
|
|
result = window(before, after, args.metric, args.start_ns, args.end_ns)
|
|
result["snapshots"] = {"before_sha256": before_sha, "after_sha256": after_sha}
|
|
with args.out.open("x", encoding="utf-8") as output:
|
|
output.write(json.dumps(result, indent=2, sort_keys=True, allow_nan=False) + "\n")
|
|
except (ValueError, OSError, TypeError, AttributeError) as error:
|
|
print(f"OTLP window rejected: {error}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|