Add MONITOR_EVENT logs for scheduler lifecycle

Author: glin1993@outlook.com

src/memos/mem_scheduler/base_scheduler.py

@@ -4,7 +4,7 @@
 import time
 
 from collections.abc import Callable
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Union
 
@@ -49,6 +49,7 @@
 from memos.mem_scheduler.utils.filter_utils import (
     transform_name_to_key,
 )
+from memos.mem_scheduler.utils.monitor_event_utils import emit_monitor_event, to_iso
 from memos.mem_scheduler.utils.status_tracker import TaskStatusTracker
 from memos.mem_scheduler.webservice_modules.rabbitmq_service import RabbitMQSchedulerModule
 from memos.mem_scheduler.webservice_modules.redis_service import RedisSchedulerModule
@@ -768,7 +769,33 @@ def _message_consumer(self) -> None:
                 messages = self.memos_message_queue.get_messages(batch_size=self.consume_batch)
 
                 if messages:
+                    now = time.time()
                     for msg in messages:
+                        enqueue_ts_obj = getattr(msg, "timestamp", None)
+                        enqueue_epoch = None
+                        if isinstance(enqueue_ts_obj, (int, float)):
+                            enqueue_epoch = float(enqueue_ts_obj)
+                        elif hasattr(enqueue_ts_obj, "timestamp"):
+                            dt = enqueue_ts_obj
+                            if dt.tzinfo is None:
+                                dt = dt.replace(tzinfo=timezone.utc)
+                            enqueue_epoch = dt.timestamp()
+
+                        queue_wait_ms = None
+                        if enqueue_epoch is not None:
+                            queue_wait_ms = max(0.0, now - enqueue_epoch) * 1000
+
+                        setattr(msg, "dequeue_ts", now)
+                        emit_monitor_event(
+                            "dequeue",
+                            msg,
+                            {
+                                "enqueue_ts": to_iso(enqueue_ts_obj),
+                                "dequeue_ts": datetime.fromtimestamp(now, tz=timezone.utc).isoformat(),
+                                "queue_wait_ms": queue_wait_ms,
+                            },
+                        )
+
                         self.metrics.task_dequeued(user_id=msg.user_id, task_type=msg.label)
                     try:
                         import contextlib

src/memos/mem_scheduler/task_schedule_modules/dispatcher.py

@@ -24,6 +24,7 @@
 from memos.mem_scheduler.task_schedule_modules.redis_queue import SchedulerRedisQueue
 from memos.mem_scheduler.task_schedule_modules.task_queue import ScheduleTaskQueue
 from memos.mem_scheduler.utils.misc_utils import group_messages_by_user_and_mem_cube
+from memos.mem_scheduler.utils.monitor_event_utils import emit_monitor_event, to_iso
 from memos.mem_scheduler.utils.status_tracker import TaskStatusTracker
 
 
@@ -126,6 +127,7 @@ def _create_task_wrapper(self, handler: Callable, task_item: RunningTaskItem):
 
         def wrapped_handler(messages: list[ScheduleMessageItem]):
             start_time = time.time()
