feat(cuda.core): bridge CUDA driver logs into Python logging

cuda_core/cuda/core/utils/__init__.py

@@ -17,6 +17,10 @@
     StridedMemoryView,
     args_viewable_as_strided_memory,
 )
+from cuda.core.utils._logging import (
+    register_cuda_error_log,
+    unregister_cuda_error_log,
+)
 from cuda.core.utils._program_cache import (
     FileStreamProgramCache,
     InMemoryProgramCache,
@@ -38,4 +42,6 @@
     "discard_prefetch_batch",
     "make_program_cache_key",
     "prefetch_batch",
+    "register_cuda_error_log",
+    "unregister_cuda_error_log",
 ]

cuda_core/cuda/core/utils/_logging.py

@@ -0,0 +1,121 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Bridge CUDA driver error logs into Python's :mod:`logging` module via
+the ``cuLogs*`` API.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import logging
+import threading
+from typing import Callable
+
+from cuda.bindings import driver as _driver
+from cuda.core._utils.cuda_utils import handle_return
+
+__all__ = ["register_cuda_error_log", "unregister_cuda_error_log"]
+
+# C callback signature for cuLogsRegisterCallback:
+#     void callback(void *userData, CUlogLevel logLevel, char *message, size_t length)
+_CUlogsCallback_functype = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_size_t)
+PyLogCallback = Callable[[int, bytes], None]
+
+# Keep a reference while registered, or GC'ing it leaves the driver with
+# a dangling function pointer which will crash on the next log message.
+_lock = threading.Lock()
+_c_callback: ctypes._FuncPointer | None = None
+_callback_handle = None  # CUlogsCallbackHandle, needed to unregister
+
+
+def _make_c_callback(logger: logging.Logger, callback: PyLogCallback | None) -> ctypes._FuncPointer:
+    """Build the ctypes C callback passed to cuLogsRegisterCallback."""
+
+    def _default_callback(cu_log_level: int, message: bytes) -> None:
+        cuda_level_to_py_level = {
+            int(_driver.CUlogLevel.CU_LOG_LEVEL_ERROR): logging.ERROR,
+            int(_driver.CUlogLevel.CU_LOG_LEVEL_WARNING): logging.WARNING,
+        }
+        msg = message.decode("utf-8", errors="replace").rstrip("\n")
+        py_level = cuda_level_to_py_level.get(cu_log_level, logging.ERROR)
+        logger.log(py_level, "[CUDA Driver] %s", msg)
+
+    emit = callback if callback is not None else _default_callback
+
+    def _on_log(_user_data, log_level, message_ptr, length):
+        try:
+            message = ctypes.string_at(message_ptr, length) if message_ptr else b""
+            emit(log_level, message)
+        except Exception:
+            logger.exception("cuda_error_log: error while forwarding driver log")
+
+    return _CUlogsCallback_functype(_on_log)
+
+
+def register_cuda_error_log(
+    logger: logging.Logger | None = None,
+    callback: PyLogCallback | None = None,
+) -> logging.Logger:
+    """Register a bridge that forwards CUDA driver ``cuLogs*`` messages to
+    a Python :class:`logging.Logger`.
+
+    Parameters
+    ----------
+    logger : logging.Logger, optional
+        Target logger. If not provided, a default is created via
+        ``logging.getLogger("cuda.driver")``.
+    callback : Callable[[int, bytes], None], optional
+        Overrides the default forwarding. Called as
+        ``callback(cu_log_level, message)`` with undecoded ``message`` and
+        unmapped ``cu_log_level``; decoding/mapping is the callback's job.
+
+    Returns
+    -------
+    logging.Logger
+        The ``logger`` argument above (or the default logger it resolved
+        to). If a bridge was already registered, it is unregistered first,
+        so this logger/callback replaces it.
+
+    """
+    global _c_callback, _callback_handle
+
+    if logger is None:
+        logger = logging.getLogger("cuda.driver")
+
+    if callback is not None and not callable(callback):
+        raise TypeError(f"callback must be callable or None, got {type(callback)!r}")
+
+    with _lock:
+        if _callback_handle is not None:
+            # Already registered: unregister the old callback/handle first,
+            # then fall through to register the new logger/callback below.
+            handle_return(_driver.cuLogsUnregisterCallback(_callback_handle))
+            _c_callback = None
+            _callback_handle = None
+
+        c_callback = _make_c_callback(logger, callback)
+        addr = ctypes.cast(c_callback, ctypes.c_void_p).value
+
+        handle = handle_return(_driver.cuLogsRegisterCallback(addr, None))
+
+        # Keep the ctypes callback alive (see _c_callback comment above).
+        _c_callback = c_callback
+        _callback_handle = handle
+
+    return logger
+
+
+def unregister_cuda_error_log() -> None:
+    """Unregister the bridge installed by :func:`register_cuda_error_log`."""
+    global _c_callback, _callback_handle
+
+    with _lock:
+        if _callback_handle is None:
+            return
+
+        handle_return(_driver.cuLogsUnregisterCallback(_callback_handle))
+
+        _c_callback = None
+        _callback_handle = None

