/
/
/
1"""Profiler provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import collections
7import logging
8import os
9import platform
10import shutil
11import time
12import tracemalloc
13from collections import deque
14from typing import TYPE_CHECKING, Any
15
16import psutil
17import yappi
18from music_assistant_models.auth import Scope
19from music_assistant_models.config_entries import ConfigActionResult, ConfigEntry
20from music_assistant_models.enums import ConfigEntryType
21
22from music_assistant.helpers.datetime import utc
23from music_assistant.models.plugin import PluginProvider
24
25from .helpers import (
26 LogErrorCounter,
27 collect_memory_stats,
28 collect_object_census,
29 collect_task_stats,
30 collect_tracemalloc_stats,
31 extract_cpu_profile,
32 finalize_recorder_entry,
33 persist_report,
34)
35
36if TYPE_CHECKING:
37 from collections.abc import Callable
38
39 from music_assistant_models.event import MassEvent
40
41CONF_CPU_PROFILE_ENABLED = "cpu_profile_enabled"
42CONF_CPU_PROFILE_DURATION = "cpu_profile_duration"
43CONF_CPU_PROFILE_INTERVAL = "cpu_profile_interval"
44CONF_TRACEMALLOC_ENABLED = "tracemalloc_enabled"
45CONF_ACTION_RUN_CPU_PROFILE = "action_run_cpu_profile"
46
47REPORT_FORMAT_VERSION = 1
48LAG_MONITOR_INTERVAL = 0.5
49RECORDER_INTERVAL = 10
50# 24 hours of history at the 10 second sample interval (roughly 1-2 MB of memory)
51RECORDER_MAX_ENTRIES = 8640
52CPU_PROFILE_FIRST_DELAY = 60
53CPU_PROFILE_MIN_DURATION = 10
54CPU_PROFILE_MAX_DURATION = 300
55CPU_PROFILE_MIN_INTERVAL = 5 # minutes
56CPU_PROFILE_MAX_INTERVAL = 1440 # minutes
57
58
59class ProfilerProvider(PluginProvider):
60 """
61 Plugin provider that continuously records performance diagnostics.
62
63 While loaded it runs a lightweight flight recorder, an event-loop lag
64 monitor and event/error counters, plus (optional) periodic CPU profile
65 windows. All collected data is aggregated into a shareable report via
66 the `profiler/report` API command.
67 """
68
69 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
70 """Return Config entries to configure this provider."""
71 return (
72 ConfigEntry(
73 key="profiler_note",
74 type=ConfigEntryType.LABEL,
75 required=False,
76 ),
77 ConfigEntry(
78 key=CONF_CPU_PROFILE_ENABLED,
79 type=ConfigEntryType.BOOLEAN,
80 default_value=True,
81 required=False,
82 ),
83 ConfigEntry(
84 key=CONF_CPU_PROFILE_DURATION,
85 type=ConfigEntryType.INTEGER,
86 default_value=60,
87 range=(CPU_PROFILE_MIN_DURATION, CPU_PROFILE_MAX_DURATION),
88 required=False,
89 advanced=True,
90 depends_on=CONF_CPU_PROFILE_ENABLED,
91 ),
92 ConfigEntry(
93 key=CONF_CPU_PROFILE_INTERVAL,
94 type=ConfigEntryType.INTEGER,
95 default_value=30,
96 range=(CPU_PROFILE_MIN_INTERVAL, CPU_PROFILE_MAX_INTERVAL),
97 required=False,
98 advanced=True,
99 depends_on=CONF_CPU_PROFILE_ENABLED,
100 ),
101 ConfigEntry(
102 key=CONF_ACTION_RUN_CPU_PROFILE,
103 type=ConfigEntryType.ACTION,
104 action=CONF_ACTION_RUN_CPU_PROFILE,
105 required=False,
106 ),
107 ConfigEntry(
108 key=CONF_TRACEMALLOC_ENABLED,
109 type=ConfigEntryType.BOOLEAN,
110 default_value=False,
111 required=False,
112 ),
113 )
114
115 async def handle_config_action(
116 self, action: str
117 ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
118 """Handle a one-shot config action button press."""
119 if action == CONF_ACTION_RUN_CPU_PROFILE:
120 self.start_cpu_profile()
121 return None
122 return await super().handle_config_action(action)
123
124 async def handle_async_init(self) -> None:
125 """Handle async initialization of the provider."""
126 self._proc = psutil.Process()
127 # prime the cpu percent counter so subsequent calls return meaningful values
128 self._proc.cpu_percent()
129 self._out_dir = os.path.join(self.mass.storage_path, "profiler")
130 await asyncio.to_thread(os.makedirs, self._out_dir, exist_ok=True)
131 self._loaded_at = time.time()
132 self._recorder_ring: deque[dict[str, Any]] = deque(maxlen=RECORDER_MAX_ENTRIES)
133 self._lag_samples: deque[float] = deque(maxlen=120)
134 self._lag_window_max = 0.0
135 self._lag_max_since_load = 0.0
136 self._event_counts: collections.Counter[str] = collections.Counter()
137 self._events_since_sample = 0
138 self._errors_at_last_sample = 0
139 self._log_counter = LogErrorCounter()
140 self._last_cpu_profile: dict[str, Any] | None = None
141 self._cpu_profile_running = False
142 self._tracemalloc_prev: tracemalloc.Snapshot | None = None
143 self._started_tracemalloc = False
144 self._unsubscribe_events: Callable[[], None] | None = None
145 self._unregister_api: Callable[[], None] | None = None
146
147 async def loaded_in_mass(self) -> None:
148 """Start all measurements once the provider is fully loaded."""
149 await super().loaded_in_mass()
150 logging.getLogger().addHandler(self._log_counter)
151 self._unsubscribe_events = self.mass.subscribe(self._on_mass_event)
152 self._unregister_api = self.mass.register_api_command(
153 "profiler/report", self.get_report, required_scope=Scope.SYSTEM_MANAGE
154 )
155 if (
156 self.get_config_value(CONF_TRACEMALLOC_ENABLED, False, return_type=bool)
157 and not tracemalloc.is_tracing()
158 ):
159 tracemalloc.start(15)
160 self._started_tracemalloc = True
161 self.mass.create_task(self._loop_lag_monitor(), task_id="profiler_lag_monitor")
162 self.mass.create_task(self._flight_recorder(), task_id="profiler_flight_recorder")
163 if self.get_config_value(CONF_CPU_PROFILE_ENABLED, True, return_type=bool):
164 self.mass.create_task(self._cpu_profile_scheduler(), task_id="profiler_cpu_scheduler")
165
166 async def unload(self, is_removed: bool = False) -> None:
167 """Stop all measurements and clean up."""
168 for task_id in (
169 "profiler_lag_monitor",
170 "profiler_flight_recorder",
171 "profiler_cpu_scheduler",
172 "profiler_cpu_window",
173 ):
174 self.mass.cancel_task(task_id)
175 if self._unregister_api is not None:
176 self._unregister_api()
177 self._unregister_api = None
178 if self._unsubscribe_events is not None:
179 self._unsubscribe_events()
180 self._unsubscribe_events = None
181 logging.getLogger().removeHandler(self._log_counter)
182 if yappi.is_running():
183 yappi.stop()
184 yappi.clear_stats()
185 if self._started_tracemalloc and tracemalloc.is_tracing():
186 tracemalloc.stop()
187 self._started_tracemalloc = False
188 self._tracemalloc_prev = None
189 if is_removed:
190 await asyncio.to_thread(shutil.rmtree, self._out_dir, ignore_errors=True)
191 await super().unload(is_removed)
192
193 async def get_report(
194 self,
195 markdown: bool = False,
196 include_object_census: bool = False,
197 recorder_minutes: int = 30,
198 ) -> dict[str, Any] | str:
199 """
200 Generate the full diagnostics report (also written to the profiler storage folder).
201
202 :param markdown: Return the report as markdown text instead of a JSON object.
203 :param include_object_census: Include a census of all live Python objects by type.
204 This walks the entire object heap, which can stall the server for several
205 seconds on large installations - only use this when hunting a memory leak.
206 :param recorder_minutes: Minutes of flight-recorder history to include (1-1440).
207 """
208 report = await self._build_report(include_object_census, recorder_minutes)
209 report_md = await asyncio.to_thread(persist_report, self._out_dir, report)
210 return report_md if markdown else report
211
212 def start_cpu_profile(self) -> None:
213 """Start a single on-demand CPU profile window (ignored if one is already running)."""
214 self.mass.create_task(self._run_cpu_profile_window(), task_id="profiler_cpu_window")
215
216 async def _build_report(
217 self, include_object_census: bool, recorder_minutes: int
218 ) -> dict[str, Any]:
219 """Assemble the report dict from all collectors."""
220 memory = await asyncio.to_thread(collect_memory_stats, self._proc)
221 memory["asyncio_tasks"] = len(asyncio.all_tasks(self.mass.loop))
222 memory["tracked_tasks"] = len(self.mass._tracked_tasks)
223 memory["tracked_timers"] = len(self.mass._tracked_timers)
224 memory["event_subscribers"] = len(self.mass._subscribers)
225 if include_object_census:
226 memory["object_census_top"] = await asyncio.to_thread(collect_object_census)
227 if tracemalloc.is_tracing():
228 memory["tracemalloc"], self._tracemalloc_prev = await asyncio.to_thread(
229 collect_tracemalloc_stats, self._tracemalloc_prev
230 )
231 recorder_minutes = max(1, min(recorder_minutes, 1440))
232 cutoff = time.time() - recorder_minutes * 60
233 lag_samples = list(self._lag_samples)
234 return {
235 "report_format_version": REPORT_FORMAT_VERSION,
236 "generated_at": utc().isoformat(timespec="seconds"),
237 "server": {
238 "version": self.mass.version,
239 "python": platform.python_version(),
240 "platform": platform.platform(),
241 "machine": platform.machine(),
242 "uptime_s": int(time.time() - self._proc.create_time()),
243 "profiler_loaded_for_s": int(time.time() - self._loaded_at),
244 "running_as_hass_addon": self.mass.running_as_hass_addon,
245 "tracemalloc_active": tracemalloc.is_tracing(),
246 },
247 "config_summary": {
248 "providers_by_type": dict(
249 collections.Counter(prov.type.value for prov in self.mass.providers)
250 ),
251 "provider_domains": sorted({prov.domain for prov in self.mass.providers}),
252 "players_total": len(self.mass.players.all_players(True, True)),
253 "players_available": len(self.mass.players.all_players(False, False)),
254 "web_clients_connected": len(self.mass.webserver.clients),
255 "library_counts": await self._get_library_counts(),
256 },
257 "memory": memory,
258 "event_loop": {
259 "sample_interval_s": LAG_MONITOR_INTERVAL,
260 "lag_avg_ms_1min": (
261 round(sum(lag_samples) / len(lag_samples), 2) if lag_samples else None
262 ),
263 "lag_max_ms_1min": round(max(lag_samples), 2) if lag_samples else None,
264 "lag_max_ms_since_load": round(self._lag_max_since_load, 2),
265 },
266 "asyncio_tasks": collect_task_stats(asyncio.all_tasks(self.mass.loop)),
267 "events": {
268 "total_since_load": sum(self._event_counts.values()),
269 "per_type_top": [
270 {"event": event, "count": count}
271 for event, count in self._event_counts.most_common(30)
272 ],
273 },
274 "log_errors": self._log_counter.summarize(),
275 "cpu_profile": self._last_cpu_profile,
276 "flight_recorder": {
277 "sample_interval_s": RECORDER_INTERVAL,
278 "window_minutes": recorder_minutes,
279 "entries": [entry for entry in self._recorder_ring if entry["ts_unix"] >= cutoff],
280 },
281 }
282
283 async def _get_library_counts(self) -> dict[str, int]:
284 """Return the number of library items per media type."""
285 counts: dict[str, int] = {}
286 for controller in (
287 self.mass.music.artists,
288 self.mass.music.albums,
289 self.mass.music.tracks,
290 self.mass.music.playlists,
291 self.mass.music.radio,
292 self.mass.music.audiobooks,
293 self.mass.music.podcasts,
294 ):
295 # raw table sizes: a profiler report needs true totals, not the filtered
296 # subset visible to the (admin) user who requested it
297 counts[controller.media_type.value] = await self.mass.music.database.get_count(
298 controller.db_table
299 )
300 return counts
301
302 def _on_mass_event(self, event: MassEvent) -> None:
303 """Count the event (cheap increment, runs for every event on the bus)."""
304 self._event_counts[event.event.value] += 1
305 self._events_since_sample += 1
306
307 async def _loop_lag_monitor(self) -> None:
308 """Continuously sample event-loop scheduling delay via sleep drift."""
309 while True:
310 start = time.monotonic()
311 await asyncio.sleep(LAG_MONITOR_INTERVAL)
312 lag_ms = max((time.monotonic() - start - LAG_MONITOR_INTERVAL) * 1000, 0.0)
313 self._lag_samples.append(lag_ms)
314 self._lag_window_max = max(self._lag_window_max, lag_ms)
315 self._lag_max_since_load = max(self._lag_max_since_load, lag_ms)
316
317 async def _flight_recorder(self) -> None:
318 """Sample server health every few seconds into the bounded in-memory ring."""
319 csv_path = os.path.join(self._out_dir, "stats.csv")
320 while True:
321 await asyncio.sleep(RECORDER_INTERVAL)
322 try:
323 entry = self._collect_recorder_entry()
324 entry = await asyncio.to_thread(
325 finalize_recorder_entry, self._proc, entry, csv_path
326 )
327 self._recorder_ring.append(entry)
328 except Exception as err:
329 self.logger.debug("Flight recorder sample failed: %s", err)
330
331 def _collect_recorder_entry(self) -> dict[str, Any]:
332 """Collect the event-loop side of a flight-recorder sample."""
333 lag_samples = list(self._lag_samples)
334 entry = {
335 "ts_unix": int(time.time()),
336 "loop_lag_avg_ms": (
337 round(sum(lag_samples) / len(lag_samples), 2) if lag_samples else 0.0
338 ),
339 "loop_lag_max_ms": round(self._lag_window_max, 2),
340 "asyncio_tasks": len(asyncio.all_tasks(self.mass.loop)),
341 "tracked_tasks": len(self.mass._tracked_tasks),
342 "tracked_timers": len(self.mass._tracked_timers),
343 "event_subscribers": len(self.mass._subscribers),
344 "ws_clients": len(self.mass.webserver.clients),
345 "events_per_s": round(self._events_since_sample / RECORDER_INTERVAL, 2),
346 "log_errors_per_s": round(
347 (self._log_counter.total - self._errors_at_last_sample) / RECORDER_INTERVAL, 2
348 ),
349 }
350 self._lag_window_max = 0.0
351 self._events_since_sample = 0
352 self._errors_at_last_sample = self._log_counter.total
353 return entry
354
355 async def _cpu_profile_scheduler(self) -> None:
356 """Periodically capture a CPU profile window."""
357 interval_minutes = min(
358 max(
359 self.get_config_value(CONF_CPU_PROFILE_INTERVAL, 30, return_type=int),
360 CPU_PROFILE_MIN_INTERVAL,
361 ),
362 CPU_PROFILE_MAX_INTERVAL,
363 )
364 # small initial delay so a fresh install has profile data available quickly
365 await asyncio.sleep(CPU_PROFILE_FIRST_DELAY)
366 while True:
367 await self._run_cpu_profile_window()
368 await asyncio.sleep(interval_minutes * 60)
369
370 async def _run_cpu_profile_window(self) -> None:
371 """Capture a single CPU profile window and store the extracted result."""
372 if self._cpu_profile_running or yappi.is_running():
373 return
374 duration = min(
375 max(
376 self.get_config_value(CONF_CPU_PROFILE_DURATION, 60, return_type=int),
377 CPU_PROFILE_MIN_DURATION,
378 ),
379 CPU_PROFILE_MAX_DURATION,
380 )
381 self.logger.info("Capturing CPU profile window of %s seconds...", duration)
382 self._cpu_profile_running = True
383 started = time.monotonic()
384 try:
385 yappi.set_clock_type("cpu")
386 # clear any leftovers from a previously aborted window as yappi accumulates
387 yappi.clear_stats()
388 yappi.start(builtins=False)
389 await asyncio.sleep(duration)
390 finally:
391 yappi.stop()
392 self._cpu_profile_running = False
393 self._last_cpu_profile = await asyncio.to_thread(
394 extract_cpu_profile, time.monotonic() - started, self._out_dir
395 )
396 self.logger.info("CPU profile window completed")
397