rustfs/scripts/test_scanner_otlp_window.py

#!/usr/bin/env python3
"""Validate measurement boundaries using synthetic OTLP, without a server."""

import copy
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

import scanner_otlp_window as collector

METRIC = "rustfs_heal_admission_start_duration_seconds"
START = 1_800_000_000_000_000_001


def export(end, buckets):
    return {"resourceMetrics": [{
        "resource": {"attributes": [{"key": "host.name", "value": {"stringValue": "node-a"}}]},
        "scopeMetrics": [{"scope": {"name": "rustfs", "version": "1"}, "metrics": [{
            "name": METRIC, "unit": "s", "histogram": {
                "aggregationTemporality": "AGGREGATION_TEMPORALITY_CUMULATIVE",
                "dataPoints": [{"startTimeUnixNano": str(START), "timeUnixNano": str(end),
                                "count": str(sum(buckets)), "bucketCounts": list(map(str, buckets)),
                                "explicitBounds": [0.001, 0.01], "max": 999,
                                "attributes": [{"key": "result", "value": {"stringValue": "accepted"}}]}],
            },
        }]}],
    }]}


def metric(document):
    return document["resourceMetrics"][0]["scopeMetrics"][0]["metrics"][0]


def point(document):
    return metric(document)["histogram"]["dataPoints"][0]


