[DOP-22268] - move spark logs to json format

Author: maxim-lixakov

syncmaster/settings/log/json.yml

@@ -9,6 +9,9 @@ filters:
     (): asgi_correlation_id.CorrelationIdFilter
     uuid_length: 32
     default_value: '-'
+  celery_tracing:
+    (): asgi_correlation_id.CeleryTracingIdsFilter
+    uuid_length: 32
 
 formatters:
   json:
@@ -26,7 +29,7 @@ handlers:
   celery:
     class: logging.StreamHandler
     formatter: json
-    filters: [correlation_id]
+    filters: [correlation_id, celery_tracing]
     stream: ext://sys.stdout
 
 loggers:

syncmaster/worker/settings/__init__.py

@@ -24,7 +24,7 @@ class WorkerSettings(BaseSettings):
     """
 
     CORRELATION_CELERY_HEADER_ID: str = Field(
-        "CORRELATION_CELERY_HEADER_ID",
+        "CORRELATION_ID",
         description="Header ID for correlation in Celery",
     )
     CREATE_SPARK_SESSION_FUNCTION: ImportString = Field(

syncmaster/worker/transfer.py

@@ -1,12 +1,14 @@
 # SPDX-FileCopyrightText: 2023-2024 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
-import logging
 from datetime import datetime, timezone
 
-import onetl
 from asgi_correlation_id import correlation_id
+from asgi_correlation_id.extensions.celery import (
+    load_celery_current_and_parent_ids,
+    load_correlation_ids,
+)
 from celery import Celery
-from celery.signals import after_setup_logger, before_task_publish, task_prerun
+from celery.signals import after_setup_task_logger, before_task_publish, task_prerun
 from celery.utils.log import get_task_logger
 from sqlalchemy import select
 from sqlalchemy.orm import Session, selectinload
@@ -20,14 +22,15 @@
 from syncmaster.worker.settings import WorkerAppSettings
 
 logger = get_task_logger(__name__)
+load_correlation_ids()
+load_celery_current_and_parent_ids()
 
 WORKER_SETTINGS = WorkerAppSettings()
 CORRELATION_CELERY_HEADER_ID = WORKER_SETTINGS.worker.CORRELATION_CELERY_HEADER_ID
 
 
 @celery.task(name="run_transfer_task", bind=True, track_started=True)
 def run_transfer_task(self: Celery, run_id: int) -> None:
-    onetl.log.setup_logging(level=logging.INFO)
     with Session(self.engine) as session:
         run_transfer(
             session=session,
@@ -83,7 +86,7 @@ def run_transfer(session: Session, run_id: int, settings: WorkerAppSettings):
     session.commit()
 
 
-@after_setup_logger.connect
+@after_setup_task_logger.connect
 def setup_loggers(*args, **kwargs):
     setup_logging(WorkerAppSettings().logging.get_log_config_path())