cuda_core/docs/source/api.rst

@@ -126,6 +126,21 @@ avoid recompiling identical source + options + target without writing the
    make_program_cache_key
 
 
+Driver logging
+``````````````
+
+:func:`~cuda.core.utils.register_cuda_error_log` bridges CUDA driver
+``cuLogs*`` messages into Python's standard :mod:`logging` module.
+
+.. currentmodule:: cuda.core.utils
+
+.. autosummary::
+   :toctree: generated/
+
+   register_cuda_error_log
+   unregister_cuda_error_log
+
+
 CUDA graphs
 -----------
 

cuda_core/tests/test_cuda_log_bridge.py

@@ -0,0 +1,176 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for bridging CUDA driver ``cuLogs*`` messages into Python
+``logging`` (see design doc / GitHub issue #671).
+"""
+
+import logging
+
+import pytest
+
+from cuda.bindings import driver
+from cuda.core.utils import register_cuda_error_log, unregister_cuda_error_log
+
+
+@pytest.fixture(autouse=True)
+def teardown_bridge():
+    yield
+    # Unregister after every test so a leaked bridge can't leak into others.
+    unregister_cuda_error_log()
+
+
+# Usage 1: show driver error log interactively
+def test_cuda_driver_log_captured_via_caplog(caplog):
+    driver.cuInit(0)
+
+    logger_name = "cuda.driver.caplog_test"
+    logger_obj = logging.getLogger(logger_name)
+    register_cuda_error_log(logger_obj)
+
+    # cuDeviceGet(9999) is out of range; driver writes two ERROR logs
+    with caplog.at_level(logging.ERROR, logger=logger_name):
+        driver.cuDeviceGet(9999)
+
+    messages = [r.message for r in caplog.records if r.name == logger_name]
+
+    assert len(messages) == 2, f"Expected exactly 2 driver log messages, got:\n{messages}"
+    assert "Parameter ordinal must be between 0 and 1" in messages[0]
+    assert "CUDA_ERROR_INVALID_DEVICE" in messages[1] and "cuDeviceGet" in messages[1]
+
+    # After unregistering, driver logs are no longer forwarded
+    unregister_cuda_error_log()
+    caplog.clear()
+    with caplog.at_level(logging.ERROR, logger=logger_name):
+        driver.cuDeviceGet(9999)
+    assert not any(r.name == logger_name for r in caplog.records)
+    assert "Parameter ordinal must be between 0 and 1" not in caplog.text
+
+
+# Usage 2: redirect driver log to a file
+def test_cuda_driver_log_written_to_file(tmp_path):
+    driver.cuInit(0)
+
+    log_file = tmp_path / "cuda_driver.log"
+    logger_name = "cuda.driver.file_test"
+    logger_obj = logging.getLogger(logger_name)
+    logger_obj.setLevel(logging.WARNING)
+
+    file_handler = logging.FileHandler(log_file, encoding="utf-8")
+    file_handler.setLevel(logging.WARNING)
+    file_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
+    logger_obj.addHandler(file_handler)
+
+    try:
+        register_cuda_error_log(logger_obj)
+        driver.cuDeviceGet(9999)
+    finally:
+        logger_obj.removeHandler(file_handler)
+        file_handler.close()
+
+    content = log_file.read_text(encoding="utf-8")
+
+    assert "Parameter ordinal must be between 0 and 1" in content
+    assert "CUDA_ERROR_INVALID_DEVICE" in content and "cuDeviceGet" in content
+
+
+# Usage 3: register -> unregister -> rebind to another logger
+def test_cuda_driver_log_rebind_logger(caplog):
+    driver.cuInit(0)
+
+    name_a, name_b = "cuda.driver.rebind_a", "cuda.driver.rebind_b"
+
+    register_cuda_error_log(logging.getLogger(name_a))
+    with caplog.at_level(logging.ERROR, logger=name_a):
+        driver.cuDeviceGet(9999)
+    assert any(r.name == name_a for r in caplog.records)
+
+    # Must unregister first to free the driver-side slot, then bind the new logger
+    unregister_cuda_error_log()
+    caplog.clear()
+    register_cuda_error_log(logging.getLogger(name_b))
+    with caplog.at_level(logging.ERROR, logger=name_b):
+        driver.cuDeviceGet(9999)
+
+    assert not any(r.name == name_a for r in caplog.records)
+    assert any(r.name == name_b for r in caplog.records)
+
+
+# Usage 4: the logger's level threshold applies to driver logs too
+def test_cuda_driver_log_level_filtering(caplog):
+    driver.cuInit(0)
+
+    logger_name = "cuda.driver.level_test"
+    register_cuda_error_log(logging.getLogger(logger_name))
+
+    # CRITICAL threshold filters out ERROR-level driver logs
+    with caplog.at_level(logging.CRITICAL, logger=logger_name):
+        driver.cuDeviceGet(9999)
+    assert not any(r.name == logger_name for r in caplog.records)
+
+    # Lowering to ERROR threshold lets them through
+    with caplog.at_level(logging.ERROR, logger=logger_name):
+        driver.cuDeviceGet(9999)
+    assert any(r.name == logger_name for r in caplog.records)
+
+
+# Usage 5: custom Handler counts errors, for monitoring/alerting integration
+def test_cuda_driver_log_alert_counter():
+    driver.cuInit(0)
+
+    class ErrorCounterHandler(logging.Handler):
+        def __init__(self):
+            super().__init__(level=logging.ERROR)
+            self.count = 0
+
+        def emit(self, record):
+            self.count += 1
+
+    logger_obj = logging.getLogger("cuda.driver.alert_test")
+    counter_handler = ErrorCounterHandler()
+    logger_obj.addHandler(counter_handler)
+
+    try:
+        register_cuda_error_log(logger_obj)
+        driver.cuDeviceGet(9999)
+    finally:
+        logger_obj.removeHandler(counter_handler)
+
+    assert counter_handler.count == 2, "Expected exactly 2 ERROR-level driver logs to be counted"
+
+
+# Usage 6: use a custom python callback function to change log level and log format
+def test_cuda_driver_log_custom_callback_overrides_default(caplog):
+    driver.cuInit(0)
+
+    logger_name = "my_library.cuda-driver"
+    logger_obj = logging.getLogger(logger_name)
+
+    # Log everything as DEBUG, prefixed with the logger name instead of "[CUDA Driver] "
+    def my_callback(cu_log_level, message):
+        msg = message.decode("utf-8", errors="replace").rstrip("\n")
+        logger_obj.debug("%s: %s", logger_obj.name, msg)
+
+    register_cuda_error_log(logger_obj, callback=my_callback)
+
+    # Default WARNING threshold filters out the DEBUG-level driver logs
+    with caplog.at_level(logging.WARNING, logger=logger_name):
+        driver.cuDeviceGet(9999)
+    assert not caplog.records
+
+    caplog.clear()
+
+    # Lowering the threshold to DEBUG lets them through, with the custom level/format
+    with caplog.at_level(logging.DEBUG, logger=logger_name):
+        driver.cuDeviceGet(9999)
+
+    assert len(caplog.records) == 2
+    assert all(r.levelno == logging.DEBUG for r in caplog.records)
+    assert all(r.message.startswith(f"{logger_name}: ") for r in caplog.records)
+
+
+# callback must be callable or None, else fail fast
+def test_cuda_driver_log_callback_must_be_callable_or_none():
+    with pytest.raises(TypeError):
+        register_cuda_error_log(logging.getLogger("cuda.driver.bad_callback_test"), callback="not-callable")