Coverage for agentos/observability/logging.py: 0%

1""" 

2AgentOS Structured Logging — JSON-formatted production logging with context propagation. 

3 

4Provides: 

5- JsonFormatter: Emits JSON lines with standard fields (timestamp, level, logger, etc.) 

6- ContextLogger: Logger wrapper that injects request_id, tenant_id, session_id into every log. 

7- configure_logging(): One-call setup for production logging. 

8 

9Usage: 

10 from agentos.observability.logging import configure_logging, get_logger 

11 

12 configure_logging(level="INFO", log_file="/app/logs/agentos.log") 

13 logger = get_logger(__name__) 

14 logger.info("Agent started", extra={"agent_id": "abc123"}) 

15""" 

16 

17from __future__ import annotations 

18 

19import json 

20import logging 

21import os 

22import sys 

23from contextvars import ContextVar 

24from datetime import datetime, timezone 

25from typing import Any, Optional 

26 

27# ── Context propagation ────────────────────────────────────────────────── 

28 

29_request_id: ContextVar[str] = ContextVar("request_id", default="") 

30_tenant_id: ContextVar[str] = ContextVar("tenant_id", default="") 

31_session_id: ContextVar[str] = ContextVar("session_id", default="") 

32 

33STANDARD_FIELDS = { 

34 "timestamp", "level", "logger", "module", "function", "line", 

35 "message", "request_id", "tenant_id", "session_id", "pid", 

36} 

37 

38_INTERNAL_RECORD_FIELDS = { 

39 "name", "msg", "args", "levelname", "levelno", "pathname", 

40 "filename", "funcName", "exc_info", "exc_text", "stack_info", 

41 "lineno", "created", "msecs", "relativeCreated", "thread", 

42 "threadName", "processName", "process", "taskName", "extra", 

43} 

44 

45 

46class JsonFormatter(logging.Formatter): 

47 """JSON Lines formatter for structured logging. 

48 

49 Output fields: 

50 timestamp, level, logger, module, function, line, message, 

51 request_id, tenant_id, session_id, pid, + any extra fields. 

52 """ 

53 

54 def __init__(self, include_extra: bool = True): 

55 super().__init__() 

56 self.include_extra = include_extra 

57 

58 def format(self, record: logging.LogRecord) -> str: 

59 log_entry: dict[str, Any] = { 

60 "timestamp": datetime.fromtimestamp( 

61 record.created, tz=timezone.utc 

62 ).isoformat(), 

63 "level": record.levelname, 

64 "logger": record.name, 

65 "module": record.module, 

66 "function": record.funcName, 

67 "line": record.lineno, 

68 "message": record.getMessage(), 

69 "request_id": _request_id.get(""), 

70 "tenant_id": _tenant_id.get(""), 

71 "session_id": _session_id.get(""), 

72 "pid": os.getpid(), 

73 } 

74 

75 # Exception info (only if it's a real sys.exc_info() tuple, not a bool) 

76 if record.exc_info and record.exc_info is not True and record.exc_info[1]: 

77 log_entry["exception"] = { 

78 "type": type(record.exc_info[1]).__name__, 

79 "message": str(record.exc_info[1]), 

80 } 

81 

82 # Stack trace 

83 if record.stack_info: 

84 log_entry["stack"] = record.stack_info 

85 

86 # Extra fields from record.__dict__ (filter out internal log record fields) 

87 if self.include_extra: 

88 for key, value in record.__dict__.items(): 

89 if key not in _INTERNAL_RECORD_FIELDS and not key.startswith("_"): 

90 try: 

91 json.dumps({key: value}) # ensure serializable 

92 log_entry[key] = value 

93 except (TypeError, ValueError): 

94 log_entry[key] = str(value) 

95 

96 return json.dumps(log_entry, default=str, ensure_ascii=False) 

97 

98 

99class ContextLogger: 

100 """Logger wrapper that automatically injects context into extra fields. 

101 

102 Usage: 

103 logger = ContextLogger(logging.getLogger(__name__)) 

104 logger.info("Processing task", task_id="t-42") 

105 """ 

106 

107 def __init__(self, logger: logging.Logger): 

