diff --git a/src/pmmcp/tools/_fetch.py b/src/pmmcp/tools/_fetch.py index 91769b2..92dfdcc 100644 --- a/src/pmmcp/tools/_fetch.py +++ b/src/pmmcp/tools/_fetch.py @@ -81,6 +81,21 @@ async def fetch_instances(): return name_by_series, instance_name_by_series +async def _fetch_descs(client: PmproxyClient, series_ids: list[str]) -> dict[str, str]: + """Fetch metric semantics from /series/descs, returning {series_id: semantics_str}. + + Swallows PmproxyError on failure — callers fall back to treating metrics + as instant (the safe default). + """ + if not series_ids: + return {} + try: + raw = await client.series_descs(series_ids) + return {entry["series"]: entry.get("semantics", "instant") for entry in raw} + except PmproxyError: + return {} + + async def _fetch_window( client: PmproxyClient, exprs: list[str], diff --git a/src/pmmcp/tools/_stats.py b/src/pmmcp/tools/_stats.py index 71d8f36..ecf6106 100644 --- a/src/pmmcp/tools/_stats.py +++ b/src/pmmcp/tools/_stats.py @@ -59,6 +59,28 @@ def pearson_correlation(xs: list[float], ys: list[float]) -> float: return cov / denom +def _compute_rates(samples: list[dict]) -> list[float]: + """Convert timestamped counter samples to per-second rate-of-change values. + + Each sample is ``{"timestamp": float, "value": float}``. For consecutive + pairs the rate is ``max(0, delta_v / delta_t)`` — clamped to zero on counter + wraps. Pairs with zero ``delta_t`` are silently skipped (avoids div-by-zero). + + Client-side rate conversion because pmproxy's Series API (our only path to + historical windows) doesn't support ``rate()``. The PMAPI surface has it via + ``/pmapi/derive``, but that's live-only. See #33 for the upstream feature + request to add rate support to the Series API. + """ + rates: list[float] = [] + for i in range(1, len(samples)): + dt = samples[i]["timestamp"] - samples[i - 1]["timestamp"] + if dt == 0.0: + continue + dv = samples[i]["value"] - samples[i - 1]["value"] + rates.append(max(0.0, dv / dt)) + return rates + + def outlier_flag(values: list[float], threshold: float = 2.0) -> list[bool]: """Flag values that deviate more than *threshold* standard deviations from the mean. diff --git a/src/pmmcp/tools/comparison.py b/src/pmmcp/tools/comparison.py index 6a8e1ce..d2e9476 100644 --- a/src/pmmcp/tools/comparison.py +++ b/src/pmmcp/tools/comparison.py @@ -8,8 +8,8 @@ from pmmcp.server import get_client, mcp from pmmcp.tools._errors import _mcp_error from pmmcp.tools._expr import build_series_expr -from pmmcp.tools._fetch import _fetch_window, _resolve_series_ids -from pmmcp.tools._stats import _compute_stats +from pmmcp.tools._fetch import _fetch_descs, _fetch_metadata, _fetch_window, _resolve_series_ids +from pmmcp.tools._stats import _compute_rates, _compute_stats from pmmcp.utils import resolve_interval logger = logging.getLogger(__name__) @@ -38,6 +38,19 @@ async def _compare_windows_impl( try: series_ids = await _resolve_series_ids(client, [expr]) + + # Fetch metric semantics to detect counters vs instant/discrete + descs_by_series = await _fetch_descs(client, series_ids) + name_by_series, instance_name_by_series = await _fetch_metadata(client, series_ids) + + # Build semantics lookup keyed the same way _fetch_window keys its results + semantics_by_key: dict[tuple[str, str | None], str] = {} + for sid in series_ids: + metric_name = name_by_series.get(sid, sid) + inst_name = instance_name_by_series.get(sid) or None + key = (metric_name, inst_name) + semantics_by_key[key] = descs_by_series.get(sid, "instant") + values_a, samples_a = await _fetch_window( client, exprs=[], @@ -77,8 +90,17 @@ async def _compare_windows_impl( va = values_a.get(key, []) vb = values_b.get(key, []) - stats_a = _compute_stats(va) - stats_b = _compute_stats(vb) + semantics = semantics_by_key.get(key, "instant") + + if semantics == "counter": + # Counters are monotonic odometers — compare rates, not raw values + rates_a = _compute_rates(samples_a.get(key, [])) + rates_b = _compute_rates(samples_b.get(key, [])) + stats_a = _compute_stats(rates_a) + stats_b = _compute_stats(rates_b) + else: + stats_a = _compute_stats(va) + stats_b = _compute_stats(vb) mean_change = stats_b["mean"] - stats_a["mean"] mean_change_pct = (mean_change / stats_a["mean"] * 100) if stats_a["mean"] != 0 else 0.0 @@ -89,6 +111,7 @@ async def _compare_windows_impl( comp: dict = { "metric": metric_name, "instance": instance_name, + "semantics": semantics, "window_a": stats_a, "window_b": stats_b, "delta": { diff --git a/tests/integration/test_comparison.py b/tests/integration/test_comparison.py index 598936b..16a9ab8 100644 --- a/tests/integration/test_comparison.py +++ b/tests/integration/test_comparison.py @@ -44,6 +44,7 @@ async def test_compare_windows_happy_path(mcp_session): ) ) mock.get("/series/instances").mock(return_value=httpx.Response(200, json=[])) + mock.get("/series/descs").mock(return_value=httpx.Response(200, json=[])) result = await mcp_session.call_tool( "pcp_compare_windows", diff --git a/tests/unit/test_comparison.py b/tests/unit/test_comparison.py index 375379c..fd9b663 100644 --- a/tests/unit/test_comparison.py +++ b/tests/unit/test_comparison.py @@ -194,6 +194,189 @@ def values_side_effect(request): await client.close() +@respx.mock +async def test_compare_windows_counter_metric_uses_rates(config): + """Counter metrics should compare rates (per-second), not raw cumulative values. + + Two windows with identical constant rate (1.0/s) should show no significant change, + even though raw counter values are wildly different between windows. + """ + # Window A: counter from 1000 to 1240 (rate = 1.0/s at 60s intervals) + window_a_counter = [1000.0, 1060.0, 1120.0, 1180.0, 1240.0] + # Window B: counter from 5000 to 5240 (same rate = 1.0/s) + window_b_counter = [5000.0, 5060.0, 5120.0, 5180.0, 5240.0] + + call_count = 0 + + def values_side_effect(request): + nonlocal call_count + call_count += 1 + if call_count == 1: + return httpx.Response(200, json=_make_values(TEST_SERIES, window_a_counter)) + return httpx.Response( + 200, json=_make_values(TEST_SERIES, window_b_counter, base_ts=1547570046.0) + ) + + respx.get(f"{PMPROXY_BASE}/series/query").mock( + return_value=httpx.Response(200, json=[TEST_SERIES]) + ) + respx.get(f"{PMPROXY_BASE}/series/values").mock(side_effect=values_side_effect) + respx.get(f"{PMPROXY_BASE}/series/labels").mock( + return_value=httpx.Response( + 200, + json=[{"series": TEST_SERIES, "labels": {"metric.name": "network.interface.in.bytes"}}], + ) + ) + respx.get(f"{PMPROXY_BASE}/series/instances").mock(return_value=httpx.Response(200, json=[])) + respx.get(f"{PMPROXY_BASE}/series/descs").mock( + return_value=httpx.Response(200, json=[{"series": TEST_SERIES, "semantics": "counter"}]) + ) + + client = PmproxyClient(config) + try: + from pmmcp.tools.comparison import _compare_windows_impl + + result = await _compare_windows_impl( + client, + names=["network.interface.in.bytes"], + window_a_start="-2hours", + window_a_end="-1hour", + window_b_start="-1hour", + window_b_end="now", + host="", + instances=[], + interval="5min", + include_samples=False, + ) + assert isinstance(result, list) + assert len(result) > 0 + comp = result[0] + # Rate is identical in both windows → not significant + assert comp["delta"]["significant"] is False + # Should report counter semantics + assert comp["semantics"] == "counter" + finally: + await client.close() + + +@respx.mock +async def test_compare_windows_instant_metric_uses_raw_values(config): + """Instant metrics should compare raw values directly (existing behaviour).""" + window_a_values = [10.0, 12.0, 11.0, 13.0, 10.0] + window_b_values = [50.0, 55.0, 52.0, 58.0, 51.0] + + call_count = 0 + + def values_side_effect(request): + nonlocal call_count + call_count += 1 + if call_count == 1: + return httpx.Response(200, json=_make_values(TEST_SERIES, window_a_values)) + return httpx.Response( + 200, json=_make_values(TEST_SERIES, window_b_values, base_ts=1547570046.0) + ) + + respx.get(f"{PMPROXY_BASE}/series/query").mock( + return_value=httpx.Response(200, json=[TEST_SERIES]) + ) + respx.get(f"{PMPROXY_BASE}/series/values").mock(side_effect=values_side_effect) + respx.get(f"{PMPROXY_BASE}/series/labels").mock( + return_value=httpx.Response( + 200, + json=[{"series": TEST_SERIES, "labels": {"metric.name": "kernel.all.cpu.user"}}], + ) + ) + respx.get(f"{PMPROXY_BASE}/series/instances").mock(return_value=httpx.Response(200, json=[])) + respx.get(f"{PMPROXY_BASE}/series/descs").mock( + return_value=httpx.Response(200, json=[{"series": TEST_SERIES, "semantics": "instant"}]) + ) + + client = PmproxyClient(config) + try: + from pmmcp.tools.comparison import _compare_windows_impl + + result = await _compare_windows_impl( + client, + names=["kernel.all.cpu.user"], + window_a_start="-2hours", + window_a_end="-1hour", + window_b_start="-1hour", + window_b_end="now", + host="", + instances=[], + interval="5min", + include_samples=False, + ) + assert isinstance(result, list) + assert len(result) > 0 + comp = result[0] + # Instant metric with very different values → significant + assert comp["delta"]["significant"] is True + assert comp["semantics"] == "instant" + finally: + await client.close() + + +@respx.mock +async def test_compare_windows_descs_failure_falls_back_to_instant(config): + """When /series/descs returns HTTP 500, fall back to instant (raw values).""" + window_a_values = [10.0, 12.0, 11.0, 13.0, 10.0] + window_b_values = [50.0, 55.0, 52.0, 58.0, 51.0] + + call_count = 0 + + def values_side_effect(request): + nonlocal call_count + call_count += 1 + if call_count == 1: + return httpx.Response(200, json=_make_values(TEST_SERIES, window_a_values)) + return httpx.Response( + 200, json=_make_values(TEST_SERIES, window_b_values, base_ts=1547570046.0) + ) + + respx.get(f"{PMPROXY_BASE}/series/query").mock( + return_value=httpx.Response(200, json=[TEST_SERIES]) + ) + respx.get(f"{PMPROXY_BASE}/series/values").mock(side_effect=values_side_effect) + respx.get(f"{PMPROXY_BASE}/series/labels").mock( + return_value=httpx.Response( + 200, + json=[{"series": TEST_SERIES, "labels": {"metric.name": "kernel.all.cpu.user"}}], + ) + ) + respx.get(f"{PMPROXY_BASE}/series/instances").mock(return_value=httpx.Response(200, json=[])) + # Descs endpoint fails + respx.get(f"{PMPROXY_BASE}/series/descs").mock( + return_value=httpx.Response(500, json={"message": "internal error"}) + ) + + client = PmproxyClient(config) + try: + from pmmcp.tools.comparison import _compare_windows_impl + + result = await _compare_windows_impl( + client, + names=["kernel.all.cpu.user"], + window_a_start="-2hours", + window_a_end="-1hour", + window_b_start="-1hour", + window_b_end="now", + host="", + instances=[], + interval="5min", + include_samples=False, + ) + # Should still work — falls back to raw values (instant behaviour) + assert isinstance(result, list) + assert len(result) > 0 + comp = result[0] + assert comp["delta"]["significant"] is True + # Fallback semantics should be "instant" + assert comp["semantics"] == "instant" + finally: + await client.close() + + @respx.mock async def test_compare_windows_same_resolved_interval_for_both(config): """pcp_compare_windows applies resolve_interval once, uses same interval for both windows.""" diff --git a/tests/unit/test_fetch.py b/tests/unit/test_fetch.py index 13e5b86..a2b31de 100644 --- a/tests/unit/test_fetch.py +++ b/tests/unit/test_fetch.py @@ -46,6 +46,84 @@ def test_chunked_exact_multiple(): # _fetch_window batching tests # --------------------------------------------------------------------------- +TEST_SERIES = "605fc77742cd0317597291329561ac4e50c0dd12" + + +# --------------------------------------------------------------------------- +# _fetch_descs tests +# --------------------------------------------------------------------------- + + +class TestFetchDescs: + """Tests for _fetch_descs — fetches metric semantics from /series/descs.""" + + @respx.mock + async def test_returns_semantics_mapping(self, config): + """Happy path: maps series ID to its semantics string.""" + respx.get(f"{PMPROXY_BASE}/series/descs").mock( + return_value=httpx.Response( + 200, + json=[{"series": TEST_SERIES, "semantics": "counter"}], + ) + ) + + client = PmproxyClient(config) + try: + from pmmcp.tools._fetch import _fetch_descs + + result = await _fetch_descs(client, [TEST_SERIES]) + assert result == {TEST_SERIES: "counter"} + finally: + await client.close() + + @respx.mock + async def test_empty_input_returns_empty(self, config): + """No series IDs → no pmproxy call, empty dict.""" + from pmmcp.tools._fetch import _fetch_descs + + client = PmproxyClient(config) + try: + result = await _fetch_descs(client, []) + assert result == {} + finally: + await client.close() + + @respx.mock + async def test_pmproxy_error_returns_empty(self, config): + """On pmproxy HTTP error, swallow and return empty dict.""" + respx.get(f"{PMPROXY_BASE}/series/descs").mock( + return_value=httpx.Response(500, json={"message": "internal error"}) + ) + + client = PmproxyClient(config) + try: + from pmmcp.tools._fetch import _fetch_descs + + result = await _fetch_descs(client, [TEST_SERIES]) + assert result == {} + finally: + await client.close() + + @respx.mock + async def test_missing_semantics_defaults_to_instant(self, config): + """When descs response omits 'semantics', default to 'instant'.""" + respx.get(f"{PMPROXY_BASE}/series/descs").mock( + return_value=httpx.Response( + 200, + json=[{"series": TEST_SERIES}], + ) + ) + + client = PmproxyClient(config) + try: + from pmmcp.tools._fetch import _fetch_descs + + result = await _fetch_descs(client, [TEST_SERIES]) + assert result == {TEST_SERIES: "instant"} + finally: + await client.close() + + SERIES_IDS = [f"series{i:04d}" for i in range(25)] diff --git a/tests/unit/test_stats.py b/tests/unit/test_stats.py index 339c03f..c966cc4 100644 --- a/tests/unit/test_stats.py +++ b/tests/unit/test_stats.py @@ -152,6 +152,68 @@ def test_symmetric_outliers(self): assert flags[-2] is True +class TestComputeRates: + """Tests for _compute_rates — converts timestamped counter samples to per-second rates.""" + + def test_empty_samples_returns_empty(self): + from pmmcp.tools._stats import _compute_rates + + assert _compute_rates([]) == [] + + def test_single_sample_returns_empty(self): + from pmmcp.tools._stats import _compute_rates + + assert _compute_rates([{"timestamp": 0.0, "value": 100.0}]) == [] + + def test_monotonic_counter_rate(self): + """600 units over 60 seconds = 10.0 per second.""" + from pmmcp.tools._stats import _compute_rates + + samples = [ + {"timestamp": 0.0, "value": 0.0}, + {"timestamp": 60.0, "value": 600.0}, + ] + rates = _compute_rates(samples) + assert len(rates) == 1 + assert abs(rates[0] - 10.0) < 1e-9 + + def test_counter_wrap_clamps_to_zero(self): + """When counter value decreases (wrap), rate should be 0.0.""" + from pmmcp.tools._stats import _compute_rates + + samples = [ + {"timestamp": 0.0, "value": 1000.0}, + {"timestamp": 60.0, "value": 500.0}, + ] + rates = _compute_rates(samples) + assert len(rates) == 1 + assert rates[0] == 0.0 + + def test_zero_dt_skipped(self): + """Pairs with identical timestamps are skipped to avoid division by zero.""" + from pmmcp.tools._stats import _compute_rates + + samples = [ + {"timestamp": 100.0, "value": 0.0}, + {"timestamp": 100.0, "value": 10.0}, + {"timestamp": 160.0, "value": 70.0}, + ] + rates = _compute_rates(samples) + # First pair skipped (dt=0), second pair: (70-10)/60 = 1.0 + assert len(rates) == 1 + assert abs(rates[0] - 1.0) < 1e-9 + + def test_steady_rate_uniform_output(self): + """A counter incrementing at a constant rate produces uniform rates.""" + from pmmcp.tools._stats import _compute_rates + + samples = [{"timestamp": float(i * 60), "value": float(i * 120)} for i in range(5)] + rates = _compute_rates(samples) + assert len(rates) == 4 + for r in rates: + assert abs(r - 2.0) < 1e-9 # 120/60 = 2.0 per second + + class TestExpandTimeUnitsInUtils: """Verify _expand_time_units was correctly relocated to utils."""