Compare commits
2 Commits
02f5ef93b0
...
6a3da82ddc
| Author | SHA1 | Date | |
|---|---|---|---|
| 6a3da82ddc | |||
| 798b6e38bd |
@@ -26,7 +26,7 @@ jobs:
|
||||
echo "APP_PORT=8004"
|
||||
echo "POSTGRES_PORT=5433"
|
||||
echo "REDIS_PORT=6380"
|
||||
echo 'METRICS_FILTER_EXPRESSION=not IsMatch(resource.attributes["container.name"], "-stage$")'
|
||||
echo 'METRICS_FILTER_EXPRESSION=resource.attributes["container.name"] != nil and not IsMatch(resource.attributes["container.name"], "-stage$")'
|
||||
echo "COMPOSE_PROJECT=g2s-aggregator-stage"
|
||||
echo "DEPLOY_DIR=/home/deploy/g2s-aggregator-stage"
|
||||
} >> "$GITHUB_ENV"
|
||||
@@ -37,7 +37,7 @@ jobs:
|
||||
echo "APP_PORT=8003"
|
||||
echo "POSTGRES_PORT=5432"
|
||||
echo "REDIS_PORT=6379"
|
||||
echo 'METRICS_FILTER_EXPRESSION=IsMatch(resource.attributes["container.name"], "-stage$")'
|
||||
echo 'METRICS_FILTER_EXPRESSION=resource.attributes["container.name"] != nil and IsMatch(resource.attributes["container.name"], "-stage$")'
|
||||
echo "COMPOSE_PROJECT=g2s-aggregator"
|
||||
echo "DEPLOY_DIR=/home/deploy/g2s-aggregator"
|
||||
} >> "$GITHUB_ENV"
|
||||
@@ -125,5 +125,6 @@ jobs:
|
||||
-u ${{ secrets.REGISTRY_USER }} \
|
||||
-p ${{ secrets.REGISTRY_TOKEN }} &&
|
||||
docker compose -p $COMPOSE_PROJECT pull &&
|
||||
docker compose -p $COMPOSE_PROJECT up -d --force-recreate otel-collector &&
|
||||
docker compose -p $COMPOSE_PROJECT up -d
|
||||
"
|
||||
|
||||
@@ -18,6 +18,7 @@ from app.config import Settings
|
||||
from app.controllers.http_client import build_controller_http_client
|
||||
from app.repositories.cache.redis_cache import PriceCache
|
||||
from app.repositories.order import OrderRepository
|
||||
from app.runtime.metrics import get_cache_metrics
|
||||
from app.schemas.payment import (
|
||||
InitPaymentRequest,
|
||||
InitPaymentResponse,
|
||||
@@ -75,7 +76,10 @@ def _build_aggregator_service(settings: Settings) -> AggregatorService:
|
||||
config=settings.address_suggestions.tomtom,
|
||||
)
|
||||
providers = (cdek_provider, cse_provider)
|
||||
cache = PriceCache.from_repository_config(settings.repository)
|
||||
cache = PriceCache.from_repository_config(
|
||||
settings.repository,
|
||||
metrics=get_cache_metrics(),
|
||||
)
|
||||
postgres_engine = create_postgres_engine(settings.postgres)
|
||||
postgres_session_factory = create_postgres_session_factory(postgres_engine)
|
||||
order_repository = OrderRepository(session_factory=postgres_session_factory)
|
||||
|
||||
@@ -5,12 +5,14 @@ from fastapi import FastAPI
|
||||
from app.config import Settings, get_settings
|
||||
from app.controllers.v1.delivery import router as delivery_router
|
||||
from app.runtime.logging import configure_logging
|
||||
from app.runtime.metrics import configure_metrics
|
||||
from app.runtime.tracing import configure_tracing
|
||||
|
||||
|
||||
def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
configure_logging()
|
||||
resolved_settings = settings or get_settings()
|
||||
configure_metrics(resolved_settings.observability)
|
||||
|
||||
application = FastAPI(title="G2S Aggregator", version="0.1.0")
|
||||
application.state.settings = resolved_settings
|
||||
|
||||
+52
-2
@@ -7,10 +7,14 @@ import json
|
||||
from typing import Any, Callable, Protocol
|
||||
|
||||
from pydantic import BaseModel
|
||||
import structlog
|
||||
|
||||
from app.config import RepositoryConfig
|
||||
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
class RedisClientProtocol(Protocol):
|
||||
async def get(self, key: str) -> bytes | str | None: ...
|
||||
|
||||
@@ -19,6 +23,14 @@ class RedisClientProtocol(Protocol):
|
||||
async def delete(self, key: str) -> Any: ...
|
||||
|
||||
|
||||
class CacheMetricsProtocol(Protocol):
|
||||
def record_request(self, *, operation: str, outcome: str) -> None: ...
|
||||
|
||||
def record_error(self, *, operation: str, error_type: str) -> None: ...
|
||||
|
||||
def record_hit(self, *, hit: bool) -> None: ...
|
||||
|
||||
|
||||
class PriceCacheRepositoryError(RuntimeError):
|
||||
"""Deterministic repository error raised on Redis operation failures."""
|
||||
|
||||
@@ -26,9 +38,16 @@ class PriceCacheRepositoryError(RuntimeError):
|
||||
class PriceCache:
|
||||
"""Repository for cached provider payloads."""
|
||||
|
||||
def __init__(self, redis_client: RedisClientProtocol, *, ttl_seconds: int) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
redis_client: RedisClientProtocol,
|
||||
*,
|
||||
ttl_seconds: int,
|
||||
metrics: CacheMetricsProtocol | None = None,
|
||||
) -> None:
|
||||
self._redis_client = redis_client
|
||||
self._ttl_seconds = ttl_seconds
|
||||
self._metrics = metrics
|
||||
|
||||
@classmethod
|
||||
def from_repository_config(
|
||||
@@ -36,20 +55,25 @@ class PriceCache:
|
||||
repository_config: RepositoryConfig,
|
||||
*,
|
||||
client_factory: Callable[[str], RedisClientProtocol] | None = None,
|
||||
metrics: CacheMetricsProtocol | None = None,
|
||||
) -> "PriceCache":
|
||||
resolved_factory = client_factory or _default_redis_client_factory
|
||||
redis_client = resolved_factory(repository_config.redis_dsn)
|
||||
return cls(
|
||||
redis_client=redis_client,
|
||||
ttl_seconds=repository_config.price_cache_ttl_seconds,
|
||||
metrics=metrics,
|
||||
)
|
||||
|
||||
async def get(self, key: str) -> Any | None:
|
||||
try:
|
||||
payload = await self._redis_client.get(key)
|
||||
except Exception as exc:
|
||||
self._record_failure(operation="get", error=exc)
|
||||
raise PriceCacheRepositoryError("Redis cache read failed.") from exc
|
||||
if payload is None:
|
||||
self._record_request(operation="get", outcome="miss")
|
||||
self._record_hit(hit=False)
|
||||
return None
|
||||
|
||||
if isinstance(payload, bytes):
|
||||
@@ -59,7 +83,10 @@ class PriceCache:
|
||||
else:
|
||||
raise TypeError("Redis cache payload must be bytes or str.")
|
||||
|
||||
return json.loads(serialized_payload)
|
||||
result = json.loads(serialized_payload)
|
||||
self._record_request(operation="get", outcome="hit")
|
||||
self._record_hit(hit=True)
|
||||
return result
|
||||
|
||||
async def set(self, key: str, value: Any, ttl: int | None = None) -> None:
|
||||
serialized_payload = _serialize(value)
|
||||
@@ -67,13 +94,36 @@ class PriceCache:
|
||||
try:
|
||||
await self._redis_client.set(key, serialized_payload, ex=resolved_ttl)
|
||||
except Exception as exc:
|
||||
self._record_failure(operation="set", error=exc)
|
||||
raise PriceCacheRepositoryError("Redis cache write failed.") from exc
|
||||
self._record_request(operation="set", outcome="success")
|
||||
|
||||
async def invalidate(self, key: str) -> None:
|
||||
try:
|
||||
await self._redis_client.delete(key)
|
||||
except Exception as exc:
|
||||
self._record_failure(operation="invalidate", error=exc)
|
||||
raise PriceCacheRepositoryError("Redis cache invalidate failed.") from exc
|
||||
self._record_request(operation="invalidate", outcome="success")
|
||||
|
||||
def _record_failure(self, *, operation: str, error: Exception) -> None:
|
||||
error_type = type(error).__name__
|
||||
logger.warning(
|
||||
"cache.operation_failed",
|
||||
operation=operation,
|
||||
error_type=error_type,
|
||||
)
|
||||
if self._metrics is not None:
|
||||
self._metrics.record_request(operation=operation, outcome="error")
|
||||
self._metrics.record_error(operation=operation, error_type=error_type)
|
||||
|
||||
def _record_request(self, *, operation: str, outcome: str) -> None:
|
||||
if self._metrics is not None:
|
||||
self._metrics.record_request(operation=operation, outcome=outcome)
|
||||
|
||||
def _record_hit(self, *, hit: bool) -> None:
|
||||
if self._metrics is not None:
|
||||
self._metrics.record_hit(hit=hit)
|
||||
|
||||
|
||||
def _default_redis_client_factory(redis_dsn: str) -> RedisClientProtocol:
|
||||
|
||||
+29
-2
@@ -7,6 +7,7 @@ from collections.abc import Iterable
|
||||
from typing import Any, TextIO
|
||||
|
||||
import structlog
|
||||
from opentelemetry import trace
|
||||
|
||||
|
||||
_UVICORN_LOGGER_NAMES = ("uvicorn", "uvicorn.error", "uvicorn.access")
|
||||
@@ -19,7 +20,11 @@ def configure_logging(
|
||||
) -> None:
|
||||
resolved_stream = sys.stdout if stream is None else stream
|
||||
formatter = structlog.stdlib.ProcessorFormatter(
|
||||
foreign_pre_chain=list(_shared_processors()),
|
||||
foreign_pre_chain=[
|
||||
_add_trace_context,
|
||||
_render_context_in_event,
|
||||
*_shared_processors(),
|
||||
],
|
||||
processors=[
|
||||
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
|
||||
structlog.processors.JSONRenderer(sort_keys=True, ensure_ascii=False),
|
||||
@@ -33,6 +38,7 @@ def configure_logging(
|
||||
structlog.configure(
|
||||
processors=[
|
||||
structlog.stdlib.PositionalArgumentsFormatter(),
|
||||
_add_trace_context,
|
||||
_render_context_in_event,
|
||||
*_shared_processors(),
|
||||
structlog.processors.StackInfoRenderer(),
|
||||
@@ -62,6 +68,20 @@ def _shared_processors() -> tuple[structlog.types.Processor, ...]:
|
||||
)
|
||||
|
||||
|
||||
def _add_trace_context(
|
||||
_logger: Any,
|
||||
_method_name: str,
|
||||
event_dict: structlog.types.EventDict,
|
||||
) -> structlog.types.EventDict:
|
||||
span_context = trace.get_current_span().get_span_context()
|
||||
if not span_context.is_valid:
|
||||
return event_dict
|
||||
|
||||
event_dict["trace_id"] = format(span_context.trace_id, "032x")
|
||||
event_dict["span_id"] = format(span_context.span_id, "016x")
|
||||
return event_dict
|
||||
|
||||
|
||||
def _render_context_in_event(
|
||||
_logger: Any,
|
||||
_method_name: str,
|
||||
@@ -71,7 +91,14 @@ def _render_context_in_event(
|
||||
context = " ".join(
|
||||
f"{key}={_serialize_log_value(value)}"
|
||||
for key, value in sorted(event_dict.items())
|
||||
if key not in {"event", "exc_info", "stack_info"}
|
||||
if key
|
||||
not in {
|
||||
"event",
|
||||
"exc_info",
|
||||
"stack_info",
|
||||
"_record",
|
||||
"_from_structlog",
|
||||
}
|
||||
)
|
||||
event_dict["event"] = f"{event} {context}" if context else event
|
||||
return event_dict
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
"""OpenTelemetry metrics bootstrap and application instruments."""
|
||||
|
||||
from opentelemetry import metrics
|
||||
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
|
||||
from opentelemetry.sdk.metrics import MeterProvider
|
||||
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
|
||||
from app.config import ObservabilityConfig
|
||||
|
||||
|
||||
_METER_PROVIDER: MeterProvider | None = None
|
||||
_CACHE_METRICS: "CacheMetrics | None" = None
|
||||
|
||||
|
||||
class CacheMetrics:
|
||||
"""Metrics emitted by the Redis-backed price cache."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
meter = metrics.get_meter("app.repositories.cache")
|
||||
self._requests = meter.create_counter(
|
||||
"cache_requests_total",
|
||||
unit="{request}",
|
||||
description="Number of cache operations by operation and outcome.",
|
||||
)
|
||||
self._errors = meter.create_counter(
|
||||
"cache_errors_total",
|
||||
unit="{error}",
|
||||
description="Number of failed cache operations by error type.",
|
||||
)
|
||||
self._hit_ratio = meter.create_histogram(
|
||||
"cache_hit_ratio",
|
||||
unit="1",
|
||||
description="Cache lookup result: 1 for a hit and 0 for a miss.",
|
||||
)
|
||||
|
||||
def record_request(self, *, operation: str, outcome: str) -> None:
|
||||
self._requests.add(1, {"operation": operation, "outcome": outcome})
|
||||
|
||||
def record_error(self, *, operation: str, error_type: str) -> None:
|
||||
self._errors.add(1, {"operation": operation, "error.type": error_type})
|
||||
|
||||
def record_hit(self, *, hit: bool) -> None:
|
||||
self._hit_ratio.record(1.0 if hit else 0.0)
|
||||
|
||||
|
||||
def configure_metrics(observability: ObservabilityConfig) -> None:
|
||||
"""Configure the process-wide OTLP metrics exporter once."""
|
||||
|
||||
global _METER_PROVIDER
|
||||
|
||||
if not observability.enabled or _METER_PROVIDER is not None:
|
||||
return
|
||||
|
||||
meter_provider = _build_meter_provider(observability)
|
||||
metrics.set_meter_provider(meter_provider)
|
||||
_METER_PROVIDER = meter_provider
|
||||
|
||||
|
||||
def get_cache_metrics() -> CacheMetrics:
|
||||
"""Return the process-wide cache instruments."""
|
||||
|
||||
global _CACHE_METRICS
|
||||
|
||||
if _CACHE_METRICS is None:
|
||||
_CACHE_METRICS = CacheMetrics()
|
||||
return _CACHE_METRICS
|
||||
|
||||
|
||||
def _build_meter_provider(observability: ObservabilityConfig) -> MeterProvider:
|
||||
exporter = OTLPMetricExporter(
|
||||
endpoint=observability.otlp_endpoint,
|
||||
insecure=observability.otlp_insecure,
|
||||
)
|
||||
reader = PeriodicExportingMetricReader(exporter)
|
||||
resource = Resource.create({"service.name": observability.service_name})
|
||||
return MeterProvider(resource=resource, metric_readers=[reader])
|
||||
@@ -11,7 +11,7 @@ business_logic:
|
||||
provider_price_multiplier: 1.0
|
||||
|
||||
repository:
|
||||
redis_dsn: "redis://localhost:6379/0"
|
||||
redis_dsn: "redis://redis:6379/0"
|
||||
price_cache_ttl_seconds: 900
|
||||
|
||||
adapter:
|
||||
|
||||
@@ -47,6 +47,12 @@ receivers:
|
||||
parse_from: attributes.log
|
||||
parse_to: attributes
|
||||
on_error: send
|
||||
- type: trace_parser
|
||||
trace_id:
|
||||
parse_from: attributes.trace_id
|
||||
span_id:
|
||||
parse_from: attributes.span_id
|
||||
on_error: send
|
||||
- type: time_parser
|
||||
parse_from: attributes.timestamp
|
||||
layout: '%Y-%m-%dT%H:%M:%S.%fZ'
|
||||
@@ -104,6 +110,6 @@ service:
|
||||
processors: [resource/env, resource/version]
|
||||
exporters: [otlp]
|
||||
metrics:
|
||||
receivers: [docker_stats]
|
||||
receivers: [otlp, docker_stats]
|
||||
processors: [filter/environment, resource/env]
|
||||
exporters: [otlp]
|
||||
|
||||
@@ -122,8 +122,11 @@ def test_post_address_suggest_uses_registered_provider_in_default_dependency(
|
||||
|
||||
class StubPriceCache:
|
||||
@classmethod
|
||||
def from_repository_config(cls, repository_config) -> StubCache:
|
||||
def from_repository_config(
|
||||
cls, repository_config, *, metrics
|
||||
) -> StubCache:
|
||||
_ = repository_config
|
||||
assert metrics is delivery_controller.get_cache_metrics()
|
||||
return StubCache()
|
||||
|
||||
stub_http_client = StubHttpClient()
|
||||
|
||||
@@ -138,8 +138,11 @@ def test_post_delivery_price_uses_registered_provider_in_default_dependency(
|
||||
|
||||
class StubPriceCache:
|
||||
@classmethod
|
||||
def from_repository_config(cls, repository_config) -> StubCache:
|
||||
def from_repository_config(
|
||||
cls, repository_config, *, metrics
|
||||
) -> StubCache:
|
||||
_ = repository_config
|
||||
assert metrics is delivery_controller.get_cache_metrics()
|
||||
return stub_cache
|
||||
|
||||
stub_http_client = StubHttpClient()
|
||||
|
||||
@@ -3,6 +3,8 @@ import logging
|
||||
from io import StringIO
|
||||
|
||||
import structlog
|
||||
from opentelemetry import trace
|
||||
from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags
|
||||
|
||||
from app.runtime.logging import configure_logging
|
||||
|
||||
@@ -20,6 +22,8 @@ def test_application_log_is_serialized_as_json() -> None:
|
||||
assert payload["level"] == "info"
|
||||
assert payload["logger"] == "app.test"
|
||||
assert isinstance(payload["timestamp"], str)
|
||||
assert "trace_id" not in payload
|
||||
assert "span_id" not in payload
|
||||
|
||||
_reset_logging()
|
||||
|
||||
@@ -75,6 +79,49 @@ def test_uvicorn_loggers_use_shared_json_logging_setup() -> None:
|
||||
_reset_logging()
|
||||
|
||||
|
||||
def test_active_trace_context_is_added_to_structlog_and_standard_logs() -> None:
|
||||
stream = StringIO()
|
||||
trace_id = 0x1234567890ABCDEF1234567890ABCDEF
|
||||
span_id = 0x1234567890ABCDEF
|
||||
span = NonRecordingSpan(
|
||||
SpanContext(
|
||||
trace_id=trace_id,
|
||||
span_id=span_id,
|
||||
is_remote=False,
|
||||
trace_flags=TraceFlags.SAMPLED,
|
||||
)
|
||||
)
|
||||
|
||||
configure_logging(stream=stream)
|
||||
|
||||
with trace.use_span(span, end_on_exit=False):
|
||||
structlog.get_logger("app.test").info("structured_event")
|
||||
logging.getLogger("app.test").info("standard_event")
|
||||
|
||||
records = [json.loads(line) for line in stream.getvalue().splitlines()]
|
||||
expected_trace_id = "1234567890abcdef1234567890abcdef"
|
||||
expected_span_id = "1234567890abcdef"
|
||||
|
||||
assert [record["trace_id"] for record in records] == [
|
||||
expected_trace_id,
|
||||
expected_trace_id,
|
||||
]
|
||||
assert [record["span_id"] for record in records] == [
|
||||
expected_span_id,
|
||||
expected_span_id,
|
||||
]
|
||||
assert records[0]["event"] == (
|
||||
f'structured_event span_id="{expected_span_id}" '
|
||||
f'trace_id="{expected_trace_id}"'
|
||||
)
|
||||
assert records[1]["event"] == (
|
||||
f'standard_event span_id="{expected_span_id}" '
|
||||
f'trace_id="{expected_trace_id}"'
|
||||
)
|
||||
|
||||
_reset_logging()
|
||||
|
||||
|
||||
def _reset_logging() -> None:
|
||||
structlog.reset_defaults()
|
||||
root_logger = logging.getLogger()
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
import importlib
|
||||
|
||||
from app.config import ObservabilityConfig
|
||||
|
||||
|
||||
def _reload_metrics_module():
|
||||
metrics_module = importlib.import_module("app.runtime.metrics")
|
||||
return importlib.reload(metrics_module)
|
||||
|
||||
|
||||
def _make_observability_config(**overrides) -> ObservabilityConfig:
|
||||
payload = {
|
||||
"enabled": True,
|
||||
"service_name": "g2s-aggregator",
|
||||
"otlp_endpoint": "http://collector:4317",
|
||||
"otlp_insecure": True,
|
||||
}
|
||||
payload.update(overrides)
|
||||
return ObservabilityConfig(**payload)
|
||||
|
||||
|
||||
def test_configure_metrics_skips_bootstrap_when_disabled(monkeypatch) -> None:
|
||||
metrics_module = _reload_metrics_module()
|
||||
calls: list[object] = []
|
||||
monkeypatch.setattr(
|
||||
metrics_module,
|
||||
"_build_meter_provider",
|
||||
lambda observability: calls.append(observability),
|
||||
)
|
||||
|
||||
metrics_module.configure_metrics(_make_observability_config(enabled=False))
|
||||
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_build_meter_provider_uses_otlp_exporter_and_service_name(monkeypatch) -> None:
|
||||
metrics_module = _reload_metrics_module()
|
||||
created: dict[str, object] = {}
|
||||
|
||||
class FakeExporter:
|
||||
def __init__(self, *, endpoint: str, insecure: bool) -> None:
|
||||
created["endpoint"] = endpoint
|
||||
created["insecure"] = insecure
|
||||
|
||||
class FakeReader:
|
||||
def __init__(self, exporter: object) -> None:
|
||||
created["exporter"] = exporter
|
||||
|
||||
class FakeMeterProvider:
|
||||
def __init__(self, *, resource, metric_readers: list[object]) -> None:
|
||||
self.resource = resource
|
||||
self.metric_readers = metric_readers
|
||||
|
||||
monkeypatch.setattr(metrics_module, "OTLPMetricExporter", FakeExporter)
|
||||
monkeypatch.setattr(metrics_module, "PeriodicExportingMetricReader", FakeReader)
|
||||
monkeypatch.setattr(metrics_module, "MeterProvider", FakeMeterProvider)
|
||||
|
||||
provider = metrics_module._build_meter_provider(
|
||||
_make_observability_config(
|
||||
otlp_endpoint="http://metrics-backend:4317",
|
||||
otlp_insecure=False,
|
||||
)
|
||||
)
|
||||
|
||||
assert created["endpoint"] == "http://metrics-backend:4317"
|
||||
assert created["insecure"] is False
|
||||
assert provider.resource.attributes["service.name"] == "g2s-aggregator"
|
||||
assert len(provider.metric_readers) == 1
|
||||
assert created["exporter"] is not None
|
||||
|
||||
|
||||
def test_cache_metrics_create_and_record_expected_instruments(monkeypatch) -> None:
|
||||
metrics_module = _reload_metrics_module()
|
||||
created: dict[str, object] = {}
|
||||
|
||||
class FakeInstrument:
|
||||
def __init__(self, name: str) -> None:
|
||||
self.name = name
|
||||
self.records: list[tuple[float, dict[str, str] | None]] = []
|
||||
|
||||
def add(self, value: float, attributes: dict[str, str]) -> None:
|
||||
self.records.append((value, attributes))
|
||||
|
||||
def record(
|
||||
self, value: float, attributes: dict[str, str] | None = None
|
||||
) -> None:
|
||||
self.records.append((value, attributes))
|
||||
|
||||
class FakeMeter:
|
||||
def create_counter(self, name: str, **kwargs) -> FakeInstrument:
|
||||
del kwargs
|
||||
instrument = FakeInstrument(name)
|
||||
created[name] = instrument
|
||||
return instrument
|
||||
|
||||
def create_histogram(self, name: str, **kwargs) -> FakeInstrument:
|
||||
del kwargs
|
||||
instrument = FakeInstrument(name)
|
||||
created[name] = instrument
|
||||
return instrument
|
||||
|
||||
monkeypatch.setattr(metrics_module.metrics, "get_meter", lambda _name: FakeMeter())
|
||||
|
||||
cache_metrics = metrics_module.CacheMetrics()
|
||||
cache_metrics.record_request(operation="get", outcome="hit")
|
||||
cache_metrics.record_error(operation="set", error_type="ConnectionError")
|
||||
cache_metrics.record_hit(hit=True)
|
||||
cache_metrics.record_hit(hit=False)
|
||||
|
||||
assert set(created) == {
|
||||
"cache_requests_total",
|
||||
"cache_errors_total",
|
||||
"cache_hit_ratio",
|
||||
}
|
||||
assert created["cache_requests_total"].records == [
|
||||
(1, {"operation": "get", "outcome": "hit"})
|
||||
]
|
||||
assert created["cache_errors_total"].records == [
|
||||
(1, {"operation": "set", "error.type": "ConnectionError"})
|
||||
]
|
||||
assert created["cache_hit_ratio"].records == [(1.0, None), (0.0, None)]
|
||||
+67
@@ -69,6 +69,22 @@ class FailingSetRedisClient:
|
||||
return 1
|
||||
|
||||
|
||||
class RecordingCacheMetrics:
|
||||
def __init__(self) -> None:
|
||||
self.requests: list[tuple[str, str]] = []
|
||||
self.errors: list[tuple[str, str]] = []
|
||||
self.hits: list[bool] = []
|
||||
|
||||
def record_request(self, *, operation: str, outcome: str) -> None:
|
||||
self.requests.append((operation, outcome))
|
||||
|
||||
def record_error(self, *, operation: str, error_type: str) -> None:
|
||||
self.errors.append((operation, error_type))
|
||||
|
||||
def record_hit(self, *, hit: bool) -> None:
|
||||
self.hits.append(hit)
|
||||
|
||||
|
||||
def test_cache_miss_returns_none() -> None:
|
||||
cache = PriceCache(InMemoryRedisClient(), ttl_seconds=120)
|
||||
|
||||
@@ -263,3 +279,54 @@ def test_set_raises_deterministic_repository_error_on_redis_failure() -> None:
|
||||
|
||||
with pytest.raises(PriceCacheRepositoryError, match="Redis cache write failed."):
|
||||
asyncio.run(cache.set("key", {"v": 1}))
|
||||
|
||||
|
||||
def test_cache_metrics_record_hits_misses_and_successful_writes() -> None:
|
||||
redis_client = InMemoryRedisClient()
|
||||
metrics = RecordingCacheMetrics()
|
||||
cache = PriceCache(redis_client, ttl_seconds=10, metrics=metrics)
|
||||
|
||||
async def scenario() -> None:
|
||||
await cache.get("missing")
|
||||
await cache.set("key", {"v": 1})
|
||||
await cache.get("key")
|
||||
await cache.invalidate("key")
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
assert metrics.requests == [
|
||||
("get", "miss"),
|
||||
("set", "success"),
|
||||
("get", "hit"),
|
||||
("invalidate", "success"),
|
||||
]
|
||||
assert metrics.errors == []
|
||||
assert metrics.hits == [False, True]
|
||||
|
||||
|
||||
def test_cache_failure_records_metrics_and_structured_log(monkeypatch) -> None:
|
||||
events: list[tuple[str, dict[str, object]]] = []
|
||||
metrics = RecordingCacheMetrics()
|
||||
cache = PriceCache(FailingGetRedisClient(), ttl_seconds=10, metrics=metrics)
|
||||
|
||||
class StubLogger:
|
||||
def warning(self, event: str, **context: object) -> None:
|
||||
events.append((event, context))
|
||||
|
||||
monkeypatch.setattr(
|
||||
"app.repositories.cache.redis_cache.logger",
|
||||
StubLogger(),
|
||||
)
|
||||
|
||||
with pytest.raises(PriceCacheRepositoryError, match="Redis cache read failed."):
|
||||
asyncio.run(cache.get("key"))
|
||||
|
||||
assert metrics.requests == [("get", "error")]
|
||||
assert metrics.errors == [("get", "RuntimeError")]
|
||||
assert metrics.hits == []
|
||||
assert events == [
|
||||
(
|
||||
"cache.operation_failed",
|
||||
{"operation": "get", "error_type": "RuntimeError"},
|
||||
)
|
||||
]
|
||||
|
||||
@@ -7,6 +7,7 @@ from fastapi import FastAPI
|
||||
from app import config as config_module
|
||||
from app.config import Settings, get_settings
|
||||
from app.runtime import logging as logging_module
|
||||
from app.runtime import metrics as metrics_module
|
||||
from app.runtime import tracing as tracing_module
|
||||
|
||||
TEST_CONFIG_FILE = Path(__file__).resolve().parents[2] / "config.test.yaml"
|
||||
@@ -21,6 +22,7 @@ def _import_main_module(monkeypatch):
|
||||
|
||||
def test_app_import_smoke(monkeypatch) -> None:
|
||||
logging_bootstrap_calls: list[None] = []
|
||||
metrics_bootstrap_calls: list[object] = []
|
||||
tracing_bootstrap_calls: list[tuple[FastAPI, object]] = []
|
||||
|
||||
def fake_configure_logging() -> None:
|
||||
@@ -29,7 +31,11 @@ def test_app_import_smoke(monkeypatch) -> None:
|
||||
def fake_configure_tracing(app: FastAPI, observability: object) -> None:
|
||||
tracing_bootstrap_calls.append((app, observability))
|
||||
|
||||
def fake_configure_metrics(observability: object) -> None:
|
||||
metrics_bootstrap_calls.append(observability)
|
||||
|
||||
monkeypatch.setattr(logging_module, "configure_logging", fake_configure_logging)
|
||||
monkeypatch.setattr(metrics_module, "configure_metrics", fake_configure_metrics)
|
||||
monkeypatch.setattr(tracing_module, "configure_tracing", fake_configure_tracing)
|
||||
main_module = _import_main_module(monkeypatch)
|
||||
|
||||
@@ -39,6 +45,7 @@ def test_app_import_smoke(monkeypatch) -> None:
|
||||
route.path for route in main_module.app.routes
|
||||
}
|
||||
assert logging_bootstrap_calls == [None]
|
||||
assert metrics_bootstrap_calls == [main_module.app.state.settings.observability]
|
||||
assert tracing_bootstrap_calls == [
|
||||
(main_module.app, main_module.app.state.settings.observability)
|
||||
]
|
||||
@@ -47,6 +54,7 @@ def test_app_import_smoke(monkeypatch) -> None:
|
||||
|
||||
def test_app_created_with_provided_settings(monkeypatch) -> None:
|
||||
logging_bootstrap_calls: list[None] = []
|
||||
metrics_bootstrap_calls: list[object] = []
|
||||
tracing_bootstrap_calls: list[tuple[FastAPI, object]] = []
|
||||
|
||||
def fake_configure_logging() -> None:
|
||||
@@ -55,7 +63,11 @@ def test_app_created_with_provided_settings(monkeypatch) -> None:
|
||||
def fake_configure_tracing(app: FastAPI, observability: object) -> None:
|
||||
tracing_bootstrap_calls.append((app, observability))
|
||||
|
||||
def fake_configure_metrics(observability: object) -> None:
|
||||
metrics_bootstrap_calls.append(observability)
|
||||
|
||||
monkeypatch.setattr(logging_module, "configure_logging", fake_configure_logging)
|
||||
monkeypatch.setattr(metrics_module, "configure_metrics", fake_configure_metrics)
|
||||
monkeypatch.setattr(tracing_module, "configure_tracing", fake_configure_tracing)
|
||||
main_module = _import_main_module(monkeypatch)
|
||||
monkeypatch.setattr(config_module, "TEST_CONFIG_FILE", str(TEST_CONFIG_FILE))
|
||||
@@ -65,6 +77,10 @@ def test_app_created_with_provided_settings(monkeypatch) -> None:
|
||||
|
||||
assert instance.state.settings is settings
|
||||
assert logging_bootstrap_calls == [None, None]
|
||||
assert metrics_bootstrap_calls == [
|
||||
main_module.app.state.settings.observability,
|
||||
settings.observability,
|
||||
]
|
||||
assert tracing_bootstrap_calls == [
|
||||
(main_module.app, main_module.app.state.settings.observability),
|
||||
(instance, settings.observability),
|
||||
|
||||
@@ -6,6 +6,8 @@ import yaml
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
COMPOSE_FILE = PROJECT_ROOT / "docker-compose.yml"
|
||||
INFRA_README = PROJECT_ROOT / "infra" / "README.md"
|
||||
DEPLOY_WORKFLOW = PROJECT_ROOT / ".gitea" / "workflows" / "deploy.yml"
|
||||
COLLECTOR_TEMPLATE = PROJECT_ROOT / "otel-collector-config.template.yaml"
|
||||
|
||||
|
||||
def _load_compose() -> dict:
|
||||
@@ -28,3 +30,26 @@ def test_smoke_command_sequence_is_documented() -> None:
|
||||
assert "docker compose config" in readme
|
||||
assert "docker compose up -d redis" in readme
|
||||
assert "docker compose ps" in readme
|
||||
|
||||
|
||||
def test_deploy_recreates_collector_after_config_update() -> None:
|
||||
workflow = DEPLOY_WORKFLOW.read_text(encoding="utf-8")
|
||||
|
||||
recreate_command = (
|
||||
"docker compose -p $COMPOSE_PROJECT up -d --force-recreate otel-collector"
|
||||
)
|
||||
application_command = "docker compose -p $COMPOSE_PROJECT up -d"
|
||||
|
||||
assert recreate_command in workflow
|
||||
assert workflow.index(recreate_command) < workflow.rindex(application_command)
|
||||
|
||||
|
||||
def test_collector_promotes_log_trace_attributes_to_trace_context() -> None:
|
||||
config = yaml.safe_load(COLLECTOR_TEMPLATE.read_text(encoding="utf-8"))
|
||||
operators = config["receivers"]["filelog"]["operators"]
|
||||
trace_parser = next(
|
||||
operator for operator in operators if operator["type"] == "trace_parser"
|
||||
)
|
||||
|
||||
assert trace_parser["trace_id"]["parse_from"] == "attributes.trace_id"
|
||||
assert trace_parser["span_id"]["parse_from"] == "attributes.span_id"
|
||||
|
||||
Reference in New Issue
Block a user