[feat] add observability runtime
This commit is contained in:
Azamat
parent
cb4560db5c
commit
741e63e978
5 changed files with 335 additions and 0 deletions
0
adapter/observability/__init__.py
Normal file
0
adapter/observability/__init__.py
Normal file
101
adapter/observability/factory.py
Normal file
101
adapter/observability/factory.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
from collections.abc import Callable
|
||||
|
||||
from opentelemetry.sdk.metrics import MeterProvider
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
|
||||
from adapter.config.model import AppConfig
|
||||
from adapter.otel.bootstrap import OtelRuntime, setup_otel
|
||||
from usecase.interface import Logger, Metrics, Tracer
|
||||
|
||||
from .logging import FileLogger, StdoutLogger
|
||||
from .noop import NoopMetrics, NoopTracer
|
||||
from .runtime import ObservabilityRuntime
|
||||
|
||||
|
||||
def build_observability(config: AppConfig) -> ObservabilityRuntime:
|
||||
otel_runtime = _setup_otel_runtime(config)
|
||||
|
||||
try:
|
||||
logger, logger_shutdown = _build_logger(config, otel_runtime)
|
||||
except Exception:
|
||||
if otel_runtime is not None:
|
||||
otel_runtime.shutdown()
|
||||
raise
|
||||
|
||||
metrics: Metrics = NoopMetrics()
|
||||
tracer: Tracer = NoopTracer()
|
||||
meter_provider: MeterProvider | None = None
|
||||
tracer_provider: TracerProvider | None = None
|
||||
shutdown_callback_list: list[Callable[[], None]] = []
|
||||
|
||||
if logger_shutdown is not None:
|
||||
shutdown_callback_list.append(logger_shutdown)
|
||||
|
||||
if otel_runtime is not None:
|
||||
meter_provider = otel_runtime.meter_provider
|
||||
tracer_provider = otel_runtime.tracer_provider
|
||||
shutdown_callback_list.append(otel_runtime.shutdown)
|
||||
|
||||
if config.metrics.enabled:
|
||||
if otel_runtime is None or otel_runtime.metrics is None:
|
||||
raise ValueError('missing otel metrics')
|
||||
metrics = otel_runtime.metrics
|
||||
|
||||
if config.tracing.enabled:
|
||||
if otel_runtime is None or otel_runtime.tracer is None:
|
||||
raise ValueError('missing otel tracer')
|
||||
tracer = otel_runtime.tracer
|
||||
|
||||
return ObservabilityRuntime(
|
||||
logger=logger,
|
||||
metrics=metrics,
|
||||
tracer=tracer,
|
||||
meter_provider=meter_provider,
|
||||
tracer_provider=tracer_provider,
|
||||
_shutdown_callbacks=tuple(shutdown_callback_list),
|
||||
)
|
||||
|
||||
|
||||
def _setup_otel_runtime(config: AppConfig) -> OtelRuntime | None:
|
||||
enable_logs = config.logging.output == 'otel'
|
||||
enable_metrics = config.metrics.enabled
|
||||
enable_tracing = config.tracing.enabled
|
||||
|
||||
if not any((enable_logs, enable_metrics, enable_tracing)):
|
||||
return None
|
||||
|
||||
return setup_otel(
|
||||
config,
|
||||
enable_logs=enable_logs,
|
||||
enable_metrics=enable_metrics,
|
||||
enable_tracing=enable_tracing,
|
||||
)
|
||||
|
||||
|
||||
def _build_logger(
|
||||
config: AppConfig,
|
||||
otel_runtime: OtelRuntime | None,
|
||||
) -> tuple[Logger, Callable[[], None] | None]:
|
||||
if config.logging.output == 'stdout':
|
||||
stdout_logger = StdoutLogger(
|
||||
f'{config.app.name}.stdout',
|
||||
config.logging.level,
|
||||
config.logging.format,
|
||||
)
|
||||
return stdout_logger, stdout_logger.shutdown
|
||||
if config.logging.output == 'file':
|
||||
file_path = config.logging.file_path
|
||||
if file_path is None:
|
||||
raise ValueError('missing logging.file_path')
|
||||
file_logger = FileLogger(
|
||||
file_path,
|
||||
f'{config.app.name}.file',
|
||||
config.logging.level,
|
||||
config.logging.format,
|
||||
)
|
||||
return file_logger, file_logger.shutdown
|
||||
if config.logging.output == 'otel':
|
||||
if otel_runtime is None or otel_runtime.logger is None:
|
||||
raise ValueError('missing otel logger')
|
||||
return otel_runtime.logger, None
|
||||
raise ValueError('invalid logging output')
|
||||
152
adapter/observability/logging.py
Normal file
152
adapter/observability/logging.py
Normal file
|
|
@ -0,0 +1,152 @@
|
|||
import json
|
||||
import logging
|
||||
import sys
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import override
|
||||
|
||||
from usecase.interface import Attrs, AttrValue
|
||||
|
||||
_LEVELS = {
|
||||
'DEBUG': logging.DEBUG,
|
||||
'INFO': logging.INFO,
|
||||
'WARN': logging.WARNING,
|
||||
'WARNING': logging.WARNING,
|
||||
'ERROR': logging.ERROR,
|
||||
'FATAL': logging.CRITICAL,
|
||||
'CRITICAL': logging.CRITICAL,
|
||||
}
|
||||
|
||||
|
||||
class BaseLogger:
|
||||
def __init__(self, logger: logging.Logger, handler: logging.Handler) -> None:
|
||||
self._logger = logger
|
||||
self._handler = handler
|
||||
|
||||
def debug(self, message: str, attrs: Attrs | None = None) -> None:
|
||||
self._emit(logging.DEBUG, message, attrs)
|
||||
|
||||
def info(self, message: str, attrs: Attrs | None = None) -> None:
|
||||
self._emit(logging.INFO, message, attrs)
|
||||
|
||||
def warning(self, message: str, attrs: Attrs | None = None) -> None:
|
||||
self._emit(logging.WARNING, message, attrs)
|
||||
|
||||
def error(self, message: str, attrs: Attrs | None = None) -> None:
|
||||
self._emit(logging.ERROR, message, attrs)
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self._logger.removeHandler(self._handler)
|
||||
try:
|
||||
self._handler.close()
|
||||
except (OSError, ValueError):
|
||||
return
|
||||
|
||||
def _emit(self, level: int, message: str, attrs: Attrs | None) -> None:
|
||||
extra = {'attrs': None if attrs is None else dict(attrs)}
|
||||
self._logger.log(level, message, extra=extra)
|
||||
|
||||
|
||||
class StdoutLogger(BaseLogger):
|
||||
def __init__(self, name: str, level: str, output_format: str) -> None:
|
||||
handler = SafeStreamHandler(sys.stdout)
|
||||
super().__init__(_build_logger(name, level, output_format, handler), handler)
|
||||
|
||||
|
||||
class FileLogger(BaseLogger):
|
||||
def __init__(self, path: str, name: str, level: str, output_format: str) -> None:
|
||||
file_path = Path(path)
|
||||
try:
|
||||
file_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
handler = SafeFileHandler(file_path, mode='a', encoding='utf-8')
|
||||
except OSError as exc:
|
||||
raise ValueError('invalid logging.file_path') from exc
|
||||
|
||||
try:
|
||||
logger = _build_logger(name, level, output_format, handler)
|
||||
except Exception:
|
||||
handler.close()
|
||||
raise
|
||||
|
||||
super().__init__(logger, handler)
|
||||
|
||||
|
||||
class SafeStreamHandler(logging.StreamHandler):
|
||||
@override
|
||||
def handleError(self, record: logging.LogRecord) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class SafeFileHandler(logging.FileHandler):
|
||||
@override
|
||||
def handleError(self, record: logging.LogRecord) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class TextFormatter(logging.Formatter):
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
parts = [_timestamp(record.created), record.levelname, record.getMessage()]
|
||||
attrs = _record_attrs(record)
|
||||
if attrs is not None:
|
||||
for key in sorted(attrs):
|
||||
parts.append(f'{key}={json.dumps(attrs[key], separators=(",", ":"))}')
|
||||
return ' '.join(parts)
|
||||
|
||||
|
||||
class JsonFormatter(logging.Formatter):
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
payload: dict[str, object] = {
|
||||
'timestamp': _timestamp(record.created),
|
||||
'level': record.levelname,
|
||||
'message': record.getMessage(),
|
||||
}
|
||||
attrs = _record_attrs(record)
|
||||
if attrs is not None:
|
||||
payload['attrs'] = attrs
|
||||
return json.dumps(payload, separators=(',', ':'), sort_keys=True)
|
||||
|
||||
|
||||
def _build_logger(
|
||||
name: str,
|
||||
level: str,
|
||||
output_format: str,
|
||||
handler: logging.Handler,
|
||||
) -> logging.Logger:
|
||||
logger = logging.Logger(name)
|
||||
logger.setLevel(_log_level_value(level))
|
||||
logger.propagate = False
|
||||
handler.setLevel(_log_level_value(level))
|
||||
handler.setFormatter(_formatter(output_format))
|
||||
logger.addHandler(handler)
|
||||
return logger
|
||||
|
||||
|
||||
def _formatter(output_format: str) -> logging.Formatter:
|
||||
normalized = output_format.strip().lower()
|
||||
if normalized not in {'text', 'json'}:
|
||||
raise ValueError('invalid log format')
|
||||
if normalized == 'json':
|
||||
return JsonFormatter()
|
||||
return TextFormatter()
|
||||
|
||||
|
||||
def _record_attrs(record: logging.LogRecord) -> dict[str, AttrValue] | None:
|
||||
attrs = getattr(record, 'attrs', None)
|
||||
if not isinstance(attrs, dict):
|
||||
return None
|
||||
return attrs
|
||||
|
||||
|
||||
def _timestamp(created: float) -> str:
|
||||
return (
|
||||
datetime.fromtimestamp(created, tz=UTC)
|
||||
.isoformat(timespec='milliseconds')
|
||||
.replace('+00:00', 'Z')
|
||||
)
|
||||
|
||||
|
||||
def _log_level_value(level: str) -> int:
|
||||
normalized = level.strip().upper()
|
||||
if normalized not in _LEVELS:
|
||||
raise ValueError('invalid log level')
|
||||
return _LEVELS[normalized]
|
||||
54
adapter/observability/noop.py
Normal file
54
adapter/observability/noop.py
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
from types import TracebackType
|
||||
|
||||
from usecase.interface import Attrs, AttrValue
|
||||
|
||||
|
||||
class NoopMetrics:
|
||||
def increment(
|
||||
self,
|
||||
name: str,
|
||||
value: int = 1,
|
||||
attrs: Attrs | None = None,
|
||||
) -> None:
|
||||
return None
|
||||
|
||||
def record(
|
||||
self,
|
||||
name: str,
|
||||
value: float,
|
||||
attrs: Attrs | None = None,
|
||||
) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class NoopSpan:
|
||||
def set_attribute(self, name: str, value: AttrValue) -> None:
|
||||
return None
|
||||
|
||||
def record_error(self, error: Exception) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class NoopSpanContext:
|
||||
def __init__(self) -> None:
|
||||
self._span = NoopSpan()
|
||||
|
||||
def __enter__(self) -> NoopSpan:
|
||||
return self._span
|
||||
|
||||
def __exit__(
|
||||
self,
|
||||
exc_type: type[BaseException] | None,
|
||||
exc: BaseException | None,
|
||||
traceback: TracebackType | None,
|
||||
) -> bool | None:
|
||||
return None
|
||||
|
||||
|
||||
class NoopTracer:
|
||||
def start_span(
|
||||
self,
|
||||
name: str,
|
||||
attrs: Attrs | None = None,
|
||||
) -> NoopSpanContext:
|
||||
return NoopSpanContext()
|
||||
28
adapter/observability/runtime.py
Normal file
28
adapter/observability/runtime.py
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from opentelemetry.sdk.metrics import MeterProvider
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
|
||||
from usecase.interface import Logger, Metrics, Tracer
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ObservabilityRuntime:
|
||||
logger: Logger
|
||||
metrics: Metrics
|
||||
tracer: Tracer
|
||||
meter_provider: MeterProvider | None = None
|
||||
tracer_provider: TracerProvider | None = None
|
||||
_shutdown_callbacks: tuple[Callable[[], None], ...] = ()
|
||||
_is_shutdown: bool = field(default=False, init=False, repr=False)
|
||||
|
||||
def shutdown(self) -> None:
|
||||
if self._is_shutdown:
|
||||
return
|
||||
|
||||
try:
|
||||
for callback in self._shutdown_callbacks:
|
||||
callback()
|
||||
finally:
|
||||
self._is_shutdown = True
|
||||
Loading…
Add table
Add a link
Reference in a new issue