From 798b6e38bd5e8a9ca9d77a124b5001f55aa5270e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A0=D0=B0=D0=B8=D1=81=20=D0=AE=D1=81=D1=83=D0=BF=D0=B0?= =?UTF-8?q?=D0=BB=D0=B8=D0=B5=D0=B2?= Date: Sat, 20 Jun 2026 19:35:23 +0300 Subject: [PATCH] =?UTF-8?q?=D0=BF=D0=BE=D1=84=D0=B8=D0=BA=D1=88=D0=B5?= =?UTF-8?q?=D0=BD=20redis=20=D0=B8=20=D0=BB=D0=BE=D0=B3=D0=B8=20redis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitea/workflows/deploy.yml | 4 +- app/controllers/v1/delivery.py | 6 +- app/main.py | 2 + app/repositories/cache/redis_cache.py | 54 +++++++- app/runtime/metrics.py | 77 +++++++++++ config.template.yaml | 2 +- otel-collector-config.template.yaml | 2 +- .../v1/test_address_suggestions.py | 5 +- tests/controllers/v1/test_delivery.py | 5 +- tests/metrics/test_bootstrap.py | 121 ++++++++++++++++++ tests/repositories/cache/test_redis_cache.py | 67 ++++++++++ tests/smoke/test_app_import.py | 16 +++ 12 files changed, 352 insertions(+), 9 deletions(-) create mode 100644 app/runtime/metrics.py create mode 100644 tests/metrics/test_bootstrap.py diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 087268b..15aff59 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -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" diff --git a/app/controllers/v1/delivery.py b/app/controllers/v1/delivery.py index 7482408..62f5099 100644 --- a/app/controllers/v1/delivery.py +++ b/app/controllers/v1/delivery.py @@ -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) diff --git a/app/main.py b/app/main.py index f36c79b..9a8d011 100644 --- a/app/main.py +++ b/app/main.py @@ -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 diff --git a/app/repositories/cache/redis_cache.py b/app/repositories/cache/redis_cache.py index 3c4276c..c997ad8 100644 --- a/app/repositories/cache/redis_cache.py +++ b/app/repositories/cache/redis_cache.py @@ -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: diff --git a/app/runtime/metrics.py b/app/runtime/metrics.py new file mode 100644 index 0000000..c7c0470 --- /dev/null +++ b/app/runtime/metrics.py @@ -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]) diff --git a/config.template.yaml b/config.template.yaml index 5ce850e..3f647cc 100644 --- a/config.template.yaml +++ b/config.template.yaml @@ -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: diff --git a/otel-collector-config.template.yaml b/otel-collector-config.template.yaml index f29c5e5..7ddb655 100644 --- a/otel-collector-config.template.yaml +++ b/otel-collector-config.template.yaml @@ -104,6 +104,6 @@ service: processors: [resource/env, resource/version] exporters: [otlp] metrics: - receivers: [docker_stats] + receivers: [otlp, docker_stats] processors: [filter/environment, resource/env] exporters: [otlp] diff --git a/tests/controllers/v1/test_address_suggestions.py b/tests/controllers/v1/test_address_suggestions.py index 9073adb..df5459e 100644 --- a/tests/controllers/v1/test_address_suggestions.py +++ b/tests/controllers/v1/test_address_suggestions.py @@ -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() diff --git a/tests/controllers/v1/test_delivery.py b/tests/controllers/v1/test_delivery.py index fb4fdc9..8f8324e 100644 --- a/tests/controllers/v1/test_delivery.py +++ b/tests/controllers/v1/test_delivery.py @@ -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() diff --git a/tests/metrics/test_bootstrap.py b/tests/metrics/test_bootstrap.py new file mode 100644 index 0000000..9a206c4 --- /dev/null +++ b/tests/metrics/test_bootstrap.py @@ -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)] diff --git a/tests/repositories/cache/test_redis_cache.py b/tests/repositories/cache/test_redis_cache.py index 26bcfe2..7b46b8e 100644 --- a/tests/repositories/cache/test_redis_cache.py +++ b/tests/repositories/cache/test_redis_cache.py @@ -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"}, + ) + ] diff --git a/tests/smoke/test_app_import.py b/tests/smoke/test_app_import.py index 4d80bdc..bc866cb 100644 --- a/tests/smoke/test_app_import.py +++ b/tests/smoke/test_app_import.py @@ -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),