Initial prototype of telemetry hook

modin/logging/__init__.py

@@ -20,4 +20,7 @@
     "get_logger",
     "enable_logging",
     "disable_logging",
+    "emit_telemetry_event",
+    "add_telemetry_handler",
+    "clear_telemetry_handler"
 ]

modin/logging/logger_decorator.py

@@ -16,14 +16,15 @@
 
 ``enable_logging`` is used for decorating individual Modin functions or classes.
 """
-
+import time
 from __future__ import annotations
 
 from functools import wraps
 from types import FunctionType, MethodType
 from typing import Any, Callable, Dict, Optional, Tuple, TypeVar, overload
 
 from modin.config import LogMode
+from modin.logging.telemetry import emit_telemetry_event
 
 from .config import LogLevel, get_logger
 
@@ -121,8 +122,9 @@
 
         assert isinstance(modin_layer, str), "modin_layer is somehow not a string!"
 
-        start_line = f"START::{modin_layer.upper()}::{name or obj.__name__}"
-        stop_line = f"STOP::{modin_layer.upper()}::{name or obj.__name__}"
+        api_call_name = f"{modin_layer.upper()}::{name or obj.__name__}"
+        start_line = f"START::{api_call_name}"
+        stop_line = f"STOP::{api_call_name}"
 
         @wraps(obj)
         def run_and_log(*args: Tuple, **kwargs: Dict) -> Any:
@@ -145,8 +147,10 @@
 
             logger = get_logger()
             logger.log(log_level, start_line)
+            start_time = time.time()
             try:
                 result = obj(*args, **kwargs)
+                emit_telemetry_event('', api_call_name, time.time() - start_time)
             except BaseException as e:
                 # Only log the exception if a deeper layer of the modin stack has not
                 # already logged it.

modin/logging/telemetry.py

@@ -0,0 +1,27 @@
+import logging
+from modin.utils import timeout
+
+_telemetry_handlers = []
+
+# Telemetry hooks can be implemented by plugin engines
+# to collect discrete data on how modin is performing at the
+# high level moden layer.
+def emit_telemetry_event(msg:str, value:int|float):
+    _telemetry_handlers
+    handlers = _telemetry_handlers.copy()
+    for fn in handlers:
+        try:
+            with timeout(seconds=1):
+                fn(f"modin::{msg}", value)
+        except:
+            logging.ERROR("telemetry handler threw exception, removing handler: " + e)
+            _telemetry_handlers.remove(fn)
+    pass
+
+def add_telemetry_handler(handler:callable):
+    _telemetry_handlers.append(handler)
+
+def clear_telemetry_handler(handler):
+    _telemetry_handlers = []
+
+

modin/utils.py

@@ -20,6 +20,7 @@
 import json
 import os
 import re
+import signal
 import sys
 import types
 import warnings
@@ -980,3 +981,16 @@ def reload_modin() -> None:
     for name, module in modules.items():
         if name.startswith("modin"):
             importlib.reload(module)
+
+# context manager for timing out functions
+class timeout:
+    def __init__(self, seconds=1, error_message='Timeout'):
+        self.seconds = seconds
+        self.error_message = error_message
+    def handle_timeout(self, signum, frame):
+        raise TimeoutError(self.error_message)
+    def __enter__(self):
+        signal.signal(signal.SIGALRM, self.handle_timeout)
+        signal.alarm(self.seconds)
+    def __exit__(self, type, value, traceback):
+        signal.alarm(0)