platform/master
5
2
Fork 0
Code Issues 6 Pull requests Projects 1 Releases Packages Wiki Activity Actions

[feat] add observability config

Browse source
This commit is contained in:
Azamat 2026-03-20 21:38:28 +03:00
parent a930185754
commit cb4560db5c
4 changed files with 268 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

225
adapter/config/loader.py
View file

@ -10,8 +10,10 @@ from .model import (
AppSectionConfig,
HttpConfig,
LoggingConfig,
MetricsConfig,
OtelConfig,
SecurityConfig,
TracingConfig,
)
ROOT_DIR = Path(__file__).resolve().parents[2]
@ -34,9 +36,44 @@ def load_config(
app_section = _section(yaml_data, 'app')
http_section = _section(yaml_data, 'http')
logging_section = _section(yaml_data, 'logging')
otel_section = _section(yaml_data, 'otel')
metrics_section = _section(yaml_data, 'metrics')
tracing_section = _section(yaml_data, 'tracing')
security_section = _section(yaml_data, 'security')
logging_output = _yaml_or_env_choice(
logging_section,
'output',
'logging.output',
env_values,
{'stdout', 'file', 'otel'},
'APP_LOGGING_OUTPUT',
)
logging_format = _yaml_or_env_choice(
logging_section,
'format',
'logging.format',
env_values,
{'text', 'json'},
'APP_LOGGING_FORMAT',
)
metrics_enabled = _yaml_or_env_bool(
metrics_section,
'enabled',
'metrics.enabled',
env_values,
'APP_METRICS_ENABLED',
)
tracing_enabled = _yaml_or_env_bool(
tracing_section,
'enabled',
'tracing.enabled',
env_values,
'APP_TRACING_ENABLED',
)
return AppConfig(
app=AppSectionConfig(
name=_yaml_or_env_str(
@ -68,43 +105,28 @@ def load_config(
env_values,
'APP_LOGGING_LEVEL',
),
output=logging_output,
format=logging_format,
file_path=(
None
if logging_output != 'file'
else _yaml_or_env_str(
logging_section,
'file_path',
'logging.file_path',
env_values,
'APP_LOGGING_FILE_PATH',
)
),
),
otel=OtelConfig(
service_name=_yaml_or_env_str(
otel_section,
'service_name',
'otel.service_name',
env_values,
'APP_OTEL_SERVICE_NAME',
),
logs_endpoint=_yaml_or_env_str(
otel_section,
'logs_endpoint',
'otel.logs_endpoint',
env_values,
'APP_OTEL_LOGS_ENDPOINT',
),
metrics_endpoint=_yaml_or_env_str(
otel_section,
'metrics_endpoint',
'otel.metrics_endpoint',
env_values,
'APP_OTEL_METRICS_ENDPOINT',
),
traces_endpoint=_yaml_or_env_str(
otel_section,
'traces_endpoint',
'otel.traces_endpoint',
env_values,
'APP_OTEL_TRACES_ENDPOINT',
),
metric_export_interval=_yaml_or_env_int(
otel_section,
'metric_export_interval',
'otel.metric_export_interval',
env_values,
'OTEL_METRIC_EXPORT_INTERVAL',
),
metrics=MetricsConfig(enabled=metrics_enabled),
tracing=TracingConfig(enabled=tracing_enabled),
otel=_load_otel_config(
yaml_data,
env_values,
enable_logs=logging_output == 'otel',
enable_metrics=metrics_enabled,
enable_tracing=tracing_enabled,
),
security=SecurityConfig(
token_header=_yaml_or_env_str(
@ -184,6 +206,94 @@ def _section(data: Mapping[str, object], name: str) -> dict[str, object]:
return section
def _optional_section(data: Mapping[str, object], name: str) -> dict[str, object]:
value = data.get(name)
if value is None:
return {}
if not isinstance(value, dict):
raise ConfigError(f'invalid {name}')
section: dict[str, object] = {}
for key, item in value.items():
if not isinstance(key, str):
raise ConfigError(f'invalid {name}')
section[key] = item
return section
def _load_otel_config(
data: Mapping[str, object],
env: Mapping[str, str],
*,
enable_logs: bool,
enable_metrics: bool,
enable_tracing: bool,
) -> OtelConfig:
if not any((enable_logs, enable_metrics, enable_tracing)):
return OtelConfig(
service_name='',
logs_endpoint='',
metrics_endpoint='',
traces_endpoint='',
metric_export_interval=0,
)
otel_section = _optional_section(data, 'otel')
service_name = _yaml_or_env_str(
otel_section,
'service_name',
'otel.service_name',
env,
'APP_OTEL_SERVICE_NAME',
)
logs_endpoint = ''
metrics_endpoint = ''
traces_endpoint = ''
metric_export_interval = 0
if enable_logs:
logs_endpoint = _yaml_or_env_str(
otel_section,
'logs_endpoint',
'otel.logs_endpoint',
env,
'APP_OTEL_LOGS_ENDPOINT',
)
if enable_metrics:
metrics_endpoint = _yaml_or_env_str(
otel_section,
'metrics_endpoint',
'otel.metrics_endpoint',
env,
'APP_OTEL_METRICS_ENDPOINT',
)
metric_export_interval = _yaml_or_env_int(
otel_section,
'metric_export_interval',
'otel.metric_export_interval',
env,
'OTEL_METRIC_EXPORT_INTERVAL',
)
if enable_tracing:
traces_endpoint = _yaml_or_env_str(
otel_section,
'traces_endpoint',
'otel.traces_endpoint',
env,
'APP_OTEL_TRACES_ENDPOINT',
)
return OtelConfig(
service_name=service_name,
logs_endpoint=logs_endpoint,
metrics_endpoint=metrics_endpoint,
traces_endpoint=traces_endpoint,
metric_export_interval=metric_export_interval,
)
def _yaml_or_env_str(
section: Mapping[str, object],
field_name: str,
@ -212,6 +322,35 @@ def _yaml_or_env_int(
return _as_int(section[field_name], label)
def _yaml_or_env_bool(
section: Mapping[str, object],
field_name: str,
label: str,
env: Mapping[str, str],
env_name: str | None = None,
) -> bool:
if env_name is not None and env_name in env:
return _as_bool(env[env_name], env_name)
if field_name not in section:
raise ConfigError(f'missing {label}')
return _as_bool(section[field_name], label)
def _yaml_or_env_choice(
section: Mapping[str, object],
field_name: str,
label: str,
env: Mapping[str, str],
choices: set[str],
env_name: str | None = None,
) -> str:
value = _yaml_or_env_str(section, field_name, label, env, env_name)
normalized = value.lower()
if normalized not in choices:
raise ConfigError(f'invalid {label}')
return normalized
def _env_str(env: Mapping[str, str], name: str) -> str:
if name not in env:
raise ConfigError(f'missing {name}')
@ -244,6 +383,18 @@ def _as_int(value: object, label: str) -> int:
raise ConfigError(f'invalid {label}')
def _as_bool(value: object, label: str) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
text = value.strip().lower()
if text == 'true':
return True
if text == 'false':
return False
raise ConfigError(f'invalid {label}')
def _display_path(path: Path) -> str:
if path.is_relative_to(ROOT_DIR):
return str(path.relative_to(ROOT_DIR))