Fix issue 2098: Add opt-in structured logging, metrics, and tracing

qlib/config.py

@@ -111,13 +111,17 @@ def set_conf_from_C(self, config_c):
     @staticmethod
     def register_from_C(config, skip_register=True):
         from .utils import set_log_with_config  # pylint: disable=C0415
+        from .metrics import configure_metrics  # pylint: disable=C0415
+        from .trace import configure_tracing  # pylint: disable=C0415
 
         if C.registered and skip_register:
             return
 
         C.set_conf_from_C(config)
         if C.logging_config:
             set_log_with_config(C.logging_config)
+        configure_metrics(C.get("metrics_config", {}))
+        configure_tracing(C.get("tracing_config", {}))
         C.register()
 
 
@@ -183,6 +187,14 @@ def register_from_C(config, skip_register=True):
     "redis_password": None,
     # This value can be reset via qlib.init
     "logging_level": logging.INFO,
+    # Optional process-local observability metrics. Disabled by default.
+    "metrics_config": {
+        "enabled": False,
+    },
+    # Optional lightweight workflow tracing. Disabled by default.
+    "tracing_config": {
+        "enabled": False,
+    },
     # Global configuration of qlib log
     # logging_level can control the logging level more finely
     "logging_config": {
@@ -440,6 +452,8 @@ def set(self, default_conf: str = "client", **kwargs):
             the default config template chosen by user: "server", "client"
         """
         from .utils import set_log_with_config, get_module_logger, can_use_cache  # pylint: disable=C0415
+        from .metrics import configure_metrics  # pylint: disable=C0415
+        from .trace import configure_tracing  # pylint: disable=C0415
 
         self.reset()
 
@@ -460,6 +474,9 @@ def set(self, default_conf: str = "client", **kwargs):
                 logger.warning("Unrecognized config %s" % k)
             self[k] = v
 
+        configure_metrics(self.get("metrics_config", {}))
+        configure_tracing(self.get("tracing_config", {}))
+
         self.resolve_path()
 
         if not (self["expression_cache"] is None and self["dataset_cache"] is None):

qlib/data/cache.py

@@ -33,6 +33,8 @@
 from ..utils.pickle_utils import restricted_pickle_load
 
 from ..log import get_module_logger
+from ..metrics import get_metrics_recorder
+from ..trace import trace_span
 from .base import Feature
 from .ops import Operators  # pylint: disable=W0611  # noqa: F401
 
@@ -1084,6 +1086,17 @@ def _uri(self, instruments, fields, start_time, end_time, freq, disk_cache=1, in
 
     def _dataset(
         self, instruments, fields, start_time=None, end_time=None, freq="day", disk_cache=1, inst_processors=[]
+    ):
+        metrics = get_metrics_recorder()
+        metric_tags = {"cache": "simple", "freq": freq}
+        with trace_span("cache.dataset"):
+            with metrics.timer("qlib.cache.dataset.load_seconds", tags=metric_tags):
+                return self._dataset_with_metrics(
+                    instruments, fields, start_time, end_time, freq, disk_cache, inst_processors, metrics, metric_tags
+                )
+
+    def _dataset_with_metrics(
+        self, instruments, fields, start_time, end_time, freq, disk_cache, inst_processors, metrics, metric_tags
     ):
         if disk_cache == 0:
             # In this case, data_set cache is configured but will not be used.
@@ -1099,6 +1112,7 @@ def _dataset(
         if cache_file.exists():
             if disk_cache == 1:
                 # use cache
+                metrics.increment("qlib.cache.dataset.hit", tags=metric_tags)
                 df = pd.read_pickle(cache_file)
                 return self.cache_to_origin_data(df, fields)
             elif disk_cache == 2:
@@ -1108,6 +1122,7 @@ def _dataset(
             gen_flag = True
 
         if gen_flag:
+            metrics.increment("qlib.cache.dataset.miss", tags=metric_tags)
             data = self.provider.dataset(
                 instruments, normalize_cache_fields(fields), start_time, end_time, freq, inst_processors=inst_processors
             )
@@ -1183,15 +1198,21 @@ class CalendarCache(BaseProviderCache):
 
 class MemoryCalendarCache(CalendarCache):
     def calendar(self, start_time=None, end_time=None, freq="day", future=False):
+        metrics = get_metrics_recorder()
+        metric_tags = {"cache": "memory", "freq": freq}
         uri = self._uri(start_time, end_time, freq, future)
-        result, expire = MemCacheExpire.get_cache(H["c"], uri)
-        if result is None or expire:
-            result = self.provider.calendar(start_time, end_time, freq, future)
-            MemCacheExpire.set_cache(H["c"], uri, result)
-
-            get_module_logger("data").debug(f"get calendar from {C.calendar_provider}")
-        else:
-            get_module_logger("data").debug("get calendar from local cache")
+        with trace_span("cache.calendar"):
+            with metrics.timer("qlib.cache.calendar.load_seconds", tags=metric_tags):
+                result, expire = MemCacheExpire.get_cache(H["c"], uri)
+                if result is None or expire:
+                    metrics.increment("qlib.cache.calendar.miss", tags=metric_tags)
+                    result = self.provider.calendar(start_time, end_time, freq, future)
+                    MemCacheExpire.set_cache(H["c"], uri, result)
+
+                    get_module_logger("data").debug(f"get calendar from {C.calendar_provider}")
+                else:
+                    metrics.increment("qlib.cache.calendar.hit", tags=metric_tags)
+                    get_module_logger("data").debug("get calendar from local cache")
 
         return result
 

qlib/log.py

@@ -2,6 +2,8 @@
 # Licensed under the MIT License.
 
 
+import copy
+import json
 import logging
 from typing import Optional, Text, Dict, Any
 import re
@@ -12,6 +14,79 @@
 from .config import C
 
 
+_LOG_RECORD_ATTRIBUTES = frozenset(logging.makeLogRecord({}).__dict__)
+_STRUCTURED_LOGGING_FORMAT = "json"
+
+
+class JSONFormatter(logging.Formatter):
+    """Format log records as JSON, including fields supplied through ``extra``."""
+
+    def format(self, record: logging.LogRecord) -> str:
+        log_data = {
+            "timestamp": self.formatTime(record, self.datefmt),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+        }
+        try:
+            from .trace import get_current_trace_context  # pylint: disable=C0415
+
+            log_data.update(get_current_trace_context())
+        except Exception:
+            pass
+
+        log_data.update(
+            {
+                key: value
+                for key, value in record.__dict__.items()
+                if key not in _LOG_RECORD_ATTRIBUTES and key not in log_data
+            }
+        )
+
+        if record.exc_info:
+            log_data["exception"] = self.formatException(record.exc_info)
+
+        return json.dumps(log_data, default=str)
+
+
+def _normalize_logging_config(log_config: Dict[Text, Any]) -> Dict[Text, Any]:
+    """Normalize Qlib's logging config before passing it to ``dictConfig``.
+
+    Qlib continues to accept a standard ``logging.config.dictConfig`` mapping.
+    In addition, users can opt into structured JSON logs with a compact config:
+
+    .. code-block:: python
+
+        qlib.init(logging_config={"structured": True, "format": "json"})
+
+    The compact config reuses Qlib's default handlers and only swaps their
+    formatter to :class:`JSONFormatter`.
+    """
+    if not isinstance(log_config, dict):
+        return log_config
+
+    is_dict_config = "version" in log_config
+    structured = bool(log_config.get("structured", False))
+
+    if not structured:
+        return copy.deepcopy(log_config)
+
+    log_format = log_config.get("format", _STRUCTURED_LOGGING_FORMAT)
+    if log_format != _STRUCTURED_LOGGING_FORMAT:
+        raise ValueError(f"Unsupported structured logging format: {log_format}")
+
+    default_logging_config = C.__dict__["_default_config"]["logging_config"]
+    normalized_config = copy.deepcopy(log_config if is_dict_config else default_logging_config)
+    normalized_config.pop("structured", None)
+    normalized_config.pop("format", None)
+
+    normalized_config.setdefault("formatters", {})[_STRUCTURED_LOGGING_FORMAT] = {"()": "qlib.log.JSONFormatter"}
+    for handler_config in normalized_config.get("handlers", {}).values():
+        handler_config["formatter"] = _STRUCTURED_LOGGING_FORMAT
+
+    return normalized_config
+
+
 class MetaLogger(type):
     def __new__(mcs, name, bases, attrs):  # pylint: disable=C0204
         wrapper_dict = logging.Logger.__dict__.copy()
@@ -155,6 +230,7 @@ def set_log_with_config(log_config: Dict[Text, Any]):
     :param log_config:
     :return:
     """
+    log_config = _normalize_logging_config(log_config)
     logging_config.dictConfig(log_config)