108 self._logger = logger 

109 

110 def _make_extra(self, extra: Optional[dict] = None) -> dict: 

111 base = { 

112 "request_id": _request_id.get(""), 

113 "tenant_id": _tenant_id.get(""), 

114 "session_id": _session_id.get(""), 

115 } 

116 if extra: 

117 base.update(extra) 

118 return base 

119 

120 def debug(self, msg: str, **extra): 

121 self._logger.debug(msg, extra=self._make_extra(extra)) 

122 

123 def info(self, msg: str, **extra): 

124 self._logger.info(msg, extra=self._make_extra(extra)) 

125 

126 def warning(self, msg: str, **extra): 

127 self._logger.warning(msg, extra=self._make_extra(extra)) 

128 

129 def error(self, msg: str, **extra): 

130 self._logger.error(msg, extra=self._make_extra(extra)) 

131 

132 def critical(self, msg: str, **extra): 

133 self._logger.critical(msg, extra=self._make_extra(extra)) 

134 

135 def exception(self, msg: str, **extra): 

136 self._logger.exception(msg, extra=self._make_extra(extra)) 

137 

138 @property 

139 def raw(self) -> logging.Logger: 

140 return self._logger 

141 

142 

143# ── Context management ──────────────────────────────────────────────────── 

144 

145def set_request_id(rid: str) -> None: 

146 _request_id.set(rid) 

147 

148 

149def set_tenant_id(tid: str) -> None: 

150 _tenant_id.set(tid) 

151 

152 

153def set_session_id(sid: str) -> None: 

154 _session_id.set(sid) 

155 

156 

157def get_request_id() -> str: 

158 return _request_id.get("") 

159 

160 

161def get_tenant_id() -> str: 

162 return _tenant_id.get("") 

163 

164 

165def get_session_id() -> str: 

166 return _session_id.get("") 

167 

168 

169# ── Configuration ───────────────────────────────────────────────────────── 

170 

171def configure_logging( 

172 level: str = "INFO", 

173 log_file: Optional[str] = None, 

174 json_format: bool = True, 

175 include_extra: bool = True, 

176 root_logger_name: str = "agentos", 

177) -> None: 

178 """Configure structured logging for the entire application. 

179 

180 Args: 

181 level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). 

182 log_file: Optional file path for log output. stdout if None. 

183 json_format: Use JSON formatter. If False, use standard format. 

184 include_extra: Include extra fields in JSON output. 

185 root_logger_name: Root logger to configure. 

186 """ 

187 root = logging.getLogger(root_logger_name) 

188 root.setLevel(getattr(logging, level.upper(), logging.INFO)) 

189 

190 # Remove existing handlers 

191 root.handlers.clear() 

192 

193 # Create handler 

194 if log_file: 

195 os.makedirs(os.path.dirname(log_file), exist_ok=True) 

196 handler = logging.FileHandler(log_file) 

197 else: 

198 handler = logging.StreamHandler(sys.stdout) 

199 

200 # Set formatter 

201 if json_format: 

202 handler.setFormatter(JsonFormatter(include_extra=include_extra)) 

203 else: 

204 handler.setFormatter(logging.Formatter( 

205 "[%(asctime)s] %(levelname)s %(name)s:%(lineno)d - %(message)s", 

206 datefmt="%Y-%m-%d %H:%M:%S", 

207 )) 

208 

209 root.addHandler(handler) 

210 

211 # Also configure uvicorn access log 

212 uvicorn_logger = logging.getLogger("uvicorn.access") 

213 uvicorn_logger.handlers.clear() 

214 uvicorn_logger.addHandler(handler) 

215 uvicorn_logger.propagate = False 

216 

217 

218def get_logger(name: str) -> ContextLogger: 

219 """Get a structured context logger.""" 

220 return ContextLogger(logging.getLogger(name)) 

221 

222 

223__all__ = [ 

224 "JsonFormatter", 

225 "ContextLogger", 

226 "configure_logging", 

227 "get_logger", 

228 "set_request_id", 

229 "set_tenant_id", 

230 "set_session_id", 

231 "get_request_id", 

232 "get_tenant_id", 

233 "get_session_id", 

234]