[nightshift] iris: hoist logging.py imports and simplify callers

Move ``from iris.rpc import logging_pb2`` and the ``_STR_TO_ENUM`` mapping
to module scope so each ``str_to_log_level`` call no longer re-executes
the import and rebuilds the dict. The deferred import had no
corresponding circular dependency (``iris.rpc.logging_pb2`` does not
import ``iris.logging``) and every file that imports ``str_to_log_level``
already imports ``logging_pb2`` directly. Widen the parameter type to
``str | None`` and fold the falsy-check into the function itself, then
drop the redundant ``if level_name else 0`` guards at four call sites.
Update the stale CALLABLE_RUNNER comment that pointed at ``iris.logging``
for ``LevelPrefixFormatter``/``_LEVEL_PREFIX`` — those moved to
``rigging.log_setup``.

Author: Nightshift Scout

lib/iris/src/iris/cluster/log_store/duckdb_store.py

@@ -525,7 +525,7 @@ def get_logs(
         regular expression and matched with DuckDB's ``regexp_matches()``.
         Otherwise it is treated as an exact key lookup.
         """
-        min_level_enum = str_to_log_level(min_level) if min_level else 0
+        min_level_enum = str_to_log_level(min_level)
         is_pattern = bool(REGEX_META_RE.search(key))
 
         if not is_pattern:

lib/iris/src/iris/cluster/log_store/mem_store.py

@@ -103,7 +103,7 @@ def get_logs(
         tail: bool = False,
         min_level: str = "",
     ) -> LogReadResult:
-        min_level_enum = str_to_log_level(min_level) if min_level else 0
+        min_level_enum = str_to_log_level(min_level)
         is_pattern = bool(REGEX_META_RE.search(key))
 
         if is_pattern:

lib/iris/src/iris/cluster/providers/k8s/tasks.py

@@ -525,7 +525,7 @@ def _build_pod_manifest(
 
 def _kubectl_log_line_to_log_entry(kll: KubectlLogLine, attempt_id: int) -> logging_pb2.LogEntry:
     level_name = parse_log_level(kll.data)
-    level = str_to_log_level(level_name) if level_name else 0
+    level = str_to_log_level(level_name)
     entry = logging_pb2.LogEntry(source=kll.stream, data=kll.data, attempt_id=attempt_id, level=level)
     entry.timestamp.CopyFrom(timestamp_to_proto(Timestamp.from_seconds(kll.timestamp.timestamp())))
     return entry

lib/iris/src/iris/cluster/types.py

@@ -481,9 +481,9 @@ def to_proto(self) -> job_pb2.ResourceSpecProto:
 
 # Reinitialize logging with the unified Iris format.
 # Uses single-letter level prefix: I=INFO, W=WARNING, E=ERROR, D=DEBUG, C=CRITICAL.
-# NOTE: This duplicates LevelPrefixFormatter and _LEVEL_PREFIX from iris.logging
+# NOTE: This duplicates LevelPrefixFormatter and _LEVEL_PREFIX from rigging.log_setup
 # because CALLABLE_RUNNER executes inside an isolated task container that may not
-# have the iris package installed (e.g. user-provided Docker images).
+# have the rigging package installed (e.g. user-provided Docker images).
 _LEVEL_PREFIX = {"DEBUG": "D", "INFO": "I", "WARNING": "W", "ERROR": "E", "CRITICAL": "C"}
 
 class _LevelPrefixFormatter(logging.Formatter):

lib/iris/src/iris/cluster/worker/task_attempt.py

@@ -898,7 +898,7 @@ def _monitor_loop(
     def _make_log_entry(self, *, source: str, data: str) -> logging_pb2.LogEntry:
         """Build a LogEntry proto from a source/data pair, parsing the level prefix."""
         level_name = parse_log_level(data)
-        level = str_to_log_level(level_name) if level_name else 0
+        level = str_to_log_level(level_name)
         entry = logging_pb2.LogEntry(source=source, data=data, level=level)
         entry.timestamp.epoch_ms = Timestamp.now().epoch_ms()
         return entry

lib/iris/src/iris/logging.py

@@ -8,24 +8,23 @@
 retains only the functions that depend on ``iris.rpc.logging_pb2``.
 """
 
+from iris.rpc import logging_pb2
 
-def str_to_log_level(level_name: str) -> int:
+_STR_TO_ENUM = {
+    "DEBUG": logging_pb2.LOG_LEVEL_DEBUG,
+    "INFO": logging_pb2.LOG_LEVEL_INFO,
+    "WARNING": logging_pb2.LOG_LEVEL_WARNING,
+    "ERROR": logging_pb2.LOG_LEVEL_ERROR,
+    "CRITICAL": logging_pb2.LOG_LEVEL_CRITICAL,
+}
+
+
+def str_to_log_level(level_name: str | None) -> int:
     """Convert a canonical level name (e.g. "INFO") to the LogLevel proto enum value.
 
-    Returns LOG_LEVEL_UNKNOWN (0) for unrecognized names.
-    Uses lazy import to avoid pulling in protobuf at module load time.
+    Returns ``LOG_LEVEL_UNKNOWN`` (0) for ``None``, empty strings, and
+    unrecognized names.
     """
-    from iris.rpc import logging_pb2
-
-    _STR_TO_ENUM = {
-        "DEBUG": logging_pb2.LOG_LEVEL_DEBUG,
-        "INFO": logging_pb2.LOG_LEVEL_INFO,
-        "WARNING": logging_pb2.LOG_LEVEL_WARNING,
-        "ERROR": logging_pb2.LOG_LEVEL_ERROR,
-        "CRITICAL": logging_pb2.LOG_LEVEL_CRITICAL,
-    }
-    return (
-        _STR_TO_ENUM.get(level_name.upper(), logging_pb2.LOG_LEVEL_UNKNOWN)
-        if level_name
-        else logging_pb2.LOG_LEVEL_UNKNOWN
-    )
+    if not level_name:
+        return logging_pb2.LOG_LEVEL_UNKNOWN
+    return _STR_TO_ENUM.get(level_name.upper(), logging_pb2.LOG_LEVEL_UNKNOWN)