Add telemetry plugin system for sampling profiler

Lib/profiling/sampling/binary_collector.py

@@ -5,6 +5,7 @@
 import _remote_debugging
 
 from .collector import Collector
+from .telemetry.chunks import TelemetryChunkWriter
 
 # Compression type constants (must match binary_io.h)
 COMPRESSION_NONE = 0
@@ -67,6 +68,21 @@ def __init__(self, filename, sample_interval_usec, *, skip_idle=False,
         self._writer = _remote_debugging.BinaryWriter(
             filename, sample_interval_usec, start_time_us, compression=compression_type
         )
+        self._telemetry_chunk_writer = None
+
+    def set_plugin_enabled(self, plugin_id):
+        if plugin_id and self._telemetry_chunk_writer is None:
+            self._telemetry_chunk_writer = TelemetryChunkWriter(self.filename)
+
+    def set_plugin_metadata(self, plugin_id, metadata):
+        if self._telemetry_chunk_writer is None:
+            return
+        self._telemetry_chunk_writer.write_record(plugin_id, "metadata", metadata)
+
+    def collect_plugin_event(self, plugin_id, event_type, payload):
+        if self._telemetry_chunk_writer is None:
+            return
+        self._telemetry_chunk_writer.write_record(plugin_id, event_type, payload)
 
     def collect(self, stack_frames, timestamp_us=None):
         """Collect profiling data from stack frames.
@@ -94,6 +110,8 @@ def export(self, filename=None):
             filename: Ignored (binary files are written incrementally)
         """
         self._writer.finalize()
+        if self._telemetry_chunk_writer is not None:
+            self._telemetry_chunk_writer.close()
 
     @property
     def total_samples(self):
@@ -115,6 +133,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         """Context manager exit - finalize unless there was an error."""
         if exc_type is None:
             self._writer.finalize()
+            if self._telemetry_chunk_writer is not None:
+                self._telemetry_chunk_writer.close()
         else:
             self._writer.close()
+            if self._telemetry_chunk_writer is not None:
+                self._telemetry_chunk_writer.close()
         return False

Lib/profiling/sampling/binary_reader.py

@@ -5,6 +5,7 @@
 from .gecko_collector import GeckoCollector
 from .stack_collector import FlamegraphCollector, CollapsedStackCollector
 from .pstats_collector import PstatsCollector
+from .telemetry import replay_sidecar_to_sink
 
 
 class BinaryReader:
@@ -70,7 +71,9 @@ def replay_samples(self, collector, progress_callback=None):
         """
         if self._reader is None:
             raise RuntimeError("Reader not open. Use as context manager.")
-        return self._reader.replay(collector, progress_callback)
+        replayed = self._reader.replay(collector, progress_callback)
+        replay_sidecar_to_sink(self.filename, collector)
+        return replayed
 
     @property
     def sample_count(self):

Lib/profiling/sampling/cli.py

@@ -21,6 +21,7 @@
 from .gecko_collector import GeckoCollector
 from .binary_collector import BinaryCollector
 from .binary_reader import BinaryReader
+from .telemetry.plugin_registry import resolve_helper_config
 from .constants import (
     MICROSECONDS_PER_SECOND,
     PROFILING_MODE_ALL,
@@ -159,6 +160,11 @@ def _build_child_profiler_args(args):
     if mode != "wall":
         child_args.extend(["--mode", mode])
 
+    # Generic telemetry plugin options
+    child_plugins = getattr(args, "plugins", None) or []
+    for plugin_id in child_plugins:
+        child_args.extend(["--plugin", plugin_id])
+
     # Format options (skip pstats as it's the default)
     if args.format != "pstats":
         child_args.append(f"--{args.format}")
@@ -416,8 +422,13 @@ def _add_sampling_options(parser):
         "Uses thread_suspend on macOS and ptrace on Linux. Adds overhead but ensures memory "
         "reads are from a frozen state.",
     )
-
-
+    sampling_group.add_argument(
+        "--plugin",
+        dest="plugins",
+        action="append",
+        default=[],
+        help="Enable a telemetry plugin (repeatable)",
+    )
 def _add_mode_options(parser):
     """Add mode options to a parser."""
     mode_group = parser.add_argument_group("Mode options")
@@ -789,6 +800,10 @@ def _validate_args(args, parser):
     if getattr(args, 'command', None) == "replay":
         return
 
+    # Deduplicate while preserving order
+    if getattr(args, "plugins", None):
+        args.plugins = list(dict.fromkeys(args.plugins))
+
     # Warn about blocking mode with aggressive sampling intervals
     if args.blocking and args.sample_interval_usec < 100:
         print(
@@ -859,6 +874,10 @@ def _validate_args(args, parser):
             )
         return
 
+    # Non-live GPU integration currently supports binary capture only.
+    if getattr(args, "plugins", None) and not args.live and getattr(args, "format", "pstats") != "binary":
+        parser.error("Telemetry plugins currently require --binary for non-live mode.")
+
     # Validate gecko mode doesn't use non-wall mode
     if args.format == "gecko" and getattr(args, 'mode', 'wall') != "wall":
         parser.error(
@@ -1067,6 +1086,7 @@ def _handle_attach(args):
         compression=getattr(args, 'compression', 'auto'),
         diff_baseline=args.diff_baseline
     )
+    telemetry_plugins = _build_telemetry_plugins(args)
 
     with _get_child_monitor_context(args, args.pid):
         collector = sample(
@@ -1081,6 +1101,7 @@ def _handle_attach(args):
             gc=args.gc,
             opcodes=args.opcodes,
             blocking=args.blocking,
+            telemetry_plugins=telemetry_plugins,
         )
         _handle_output(collector, args, args.pid, mode)
 
@@ -1146,6 +1167,7 @@ def _handle_run(args):
         compression=getattr(args, 'compression', 'auto'),
         diff_baseline=args.diff_baseline
     )
+    telemetry_plugins = _build_telemetry_plugins(args)
 
     with _get_child_monitor_context(args, process.pid):
         try:
@@ -1161,6 +1183,7 @@ def _handle_run(args):
                 gc=args.gc,
                 opcodes=args.opcodes,
                 blocking=args.blocking,
+                telemetry_plugins=telemetry_plugins,
             )
             _handle_output(collector, args, process.pid, mode)
         finally:
@@ -1192,7 +1215,9 @@ def _handle_live_attach(args, pid):
         mode=mode,
         opcodes=args.opcodes,
         async_aware=args.async_mode if args.async_aware else None,
+        telemetry_plugin_ids=getattr(args, "plugins", []),
     )
+    telemetry_plugins = _build_telemetry_plugins(args)
 
     # Sample in live mode
     sample_live(
@@ -1207,6 +1232,7 @@ def _handle_live_attach(args, pid):
         gc=args.gc,
         opcodes=args.opcodes,
         blocking=args.blocking,
+        telemetry_plugins=telemetry_plugins,
     )
 
 
@@ -1239,7 +1265,9 @@ def _handle_live_run(args):
         mode=mode,
         opcodes=args.opcodes,
         async_aware=args.async_mode if args.async_aware else None,
+        telemetry_plugin_ids=getattr(args, "plugins", []),
     )
+    telemetry_plugins = _build_telemetry_plugins(args)
 
     # Profile the subprocess in live mode
     try:
@@ -1255,6 +1283,7 @@ def _handle_live_run(args):
             gc=args.gc,
             opcodes=args.opcodes,
             blocking=args.blocking,
+            telemetry_plugins=telemetry_plugins,
         )
     finally:
         # Clean up the subprocess and get any error output
@@ -1294,5 +1323,13 @@ def _handle_replay(args):
         sys.exit(f"Error: {exc}")
 
 
+def _build_telemetry_plugins(args):
+    plugins = []
+    for plugin_id in getattr(args, "plugins", []) or []:
+        config = resolve_helper_config(plugin_id)
+        plugins.append({"id": plugin_id, "config": config})
+    return plugins or None
+
+
 if __name__ == "__main__":
     main()

Lib/profiling/sampling/live_collector/__init__.py

@@ -110,6 +110,7 @@
     TableWidget,
     FooterWidget,
     HelpWidget,
+    TelemetryPanelWidget,
 )
 from .constants import (
     MICROSECONDS_PER_SECOND,
@@ -137,6 +138,7 @@
     MAX_EFFICIENCY_BAR_WIDTH,
     MIN_SAMPLE_RATE_FOR_SCALING,
     FINISHED_BANNER_EXTRA_LINES,
+    TELEMETRY_PANEL_HEIGHT,
     COLOR_PAIR_HEADER_BG,
     COLOR_PAIR_CYAN,
     COLOR_PAIR_YELLOW,
@@ -162,6 +164,7 @@
     "TableWidget",
     "FooterWidget",
     "HelpWidget",
+    "TelemetryPanelWidget",
     # Constants
     "MICROSECONDS_PER_SECOND",
     "DISPLAY_UPDATE_HZ",
@@ -188,6 +191,7 @@
     "MAX_EFFICIENCY_BAR_WIDTH",
     "MIN_SAMPLE_RATE_FOR_SCALING",
     "FINISHED_BANNER_EXTRA_LINES",
+    "TELEMETRY_PANEL_HEIGHT",
     "COLOR_PAIR_HEADER_BG",
     "COLOR_PAIR_CYAN",
     "COLOR_PAIR_YELLOW",

Lib/profiling/sampling/live_collector/collector.py

@@ -43,10 +43,12 @@
     COLOR_PAIR_MAGENTA,
     COLOR_PAIR_RED,
     COLOR_PAIR_SORTED_HEADER,
+    TELEMETRY_PANEL_HEIGHT,
 )
 from .display import CursesDisplay
-from .widgets import HeaderWidget, TableWidget, FooterWidget, HelpWidget, OpcodePanel
+from .widgets import HeaderWidget, TableWidget, FooterWidget, HelpWidget, OpcodePanel, TelemetryPanelWidget
 from .trend_tracker import TrendTracker
+from ..telemetry.plugin_registry import create_live_plugin
 
 
 @dataclass
@@ -118,6 +120,7 @@ def __init__(
         mode=None,
         opcodes=False,
         async_aware=None,
+        telemetry_plugin_ids=None,
     ):
         """
         Initialize the live stats collector.
@@ -176,6 +179,13 @@ def __init__(
         # Opcode statistics: {location: {opcode: count}}
         self.opcode_stats = collections.defaultdict(lambda: collections.defaultdict(int))
         self.show_opcodes = opcodes  # Show opcode panel when --opcodes flag is passed
+        self.show_telemetry_panel = bool(telemetry_plugin_ids)
+        self.telemetry_plugins = {}  # plugin_id -> plugin runtime
+        self.telemetry_plugin_order = []
+        self.current_telemetry_plugin_idx = 0
+        self.current_telemetry_mode_idx = 0
+        for plugin_id in telemetry_plugin_ids or []:
+            self.set_plugin_enabled(plugin_id)
         self.selected_row = 0  # Currently selected row in table for opcode view
         self.scroll_offset = 0  # Scroll offset for table when in opcode mode
 
@@ -206,6 +216,7 @@ def __init__(
         self.footer_widget = None
         self.help_widget = None
         self.opcode_panel = None
+        self.telemetry_panel = None
 
         # Color mode
         self._can_colorize = _colorize.can_colorize()
@@ -431,12 +442,14 @@ def _prepare_display_data(self, height):
         extra_header_lines = (
             FINISHED_BANNER_EXTRA_LINES if self.finished else 0
         )
+        extra_telemetry_lines = TELEMETRY_PANEL_HEIGHT if self.show_telemetry_panel else 0
         max_stats_lines = max(
             0,
             height
             - HEADER_LINES
             - extra_header_lines
             - FOOTER_LINES
+            - extra_telemetry_lines
             - SAFETY_MARGIN,
         )
         stats_list = stats_list[:max_stats_lines]
@@ -455,6 +468,7 @@ def _initialize_widgets(self, colors):
             self.footer_widget = FooterWidget(self.display, colors, self)
             self.help_widget = HelpWidget(self.display, colors)
             self.opcode_panel = OpcodePanel(self.display, colors, self)
+            self.telemetry_panel = TelemetryPanelWidget(self.display, colors, self)
 
     def _render_display_sections(
         self, height, width, elapsed, stats_list, colors
@@ -480,6 +494,10 @@ def _render_display_sections(
                 line = self.opcode_panel.render(
                     line, width, height=height, stats_list=stats_list
                 )
+            if self.show_telemetry_panel:
+                line = self.telemetry_panel.render(
+                    line, width, height=height
+                )
 
         except curses.error:
             pass
@@ -1003,6 +1021,12 @@ def _handle_input(self):
             # Toggle trend colors on/off
             if self._trend_tracker is not None:
                 self._trend_tracker.toggle()
+        elif ch == ord("g") or ch == ord("G"):
+            self.show_telemetry_panel = not self.show_telemetry_panel
+        elif ch == ord("m") or ch == ord("M"):
+            self._cycle_telemetry_mode()
+        elif ch == ord("n") or ch == ord("N"):
+            self._cycle_telemetry_plugin()
 
         elif ch == ord("j") or ch == ord("J"):
             # Move selection down in opcode mode (with scrolling)
@@ -1109,3 +1133,42 @@ def export(self, filename):
             "Export to file is not supported in live mode. "
             "Use the live TUI to view statistics in real-time."
         )
+
+    def set_plugin_enabled(self, plugin_id):
+        if not plugin_id or plugin_id in self.telemetry_plugins:
+            return
+        plugin = create_live_plugin(plugin_id)
+        if plugin is None:
+            return
+        self.telemetry_plugins[plugin_id] = plugin
+        self.telemetry_plugin_order.append(plugin_id)
+
+    def collect_plugin_event(self, plugin_id, event_type, payload):
+        plugin = self.telemetry_plugins.get(plugin_id)
+        if plugin is None:
+            self.set_plugin_enabled(plugin_id)
+            plugin = self.telemetry_plugins.get(plugin_id)
+        if plugin is not None:
+            plugin.ingest(event_type, payload)
+
+    def _get_current_telemetry_plugin(self):
+        if not self.telemetry_plugin_order:
+            return None, None
+        idx = self.current_telemetry_plugin_idx % len(self.telemetry_plugin_order)
+        plugin_id = self.telemetry_plugin_order[idx]
+        return plugin_id, self.telemetry_plugins.get(plugin_id)
+
+    def _cycle_telemetry_mode(self):
+        _, plugin = self._get_current_telemetry_plugin()
+        if plugin is None:
+            return
+        modes = list(getattr(plugin, "panel_modes", ("default",)))
+        if not modes:
+            return
+        self.current_telemetry_mode_idx = (self.current_telemetry_mode_idx + 1) % len(modes)
+
+    def _cycle_telemetry_plugin(self):
+        if not self.telemetry_plugin_order:
+            return
+        self.current_telemetry_plugin_idx = (self.current_telemetry_plugin_idx + 1) % len(self.telemetry_plugin_order)
+        self.current_telemetry_mode_idx = 0

Lib/profiling/sampling/live_collector/constants.py

@@ -45,8 +45,9 @@
 # Finished banner display
 FINISHED_BANNER_EXTRA_LINES = 3  # Blank line + banner + blank line
 
-# Opcode panel display
+# Panel display
 OPCODE_PANEL_HEIGHT = 12  # Height reserved for opcode statistics panel
+TELEMETRY_PANEL_HEIGHT = 10  # Height reserved for telemetry plugin panel
 
 # Color pair IDs
 COLOR_PAIR_SAMPLES = 1