Compare commits

...

2 Commits

Author SHA1 Message Date
Раис Юсупалиев 6a3da82ddc span_id и trace_id
Deploy / deploy (push) Successful in 51s
2026-06-20 19:56:48 +03:00
Раис Юсупалиев 798b6e38bd пофикшен redis и логи redis
Deploy / deploy (push) Successful in 52s
2026-06-20 19:35:23 +03:00
15 changed files with 460 additions and 11 deletions
+3 -2
View File
@@ -26,7 +26,7 @@ jobs:
echo "APP_PORT=8004" echo "APP_PORT=8004"
echo "POSTGRES_PORT=5433" echo "POSTGRES_PORT=5433"
echo "REDIS_PORT=6380" 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 "COMPOSE_PROJECT=g2s-aggregator-stage"
echo "DEPLOY_DIR=/home/deploy/g2s-aggregator-stage" echo "DEPLOY_DIR=/home/deploy/g2s-aggregator-stage"
} >> "$GITHUB_ENV" } >> "$GITHUB_ENV"
@@ -37,7 +37,7 @@ jobs:
echo "APP_PORT=8003" echo "APP_PORT=8003"
echo "POSTGRES_PORT=5432" echo "POSTGRES_PORT=5432"
echo "REDIS_PORT=6379" 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 "COMPOSE_PROJECT=g2s-aggregator"
echo "DEPLOY_DIR=/home/deploy/g2s-aggregator" echo "DEPLOY_DIR=/home/deploy/g2s-aggregator"
} >> "$GITHUB_ENV" } >> "$GITHUB_ENV"
@@ -125,5 +125,6 @@ jobs:
-u ${{ secrets.REGISTRY_USER }} \ -u ${{ secrets.REGISTRY_USER }} \
-p ${{ secrets.REGISTRY_TOKEN }} && -p ${{ secrets.REGISTRY_TOKEN }} &&
docker compose -p $COMPOSE_PROJECT pull && docker compose -p $COMPOSE_PROJECT pull &&
docker compose -p $COMPOSE_PROJECT up -d --force-recreate otel-collector &&
docker compose -p $COMPOSE_PROJECT up -d docker compose -p $COMPOSE_PROJECT up -d
" "
+5 -1
View File
@@ -18,6 +18,7 @@ from app.config import Settings
from app.controllers.http_client import build_controller_http_client from app.controllers.http_client import build_controller_http_client
from app.repositories.cache.redis_cache import PriceCache from app.repositories.cache.redis_cache import PriceCache
from app.repositories.order import OrderRepository from app.repositories.order import OrderRepository
from app.runtime.metrics import get_cache_metrics
from app.schemas.payment import ( from app.schemas.payment import (
InitPaymentRequest, InitPaymentRequest,
InitPaymentResponse, InitPaymentResponse,
@@ -75,7 +76,10 @@ def _build_aggregator_service(settings: Settings) -> AggregatorService:
config=settings.address_suggestions.tomtom, config=settings.address_suggestions.tomtom,
) )
providers = (cdek_provider, cse_provider) 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_engine = create_postgres_engine(settings.postgres)
postgres_session_factory = create_postgres_session_factory(postgres_engine) postgres_session_factory = create_postgres_session_factory(postgres_engine)
order_repository = OrderRepository(session_factory=postgres_session_factory) order_repository = OrderRepository(session_factory=postgres_session_factory)
+2
View File
@@ -5,12 +5,14 @@ from fastapi import FastAPI
from app.config import Settings, get_settings from app.config import Settings, get_settings
from app.controllers.v1.delivery import router as delivery_router from app.controllers.v1.delivery import router as delivery_router
from app.runtime.logging import configure_logging from app.runtime.logging import configure_logging
from app.runtime.metrics import configure_metrics
from app.runtime.tracing import configure_tracing from app.runtime.tracing import configure_tracing
def create_app(settings: Settings | None = None) -> FastAPI: def create_app(settings: Settings | None = None) -> FastAPI:
configure_logging() configure_logging()
resolved_settings = settings or get_settings() resolved_settings = settings or get_settings()
configure_metrics(resolved_settings.observability)
application = FastAPI(title="G2S Aggregator", version="0.1.0") application = FastAPI(title="G2S Aggregator", version="0.1.0")
application.state.settings = resolved_settings application.state.settings = resolved_settings
+52 -2
View File
@@ -7,10 +7,14 @@ import json
from typing import Any, Callable, Protocol from typing import Any, Callable, Protocol
from pydantic import BaseModel from pydantic import BaseModel
import structlog
from app.config import RepositoryConfig from app.config import RepositoryConfig
logger = structlog.get_logger(__name__)
class RedisClientProtocol(Protocol): class RedisClientProtocol(Protocol):
async def get(self, key: str) -> bytes | str | None: ... async def get(self, key: str) -> bytes | str | None: ...
@@ -19,6 +23,14 @@ class RedisClientProtocol(Protocol):
async def delete(self, key: str) -> Any: ... 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): class PriceCacheRepositoryError(RuntimeError):
"""Deterministic repository error raised on Redis operation failures.""" """Deterministic repository error raised on Redis operation failures."""
@@ -26,9 +38,16 @@ class PriceCacheRepositoryError(RuntimeError):
class PriceCache: class PriceCache:
"""Repository for cached provider payloads.""" """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._redis_client = redis_client
self._ttl_seconds = ttl_seconds self._ttl_seconds = ttl_seconds
self._metrics = metrics
@classmethod @classmethod
def from_repository_config( def from_repository_config(
@@ -36,20 +55,25 @@ class PriceCache:
repository_config: RepositoryConfig, repository_config: RepositoryConfig,
*, *,
client_factory: Callable[[str], RedisClientProtocol] | None = None, client_factory: Callable[[str], RedisClientProtocol] | None = None,
metrics: CacheMetricsProtocol | None = None,
) -> "PriceCache": ) -> "PriceCache":
resolved_factory = client_factory or _default_redis_client_factory resolved_factory = client_factory or _default_redis_client_factory
redis_client = resolved_factory(repository_config.redis_dsn) redis_client = resolved_factory(repository_config.redis_dsn)
return cls( return cls(
redis_client=redis_client, redis_client=redis_client,
ttl_seconds=repository_config.price_cache_ttl_seconds, ttl_seconds=repository_config.price_cache_ttl_seconds,
metrics=metrics,
) )
async def get(self, key: str) -> Any | None: async def get(self, key: str) -> Any | None:
try: try:
payload = await self._redis_client.get(key) payload = await self._redis_client.get(key)
except Exception as exc: except Exception as exc:
self._record_failure(operation="get", error=exc)
raise PriceCacheRepositoryError("Redis cache read failed.") from exc raise PriceCacheRepositoryError("Redis cache read failed.") from exc
if payload is None: if payload is None:
self._record_request(operation="get", outcome="miss")
self._record_hit(hit=False)
return None return None
if isinstance(payload, bytes): if isinstance(payload, bytes):
@@ -59,7 +83,10 @@ class PriceCache:
else: else:
raise TypeError("Redis cache payload must be bytes or str.") 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: async def set(self, key: str, value: Any, ttl: int | None = None) -> None:
serialized_payload = _serialize(value) serialized_payload = _serialize(value)
@@ -67,13 +94,36 @@ class PriceCache:
try: try:
await self._redis_client.set(key, serialized_payload, ex=resolved_ttl) await self._redis_client.set(key, serialized_payload, ex=resolved_ttl)
except Exception as exc: except Exception as exc:
self._record_failure(operation="set", error=exc)
raise PriceCacheRepositoryError("Redis cache write failed.") from exc raise PriceCacheRepositoryError("Redis cache write failed.") from exc
self._record_request(operation="set", outcome="success")
async def invalidate(self, key: str) -> None: async def invalidate(self, key: str) -> None:
try: try:
await self._redis_client.delete(key) await self._redis_client.delete(key)
except Exception as exc: except Exception as exc:
self._record_failure(operation="invalidate", error=exc)
raise PriceCacheRepositoryError("Redis cache invalidate failed.") from 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: def _default_redis_client_factory(redis_dsn: str) -> RedisClientProtocol:
+29 -2
View File
@@ -7,6 +7,7 @@ from collections.abc import Iterable
from typing import Any, TextIO from typing import Any, TextIO
import structlog import structlog
from opentelemetry import trace
_UVICORN_LOGGER_NAMES = ("uvicorn", "uvicorn.error", "uvicorn.access") _UVICORN_LOGGER_NAMES = ("uvicorn", "uvicorn.error", "uvicorn.access")
@@ -19,7 +20,11 @@ def configure_logging(
) -> None: ) -> None:
resolved_stream = sys.stdout if stream is None else stream resolved_stream = sys.stdout if stream is None else stream
formatter = structlog.stdlib.ProcessorFormatter( formatter = structlog.stdlib.ProcessorFormatter(
foreign_pre_chain=list(_shared_processors()), foreign_pre_chain=[
_add_trace_context,
_render_context_in_event,
*_shared_processors(),
],
processors=[ processors=[
structlog.stdlib.ProcessorFormatter.remove_processors_meta, structlog.stdlib.ProcessorFormatter.remove_processors_meta,
structlog.processors.JSONRenderer(sort_keys=True, ensure_ascii=False), structlog.processors.JSONRenderer(sort_keys=True, ensure_ascii=False),
@@ -33,6 +38,7 @@ def configure_logging(
structlog.configure( structlog.configure(
processors=[ processors=[
structlog.stdlib.PositionalArgumentsFormatter(), structlog.stdlib.PositionalArgumentsFormatter(),
_add_trace_context,
_render_context_in_event, _render_context_in_event,
*_shared_processors(), *_shared_processors(),
structlog.processors.StackInfoRenderer(), 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( def _render_context_in_event(
_logger: Any, _logger: Any,
_method_name: str, _method_name: str,
@@ -71,7 +91,14 @@ def _render_context_in_event(
context = " ".join( context = " ".join(
f"{key}={_serialize_log_value(value)}" f"{key}={_serialize_log_value(value)}"
for key, value in sorted(event_dict.items()) 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 event_dict["event"] = f"{event} {context}" if context else event
return event_dict return event_dict
+77
View File
@@ -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])
+1 -1
View File
@@ -11,7 +11,7 @@ business_logic:
provider_price_multiplier: 1.0 provider_price_multiplier: 1.0
repository: repository:
redis_dsn: "redis://localhost:6379/0" redis_dsn: "redis://redis:6379/0"
price_cache_ttl_seconds: 900 price_cache_ttl_seconds: 900
adapter: adapter:
+7 -1
View File
@@ -47,6 +47,12 @@ receivers:
parse_from: attributes.log parse_from: attributes.log
parse_to: attributes parse_to: attributes
on_error: send 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 - type: time_parser
parse_from: attributes.timestamp parse_from: attributes.timestamp
layout: '%Y-%m-%dT%H:%M:%S.%fZ' layout: '%Y-%m-%dT%H:%M:%S.%fZ'
@@ -104,6 +110,6 @@ service:
processors: [resource/env, resource/version] processors: [resource/env, resource/version]
exporters: [otlp] exporters: [otlp]
metrics: metrics:
receivers: [docker_stats] receivers: [otlp, docker_stats]
processors: [filter/environment, resource/env] processors: [filter/environment, resource/env]
exporters: [otlp] exporters: [otlp]
@@ -122,8 +122,11 @@ def test_post_address_suggest_uses_registered_provider_in_default_dependency(
class StubPriceCache: class StubPriceCache:
@classmethod @classmethod
def from_repository_config(cls, repository_config) -> StubCache: def from_repository_config(
cls, repository_config, *, metrics
) -> StubCache:
_ = repository_config _ = repository_config
assert metrics is delivery_controller.get_cache_metrics()
return StubCache() return StubCache()
stub_http_client = StubHttpClient() stub_http_client = StubHttpClient()
+4 -1
View File
@@ -138,8 +138,11 @@ def test_post_delivery_price_uses_registered_provider_in_default_dependency(
class StubPriceCache: class StubPriceCache:
@classmethod @classmethod
def from_repository_config(cls, repository_config) -> StubCache: def from_repository_config(
cls, repository_config, *, metrics
) -> StubCache:
_ = repository_config _ = repository_config
assert metrics is delivery_controller.get_cache_metrics()
return stub_cache return stub_cache
stub_http_client = StubHttpClient() stub_http_client = StubHttpClient()
+47
View File
@@ -3,6 +3,8 @@ import logging
from io import StringIO from io import StringIO
import structlog import structlog
from opentelemetry import trace
from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags
from app.runtime.logging import configure_logging 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["level"] == "info"
assert payload["logger"] == "app.test" assert payload["logger"] == "app.test"
assert isinstance(payload["timestamp"], str) assert isinstance(payload["timestamp"], str)
assert "trace_id" not in payload
assert "span_id" not in payload
_reset_logging() _reset_logging()
@@ -75,6 +79,49 @@ def test_uvicorn_loggers_use_shared_json_logging_setup() -> None:
_reset_logging() _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: def _reset_logging() -> None:
structlog.reset_defaults() structlog.reset_defaults()
root_logger = logging.getLogger() root_logger = logging.getLogger()
+121
View File
@@ -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
View File
@@ -69,6 +69,22 @@ class FailingSetRedisClient:
return 1 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: def test_cache_miss_returns_none() -> None:
cache = PriceCache(InMemoryRedisClient(), ttl_seconds=120) 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."): with pytest.raises(PriceCacheRepositoryError, match="Redis cache write failed."):
asyncio.run(cache.set("key", {"v": 1})) 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"},
)
]
+16
View File
@@ -7,6 +7,7 @@ from fastapi import FastAPI
from app import config as config_module from app import config as config_module
from app.config import Settings, get_settings from app.config import Settings, get_settings
from app.runtime import logging as logging_module from app.runtime import logging as logging_module
from app.runtime import metrics as metrics_module
from app.runtime import tracing as tracing_module from app.runtime import tracing as tracing_module
TEST_CONFIG_FILE = Path(__file__).resolve().parents[2] / "config.test.yaml" 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: def test_app_import_smoke(monkeypatch) -> None:
logging_bootstrap_calls: list[None] = [] logging_bootstrap_calls: list[None] = []
metrics_bootstrap_calls: list[object] = []
tracing_bootstrap_calls: list[tuple[FastAPI, object]] = [] tracing_bootstrap_calls: list[tuple[FastAPI, object]] = []
def fake_configure_logging() -> None: 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: def fake_configure_tracing(app: FastAPI, observability: object) -> None:
tracing_bootstrap_calls.append((app, observability)) 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(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) monkeypatch.setattr(tracing_module, "configure_tracing", fake_configure_tracing)
main_module = _import_main_module(monkeypatch) 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 route.path for route in main_module.app.routes
} }
assert logging_bootstrap_calls == [None] assert logging_bootstrap_calls == [None]
assert metrics_bootstrap_calls == [main_module.app.state.settings.observability]
assert tracing_bootstrap_calls == [ assert tracing_bootstrap_calls == [
(main_module.app, main_module.app.state.settings.observability) (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: def test_app_created_with_provided_settings(monkeypatch) -> None:
logging_bootstrap_calls: list[None] = [] logging_bootstrap_calls: list[None] = []
metrics_bootstrap_calls: list[object] = []
tracing_bootstrap_calls: list[tuple[FastAPI, object]] = [] tracing_bootstrap_calls: list[tuple[FastAPI, object]] = []
def fake_configure_logging() -> None: 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: def fake_configure_tracing(app: FastAPI, observability: object) -> None:
tracing_bootstrap_calls.append((app, observability)) 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(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) monkeypatch.setattr(tracing_module, "configure_tracing", fake_configure_tracing)
main_module = _import_main_module(monkeypatch) main_module = _import_main_module(monkeypatch)
monkeypatch.setattr(config_module, "TEST_CONFIG_FILE", str(TEST_CONFIG_FILE)) 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 instance.state.settings is settings
assert logging_bootstrap_calls == [None, None] assert logging_bootstrap_calls == [None, None]
assert metrics_bootstrap_calls == [
main_module.app.state.settings.observability,
settings.observability,
]
assert tracing_bootstrap_calls == [ assert tracing_bootstrap_calls == [
(main_module.app, main_module.app.state.settings.observability), (main_module.app, main_module.app.state.settings.observability),
(instance, settings.observability), (instance, settings.observability),
+25
View File
@@ -6,6 +6,8 @@ import yaml
PROJECT_ROOT = Path(__file__).resolve().parents[2] PROJECT_ROOT = Path(__file__).resolve().parents[2]
COMPOSE_FILE = PROJECT_ROOT / "docker-compose.yml" COMPOSE_FILE = PROJECT_ROOT / "docker-compose.yml"
INFRA_README = PROJECT_ROOT / "infra" / "README.md" 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: 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 config" in readme
assert "docker compose up -d redis" in readme assert "docker compose up -d redis" in readme
assert "docker compose ps" 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"