/
/
/
1"""
2Diagnostics controller: on-demand, privacy-safe troubleshooting reports.
3
4Builds a single report combining the always-on exception/log capture (see
5helpers/diagnostics.py) with system info, an install census and pluggable
6sections contributed by core controllers, providers and external registrants.
7The report is returned on request via the `diagnostics/get` API command. All
8report building happens on request only; every string that ends up in the
9report is sanitized so it can safely be attached to a public GitHub issue.
10"""
11
12from __future__ import annotations
13
14import asyncio
15import inspect
16import platform
17import resource
18import shutil
19import sys
20import threading
21import time
22from collections import Counter
23from typing import TYPE_CHECKING, Any
24
25from music_assistant_models.auth import Scope
26
27from music_assistant.helpers.api import api_command
28from music_assistant.helpers.datetime import from_utc_timestamp, utc
29from music_assistant.helpers.diagnostics import (
30 REDACTION_NOTICE,
31 install_diagnostics_log_handler,
32 sanitize_data,
33 sanitize_text,
34)
35from music_assistant.helpers.json import json_dumps, json_loads
36from music_assistant.models.core_controller import CoreController
37from music_assistant.models.provider import Provider
38
39if TYPE_CHECKING:
40 from collections.abc import Awaitable, Callable
41
42 from music_assistant_models.config_entries import CoreConfig
43
44 from music_assistant.helpers.json import SerializableType
45 from music_assistant.mass import MusicAssistant
46
47SCHEMA_VERSION = 1
48# maximum time one section contributor may take before it is dropped from the report
49SECTION_TIMEOUT = 2.0
50
51# core controller attributes on MusicAssistant that may implement get_diagnostics
52CORE_CONTROLLER_ATTRS = (
53 "cache",
54 "discovery",
55 "metadata",
56 "music",
57 "player_queues",
58 "players",
59 "streams",
60 "tasks",
61 "translations",
62 "webserver",
63)
64
65type DiagnosticsSectionCallback = Callable[
66 [], dict[str, SerializableType] | None | Awaitable[dict[str, SerializableType] | None]
67]
68
69
70class DiagnosticsController(CoreController):
71 """Core controller that assembles privacy-safe diagnostics reports on demand."""
72
73 domain: str = "diagnostics"
74
75 def __init__(self, mass: MusicAssistant) -> None:
76 """Initialize the diagnostics controller."""
77 super().__init__(mass)
78 self.manifest.name = "Diagnostics"
79 self.manifest.description = (
80 "Provides a downloadable, privacy-safe diagnostics report for troubleshooting."
81 )
82 self.manifest.icon = "stethoscope"
83 # adopt the always-on capture handler (installed even earlier when booted
84 # through __main__, otherwise installed right here)
85 self._log_handler = install_diagnostics_log_handler()
86 self._sections: dict[str, DiagnosticsSectionCallback] = {}
87 self._started_at = time.monotonic()
88
89 async def setup(self, config: CoreConfig) -> None:
90 """Async initialize of module."""
91
92 async def close(self) -> None:
93 """Handle logic on server stop."""
94 self._sections.clear()
95
96 def register_section(
97 self, name: str, callback: DiagnosticsSectionCallback
98 ) -> Callable[[], None]:
99 """
100 Register a callback that contributes a named section to the diagnostics report.
101
102 The callback is invoked whenever a report is requested and may be sync or
103 async. Async callbacks run with a timeout; sync callbacks run on the event
104 loop and must return quickly without blocking I/O. Returns a function to
105 unregister the section again.
106
107 :param name: Unique name for the section within the report.
108 :param callback: Callable returning the section data as a (JSON-safe) dict.
109 """
110 if name in self._sections:
111 raise ValueError(f"A diagnostics section named '{name}' is already registered")
112 self._sections[name] = callback
113
114 def unregister() -> None:
115 # only remove if this exact registration still owns the name
116 if self._sections.get(name) is callback:
117 self._sections.pop(name)
118
119 return unregister
120
121 @api_command("diagnostics/get", required_scope=Scope.SYSTEM_MANAGE)
122 async def get_report(self, include_log_tail: bool = False) -> dict[str, Any]:
123 """
124 Return a full (sanitized) diagnostics report.
125
126 :param include_log_tail: Include the recent warning/error log excerpt.
127 """
128 return await self._build_report(include_log_tail=include_log_tail)
129
130 async def _build_report(self, include_log_tail: bool = False) -> dict[str, Any]:
131 """Assemble the full report; every part is isolated so the report never fails."""
132 report: dict[str, Any] = {
133 "schema_version": SCHEMA_VERSION,
134 "generated_at": utc().isoformat(timespec="seconds"),
135 "redaction_notice": REDACTION_NOTICE,
136 }
137 for key, builder in (
138 ("system", self._build_system_info),
139 ("install", self._build_install_census),
140 ("exceptions", self._build_exception_summary),
141 ("sections", self._collect_sections),
142 ):
143 try:
144 report[key] = await builder()
145 except Exception as err:
146 report[key] = {"error": sanitize_text(f"{type(err).__name__}: {err}")}
147 if include_log_tail:
148 report["log_tail"] = self._build_log_tail()
149 return report
150
151 async def _build_system_info(self) -> dict[str, Any]:
152 """Collect a point-in-time snapshot of system/runtime info."""
153 disk_info, memory_info = await asyncio.to_thread(self._probe_system_blocking)
154 return {
155 "version": self.mass.version,
156 "python_version": platform.python_version(),
157 "platform": platform.platform(),
158 "machine": platform.machine(),
159 "hass_addon": self.mass.running_as_hass_addon,
160 "safe_mode": self.mass.safe_mode,
161 "uptime_seconds": round(time.monotonic() - self._started_at),
162 "event_loop_lag_ms": await self._measure_loop_lag(),
163 "memory": memory_info,
164 "data_dir_disk": disk_info,
165 "counts": {
166 "threads": threading.active_count(),
167 "asyncio_tasks": len(asyncio.all_tasks()),
168 "tracked_tasks": len(self.mass._tracked_tasks),
169 "tracked_timers": len(self.mass._tracked_timers),
170 "event_subscribers": len(self.mass._subscribers),
171 "websocket_clients": len(self.mass.webserver.clients),
172 },
173 }
174
175 async def _build_install_census(self) -> dict[str, Any]:
176 """Collect install structure (never configuration values or credentials)."""
177 census: dict[str, Any] = {}
178 for key, builder in (
179 ("providers", self._census_providers),
180 ("players", self._census_players),
181 ("library", self._census_library),
182 ("core_config_non_default", self._census_core_config),
183 ):
184 try:
185 result = builder()
186 census[key] = await result if inspect.isawaitable(result) else result
187 except Exception as err:
188 census[key] = {"error": sanitize_text(f"{type(err).__name__}: {err}")}
189 return census
190
191 async def _census_providers(self) -> list[dict[str, Any]]:
192 """Return the configured providers with their load/availability state."""
193 providers: list[dict[str, Any]] = []
194 for prov_conf in await self.mass.config.get_provider_configs():
195 loaded = self.mass.get_provider(prov_conf.instance_id, return_unavailable=True)
196 entry: dict[str, Any] = {
197 "domain": prov_conf.domain,
198 "instance_id": prov_conf.instance_id,
199 "type": prov_conf.type.value,
200 "enabled": prov_conf.enabled,
201 "loaded": loaded is not None,
202 "available": loaded.available if loaded else False,
203 }
204 if prov_conf.last_error is not None:
205 entry["last_error"] = sanitize_text(prov_conf.last_error.message)
206 providers.append(entry)
207 providers.sort(key=lambda entry: (entry["domain"], entry["instance_id"]))
208 return providers
209
210 def _census_players(self) -> dict[str, Any]:
211 """Return player counts grouped by provider and type."""
212 by_provider: Counter[str] = Counter()
213 by_type: Counter[str] = Counter()
214 total = available = 0
215 for player in self.mass.players.all_players(
216 return_unavailable=True, return_disabled=True, return_protocol_players=True
217 ):
218 total += 1
219 available += int(player.available)
220 by_provider[player.provider.domain] += 1
221 by_type[player.type.value] += 1
222 return {
223 "total": total,
224 "available": available,
225 "by_provider": dict(sorted(by_provider.items())),
226 "by_type": dict(sorted(by_type.items())),
227 }
228
229 async def _census_library(self) -> dict[str, int]:
230 """Return the library item counts per media type."""
231 music = self.mass.music
232 # raw table sizes: a support report needs true totals, not the filtered
233 # subset visible to the (admin) user who requested it
234 return {
235 name: await music.database.get_count(controller.db_table)
236 for name, controller in (
237 ("albums", music.albums),
238 ("artists", music.artists),
239 ("audiobooks", music.audiobooks),
240 ("genres", music.genres),
241 ("playlists", music.playlists),
242 ("podcasts", music.podcasts),
243 ("radio", music.radio),
244 ("tracks", music.tracks),
245 )
246 }
247
248 async def _census_core_config(self) -> dict[str, list[str]]:
249 """Return per core module which config keys differ from default (key names only)."""
250 result: dict[str, list[str]] = {}
251 for core_conf in await self.mass.config.get_core_configs(include_values=True):
252 changed_keys = [
253 key
254 for key, entry in core_conf.values.items()
255 if entry.value is not None and entry.value != entry.default_value
256 ]
257 if changed_keys:
258 result[core_conf.domain] = sorted(changed_keys)
259 return result
260
261 async def _build_exception_summary(self) -> list[dict[str, Any]]:
262 """Return the aggregated (sanitized) exceptions, most recent first."""
263 _, exceptions = self._log_handler.snapshot()
264 exceptions.sort(key=lambda entry: entry.last_seen, reverse=True)
265 # rendering traceback text may read source files (linecache), so run off-loop
266 rendered = await asyncio.get_running_loop().run_in_executor(
267 None, lambda: [entry.render_traceback() for entry in exceptions]
268 )
269 return [
270 {
271 "type": entry.exc_type,
272 "fingerprint": entry.fingerprint,
273 "count": entry.count,
274 "first_seen": _format_timestamp(entry.first_seen),
275 "last_seen": _format_timestamp(entry.last_seen),
276 "logger": sanitize_text(entry.logger_name),
277 "level": entry.level,
278 "origin": sanitize_text(entry.origin),
279 "message": sanitize_text(entry.message),
280 "traceback": sanitize_text(traceback_text),
281 }
282 for entry, traceback_text in zip(exceptions, rendered, strict=True)
283 ]
284
285 async def _collect_sections(self) -> dict[str, Any]:
286 """Collect all pluggable sections (isolated, each bounded by a timeout)."""
287 producers: list[tuple[str, DiagnosticsSectionCallback]] = []
288 for attr_name in CORE_CONTROLLER_ATTRS:
289 controller = getattr(self.mass, attr_name, None)
290 if controller is None:
291 continue
292 # skip controllers that don't implement the optional hook
293 if type(controller).get_diagnostics is CoreController.get_diagnostics:
294 continue
295 producers.append((f"core.{attr_name}", controller.get_diagnostics))
296 for provider in sorted(self.mass.providers, key=lambda prov: prov.instance_id):
297 if type(provider).get_diagnostics is Provider.get_diagnostics:
298 continue
299 producers.append((f"provider.{provider.instance_id}", provider.get_diagnostics))
300 producers.extend(self._sections.items())
301 results = await asyncio.gather(
302 *(self._collect_section(name, producer) for name, producer in producers)
303 )
304 return {name: data for name, data in results if data is not None}
305
306 async def _collect_section(
307 self, name: str, producer: DiagnosticsSectionCallback
308 ) -> tuple[str, Any]:
309 """Run one section contributor with timeout/failure isolation and sanitize it."""
310 try:
311 async with asyncio.timeout(SECTION_TIMEOUT):
312 raw_result = producer()
313 result = await raw_result if inspect.isawaitable(raw_result) else raw_result
314 if result is None:
315 return name, None
316 # roundtrip through JSON to normalize (and validate) the data,
317 # then sanitize all string values in the result
318 return name, sanitize_data(json_loads(json_dumps(result)))
319 except Exception as err:
320 return name, {"error": sanitize_text(f"{type(err).__name__}: {err}")}
321
322 def _build_log_tail(self) -> list[dict[str, str]]:
323 """Return the sanitized recent warning/error log excerpt."""
324 records, _ = self._log_handler.snapshot()
325 return [
326 {
327 "time": _format_timestamp(record.created),
328 "level": record.level,
329 "logger": sanitize_text(record.logger_name),
330 "message": sanitize_text(record.message),
331 }
332 for record in records
333 ]
334
335 async def _measure_loop_lag(self) -> float:
336 """Spot-sample the event loop lag in milliseconds."""
337 loop = asyncio.get_running_loop()
338 start = loop.time()
339 await asyncio.sleep(0.1)
340 return max(0.0, round((loop.time() - start - 0.1) * 1000, 1))
341
342 def _probe_system_blocking(self) -> tuple[dict[str, Any], dict[str, Any]]:
343 """Collect disk and memory info (blocking, run in executor)."""
344 usage = shutil.disk_usage(self.mass.storage_path)
345 disk_info = {
346 "free_mb": usage.free // (1024 * 1024),
347 "total_mb": usage.total // (1024 * 1024),
348 }
349 return disk_info, _get_memory_info()
350
351
352def _format_timestamp(timestamp: float) -> str:
353 """Format a unix timestamp as a compact UTC ISO string."""
354 return from_utc_timestamp(timestamp).isoformat(timespec="seconds")
355
356
357def _get_memory_info() -> dict[str, Any]:
358 """Return the process memory usage (rss on Linux, peak rss elsewhere)."""
359 try:
360 with open("/proc/self/status", encoding="ascii") as status_file:
361 for line in status_file:
362 if line.startswith("VmRSS:"):
363 return {"rss_mb": round(int(line.split()[1]) / 1024, 1)}
364 except OSError, ValueError, IndexError:
365 pass
366 # ru_maxrss is in bytes on macOS, kilobytes on other platforms
367 divisor = 1024 * 1024 if sys.platform == "darwin" else 1024
368 return {"peak_rss_mb": round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / divisor, 1)}
369