jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108import importlib.utilimport jsonimport pathlibimport unittest
SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "tools/check_dashboard.py"SPEC = importlib.util.spec_from_file_location("check_dashboard", SCRIPT)assert SPEC and SPEC.loadercheck_dashboard = importlib.util.module_from_spec(SPEC)SPEC.loader.exec_module(check_dashboard)
class DashboardContractTest(unittest.TestCase): def test_pinned_dashboard_checksums_and_import_families(self): for path, expected in check_dashboard.EXPECTED_SHA256.items(): self.assertEqual(expected, check_dashboard.hashlib.sha256(path.read_bytes()).hexdigest()) raw = check_dashboard.DEFAULT_DASHBOARD.read_bytes() required = check_dashboard.dashboard_families(raw) self.assertIn("jetstream_import_jobs_total", required) self.assertIn("jetstream_import_rows_rejected_total", required)
def test_stream_runtime_row_has_no_go_or_fake_network_families(self): required = check_dashboard.dashboard_families(check_dashboard.DEFAULT_DASHBOARD.read_bytes()) self.assertFalse(any(name.startswith("go_") for name in required)) self.assertNotIn("process_network_receive_bytes_total", required) self.assertNotIn("process_network_transmit_bytes_total", required) self.assertIn("process_virtual_memory_bytes", required) self.assertIn("stream_process_threads", required) self.assertIn("process_major_page_faults_total", required) self.assertIn("process_involuntary_context_switches_total", required)
def test_non_runtime_panels_are_upstream_exact(self): upstream = json.loads(check_dashboard.UPSTREAM_DASHBOARD.read_bytes()) stream = json.loads(check_dashboard.DEFAULT_DASHBOARD.read_bytes()) runtime_ids = {69, 70, 71, 72, 73, 74, 75} stream_only_ids = {76, 77, 78, 79, 80, 81} upstream_panels = {panel["id"]: panel for panel in upstream["panels"] if panel["id"] not in runtime_ids} stream_panels = { panel["id"]: panel for panel in stream["panels"] if panel["id"] not in runtime_ids | stream_only_ids } self.assertEqual(upstream_panels, stream_panels)
def test_stream_pipeline_diagnostics_are_explicit_additions(self): stream = json.loads(check_dashboard.DEFAULT_DASHBOARD.read_bytes()) panels = {panel["id"]: panel for panel in stream["panels"]} self.assertEqual("Stream pipeline diagnostics", panels[76]["title"]) self.assertEqual("Pipeline tickets", panels[77]["title"]) self.assertEqual("Pipeline stage gaps", panels[78]["title"]) self.assertEqual("Pipeline terminal failures", panels[79]["title"]) required = check_dashboard.dashboard_families(check_dashboard.DEFAULT_DASHBOARD.read_bytes()) self.assertTrue({ "stream_pipeline_ticket", "stream_pipeline_stage_gap", "stream_pipeline_fatal", "stream_pipeline_failures_total", }.issubset(required))
def test_stream_progress_chart_names_upstream_batch_semantics(self): stream = json.loads(check_dashboard.DEFAULT_DASHBOARD.read_bytes()) stream_panels = {panel["id"]: panel for panel in stream["panels"]} progress = stream_panels[81]
self.assertEqual("timeseries", progress["type"]) self.assertEqual("Backfill repositories over time", progress["title"]) self.assertIn("pages of 1,000", progress["description"]) self.assertIn("100,000", progress["description"]) self.assertIn("not a percentage of the full network", progress["description"]) self.assertEqual( [ "jetstream_backfill_discovered_total", "jetstream_backfill_completion_queued_total", # The series legended "durably committed" must come from the # metadata store, not from a process-local counter that resets # on restart. Reading the latter under that label is what made a # restart look like progress in July. "stream_backfill_repos_durable", ], [ check_dashboard.METRIC_RE.findall(target["expr"])[0] for target in progress["targets"] ], ) self.assertEqual( ["discovered", "processed", "durably committed"], [target["legendFormat"] for target in progress["targets"]], ) self.assertTrue(all(target["range"] for target in progress["targets"])) self.assertTrue(all(not target["instant"] for target in progress["targets"])) self.assertEqual(0, progress["fieldConfig"]["defaults"]["min"]) self.assertNotIn("max", progress["fieldConfig"]["defaults"]) self.assertEqual("short", progress["fieldConfig"]["defaults"]["unit"])
def test_histogram_type_expands_queryable_children(self): present = check_dashboard.scrape_families( "# TYPE request_seconds histogram\nrequest_seconds_bucket{le=\"1\"} 2\n" ) self.assertEqual( {"request_seconds", "request_seconds_bucket", "request_seconds_count", "request_seconds_sum"}, present, )
if __name__ == "__main__": unittest.main()