улучшены логи, otel collector добавлен в git

.gitea/workflows/deploy.yml

@@ -26,6 +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 "COMPOSE_PROJECT=g2s-aggregator-stage"
               echo "DEPLOY_DIR=/home/deploy/g2s-aggregator-stage"
             } >> "$GITHUB_ENV"
@@ -36,6 +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 "COMPOSE_PROJECT=g2s-aggregator"
               echo "DEPLOY_DIR=/home/deploy/g2s-aggregator"
             } >> "$GITHUB_ENV"
@@ -106,11 +108,14 @@ jobs:
 
           render config.template.yaml config.rendered.yaml
           render docker-compose.template.yml docker-compose.rendered.yml
+          render otel-collector-config.template.yaml otel-collector-config.rendered.yaml
 
           scp -i ~/.ssh/deploy_key config.rendered.yaml \
             deploy@194.58.121.203:$DEPLOY_DIR/config.yaml
           scp -i ~/.ssh/deploy_key docker-compose.rendered.yml \
             deploy@194.58.121.203:$DEPLOY_DIR/docker-compose.yml
+          scp -i ~/.ssh/deploy_key otel-collector-config.rendered.yaml \
+            deploy@194.58.121.203:$DEPLOY_DIR/otel-collector-config.yaml
 
       - name: Deploy
         run: |

.gitignore

@@ -2,6 +2,7 @@
 *.iml
 /config.yaml
 /docker-compose.yml
+/otel-collector-config.yaml
 __pycache__
 http-client.private.env.json
 scripts

app/runtime/logging.py

@@ -1,9 +1,10 @@
 """Centralized runtime logging bootstrap."""
 
+import json
 import logging
 import sys
 from collections.abc import Iterable
-from typing import TextIO
+from typing import Any, TextIO
 
 import structlog
 
