Coverage for custom_components/remote_logger/otel/exporter.py: 91%
286 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-06 00:18 +0000
1from __future__ import annotations
3import asyncio
4import base64
5import datetime as dt
6import logging
7import math
8import re
9import time
10import typing
11from abc import abstractmethod
12from dataclasses import dataclass
14import aiohttp
15from homeassistant.components.system_log import EVENT_SYSTEM_LOG
16from homeassistant.const import CONF_HEADERS, CONF_HOST, CONF_PATH, CONF_PORT, CONF_TOKEN
17from homeassistant.const import __version__ as hass_version
18from homeassistant.helpers.aiohttp_client import async_get_clientsession
20from custom_components.remote_logger.const import (
21 CONF_BATCH_MAX_SIZE,
22 CONF_CLIENT_TIMEOUT,
23 CONF_ENCODING,
24 CONF_RESOURCE_ATTRIBUTES,
25 CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME,
26 CONF_USE_TLS,
27 DEFAULT_CLIENT_TIMEOUT,
28)
29from custom_components.remote_logger.exporter import LogExporter, LogMessage, LogSubmission
30from custom_components.remote_logger.helpers import flatten_event_data, isotimestamp
32from .const import (
33 CONF_TOKEN_TYPE,
34 DEFAULT_RESOURCE_ATTRIBUTES,
35 DEFAULT_SERVICE_NAME,
36 DEFAULT_SEVERITY,
37 ENCODING_JSON,
38 ENCODING_PROTOBUF,
39 OTLP_LOGS_PATH,
40 SCOPE_NAME,
41 SCOPE_VERSION,
42 SEVERITY_MAP,
43 TOKEN_TYPE_API_KEY,
44 TOKEN_TYPE_BASIC,
45 TOKEN_TYPE_BEARER,
46 TOKEN_TYPE_RAW_BASIC,
47 VALIDATION_EVENT_NAME,
48 VALIDATION_MESSAGE,
49)
50from .protobuf_encoder import encode_export_logs_request
52if typing.TYPE_CHECKING:
53 from collections.abc import Mapping
55 from homeassistant.config_entries import ConfigEntry
56 from homeassistant.core import HomeAssistant
58_LOGGER = logging.getLogger(__name__)
61def build_auth_header(token: str, token_type: str) -> str:
62 """Build the Authorization header value for bearer or basic auth."""
63 if token_type == TOKEN_TYPE_BASIC:
64 credentials = base64.b64encode(token.encode()).decode()
65 return f"Basic {credentials}"
66 if token_type == TOKEN_TYPE_API_KEY:
67 return f"ApiKey {token}"
68 if token_type == TOKEN_TYPE_RAW_BASIC:
69 return f"Basic {token}"
70 return f"Bearer {token}"
73def parse_resource_attributes(raw: str) -> list[tuple[str, str]]:
74 """Parse 'key1=val1,key2=val2' into a list of (key, value) tuples.
76 Raises ValueError if the format is invalid.
77 """
78 result = []
79 for pair in raw.split(","):
80 pair = pair.strip()
81 if not pair:
82 continue
83 if "=" not in pair:
84 raise ValueError(f"Invalid attribute pair: {pair!r}")
85 key, _, value = pair.partition("=")
86 key = key.strip()
87 value = value.strip()
88 if not key:
89 raise ValueError("Attribute key cannot be empty")
90 result.append((key, value))
91 return result
94def parse_headers(raw: str) -> dict[str, str]:
95 """Parse 'Name: value' entries into a dict.
97 Entries may be newline- or comma-separated (commas inside values are safe).
98 Raises ValueError if an entry is malformed.
99 """
100 result: dict[str, str] = {}
101 for line in re.split(r"\n|,(?=\s*[\w-]+\s*:)", raw):
102 line = line.strip()
103 if not line:
104 continue
105 if ":" not in line:
106 raise ValueError(f"Invalid header line: {line!r}")
107 name, _, value = line.partition(":")
108 name = name.strip()
109 if not name:
110 raise ValueError("Header name cannot be empty")
111 result[name] = value.strip()
112 return result
115def _mask_auth_headers(headers: dict[str, str]) -> dict[str, str]:
116 def _mask_credential(v: str) -> str:
117 parts = v.split(" ", 1)
118 if len(parts) == 2:
119 scheme, token = parts
120 return f"{scheme} {'*' * len(token)}"
121 return "*" * len(v)
123 return {k: _mask_credential(v) if k.lower() == "authorization" else v for k, v in headers.items()}
126def append_attr(attrs: list[dict[str, typing.Any]], key: str, value: typing.Any, force_null: bool = False) -> None:
127 attr: dict[str, typing.Any] | None = _kv(key, value, force_null=force_null)
128 if attr is not None:
129 attrs.append(attr)
132def _kv(key: str, value: typing.Any, force_null: bool = False) -> dict[str, typing.Any] | None:
133 """Build an OTLP KeyValue attribute"""
134 if value is None and not force_null:
135 return None
136 if isinstance(value, str):
137 return {"key": key, "value": {"stringValue": value}}
138 if isinstance(value, bool):
139 return {"key": key, "value": {"boolValue": value}}
140 if isinstance(value, int):
141 int_val: int | str = value if -(2**31) <= value <= 2**31 - 1 else str(value)
142 return {"key": key, "value": {"intValue": int_val}}
143 if isinstance(value, float):
144 return {"key": key, "value": {"doubleValue": None if math.isnan(value) or math.isinf(value) else value}}
145 if isinstance(value, bytes):
146 return {"key": key, "value": {"bytesValue": value}}
147 return {"key": key, "value": {"stringValue": str(value)}}
150async def validate(
151 session: aiohttp.ClientSession,
152 url: str,
153 encoding: str,
154 extra_headers: dict[str, str] | None = None,
155) -> dict[str, str]:
156 # Validate connectivity by sending a real log record, since some backends reject empty payloads
157 errors: dict[str, str] = {}
158 resource: dict[str, typing.Any] = {"attributes": []}
159 append_attr(resource["attributes"], "service.name", DEFAULT_SERVICE_NAME)
160 append_attr(resource["attributes"], "service.version", hass_version or "unknown")
161 now = str(time.time_ns())
162 severity_number, severity_text = DEFAULT_SEVERITY
163 record = OtlpMessage(
164 payload={
165 "timeUnixNano": now,
166 "observedTimeUnixNano": now,
167 "severityNumber": severity_number,
168 "severityText": severity_text,
169 "body": {"stringValue": VALIDATION_MESSAGE},
170 "attributes": [],
171 "eventName": VALIDATION_EVENT_NAME,
172 },
173 )
174 submission: OtlpSubmission
175 if encoding == ENCODING_PROTOBUF:
176 submission = OtlpProtobufSubmission(resource, [record], extra_headers)
177 elif encoding == ENCODING_JSON:
178 submission = OtlpJsonSubmission(resource, [record], extra_headers)
179 else:
180 raise ValueError(f"Unknown encoding {encoding}")
181 try:
182 async with session.post(
183 url,
184 timeout=aiohttp.ClientTimeout(total=10),
185 **submission.body(),
186 ) as resp:
187 if resp.status >= 400 and resp.status < 500:
188 errors["base"] = "cannot_connect"
189 _LOGGER.error("remote_logger: client connect failed (%s): %s", resp.status, await resp.text())
190 if resp.status >= 500:
191 errors["base"] = "cannot_connect"
192 _LOGGER.error("remote_logger: server connect failed (%s): %s", resp.status, await resp.text())
193 except aiohttp.ClientResponseError as e1:
194 errors["base"] = "cannot_connect"
195 _LOGGER.error("remote_logger: connect client response error: %s", e1)
196 except aiohttp.ClientError as e2:
197 errors["base"] = "cannot_connect"
198 _LOGGER.error("remote_logger: connect client error: %s", e2)
199 except Exception as e3: # ruff: ignore[blind-except]
200 errors["base"] = "unknown"
201 _LOGGER.error("remote_logger: connect unknown error: %s", e3)
202 return errors
205@dataclass
206class OtlpMessage(LogMessage):
207 payload: dict[str, typing.Any]
210class OtlpSubmission(LogSubmission):
211 def __init__(
212 self,
213 resource: dict[str, typing.Any],
214 records: list[OtlpMessage],
215 extra_headers: dict[str, typing.Any] | None = None,
216 ) -> None:
217 self.extra_headers = extra_headers or {}
218 self.resource: dict[str, typing.Any] = resource
219 self.request: dict[str, typing.Any] = self._build_export_request(records)
221 @abstractmethod
222 def body(self) -> dict[str, typing.Any]:
223 pass
225 def _build_export_request(self, records: list[OtlpMessage]) -> dict[str, typing.Any]:
226 """Wrap logRecords in the ExportLogsServiceRequest envelope."""
227 return {
228 "resourceLogs": [
229 {
230 "resource": self.resource,
231 "scopeLogs": [
232 {
233 "scope": {
234 "name": SCOPE_NAME,
235 "version": SCOPE_VERSION,
236 },
237 "logRecords": [r.payload for r in records],
238 },
239 ],
240 },
241 ],
242 }
245class OtlpJsonSubmission(OtlpSubmission):
246 def __init__(
247 self,
248 resource: dict[str, typing.Any],
249 records: list[OtlpMessage],
250 extra_headers: dict[str, typing.Any] | None = None,
251 ) -> None:
252 super().__init__(resource, records, extra_headers)
254 def body(self) -> dict[str, typing.Any]:
255 return {"headers": {"Content-Type": "application/json", **self.extra_headers}, "json": self.request}
257 def for_display(self) -> dict[str, typing.Any]:
258 body = self.body()
259 return {**body, "headers": _mask_auth_headers(body["headers"])}
262class OtlpProtobufSubmission(OtlpSubmission):
263 def __init__(
264 self,
265 resource: dict[str, typing.Any],
266 records: list[OtlpMessage],
267 extra_headers: dict[str, typing.Any] | None = None,
268 ) -> None:
269 super().__init__(resource, records, extra_headers)
271 def body(self) -> dict[str, typing.Any]:
272 return {
273 "headers": {"Content-Type": "application/x-protobuf", **self.extra_headers},
274 "data": encode_export_logs_request(self.request),
275 }
277 def for_display(self) -> dict[str, typing.Any]:
278 base = self.body()
279 return {
280 "headers": _mask_auth_headers(base["headers"]),
281 "data": base["data"].decode("utf-8", errors="replace").replace("\ufffd", "?"),
282 }
285class OtlpLogExporter(LogExporter):
286 """Buffers system_log_event records and flushes them as OTLP/HTTP JSON."""
288 logger_type = "otel"
290 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
291 super().__init__(hass)
292 self.name = entry.title
294 self._lock = asyncio.Lock()
295 if hass and hass.config and hass.config.api:
296 self.server_address = hass.config.api.local_ip
297 self.server_port = hass.config.api.port
298 else:
299 self.server_address = None
300 self.server_port = None
302 opts = {**entry.data, **entry.options}
304 host = opts[CONF_HOST]
305 port = opts[CONF_PORT]
306 encoding = opts[CONF_ENCODING]
307 use_tls = opts[CONF_USE_TLS]
308 scheme = "https" if use_tls else "http"
309 path = opts.get(CONF_PATH, OTLP_LOGS_PATH)
310 self.endpoint_url = f"{scheme}://{host}:{port}{path}"
311 self.destination = (host, str(port), encoding)
312 self._use_tls = use_tls
313 self._use_protobuf = encoding == ENCODING_PROTOBUF
314 self._entry = entry
315 self._batch_max_size = opts.get(CONF_BATCH_MAX_SIZE, 100)
316 self._client_timeout = opts.get(CONF_CLIENT_TIMEOUT, DEFAULT_CLIENT_TIMEOUT)
317 self._suppress_system_log_event_name = opts.get(CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME, True)
318 self._extra_headers = self._build_extra_headers(opts)
319 self._resource = self._build_resource(opts)
321 _LOGGER.info(f"remote_logger: otel configured for {self.endpoint_url}, protobuf={self._use_protobuf}")
323 def _build_extra_headers(self, opts: dict[str, typing.Any]) -> dict[str, str]:
324 headers: dict[str, str] = {}
325 token = opts.get(CONF_TOKEN, "").strip()
326 if token:
327 token_type = opts.get(CONF_TOKEN_TYPE, TOKEN_TYPE_BEARER)
328 headers["Authorization"] = build_auth_header(token, token_type)
329 raw_headers = "\n".join(opts.get(CONF_HEADERS, []))
330 if raw_headers:
331 headers.update(parse_headers(raw_headers))
332 return headers
334 def _build_resource(self, opts: dict[str, typing.Any]) -> dict[str, typing.Any]:
335 """Build the OTLP Resource object with attributes."""
336 attrs: list[dict[str, typing.Any]] = []
337 append_attr(attrs, "service.name", DEFAULT_SERVICE_NAME)
338 append_attr(attrs, "service.version", hass_version or "unknown")
340 if self.server_address:
341 append_attr(attrs, "service.address", self.server_address)
342 if self.server_port:
343 append_attr(attrs, "service.port", self.server_port)
345 raw = opts.get(CONF_RESOURCE_ATTRIBUTES, DEFAULT_RESOURCE_ATTRIBUTES)
346 if raw and raw.strip():
347 for key, value in parse_resource_attributes(raw):
348 append_attr(attrs, key, value)
350 return {"attributes": attrs}
352 def create_log_record(
353 self,
354 event_data: Mapping[str, typing.Any],
355 event_type: str | None = None,
356 time_fired: dt.datetime | None = None,
357 message_override: list[str] | None = None,
358 level_override: str | None = None,
359 state_only: bool = False,
360 ) -> OtlpMessage:
361 """Convert a system_log_event payload to an OTLP logRecord dict."""
362 """ HA System Log Event
363 "name": str
364 "message": list(str)
365 "level": str
366 "source": (str,int)
367 "timestamp": float
368 "exception": str
369 "count": int
370 "first_occurred": float
371 """
372 data: Mapping[str, typing.Any] | dict[str, typing.Any] = event_data or {}
373 time_fired = time_fired or dt.datetime.now(tz=self.tz)
374 timestamp_s: float = data.get("timestamp", time.time())
375 time_unix_nano = str(int(timestamp_s * 1_000_000_000))
376 observed_timestamp: float = time_fired.timestamp()
377 observed_time_unix_nano = str(int(observed_timestamp * 1_000_000_000))
379 level: str = level_override or data.get("level", "INFO").upper()
380 severity_number, severity_text = SEVERITY_MAP.get(level, DEFAULT_SEVERITY)
382 messages: list[str] = message_override or data.get("message", [])
383 message: str = "\n".join(messages)
385 attributes: list[dict[str, typing.Any]] = []
387 if event_type == EVENT_SYSTEM_LOG or event_type is None:
388 source = data.get("source")
389 if source and isinstance(source, tuple):
390 source_path, source_lineno = source
391 append_attr(attributes, "code.file.path", source_path)
392 append_attr(attributes, "code.line.number", source_lineno)
393 logger_name = data.get("name")
394 if data.get("count"):
395 append_attr(attributes, "exception.count", data["count"])
396 if data.get("first_occurred"):
397 append_attr(attributes, "exception.first_occurred", isotimestamp(data["first_occurred"]))
398 if logger_name:
399 append_attr(attributes, "code.function.name", logger_name)
400 exception = data.get("exception")
401 if exception:
402 append_attr(attributes, "exception.stacktrace", exception)
404 else:
405 for k, v in data.items():
406 for flat_key, flat_val in flatten_event_data(f"event.data.{k}" if k != "event.data" else k, v, state_only):
407 append_attr(attributes, flat_key, flat_val)
409 # https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/logs/v1/logs.proto
410 payload: dict[str, typing.Any] = {
411 "timeUnixNano": time_unix_nano,
412 "observedTimeUnixNano": observed_time_unix_nano,
413 "severityNumber": severity_number,
414 "severityText": severity_text,
415 "body": {"stringValue": message},
416 "attributes": attributes,
417 }
418 if event_type is not None and (event_type != EVENT_SYSTEM_LOG or not self._suppress_system_log_event_name):
419 payload["eventName"] = event_type
420 return OtlpMessage(payload=payload)
422 async def flush(self) -> None:
423 """Flush all buffered log records to the OTLP endpoint."""
424 records: list[OtlpMessage] | None = None
425 async with self._lock:
426 if not self._buffer:
427 return
428 records = typing.cast("list[OtlpMessage]", self._buffer.copy())
429 self._buffer.clear()
431 try:
432 if records:
433 if self._use_protobuf:
434 submission: OtlpSubmission = OtlpProtobufSubmission(self._resource, records, self._extra_headers)
435 else:
436 submission = OtlpJsonSubmission(self._resource, records, self._extra_headers)
437 else:
438 return
439 session: aiohttp.ClientSession = async_get_clientsession(self._hass, verify_ssl=self._use_tls)
440 timeout = aiohttp.ClientTimeout(total=self._client_timeout)
441 async with session.post(self.endpoint_url, timeout=timeout, **submission.body()) as resp:
442 if resp.status in (401, 403):
443 _LOGGER.warning("remote_logger: OTLP authentication failed (%s), triggering reauth", resp.status)
444 self._entry.async_start_reauth(self._hass)
445 return
446 if resp.status >= 400:
447 body = await resp.text()
448 _LOGGER.warning(
449 "remote_logger: OTLP endpoint returned HTTP %s: %s",
450 resp.status,
451 body[:200],
452 )
453 self.on_posting_error(body)
454 if resp.ok or (resp.status >= 400 and resp.status < 500):
455 # records were sent, or there was a client-side error
456 if records:
457 self.last_sent_payload = submission
458 self.on_success()
460 except aiohttp.ClientError as err:
461 _LOGGER.warning("remote_logger: failed to send logs: %s", err)
462 self.on_posting_error(str(err))
463 except RuntimeError as err:
464 if "Session is closed" in str(err):
465 _LOGGER.debug("remote_logger: session closed during flush (shutdown), dropping %d records", len(records or []))
466 else:
467 _LOGGER.exception("remote_logger: unexpected error sending logs, skipping records")
468 self.on_posting_error(str(err))
469 except Exception as e:
470 _LOGGER.exception("remote_logger: unexpected error sending logs, skipping records")
471 self.on_posting_error(str(e))
473 def log_direct(self, event_name: str, message: str, level: str, attributes: dict[str, typing.Any] | None = None) -> None:
474 """Buffer a custom log record without requiring a HA Event."""
475 now = time.time()
476 time_unix_nano = str(int(now * 1_000_000_000))
477 severity_number, severity_text = SEVERITY_MAP.get(level.upper(), DEFAULT_SEVERITY)
478 attrs: list[dict[str, typing.Any]] = []
479 for k, v in (attributes or {}).items():
480 append_attr(attrs, k, v)
482 record = OtlpMessage(
483 payload={
484 "timeUnixNano": time_unix_nano,
485 "observedTimeUnixNano": time_unix_nano,
486 "severityNumber": severity_number,
487 "severityText": severity_text,
488 "body": {"stringValue": message},
489 "attributes": attrs,
490 "eventName": event_name,
491 },
492 )
493 self._buffer.append(record)
494 self.on_event()
495 if len(self._buffer) >= self._batch_max_size:
496 self._hass.async_create_task(self.flush())