fix: correct the settings for OTEL and add logging in the scheduler (#904)

* fix: correct OTEL settings

* feat: add some logging in the task scheduler to keep track of its activity

Author: chaen

diracx-routers/src/diracx/routers/otel.py

@@ -124,7 +124,10 @@ def instrument_otel(app: FastAPI) -> None:
     # # override logger format which with trace id and span id
     # https://github.com/mhausenblas/ref.otel.help/blob/main/how-to/logs-collection/yoda/main.py
 
-    LoggingInstrumentor().instrument(set_logging_format=True)
+    # When set to True, the logs are too noisy to be read from the pods.
+    # Moreover, the data can't be ingested by CERN Opentelemtry
+    # setup. To be investigated
+    LoggingInstrumentor().instrument(set_logging_format=False)
 
     logger_provider = LoggerProvider(resource=resource)
     _logs.set_logger_provider(logger_provider)
@@ -136,7 +139,15 @@ def instrument_otel(app: FastAPI) -> None:
     )
     logger_provider.add_log_record_processor(BatchLogRecordProcessor(otlp_exporter))
     handler = LoggingHandler(level=logging.DEBUG, logger_provider=logger_provider)
-    handler.setFormatter(logging.Formatter(DEFAULT_LOGGING_FORMAT))
+    # We need to give some default values for these keys, otherwise the service crashes
+    # https://github.com/DIRACGrid/diracx/pull/847/
+    # handler.setFormatter(logging.Formatter(DEFAULT_LOGGING_FORMAT))
+    default_format = dict.fromkeys(
+        ["otelTraceID", "otelSpanID", "otelServiceName", "otelTraceSampled"], "Default"
+    )
+    handler.setFormatter(
+        logging.Formatter(DEFAULT_LOGGING_FORMAT, defaults=default_format)
+    )
     # Add the handler to diracx and all uvicorn logger
     # Note adding it to just 'uvicorn' or the root logger
     # is not enough because uvicorn sets propagate=False

diracx-tasks/src/diracx/tasks/plumbing/scheduler/scheduler.py