@@ -31,8 +32,9 @@ def configure_logging(
     structlog.reset_defaults()
     structlog.configure(
         processors=[
-            *_shared_processors(),
             structlog.stdlib.PositionalArgumentsFormatter(),
+            _render_context_in_event,
+            *_shared_processors(),
             structlog.processors.StackInfoRenderer(),
             structlog.processors.format_exc_info,
             structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
@@ -60,6 +62,31 @@ def _shared_processors() -> tuple[structlog.types.Processor, ...]:
     )
 
 
+def _render_context_in_event(
+    _logger: Any,
+    _method_name: str,
+    event_dict: structlog.types.EventDict,
+) -> structlog.types.EventDict:
+    event = str(event_dict.get("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"}
+    )
+    event_dict["event"] = f"{event} {context}" if context else event
+    return event_dict
+
+
+def _serialize_log_value(value: object) -> str:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+
+
 def _configure_logger(
     logger: logging.Logger,
     *,

app/services/waybill_email_sender.py

@@ -116,11 +116,11 @@ class WaybillEmailSenderService:
         while not stop_event.is_set():
             try:
                 summary = await self.poll_once()
-                logger.info(
-                    "waybill_email_tick "
-                    f"processed={summary.processed} "
-                    f"succeeded={summary.succeeded} "
-                    f"failed={summary.failed}"
+                logger.debug(
+                    "waybill_email_tick",
+                    processed=summary.processed,
+                    succeeded=summary.succeeded,
+                    failed=summary.failed,
                 )
             except Exception:
                 logger.exception("waybill_email_tick_failed")

app/services/waybill_poller.py

@@ -115,11 +115,11 @@ class WaybillPollerService:
         while not stop_event.is_set():
             try:
                 summary = await self.poll_once()
-                logger.info(
-                    "waybill_poll_tick "
-                    f"processed={summary.processed} "
-                    f"succeeded={summary.succeeded} "
-                    f"failed={summary.failed}"
+                logger.debug(
+                    "waybill_poll_tick",
+                    processed=summary.processed,
+                    succeeded=summary.succeeded,
+                    failed=summary.failed,
                 )
             except Exception:
                 logger.exception("waybill_poll_tick_failed")

otel-collector-config.template.yaml

@@ -0,0 +1,109 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+      http:
+        endpoint: 0.0.0.0:4318
+  filelog:
+    include:
+      - /var/lib/docker/containers/*/*-json.log
+    include_file_path: true
+    operators:
+      - type: json_parser
+        parse_to: attributes
+        on_error: send
+      - type: filter
+        expr: |
+          attributes["attrs"] == nil or
+          attributes["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}"
+          )
+        on_error: send
+      - type: regex_parser
+        parse_from: attributes["attrs"]["tag"]
+        regex: '^(?P<service_name>g2s-aggregator(?:-migrations|-waybill-poller|-waybill-email-sender)?)(?:-stage)?$'
+        on_error: send
+      - type: move
+        from: attributes["service_name"]
+        to: resource["service.name"]
+        on_error: send
+      - type: regex_parser
+        parse_from: attributes["log.file.path"]
+        regex: '^/var/lib/docker/containers/(?P<container_id>[^/]+)/'
+        on_error: send
+      - type: move
+        from: attributes["container_id"]
+        to: resource["service.instance.id"]
+        on_error: send
+      - type: remove
+        field: attributes["attrs"]
+        on_error: send
+      - type: json_parser
+        parse_from: attributes.log
+        parse_to: attributes
+        on_error: send
+      - type: time_parser
+        parse_from: attributes.timestamp
+        layout: '%Y-%m-%dT%H:%M:%S.%fZ'
+        on_error: send
+      - type: severity_parser
+        parse_from: attributes.level
+        on_error: send
+      - type: move
+        from: attributes.event
+        to: body
+        on_error: send
+      - type: remove
+        field: attributes.log
+        on_error: send
+  docker_stats:
+    endpoint: unix:///var/run/docker.sock
+    collection_interval: 30s
+    container_labels_as_resource_attributes: true
+    api_version: "1.43"
+
+processors:
+  resource/env:
+    attributes:
+      - key: deployment.environment
+        value: "${ENV_NAME}"
+        action: upsert
+      - key: service.namespace
+        value: g2s
+        action: upsert
+  resource/version:
+    attributes:
+      - key: service.version
+        value: "${IMAGE_TAG}"
+        action: upsert
+  filter/environment:
+    error_mode: ignore
+    metrics:
+      metric:
+        - '${METRICS_FILTER_EXPRESSION}'
+
+exporters:
+  otlp:
+    endpoint: "${SIGNOZ_OTLP_ENDPOINT}"
+    tls:
+      insecure: true
+
+service:
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [resource/env, resource/version]
+      exporters: [otlp]
+    logs:
+      receivers: [filelog]
+      processors: [resource/env, resource/version]
+      exporters: [otlp]
+    metrics:
+      receivers: [docker_stats]
+      processors: [filter/environment, resource/env]
+      exporters: [otlp]

tests/logging/test_json_logging.py

@@ -24,6 +24,33 @@ def test_application_log_is_serialized_as_json() -> None:
     _reset_logging()
 
 
+def test_structured_context_is_preserved_and_rendered_in_event() -> None:
+    stream = StringIO()
+
+    configure_logging(stream=stream)
+
+    structlog.get_logger("app.test").warning(
+        "application_event",
+        order_uuid="order-1",
+        attempt=2,
+        retryable=True,
+        details={"code": "invalid"},
+    )
+
+    payload = json.loads(stream.getvalue().splitlines()[0])
+
+    assert payload["event"] == (
+        'application_event attempt=2 details={"code":"invalid"} '
+        'order_uuid="order-1" retryable=true'
+    )
+    assert payload["order_uuid"] == "order-1"
+    assert payload["attempt"] == 2
+    assert payload["retryable"] is True
+    assert payload["details"] == {"code": "invalid"}
+
+    _reset_logging()
+
+
 def test_uvicorn_loggers_use_shared_json_logging_setup() -> None:
     stream = StringIO()