Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/services/telemetry.py: 85%

1""" 

2The Telemetry service. Sends anonymous data to Prefect to help us improve. 

3""" 

4 

5import logging 

6import os 

7import platform 

8from datetime import timedelta 

9from uuid import uuid4 

10 

11import httpx 

12from docket import Perpetual 

13 

14import prefect 

15from prefect.logging import get_logger 

16from prefect.server.database import PrefectDBInterface, provide_database_interface 

17from prefect.server.models import configuration 

18from prefect.server.schemas.core import Configuration 

19from prefect.server.services.perpetual_services import perpetual_service 

20from prefect.settings import PREFECT_DEBUG_MODE 

21from prefect.settings.context import get_current_settings 

22from prefect.types._datetime import now 

23 

24logger: logging.Logger = get_logger(__name__) 

25 

26 

27async def _fetch_or_set_telemetry_session( 

28 db: PrefectDBInterface, 

29) -> tuple[str, str]: 

30 """ 

31 Fetch or create a telemetry session in the configuration table. 

32 

33 Returns: 

34 tuple of (session_start_timestamp, session_id) 

35 """ 

36 async with db.session_context(begin_transaction=True) as session: 

37 telemetry_session = await configuration.read_configuration( 

38 session, "TELEMETRY_SESSION" 

39 ) 

40 

41 if telemetry_session is None: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 logger.debug("No telemetry session found, setting") 

43 session_id = str(uuid4()) 

44 session_start_timestamp = now("UTC").isoformat() 

45 

46 telemetry_session = Configuration( 

47 key="TELEMETRY_SESSION", 

48 value={ 

49 "session_id": session_id, 

50 "session_start_timestamp": session_start_timestamp, 

51 }, 

52 ) 

53 

54 await configuration.write_configuration(session, telemetry_session) 

55 else: 

56 logger.debug("Session information retrieved from database") 

57 session_id = telemetry_session.value["session_id"] 

58 session_start_timestamp = telemetry_session.value["session_start_timestamp"] 

59 

60 logger.debug(f"Telemetry Session: {session_id}, {session_start_timestamp}") 

61 return (session_start_timestamp, session_id) 

62 

63 

64@perpetual_service( 

65 enabled_getter=lambda: get_current_settings().server.analytics_enabled, 

66 run_in_ephemeral=True, 

67 run_in_webserver=True, 

68) 

69async def send_telemetry_heartbeat( 

70 perpetual: Perpetual = Perpetual(automatic=True, every=timedelta(seconds=600)), 

71) -> None: 

72 """ 

73 Sends anonymous telemetry data to Prefect to help us improve. 

74 

75 It can be toggled off with the PREFECT_SERVER_ANALYTICS_ENABLED setting. 

76 """ 

77 from prefect.client.constants import SERVER_API_VERSION 

78 

79 db = provide_database_interface() 

80 session_start_timestamp, session_id = await _fetch_or_set_telemetry_session(db=db) 

81 telemetry_environment = os.environ.get( 

82 "PREFECT_API_TELEMETRY_ENVIRONMENT", "production" 

83 ) 

84 

85 heartbeat = { 

86 "source": "prefect_server", 

87 "type": "heartbeat", 

88 "payload": { 

89 "platform": platform.system(), 

90 "architecture": platform.machine(), 

91 "python_version": platform.python_version(), 

92 "python_implementation": platform.python_implementation(), 

93 "environment": telemetry_environment, 

94 "ephemeral_server": bool(os.getenv("PREFECT__SERVER_EPHEMERAL", False)), 

95 "api_version": SERVER_API_VERSION, 

96 "prefect_version": prefect.__version__, 

97 "session_id": session_id, 

98 "session_start_timestamp": session_start_timestamp, 

99 }, 

100 } 

101 

102 try: 

103 async with httpx.AsyncClient() as client: 

104 result = await client.post( 

105 "https://sens-o-matic.prefect.io/", 

106 json=heartbeat, 

107 headers={"x-prefect-event": "prefect_server"}, 

108 ) 

109 result.raise_for_status() 

110 except Exception as exc: 

111 logger.error( 

112 f"Failed to send telemetry: {exc}", 

113 exc_info=PREFECT_DEBUG_MODE.value(), 

114 )