Merge pull request #314 from ABHIRAMSHIBU/fix/replay-cached-metrics-on-mci-defer

Handler: replay cached metrics when minimal_collect_interval defers a scrape

Author: akpw

mktxp/flow/collector_handler.py

@@ -14,7 +14,7 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from timeit import default_timer
 from datetime import datetime
-from threading import Event, Timer
+from threading import Event, Lock, Timer
 from mktxp.cli.config.config import config_handler
 from mktxp.cli.config.config import MKTXPConfigKeys
 
@@ -26,6 +26,8 @@ def __init__(self, entries_handler, collector_registry):
         self.entries_handler = entries_handler
         self.collector_registry = collector_registry
         self.last_collect_timestamp = 0
+        self._metrics_cache = []
+        self._cache_lock = Lock()
 
 
     def collect_sync(self):
@@ -118,19 +120,32 @@ def timeout(timeout_event):
 
     def collect(self):
         if not self._valid_collect_interval():
+            # Within minimal_collect_interval: replay the last successful collection
+            # so concurrent or closely-spaced scrapers do not see an empty registry
+            # (which would otherwise drop every mktxp_* series for that scrape and
+            # surface as phantom "down" gaps in dashboards).
+            with self._cache_lock:
+                cached = list(self._metrics_cache)
+            yield from cached
             return
 
         # bandwidth collector
-        yield from self.collector_registry.bandwidthCollector.collect()
+        collected = list(self.collector_registry.bandwidthCollector.collect())
 
         # all other collectors
         # Check whether to run in parallel by looking at the mktxp system configuration
         parallel = config_handler.system_entry.fetch_routers_in_parallel
         max_worker_threads = config_handler.system_entry.max_worker_threads
         if parallel:
-            yield from self.collect_async(max_worker_threads=max_worker_threads)
+            collected.extend(self.collect_async(max_worker_threads=max_worker_threads))
         else:
-            yield from self.collect_sync()
+            collected.extend(self.collect_sync())
+
+        if collected:
+            with self._cache_lock:
+                self._metrics_cache = collected
+
+        yield from collected
 
     def _valid_collect_interval(self):
         now = datetime.now().timestamp()

tests/flow/test_collector_handler.py

@@ -59,3 +59,94 @@ def test_collect_sync_lifecycle(is_ready, is_done_called, mock_entries_handler,
         mock_router_entry.is_done.assert_called_once()
     else:
         mock_router_entry.is_done.assert_not_called()
+
+
+
+class _StubMetric:
+    """Identifiable placeholder for a yielded metric family."""
+    def __init__(self, label):
+        self.label = label
+    def __eq__(self, other):
+        return isinstance(other, _StubMetric) and other.label == self.label
+    def __hash__(self):
+        return hash((type(self), self.label))
+    def __repr__(self):
+        return f'_StubMetric({self.label!r})'
+
+
+def _system_entry_stub(mci):
+    """Build a stand-in for config_handler.system_entry that the handler reads."""
+    return type('SysEntry', (), {
+        'minimal_collect_interval': mci,
+        'fetch_routers_in_parallel': False,
+        'max_worker_threads': 1,
+        'verbose_mode': False,
+    })()
+
+
+def _make_handler(per_call_metrics):
+    from mktxp.flow.collector_handler import CollectorHandler
+
+    router_entry = MagicMock()
+    router_entry.is_ready.return_value = True
+    router_entry.time_spent = {}
+
+    entries_handler = MagicMock()
+    entries_handler.router_entries = [router_entry]
+
+    registry = MagicMock()
+    registry.bandwidthCollector.collect.return_value = []
+
+    iterator = iter(per_call_metrics)
+    def collect_func(_entry):
+        try:
+            yield from next(iterator)
+        except StopIteration:
+            return
+
+    registry.registered_collectors = {'mock_collector': collect_func}
+    return CollectorHandler(entries_handler, registry)
+
+
+def test_collect_caches_and_replays_on_mci_defer(monkeypatch):
+    """When MCI defers, the handler must yield the cached metric families
+    instead of an empty registry — otherwise Prometheus drops every mktxp_*
+    series for that scrape and dashboards show false-down."""
+    from mktxp.flow import collector_handler as ch_mod
+
+    fresh = [_StubMetric('a'), _StubMetric('b')]
+    handler = _make_handler([fresh])
+
+    fake_cfg = MagicMock()
+    fake_cfg.system_entry = _system_entry_stub(mci=5)
+    monkeypatch.setattr(ch_mod, 'config_handler', fake_cfg)
+
+    first = list(handler.collect())
+    assert first == fresh, 'first scrape should yield freshly collected metrics'
+
+    # Second call lands well within MCI; the inner collector iterator is
+    # exhausted, so without the cache the result would be empty.
+    second = list(handler.collect())
+    assert second == fresh, 'within-MCI scrape must replay the cached metrics'
+
+
+def test_empty_collection_does_not_clobber_cache(monkeypatch):
+    """If a fresh collection produces nothing (e.g., all routers not_ready),
+    the cache must be preserved so the next within-MCI scrape still has data."""
+    from mktxp.flow import collector_handler as ch_mod
+
+    seed = [_StubMetric('seed')]
+    handler = _make_handler([seed, []])
+
+    fake_cfg = MagicMock()
+    fake_cfg.system_entry = _system_entry_stub(mci=0)
+    monkeypatch.setattr(ch_mod, 'config_handler', fake_cfg)
+
+    assert list(handler.collect()) == seed
+
+    # Force a successful but empty collection (MCI=0 → always runs fresh).
+    assert list(handler.collect()) == []
+
+    # Bump MCI so the next call defers; cache should still hold the seed.
+    fake_cfg.system_entry = _system_entry_stub(mci=60)
+    assert list(handler.collect()) == seed