Coverage for custom_components/remote_logger/remote_logger.py: 92%
106 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-07 04:46 +0000
1"""The remote_logger integration: ship HA system_log_event to an OTLP collector or syslog server."""
3from __future__ import annotations
5import asyncio
6import contextlib
7import logging
8from functools import partial
9from typing import TYPE_CHECKING, Any
11import voluptuous as vol
12from homeassistant.const import EVENT_HOMEASSISTANT_STOP
13from homeassistant.core import EVENT_HOMEASSISTANT_CLOSE, EVENT_HOMEASSISTANT_FINAL_WRITE, SupportsResponse, callback
15from .const import (
16 BACKEND_SYSLOG,
17 CONF_BACKEND,
18 CONF_CUSTOM_EVENTS,
19 CONF_LOG_HA_CORE_ACTIVITY,
20 CONF_LOG_HA_CORE_CHANGES,
21 CONF_LOG_HA_EVENT_BODY,
22 CONF_LOG_HA_FULL_STATE_CHANGES,
23 CONF_LOG_HA_LIFECYCLE,
24 CONF_LOG_HA_STATE_CHANGES,
25 CORE_ACTIVITY_EVENTS,
26 CORE_CHANGE_EVENTS,
27 CORE_STATE_EVENTS,
28 DOMAIN,
29 EVENT_SYSTEM_LOG,
30 LIFECYCLE_EVENTS,
31 PLATFORMS,
32)
33from .otel.exporter import OtlpLogExporter
34from .syslog.exporter import SyslogExporter
36REF_CANCEL_LISTENERS = "cancel_listeners"
37REF_FLUSH_TASK = "flush_task"
38REF_EXPORTER = "exporter"
40SERVICE_SEND_LOG = "send_log"
41SERVICE_SEND_LOG_SCHEMA = vol.Schema({
42 vol.Required("event"): str,
43 vol.Required("message"): str,
44 vol.Optional("level", default="INFO"): vol.In(["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]),
45 vol.Optional("attributes"): dict,
46})
48SERVICE_FLUSH = "flush"
50SERVICE_LAST_LOG = "last_log"
51SERVICE_LAST_LOG_SCHEMA = vol.Schema({
52 vol.Required("config_entry_id"): str,
53})
55if TYPE_CHECKING:
56 from collections.abc import Callable
58 from homeassistant.config_entries import ConfigEntry
59 from homeassistant.core import HomeAssistant, ServiceCall
61 from custom_components.remote_logger.exporter import LogSubmission
63_LOGGER = logging.getLogger(__name__)
66async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
67 """Reload the entry when options are updated."""
68 await hass.config_entries.async_reload(entry.entry_id)
71async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
72 """Set up remote logs from a config entry."""
73 backend = entry.data.get(CONF_BACKEND)
75 exporter: OtlpLogExporter | SyslogExporter
76 if backend == BACKEND_SYSLOG:
77 exporter = SyslogExporter(hass, entry)
78 label: str = exporter.endpoint_desc
79 else:
80 exporter = OtlpLogExporter(hass, entry)
81 label = exporter.endpoint_url
83 # Options take precedence over initial data for the three event-subscription keys
84 opts = {**entry.data, **entry.options}
86 async def _flush_on_stop(_: Any) -> None:
87 await exporter.disable_buffer()
89 cancel_listeners: list[Callable[[], None]] = [
90 hass.bus.async_listen(EVENT_SYSTEM_LOG, exporter.handle_event),
91 hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _flush_on_stop),
92 hass.bus.async_listen_once(EVENT_HOMEASSISTANT_CLOSE, _flush_on_stop),
93 hass.bus.async_listen_once(EVENT_HOMEASSISTANT_FINAL_WRITE, _flush_on_stop),
94 entry.add_update_listener(_async_update_listener),
95 ]
96 _LOGGER.info("remote_logger: listening for system_log_event, exporting %s to %s", backend, label)
98 event_body: bool = bool(opts.get(CONF_LOG_HA_EVENT_BODY))
100 if opts.get(CONF_LOG_HA_LIFECYCLE):
101 cancel_listeners.extend(
102 hass.bus.async_listen(et, partial(exporter.handle_ha_event, et, event_body=event_body)) for et in LIFECYCLE_EVENTS
103 )
104 _LOGGER.info("remote_logger: listening for HA lifecycle events")
106 if opts.get(CONF_LOG_HA_CORE_CHANGES):
107 cancel_listeners.extend(
108 hass.bus.async_listen(et, partial(exporter.handle_ha_event, et, event_body=event_body)) for et in CORE_CHANGE_EVENTS
109 )
110 _LOGGER.info("remote_logger: listening for HA core config events")
112 if opts.get(CONF_LOG_HA_STATE_CHANGES):
113 cancel_listeners.extend(
114 hass.bus.async_listen(et, partial(exporter.handle_ha_event, et, state_only=True, event_body=event_body))
115 for et in CORE_STATE_EVENTS
116 )
117 _LOGGER.info("remote_logger: listening for HA state changes")
119 if opts.get(CONF_LOG_HA_FULL_STATE_CHANGES):
120 cancel_listeners.extend(
121 hass.bus.async_listen(et, partial(exporter.handle_ha_event, et, state_only=False, event_body=event_body))
122 for et in CORE_STATE_EVENTS
123 )
124 _LOGGER.info("remote_logger: listening for HA state changes")
126 if opts.get(CONF_LOG_HA_CORE_ACTIVITY):
127 cancel_listeners.extend(
128 hass.bus.async_listen(et, partial(exporter.handle_ha_event, et, event_body=event_body))
129 for et in CORE_ACTIVITY_EVENTS
130 )
131 _LOGGER.info("remote_logger: listening for HA core activity")
133 custom_events_raw = opts.get(CONF_CUSTOM_EVENTS, "")
134 cancel_listeners.extend(
135 hass.bus.async_listen(et, partial(exporter.handle_ha_event, et, event_body=event_body))
136 for et in (e.strip() for e in custom_events_raw.splitlines() if e.strip())
137 )
139 flush_task: asyncio.Task[None] = asyncio.create_task(exporter.flush_loop())
141 hass.data.setdefault(DOMAIN, {})
142 hass.data[DOMAIN][entry.entry_id] = {
143 REF_CANCEL_LISTENERS: cancel_listeners,
144 REF_FLUSH_TASK: flush_task,
145 REF_EXPORTER: exporter,
146 }
148 if not hass.services.has_service(DOMAIN, SERVICE_SEND_LOG):
149 hass.services.async_register(
150 DOMAIN, SERVICE_SEND_LOG, partial(handle_send_log, hass.data[DOMAIN]), schema=SERVICE_SEND_LOG_SCHEMA
151 )
153 if not hass.services.has_service(DOMAIN, SERVICE_FLUSH):
154 hass.services.async_register(DOMAIN, SERVICE_FLUSH, partial(handle_flush, hass.data[DOMAIN]))
156 if not hass.services.has_service(DOMAIN, SERVICE_LAST_LOG):
157 hass.services.async_register(
158 DOMAIN,
159 SERVICE_LAST_LOG,
160 partial(handle_last_log, hass.data[DOMAIN]),
161 schema=SERVICE_LAST_LOG_SCHEMA,
162 supports_response=SupportsResponse.ONLY,
163 )
165 await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
167 return True
170async def handle_flush(domain_data: dict[str, Any], _call: ServiceCall) -> None:
171 for entry in domain_data.values():
172 await entry[REF_EXPORTER].flush()
175@callback
176def handle_last_log(domain_data: dict[str, Any], call: ServiceCall) -> dict[str, Any]:
177 entry_id: str | None = call.data.get("config_entry_id")
178 entry = None
179 if entry_id:
180 entry = domain_data.get(entry_id)
181 if entry is None:
182 return {}
183 submission: LogSubmission | None = entry[REF_EXPORTER].last_sent_payload
184 if submission is None:
185 return {}
186 return submission.for_display()
189@callback
190def handle_send_log(domain_data: dict[str, Any], call: ServiceCall) -> None:
191 message: str = call.data["message"]
192 level: str = call.data["level"]
193 event_name: str = call.data["event"]
194 attributes: dict[str, Any] | None = call.data.get("attributes")
195 for entry in domain_data.values():
196 entry[REF_EXPORTER].log_direct(event_name, message, level, attributes)
199async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
200 """Unload remote_logger config entry.
202 https://developers.home-assistant.io/docs/config_entries_index#unloading-entries
203 """
204 await hass.config_entries.async_unload_platforms(entry, ["sensor"])
205 data = hass.data[DOMAIN].pop(entry.entry_id, None)
206 if data is None:
207 return True
209 for cancel in data.get(REF_CANCEL_LISTENERS, []):
210 try:
211 cancel()
212 except Exception as e:
213 _LOGGER.warning("Failed to cancel listener on unload: %s", e)
215 if data.get(REF_FLUSH_TASK):
216 data[REF_FLUSH_TASK].cancel()
217 with contextlib.suppress(asyncio.CancelledError):
218 await data[REF_FLUSH_TASK]
220 if data.get(REF_EXPORTER):
221 await data[REF_EXPORTER].flush()
222 await data[REF_EXPORTER].close()
224 _LOGGER.info("remote_logger: unloaded, flushed remaining logs")
225 return True