diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 087268b..5a4a7b9 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -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 + docker compose -p $COMPOSE_PROJECT up -d && + docker compose -p $COMPOSE_PROJECT up -d --force-recreate otel-collector " diff --git a/app/runtime/logging.py b/app/runtime/logging.py index 2825e79..ec0427e 100644 --- a/app/runtime/logging.py +++ b/app/runtime/logging.py @@ -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,7 @@ 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_open_telemetry_context, *_shared_processors()], processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, structlog.processors.JSONRenderer(sort_keys=True, ensure_ascii=False), @@ -33,6 +34,7 @@ def configure_logging( structlog.configure( processors=[ structlog.stdlib.PositionalArgumentsFormatter(), + _add_open_telemetry_context, _render_context_in_event, *_shared_processors(), structlog.processors.StackInfoRenderer(), @@ -62,6 +64,19 @@ def _shared_processors() -> tuple[structlog.types.Processor, ...]: ) +def _add_open_telemetry_context( + _logger: Any, + _method_name: str, + event_dict: structlog.types.EventDict, +) -> structlog.types.EventDict: + span_context = trace.get_current_span().get_span_context() + if span_context.is_valid: + event_dict["trace_id"] = format(span_context.trace_id, "032x") + event_dict["span_id"] = format(span_context.span_id, "016x") + event_dict["trace_flags"] = int(span_context.trace_flags) + return event_dict + + def _render_context_in_event( _logger: Any, _method_name: str, 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..b910e32 100644 --- a/otel-collector-config.template.yaml +++ b/otel-collector-config.template.yaml @@ -13,19 +13,26 @@ receivers: - type: json_parser parse_to: attributes on_error: send + - type: filter + expr: 'attributes["attrs"] == nil' + - type: json_parser + if: 'attributes["attrs"] != nil' + parse_from: attributes["attrs"] + parse_to: attributes["docker_attrs"] + on_error: send - type: filter expr: | - attributes["attrs"] == nil or - attributes["attrs"]["tag"] == nil or + attributes["docker_attrs"] == nil or + attributes["docker_attrs"]["tag"] == nil or ( - attributes["attrs"]["tag"] != "g2s-aggregator${CONTAINER_SUFFIX}" and - attributes["attrs"]["tag"] != "g2s-aggregator-migrations${CONTAINER_SUFFIX}" and - attributes["attrs"]["tag"] != "g2s-aggregator-waybill-poller${CONTAINER_SUFFIX}" and - attributes["attrs"]["tag"] != "g2s-aggregator-waybill-email-sender${CONTAINER_SUFFIX}" + attributes["docker_attrs"]["tag"] != "g2s-aggregator${CONTAINER_SUFFIX}" and + attributes["docker_attrs"]["tag"] != "g2s-aggregator-migrations${CONTAINER_SUFFIX}" and + attributes["docker_attrs"]["tag"] != "g2s-aggregator-waybill-poller${CONTAINER_SUFFIX}" and + attributes["docker_attrs"]["tag"] != "g2s-aggregator-waybill-email-sender${CONTAINER_SUFFIX}" ) on_error: send - type: regex_parser - parse_from: attributes["attrs"]["tag"] + parse_from: attributes["docker_attrs"]["tag"] regex: '^(?Pg2s-aggregator(?:-migrations|-waybill-poller|-waybill-email-sender)?)(?:-stage)?$' on_error: send - type: move @@ -43,10 +50,25 @@ receivers: - type: remove field: attributes["attrs"] on_error: send + - type: remove + field: attributes["docker_attrs"] + on_error: send - type: json_parser parse_from: attributes.log parse_to: attributes on_error: send + - type: trace_parser + if: | + attributes.trace_id != nil and + attributes.span_id != nil and + attributes.trace_flags != nil + trace_id: + parse_from: attributes.trace_id + span_id: + parse_from: attributes.span_id + trace_flags: + parse_from: attributes.trace_flags + on_error: send - type: time_parser parse_from: attributes.timestamp layout: '%Y-%m-%dT%H:%M:%S.%fZ' diff --git a/tests/logging/test_json_logging.py b/tests/logging/test_json_logging.py index f788c9a..e4803e7 100644 --- a/tests/logging/test_json_logging.py +++ b/tests/logging/test_json_logging.py @@ -3,6 +3,7 @@ import logging from io import StringIO import structlog +from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags, use_span from app.runtime.logging import configure_logging @@ -51,6 +52,72 @@ def test_structured_context_is_preserved_and_rendered_in_event() -> None: _reset_logging() +def test_active_open_telemetry_context_is_added_to_structured_log() -> None: + stream = StringIO() + span_context = SpanContext( + trace_id=0x1234567890ABCDEF1234567890ABCDEF, + span_id=0x1234567890ABCDEF, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + + configure_logging(stream=stream) + + with use_span(NonRecordingSpan(span_context)): + structlog.get_logger("app.test").info("application_event") + + payload = json.loads(stream.getvalue().splitlines()[0]) + + assert payload["trace_id"] == "1234567890abcdef1234567890abcdef" + assert payload["span_id"] == "1234567890abcdef" + assert payload["trace_flags"] == 1 + assert payload["event"] == ( + 'application_event span_id="1234567890abcdef" trace_flags=1 ' + 'trace_id="1234567890abcdef1234567890abcdef"' + ) + + _reset_logging() + + +def test_missing_open_telemetry_context_is_not_added_to_log() -> None: + stream = StringIO() + + configure_logging(stream=stream) + + structlog.get_logger("app.test").info("application_event") + + payload = json.loads(stream.getvalue().splitlines()[0]) + + assert "trace_id" not in payload + assert "span_id" not in payload + assert "trace_flags" not in payload + + _reset_logging() + + +def test_active_open_telemetry_context_is_added_to_stdlib_log() -> None: + stream = StringIO() + span_context = SpanContext( + trace_id=0x1234567890ABCDEF1234567890ABCDEF, + span_id=0x1234567890ABCDEF, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + + configure_logging(stream=stream) + + with use_span(NonRecordingSpan(span_context)): + logging.getLogger("app.test").info("application_event") + + payload = json.loads(stream.getvalue().splitlines()[0]) + + assert payload["trace_id"] == "1234567890abcdef1234567890abcdef" + assert payload["span_id"] == "1234567890abcdef" + assert payload["trace_flags"] == 1 + + _reset_logging() + + def test_uvicorn_loggers_use_shared_json_logging_setup() -> None: stream = StringIO()