class WindowTests(unittest.TestCase):
    def setUp(self):
        self.before = export(START + 10, [200, 100, 5])
        self.after = export(START + 20, [290, 109, 6])

    def result(self):
        return collector.window(self.before, self.after, [METRIC])

    def test_subtracts_seed_samples_and_preserves_nanoseconds(self):
        result = collector.window(self.before, self.after, [METRIC], START + 10, START + 20)
        row = result["series"][0]
        self.assertTrue(result["window_checked"])
        self.assertEqual(row["window_start_ns"], START + 10)
        self.assertEqual(row["window_end_ns"], START + 20)
        self.assertEqual(row["count"], 100)
        self.assertEqual(row["bucket_counts"], [90, 9, 1])
        self.assertEqual(row["p95"], {"lower_exclusive": 0.001, "upper_inclusive": 0.01})
        self.assertEqual(row["p99"], row["p95"])
        self.assertNotIn("max", row)
        self.assertNotIn("sum", row)

    def test_empty_window_is_not_zero_latency(self):
        self.after = export(START + 20, [200, 100, 5])
        row = self.result()["series"][0]
        self.assertEqual(row["count"], 0)
        self.assertIsNone(row["p95"])
        self.assertIsNone(row["p99"])

    def test_open_tail_is_not_replaced_with_historical_max(self):
        self.after = export(START + 20, [200, 100, 105])
        self.assertEqual(self.result()["series"][0]["p95"],
                         {"lower_exclusive": 0.01, "upper_inclusive": None})

    def test_quantile_rank_does_not_round_large_integer_counts(self):
        # Float multiplication would move this p99 across a bucket boundary.
        count = 2**60 + 1
        rank = (count * 99 + 99) // 100
        self.assertEqual(collector.quantile_bounds([rank - 1, count - rank + 1], [1], 99),
                         {"lower_exclusive": 1, "upper_inclusive": None})

    def test_zero_or_unknown_collection_start_is_rejected(self):
        for start in (None, 0, "0"):
            with self.subTest(start=start):
                point(self.before)["startTimeUnixNano"] = start
                with self.assertRaises(ValueError):
                    self.result()

    def test_rejects_reset_even_when_counters_grow(self):
        point(self.after)["startTimeUnixNano"] = str(START + 1)
        with self.assertRaisesRegex(ValueError, "reset"):
            self.result()

    def test_rejects_decreasing_bucket_even_when_total_grows(self):
        self.after = export(START + 20, [199, 200, 6])
        with self.assertRaisesRegex(ValueError, "counter decreased"):
            self.result()

    def test_rejects_shape_and_count_corruption(self):
        for update in ({"count": "999"}, {"bucketCounts": []},
                       {"bucketCounts": [True, "100", "5"]},
                       {"count": 305.0}, {"timeUnixNano": float(START)},
                       {"explicitBounds": [0.01, 0.001]},
                       {"explicitBounds": [0.001, float("inf")]}, {"flags": 1}):
            with self.subTest(update=update):
                before = copy.deepcopy(self.before)
                point(before).update(update)
                with self.assertRaises(ValueError):
                    collector.window(before, self.after, [METRIC])

    def test_rejects_changed_bucket_layout(self):
        point(self.after)["explicitBounds"] = [0.001, 0.1]
        with self.assertRaisesRegex(ValueError, "boundaries changed"):
            self.result()

    def test_rejects_duplicate_series_or_missing_metric(self):
        scope = self.after["resourceMetrics"][0]["scopeMetrics"][0]
        scope["metrics"].append(copy.deepcopy(metric(self.after)))
        with self.assertRaisesRegex(ValueError, "duplicate metric"):
            self.result()
        scope["metrics"] = []
        with self.assertRaisesRegex(ValueError, "missing requested"):
            self.result()

    def test_rejects_changed_resource_scope_labels_or_unit(self):
        for change in ("resource", "scope", "labels", "unit", "dropped"):
            with self.subTest(change=change):
                after = copy.deepcopy(self.after)
                if change == "resource":
                    after["resourceMetrics"][0]["resource"]["attributes"][0]["value"]["stringValue"] = "node-b"
                elif change == "scope":
                    after["resourceMetrics"][0]["scopeMetrics"][0]["scope"]["version"] = "2"
                elif change == "labels":
                    point(after)["attributes"][0]["value"]["stringValue"] = "duplicate"
                elif change == "unit":
                    metric(after)["unit"] = "ms"
                else:
                    after["resourceMetrics"][0]["resource"]["droppedAttributesCount"] = 1
                with self.assertRaises(ValueError):
                    collector.window(self.before, after, [METRIC])

    def test_label_order_is_not_identity(self):
        for document in (self.before, self.after):
            point(document)["attributes"].append({"key": "source", "value": {"stringValue": "manual"}})
        point(self.after)["attributes"].reverse()
        self.assertEqual(self.result()["series"][0]["count"], 100)

    def test_rejects_duplicate_label_keys(self):
        point(self.after)["attributes"] *= 2
        with self.assertRaisesRegex(ValueError, "duplicate or missing attribute"):
            self.result()

    def test_nodes_remain_separate_and_cannot_claim_a_shared_window(self):
        for document in (self.before, self.after):
            node = copy.deepcopy(document["resourceMetrics"][0])
            node["resource"]["attributes"][0]["value"]["stringValue"] = "node-b"
            node["scopeMetrics"][0]["metrics"][0]["histogram"]["dataPoints"][0]["timeUnixNano"] = str(
                int(point(document)["timeUnixNano"]) + 1)
            document["resourceMetrics"].append(node)
        self.assertEqual(len(self.result()["series"]), 2)
        with self.assertRaisesRegex(ValueError, "requested measurement window"):
            collector.window(self.before, self.after, [METRIC], START + 10, START + 20)

    def test_rejects_reversed_duplicate_and_misaligned_times(self):
        for end in (START + 9, START + 10):
            after = export(end, [290, 109, 6])
            with self.assertRaisesRegex(ValueError, "snapshot time"):
                collector.window(self.before, after, [METRIC])
        with self.assertRaisesRegex(ValueError, "requested measurement window"):
            collector.window(self.before, self.after, [METRIC], START + 11, START + 20)
        with self.assertRaisesRegex(ValueError, "both expected"):
            collector.window(self.before, self.after, [METRIC], START + 10)

    def test_rejects_delta_or_gauge(self):
        metric(self.after)["histogram"]["aggregationTemporality"] = 1
        with self.assertRaisesRegex(ValueError, "not cumulative"):
            self.result()
        metric(self.after)["gauge"] = metric(self.after).pop("histogram")
        with self.assertRaisesRegex(ValueError, "unsupported metric type"):
            self.result()

    def test_rejects_ambiguous_metric_type(self):
        metric(self.after)["sum"] = {"dataPoints": []}
        with self.assertRaisesRegex(ValueError, "ambiguous"):
            self.result()

    def test_rejects_duplicate_json_fields(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "snapshot.json"
            path.write_text('{"resourceMetrics": [], "resourceMetrics": []}')
            with self.assertRaisesRegex(ValueError, "duplicate JSON field"):
                collector.load(path, "json")

    def test_cli_provenance_and_rejection_without_output(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            before, after, output = (root / name for name in ("before.json", "after.json", "window.json"))
            before.write_text(json.dumps(self.before))
            after.write_text(json.dumps(self.after))
            command = [sys.executable, collector.__file__, "--before", str(before), "--after", str(after),
                       "--metric", METRIC, "--out", str(output)]
            self.assertEqual(subprocess.run(command, capture_output=True).returncode, 0)
            result = json.loads(output.read_text())
            self.assertFalse(result["window_checked"])
            self.assertEqual(result["snapshots"]["before_sha256"], collector.load(before, "json")[1])
            original = output.read_bytes()
            self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0)
            self.assertEqual(output.read_bytes(), original)
            output.unlink()
            point(self.after)["flags"] = 1
            after.write_text(json.dumps(self.after))
            self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0)
            self.assertFalse(output.exists())


