[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/transfer.py

@@ -1,10 +1,12 @@
 # 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.utils.log import get_task_logger
@@ -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,