пофикшен redis и логи redis

.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"

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)

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

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:

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])

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:

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]

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()

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()

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)]

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"},
+        )
+    ]

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),