diff --git a/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py b/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py index c243877ac..37eff339f 100644 --- a/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py +++ b/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py @@ -225,11 +225,27 @@ def _write_snapshot(self, path, result_df): self._cache_stats["write_errors"] += 1 log.warning("xorq stat snapshot write failed for %s: %s", path, e) - def _log_cache_stats(self): - """One summary line per run when a snapshot cache is in play (#910).""" + def _log_cache_stats(self, span=None): + """Surface the per-run snapshot-cache outcome (#910, #951). + + Two channels, so the write side is never invisible: + + * ``span`` — attach the hit/miss/snapshot/byte/write-error counts to the + run's ``stat.xorq.total`` telemetry span. A bound telemetry sink then + carries the cache outcome (including ``write_errors``) to the operator + without a debug log build — the ``log.info`` line below lands in a + server log file no deployment reads (#951). + * ``log`` — one summary line per run for a local perf/debug session.""" if self.cache_storage is None: return s = self._cache_stats + if span is not None: + cs = self.cache_run_stats() + span.set_attr( + cache_status=cs["status"], cache_hits=cs["hits"], + cache_misses=cs["misses"], cache_secs=cs["secs"], + cache_snapshots=cs["snapshots"], cache_bytes=cs["bytes"], + cache_write_errors=cs["write_errors"]) base_path = getattr(getattr(self.cache_storage, "storage", None), "base_path", "?") log.info( "xorq stat cache [%s]: %d hit(s), %d miss(es), %d snapshot(s) " @@ -318,12 +334,12 @@ def process_table(self, table, skip_columns=None) -> Tuple[SDType, List[StatErro self._perf = (perf_log.PerfRecorder() if perf_log.enabled() and not self._suppress_perf_summary else None) _t0 = time.perf_counter() - with self._span("stat.xorq.total"): + with self._span("stat.xorq.total") as span: try: return self._process_table_impl(table, skip_columns=skip_columns) finally: self._cache_stats["secs"] = round(time.perf_counter() - _t0, 4) - self._log_cache_stats() + self._log_cache_stats(span) if self._perf is not None: self._perf.label = ( f"xorq cols={len(table.columns)} " diff --git a/buckaroo/xorq_buckaroo.py b/buckaroo/xorq_buckaroo.py index 75e8050b8..36f18687b 100644 --- a/buckaroo/xorq_buckaroo.py +++ b/buckaroo/xorq_buckaroo.py @@ -173,16 +173,22 @@ def _get_summary_sd(self, processed_df): processed_df, self.analysis_klasses, self.df_name, debug=self.debug, cache_storage=cache_storage, skip_columns=getattr(self, 'skip_stat_columns', None)) - # Attach the summary-stat cache hit/miss/timing signal (#944) to the - # span so a telemetry consumer learns whether the stats were cached - # — the one signal only the server observes (#943). The pandas branch - # above already returned, so stats here is always an XorqDfStatsV2, - # which always exposes cache_run_stats(); call it unconditionally so - # a missing signal fails loudly rather than silently vanishing. + # Attach the summary-stat cache signal (#944) to the span so a + # telemetry consumer learns whether the stats were cached — the one + # signal only the server observes (#943). Carries the write side too + # (snapshots/bytes/write_errors, #951) so a cache that stops writing — + # or a run with write_errors > 0 — is visible to a telemetry consumer + # rather than only to a server log no deployment reads. The pandas + # branch above already returned, so stats here is always an + # XorqDfStatsV2, which always exposes cache_run_stats(); call it + # unconditionally so a missing signal fails loudly rather than + # silently vanishing. cs = stats.cache_run_stats() span.set_attr( cache_status=cs.get("status"), cache_hits=cs.get("hits"), - cache_misses=cs.get("misses"), cache_secs=cs.get("secs")) + cache_misses=cs.get("misses"), cache_secs=cs.get("secs"), + cache_snapshots=cs.get("snapshots"), cache_bytes=cs.get("bytes"), + cache_write_errors=cs.get("write_errors")) sdf = stats.sdf if stats.errs: if self.debug: diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 0150500cd..849dd041e 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -445,6 +445,12 @@ async def test_load_expr_telemetry_emits_session_correlated_spans(self): self.assertEqual(attrs["cache_status"], "miss") self.assertEqual(attrs["cache_hits"], 0) self.assertGreater(attrs["cache_misses"], 0) + # The write side rides the span too (#951): the cold miss writes + # snapshots with no write errors, so a cache that stops writing — or a + # run with write_errors > 0 — is visible to the telemetry consumer. + self.assertGreater(attrs["cache_snapshots"], 0) + self.assertGreater(attrs["cache_bytes"], 0) + self.assertEqual(attrs["cache_write_errors"], 0) finally: shutil.rmtree(builds_root, ignore_errors=True) shutil.rmtree(cache_root, ignore_errors=True) diff --git a/tests/unit/test_xorq_stats_v2.py b/tests/unit/test_xorq_stats_v2.py index 6e33d6e30..3565ef385 100644 --- a/tests/unit/test_xorq_stats_v2.py +++ b/tests/unit/test_xorq_stats_v2.py @@ -831,6 +831,31 @@ def test_internal_spans_emit_to_telemetry_sink_with_perf_off(self): f"stat.xorq.total span must emit to the sink with perf logging off; " f"got {names}") + def test_cache_outcome_rides_total_span_to_telemetry(self, tmp_path): + """#951: the per-run snapshot-cache outcome — status, snapshots, bytes, + write_errors — rides the stat.xorq.total telemetry span, so a telemetry + consumer sees the write side without reading a server log. + + Before the fix _log_cache_stats emitted only a log.info line the server + never surfaced, so a cache that stopped writing was invisible.""" + records: list = [] + saved = perf_log._ENABLED + perf_log._ENABLED = False + try: + pipeline = XorqStatPipeline( + XORQ_STATS_V2, unit_test=False, cache_storage=self._cache(tmp_path)) + with perf_log.telemetry_context("sess-cache-write", records.append): + pipeline.process_table(self._filter_chain_table()) + finally: + perf_log._ENABLED = saved + total = next(r for r in records if r["name"] == "stat.xorq.total") + attrs = total["attrs"] + # Cold run against a fresh cache dir → a pure miss that writes snapshots. + assert attrs["cache_status"] == "miss" + assert attrs["cache_snapshots"] > 0 + assert attrs["cache_bytes"] > 0 + assert attrs["cache_write_errors"] == 0 + def test_cached_stats_match_uncached(self, tmp_path): """The cold (compute + write) and warm (snapshot-read) cache paths produce the same stats as a plain uncached run — the cache is a perf