COUNTER = "rustfs_scanner_cache_serialized_bytes_total"


def counter_export(end, value):
    document = export(end, [0, 0, 0])
    data = metric(document)
    data.update(name=COUNTER, unit="By")
    data.pop("histogram")
    data["sum"] = {"aggregationTemporality": 2, "isMonotonic": True, "dataPoints": [{
        "startTimeUnixNano": str(START), "timeUnixNano": str(end), "asInt": str(value),
        "attributes": [{"key": "cache", "value": {"stringValue": "main"}}],
    }]}
    return document


def counter_point(document):
    return metric(document)["sum"]["dataPoints"][0]


class CounterWindowTests(unittest.TestCase):
    def setUp(self):
        self.before = counter_export(START + 10, 2**60 + 3)
        self.after = counter_export(START + 20, 2**60 + 4)

    def result(self):
        return collector.window(self.before, self.after, [COUNTER], START + 10, START + 20)

    def test_counter_delta_preserves_large_integer_precision(self):
        row = self.result()["series"][0]
        self.assertEqual(row["kind"], "sum")
        self.assertEqual(row["delta"], 1)
        self.assertIs(type(row["delta"]), int)
        self.assertEqual((row["window_start_ns"], row["window_end_ns"]), (START + 10, START + 20))
        self.assertNotIn("p99", row)

    def test_observed_unchanged_counter_is_zero(self):
        for value in (0, 2**63 - 1):
            with self.subTest(value=value):
                counter_point(self.before)["asInt"] = value
                counter_point(self.after)["asInt"] = str(value)
                self.assertEqual(self.result()["series"][0]["delta"], 0)

    def test_counter_rejects_reset_and_decrease(self):
        counter_point(self.after)["startTimeUnixNano"] = str(START + 1)
        with self.assertRaisesRegex(ValueError, "reset"):
            self.result()
        counter_point(self.after)["startTimeUnixNano"] = str(START)
        counter_point(self.after)["asInt"] = "1"
        with self.assertRaisesRegex(ValueError, "counter decreased"):
            self.result()

    def test_counter_requires_monotonic_cumulative_sum(self):
        for value in (False, None, 1, "true"):
            with self.subTest(value=value):
                metric(self.after)["sum"]["isMonotonic"] = value
                with self.assertRaises(ValueError):
                    self.result()
        metric(self.after)["sum"]["isMonotonic"] = True
        metric(self.after)["sum"]["aggregationTemporality"] = 1
        with self.assertRaises(ValueError):
            self.result()

    def test_counter_rejects_non_int64_or_ambiguous_values(self):
        for value in (-1, "-1", str(2**63), True, 3.0, "3.0", None):
            with self.subTest(value=value):
                counter_point(self.after)["asInt"] = value
                with self.assertRaises(ValueError):
                    self.result()
        for values in ({}, {"asDouble": 3.0}, {"asInt": "3", "asDouble": 3.0}):
            with self.subTest(values=values):
                p = counter_point(self.after)
                p.pop("asInt", None)
                p.pop("asDouble", None)
                p.update(values)
                with self.assertRaises(ValueError):
                    self.result()

    def test_counter_rejects_missing_series_or_changed_identity(self):
        for change in ("missing", "duplicate", "labels", "unit"):
            with self.subTest(change=change):
                after = copy.deepcopy(self.after)
                if change == "missing":
                    metric(after)["sum"]["dataPoints"] = []
                elif change == "duplicate":
                    metric(after)["sum"]["dataPoints"] *= 2
                elif change == "labels":
                    counter_point(after)["attributes"][0]["value"]["stringValue"] = "backup"
                else:
                    metric(after)["unit"] = "KiB"
                with self.assertRaises(ValueError):
                    collector.window(self.before, after, [COUNTER])

    def test_counter_rejects_flags_and_invalid_times(self):
        for update in ({"flags": 1}, {"startTimeUnixNano": "0"},
                       {"timeUnixNano": str(START + 10)}, {"timeUnixNano": str(START + 21)}):
            with self.subTest(update=update):
                after = copy.deepcopy(self.after)
                counter_point(after).update(update)
                with self.assertRaises(ValueError):
                    collector.window(self.before, after, [COUNTER], START + 10, START + 20)

    def test_counter_rejects_metric_type_change(self):
        after = export(START + 20, [1, 2, 3])
        metric(after).update(name=COUNTER, unit="By")
        point(after)["attributes"] = counter_point(self.before)["attributes"]
        with self.assertRaisesRegex(ValueError, "type changed"):
            collector.window(self.before, after, [COUNTER])

    def test_counter_and_histogram_share_checked_window(self):
        for document, end, buckets in ((self.before, START + 10, [1, 2, 3]),
                                       (self.after, START + 20, [2, 4, 6])):
            document["resourceMetrics"][0]["scopeMetrics"][0]["metrics"].append(metric(export(end, buckets)))
        result = collector.window(self.before, self.after, [COUNTER, METRIC], START + 10, START + 20)
        rows = {row["name"]: row for row in result["series"]}
        self.assertEqual(rows[COUNTER]["delta"], 1)
        self.assertEqual(rows[METRIC]["count"], 6)
        self.assertTrue(result["window_checked"])
        counter_point(self.after)["timeUnixNano"] = str(START + 21)
        with self.assertRaisesRegex(ValueError, "requested measurement window"):
            collector.window(self.before, self.after, [COUNTER, METRIC], START + 10, START + 20)

    def test_counter_cli_keeps_provenance_and_exact_value(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            before, after, output = (root / name for name in ("before.json", "after.json", "window.json"))
            before.write_text(json.dumps(self.before))
            after.write_text(json.dumps(self.after))
            command = [sys.executable, collector.__file__, "--before", str(before), "--after", str(after),
                       "--metric", COUNTER, "--start-ns", str(START + 10), "--end-ns", str(START + 20),
                       "--out", str(output)]
            run = subprocess.run(command, capture_output=True)
            self.assertEqual(run.returncode, 0, run.stderr)
            result = json.loads(output.read_text())
            self.assertTrue(result["window_checked"])
            self.assertEqual(result["series"][0]["delta"], 1)
            self.assertEqual(result["snapshots"]["after_sha256"], collector.load(after, "json")[1])


if __name__ == "__main__":
    unittest.main()