Coverage for src / agent / observability.py: 88%

1# Copyright 2025-2026 Microsoft Corporation 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Observability helpers for Agent Base. 

16 

17This module provides helper utilities for OpenTelemetry integration. 

18The main observability setup is handled by agent_framework.observability.setup_observability(). 

19 

20For full observability setup, see: 

21- agent_framework.observability.setup_observability() 

22- Examples: examples/observability_example.py 

23- Documentation: docs/decisions/0014-observability-integration.md 

24""" 

25 

26import contextvars 

27import logging 

28import socket 

29from typing import Any 

30from urllib.parse import urlparse 

31 

32logger = logging.getLogger(__name__) 

33 

34# Context var to hold the current agent span for cross-task propagation 

35_current_agent_span: contextvars.ContextVar[Any] = contextvars.ContextVar( 

36 "_current_agent_span", default=None 

37) 

38 

39 

40def set_current_agent_span(span: Any) -> None: 

41 """Record the current agent span in a context variable. 

42 

43 This helps preserve parent-child relationships for tool spans when 

44 asynchronous task boundaries might lose the active span context. 

45 

46 Args: 

47 span: The agent-level span to set as current 

48 """ 

49 try: 

50 _current_agent_span.set(span) 

51 except Exception: 

52 # Best-effort: ignore issues with context vars in constrained envs 

53 pass 

54 

55 

56def get_current_agent_span() -> Any: 

57 """Retrieve the current agent span from the context variable. 

58 

59 Returns: 

60 The agent span if set, otherwise None. 

61 """ 

62 try: 

63 return _current_agent_span.get() 

64 except Exception: 

65 return None 

66 

67 

68def check_telemetry_endpoint(endpoint: str | None = None, timeout: float = 0.02) -> bool: 

69 """Check if telemetry endpoint is reachable. 

70 

71 Uses a fast socket connection test with minimal timeout to avoid startup delays. 

72 This enables auto-detection of telemetry availability without user configuration. 

73 

74 Args: 

75 endpoint: OTLP endpoint URL (default: http://localhost:4317) 

76 timeout: Connection timeout in seconds (default: 0.02 = 20ms) 

77 

78 Returns: 

79 True if endpoint is reachable, False otherwise 

80 

81 Example: 

82 >>> if check_telemetry_endpoint(): 

83 ... setup_observability() 

84 

85 Performance: 

86 - When available: ~1-2ms 

87 - When unavailable: ~20-30ms (timeout period) 

88 """ 

89 if not endpoint: 

90 endpoint = "http://localhost:4317" 

91 

92 try: 

93 # Parse endpoint URL to extract host and port 

94 parsed = urlparse(endpoint) 

95 host = parsed.hostname or "localhost" 

96 port = parsed.port or 4317 

97 

98 # Fast socket connection check 

99 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

100 sock.settimeout(timeout) 

101 result = sock.connect_ex((host, port)) 

102 sock.close() 

103 

104 return result == 0 

105 except Exception as e: 

106 logger.debug(f"Telemetry endpoint check failed: {e}") 

107 return False