@@ -66,6 +66,8 @@
 """
 
 DELAYED_ZSET_KEY = "diracx:tasks:delayed"
+SCHEDULE_DUMP_INTERVAL_SECONDS = 600
+SCHEDULE_DUMP_MAX_ENTRIES = 20
 
 
 async def schedule_delayed(
@@ -124,6 +126,8 @@ def __init__(
         )
         # Mapping of (task_class_name, vo_or_empty) -> next_scheduled_time
         self._next_runs: dict[tuple[str, str], datetime] = {}
+        self._schedule_dump_interval_seconds = SCHEDULE_DUMP_INTERVAL_SECONDS
+        self._last_schedule_dump_at: datetime | None = None
         # Cached ZSET size for OTel observable gauge
         self._delayed_zset_size: int = 0
         _meter.create_observable_gauge(
@@ -134,6 +138,7 @@ def __init__(
 
     async def startup(self) -> None:
         await self.broker.startup()
+        self._log_task_registry_awareness()
         logger.info("Scheduler started")
 
     async def shutdown(self) -> None:
@@ -201,6 +206,9 @@ async def _periodic_loop(self, finish_event: asyncio.Event) -> None:
                 await asyncio.gather(*coros)
 
             self._next_runs.update(due_updates)
+            if self._should_dump_schedule_snapshot(now):
+                self._log_next_schedules_snapshot("periodic")
+                self._last_schedule_dump_at = now
 
             try:
                 await asyncio.wait_for(finish_event.wait(), timeout=self.check_interval)
@@ -256,15 +264,24 @@ def load_vos(self) -> list[str]:
     def _compute_initial_schedules(self) -> None:
         """Compute the initial next-run times for all periodic tasks."""
         vos = self.load_vos()
+        non_periodic_count = 0
+        disabled_count = 0
+        periodic_count = 0
+        vo_aware_count = 0
+        scheduled_entries = 0
 
         for task_name, task_cls in self.task_registry.items():
             if not issubclass(task_cls, PeriodicBaseTask):
+                non_periodic_count += 1
                 continue
+            periodic_count += 1
             if not getattr(task_cls, "_enabled", True):
+                disabled_count += 1
                 continue
             schedule = task_cls.default_schedule
 
             if issubclass(task_cls, PeriodicVoAwareBaseTask):
+                vo_aware_count += 1
                 if not vos:
                     logger.warning(
                         "No VOs configured, skipping VO-aware task %s",
@@ -273,8 +290,22 @@ def _compute_initial_schedules(self) -> None:
                     continue
                 for vo in vos:
                     self.add_vo_schedule(task_name, vo, schedule.next_occurrence())
+                    scheduled_entries += 1
             else:
                 self._next_runs[(task_name, "")] = schedule.next_occurrence()
+                scheduled_entries += 1
+
+        logger.info(
+            "Initial periodic schedules computed: entries=%d periodic=%d "
+            "vo_aware=%d disabled=%d non_periodic=%d vos=%d",
+            scheduled_entries,
+            periodic_count,
+            vo_aware_count,
+            disabled_count,
+            non_periodic_count,
+            len(vos),
+        )
+        self._log_next_schedules_snapshot("initial")
 
     def add_vo_schedule(self, task_name: str, vo: str, next_run: datetime) -> None:
         """Register a VO-specific periodic task schedule."""
@@ -414,6 +445,86 @@ async def _config_watch_loop(self, finish_event: asyncio.Event) -> None:
                 if key[1] in removed:
                     del self._next_runs[key]
 
+            logger.info(
+                "Reconciled VO schedules: tracked_entries=%d added_vos=%d removed_vos=%d",
+                len(self._next_runs),
+                len(added),
+                len(removed),
+            )
+            self._log_next_schedules_snapshot("config_reconcile")
+
+    def _log_task_registry_awareness(self) -> None:
+        """Log which tasks are known to the scheduler."""
+        periodic_enabled: list[str] = []
+        vo_aware_enabled: list[str] = []
+        disabled_periodic: list[str] = []
+        non_periodic: list[str] = []
+
+        for task_name, task_cls in self.task_registry.items():
+            if not issubclass(task_cls, PeriodicBaseTask):
+                non_periodic.append(task_name)
+                continue
+            if not getattr(task_cls, "_enabled", True):
+                disabled_periodic.append(task_name)
+                continue
+            periodic_enabled.append(task_name)
+            if issubclass(task_cls, PeriodicVoAwareBaseTask):
+                vo_aware_enabled.append(task_name)
+
+        periodic_enabled.sort()
+        vo_aware_enabled.sort()
+        disabled_periodic.sort()
+        non_periodic.sort()
+
+        logger.info(
+            "Scheduler task registry: total=%d periodic_enabled=%d "
+            "vo_aware_enabled=%d periodic_disabled=%d non_periodic=%d",
+            len(self.task_registry),
+            len(periodic_enabled),
+            len(vo_aware_enabled),
+            len(disabled_periodic),
+            len(non_periodic),
+        )
+        if periodic_enabled:
+            logger.info("Scheduler periodic tasks: %s", periodic_enabled)
+        if vo_aware_enabled:
+            logger.info("Scheduler VO-aware periodic tasks: %s", vo_aware_enabled)
+        if disabled_periodic:
+            logger.info("Scheduler disabled periodic tasks: %s", disabled_periodic)
+        if non_periodic:
+            logger.info("Scheduler non-periodic tasks in registry: %s", non_periodic)
+
+    def _should_dump_schedule_snapshot(self, now: datetime) -> bool:
+        if self._last_schedule_dump_at is None:
+            return True
+        elapsed = (now - self._last_schedule_dump_at).total_seconds()
+        return elapsed >= self._schedule_dump_interval_seconds
+
+    def _log_next_schedules_snapshot(self, source: str) -> None:
+        """Log a bounded, sorted dump of upcoming schedules."""
+        if not self._next_runs:
+            logger.info("Next schedule snapshot (%s): no tracked schedules", source)
+            return
+
+        upcoming = sorted(self._next_runs.items(), key=lambda item: item[1])
+        shown = upcoming[:SCHEDULE_DUMP_MAX_ENTRIES]
+        rendered = [
+            {
+                "task": task_name,
+                "vo": vo or "N/A",
+                "next_run": next_run.isoformat(),
+            }
+            for (task_name, vo), next_run in shown
+        ]
+
+        logger.info(
+            "Next schedule snapshot (%s): tracked=%d shown=%d entries=%s",
+            source,
+            len(upcoming),
+            len(rendered),
+            rendered,
+        )
+
     # ------------------------------------------------------------------
     # OTel observable gauge callback
     # ------------------------------------------------------------------