[DOP-22268] - remove CORRELATION_CELERY_HEADER_ID setting

Author: maxim-lixakov

docs/worker/log_url.rst

@@ -9,5 +9,5 @@ The configuration parameter is:
 
   SYNCMASTER__SERVER__LOG_URL_TEMPLATE=https://grafana.example.com?correlation_id={{ correlation_id }}&run_id={{ run.id }}
 
-You can search for each run by either its correlation id ``CORRELATION_CELERY_HEADER_ID`` in http headers or the ``Run.Id``.
+You can search for each run by either its correlation id ``x-request-id`` in http headers or the ``Run.Id``.
 

syncmaster/settings/log/json.yml

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

syncmaster/worker/settings/__init__.py

@@ -20,13 +20,9 @@ class WorkerSettings(BaseSettings):
 
     .. code-block:: bash
 
-        SYNCMASTER__WORKER__CORRELATION_CELERY_HEADER_ID=CORRELATION_ID_CELERY
+        SYNCMASTER__WORKER__CREATE_SPARK_SESSION_FUNCTION=custom_syncmaster.spark.get_worker_spark_session
     """
 
-    CORRELATION_CELERY_HEADER_ID: str = Field(
-        "CORRELATION_ID",
-        description="Header ID for correlation in Celery",
-    )
     CREATE_SPARK_SESSION_FUNCTION: ImportString = Field(
         "syncmaster.worker.spark.get_worker_spark_session",
         description="Function to create Spark session for worker",
@@ -51,9 +47,6 @@ class WorkerAppSettings(BaseSettings):
 
     .. code-block:: bash
 
-        # Example of setting a CORRELATION_CELERY_HEADER_ID via environment variable
-        SYNCMASTER__WORKER__CORRELATION_CELERY_HEADER_ID=CORRELATION_ID_CELERY
-
         # Example of setting a database URL via environment variable
         SYNCMASTER__DATABASE__URL=postgresql+asyncpg://user:password@localhost:5432/dbname
 

syncmaster/worker/transfer.py

@@ -2,10 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 from datetime import datetime, timezone
 
-from asgi_correlation_id import correlation_id
 from asgi_correlation_id.extensions.celery import load_correlation_ids
 from celery import Celery
-from celery.signals import after_setup_task_logger, before_task_publish, task_prerun
+from celery.signals import after_setup_task_logger
 from celery.utils.log import get_task_logger
 from sqlalchemy import select
 from sqlalchemy.orm import Session, selectinload
@@ -22,7 +21,6 @@
 load_correlation_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)
@@ -85,18 +83,3 @@ def run_transfer(session: Session, run_id: int, settings: WorkerAppSettings):
 @after_setup_task_logger.connect
 def setup_loggers(*args, **kwargs):
     setup_logging(WorkerAppSettings().logging.get_log_config_path())
-
-
-@before_task_publish.connect()
-def transfer_correlation_id(headers, *args, **kwargs) -> None:
-    # This is called before task.delay() finishes
-    # Here we're able to transfer the correlation ID via the headers kept in our backend
-    headers[CORRELATION_CELERY_HEADER_ID] = correlation_id.get()
-
-
-@task_prerun.connect()
-def load_correlation_id(task, *args, **kwargs) -> None:
-    # This is called when the worker picks up the task
-    # Here we're able to load the correlation ID from the headers
-    id_value = task.request.get(CORRELATION_CELERY_HEADER_ID)
-    correlation_id.set(id_value)