+            start_iso = datetime.fromtimestamp(start_time, tz=timezone.utc).isoformat()
             if self.status_tracker:
                 self.status_tracker.task_started(
                     task_id=task_item.item_id, user_id=task_item.user_id
@@ -164,17 +166,49 @@ def wrapped_handler(messages: list[ScheduleMessageItem]):
                 wait_sec = max(0.0, now - enq_epoch)
                 self.metrics.observe_task_wait_duration(wait_sec, m.user_id, m.label)
 
+                dequeue_ts = getattr(first_msg, "dequeue_ts", None)
+                start_delay_ms = None
+                if isinstance(dequeue_ts, (int, float)):
+                    start_delay_ms = max(0.0, start_time - dequeue_ts) * 1000
+
+                emit_monitor_event(
+                    "start",
+                    first_msg,
+                    {
+                        "start_ts": start_iso,
+                        "start_delay_ms": start_delay_ms,
+                        "enqueue_ts": to_iso(enq_ts),
+                        "dequeue_ts": to_iso(
+                            datetime.fromtimestamp(dequeue_ts, tz=timezone.utc)
+                            if isinstance(dequeue_ts, (int, float))
+                            else None
+                        ),
+                    },
+                )
+
                 # Execute the original handler
                 result = handler(messages)
 
                 # --- mark done ---
-                duration = time.time() - start_time
+                finish_time = time.time()
+                duration = finish_time - start_time
                 self.metrics.observe_task_duration(duration, m.user_id, m.label)
                 if self.status_tracker:
                     self.status_tracker.task_completed(
                         task_id=task_item.item_id, user_id=task_item.user_id
                     )
                 self.metrics.task_completed(user_id=m.user_id, task_type=m.label)
+
+                emit_monitor_event(
+                    "finish",
+                    first_msg,
+                    {
+                        "status": "ok",
+                        "start_ts": start_iso,
+                        "finish_ts": datetime.fromtimestamp(finish_time, tz=timezone.utc).isoformat(),
+                        "exec_duration_ms": duration * 1000,
+                    },
+                )
                 # Redis ack is handled in finally to cover failure cases
 
                 # Mark task as completed and remove from tracking
@@ -187,11 +221,24 @@ def wrapped_handler(messages: list[ScheduleMessageItem]):
 
             except Exception as e:
                 m = messages[0]
+                finish_time = time.time()
                 self.metrics.task_failed(m.user_id, m.label, type(e).__name__)
                 if self.status_tracker:
                     self.status_tracker.task_failed(
                         task_id=task_item.item_id, user_id=task_item.user_id, error_message=str(e)
                     )
+                emit_monitor_event(
+                    "finish",
+                    m,
+                    {
+                        "status": "fail",
+                        "start_ts": start_iso,
+                        "finish_ts": datetime.fromtimestamp(finish_time, tz=timezone.utc).isoformat(),
+                        "exec_duration_ms": (finish_time - start_time) * 1000,
+                        "error_type": type(e).__name__,
+                        "error_msg": str(e),
+                    },
+                )
                 # Mark task as failed and remove from tracking
                 with self._task_lock:
                     if task_item.item_id in self._running_tasks:

src/memos/mem_scheduler/task_schedule_modules/task_queue.py

@@ -11,6 +11,7 @@
 from memos.mem_scheduler.task_schedule_modules.local_queue import SchedulerLocalQueue
 from memos.mem_scheduler.task_schedule_modules.redis_queue import SchedulerRedisQueue
 from memos.mem_scheduler.utils.db_utils import get_utc_now
+from memos.mem_scheduler.utils.monitor_event_utils import emit_monitor_event, to_iso
 from memos.mem_scheduler.utils.misc_utils import group_messages_by_user_and_mem_cube
 
 
@@ -77,6 +78,8 @@ def submit_messages(self, messages: ScheduleMessageItem | list[ScheduleMessageIt
         if len(messages) < 1:
             logger.error("Submit empty")
         elif len(messages) == 1:
+            enqueue_ts = to_iso(getattr(messages[0], "timestamp", None))
+            emit_monitor_event("enqueue", messages[0], {"enqueue_ts": enqueue_ts})
             self.memos_message_queue.put(messages[0])
         else:
             user_cube_groups = group_messages_by_user_and_mem_cube(messages)
@@ -99,6 +102,8 @@ def submit_messages(self, messages: ScheduleMessageItem | list[ScheduleMessageIt
                             )
                             continue
 
+                        enqueue_ts = to_iso(getattr(message, "timestamp", None))
+                        emit_monitor_event("enqueue", message, {"enqueue_ts": enqueue_ts})
                         self.memos_message_queue.put(message)
                         logger.info(
                             f"Submitted message to local queue: {message.label} - {message.content}"

src/memos/mem_scheduler/utils/monitor_event_utils.py

@@ -0,0 +1,66 @@
+import json
+import os
+import socket
+from datetime import datetime, timezone
+from typing import Any
+
+from memos.log import get_logger
+from memos.mem_scheduler.schemas.message_schemas import ScheduleMessageItem
+
+
+logger = get_logger(__name__)
+
+
+def _iso_ts_now() -> str:
+    """Return current UTC timestamp in ISO format with milliseconds."""
+    return datetime.now(timezone.utc).isoformat()
+
+
+def to_iso(ts) -> str | None:
+    """Convert datetime to ISO string; return None if not convertible."""
+    if ts is None:
+        return None
+    if isinstance(ts, datetime):
+        dt = ts
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        return dt.isoformat()
+    try:
+        return datetime.fromtimestamp(float(ts), tz=timezone.utc).isoformat()
+    except Exception:
+        return None
+
+
+def emit_monitor_event(event: str, msg: ScheduleMessageItem, extra: dict[str, Any] | None = None):
+    """
+    Emit a structured MONITOR_EVENT log line for SLS consumption.
+
+    This must be fire-and-forget: any exception here should never break the scheduler flow.
+    """
+    try:
+        payload: dict[str, Any] = {
+            "event": event,
+            "ts": _iso_ts_now(),
+            "label": getattr(msg, "label", None),
+            "user_id": getattr(msg, "user_id", None),
+            "mem_cube_id": getattr(msg, "mem_cube_id", None),
+            "item_id": getattr(msg, "item_id", None),
+            "task_id": getattr(msg, "task_id", "") or "",
+            "trace_id": getattr(msg, "trace_id", None),
+            "stream_key": getattr(msg, "stream_key", None),
+            "redis_message_id": getattr(msg, "redis_message_id", None),
+            "monitor_flag": None,
+            "host": socket.gethostname(),
+            "env": os.getenv("ENV") or os.getenv("ENVIRONMENT") or "",
+        }
+
+        info = getattr(msg, "info", None)
+        if isinstance(info, dict):
+            payload["monitor_flag"] = info.get("monitor_flag")
+
+        if extra:
+            payload.update(extra)
+
+        logger.info("MONITOR_EVENT " + json.dumps(payload, ensure_ascii=False))
+    except Exception:
+        logger.debug("Failed to emit MONITOR_EVENT", exc_info=True)