Coverage for haystack/telemetry/_telemetry.py: 86%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import datetime 

6import functools 

7import logging 

8import os 

9import uuid 

10from collections import defaultdict 

11from collections.abc import Callable 

12from pathlib import Path 

13from typing import TYPE_CHECKING, Any 

14 

15import posthog 

16import yaml 

17 

18from haystack import logging as haystack_logging 

19from haystack.core.serialization import generate_qualified_class_name 

20from haystack.telemetry._environment import collect_system_specs 

21 

22if TYPE_CHECKING: 

23 from haystack.core.pipeline import Pipeline 

24 

25 

26HAYSTACK_TELEMETRY_ENABLED = "HAYSTACK_TELEMETRY_ENABLED" 

27CONFIG_PATH = Path("~/.haystack/config.yaml").expanduser() 

28 

29#: Telemetry sends at most one event every number of seconds specified in this constant 

30MIN_SECONDS_BETWEEN_EVENTS = 60 

31 

32 

33logger = haystack_logging.getLogger(__name__) 

34 

35 

36class Telemetry: 

37 """ 

38 Haystack reports anonymous usage statistics to support continuous software improvements for all its users. 

39 

40 You can opt-out of sharing usage statistics by manually setting the environment 

41 variable `HAYSTACK_TELEMETRY_ENABLED` as described for different operating systems on the 

42 [documentation page](https://docs.haystack.deepset.ai/docs/telemetry#how-can-i-opt-out). 

43 

44 Check out the documentation for more details: [Telemetry](https://docs.haystack.deepset.ai/docs/telemetry). 

45 """ 

46 

47 def __init__(self) -> None: 

48 """ 

49 Initializes the telemetry. 

50 

51 Loads the user_id from the config file, or creates a new id and saves it if the file is not found. 

52 

53 It also collects system information which cannot change across the lifecycle 

54 of the process (for example `is_containerized()`). 

55 """ 

56 posthog.api_key = "phc_C44vUK9R1J6HYVdfJarTEPqVAoRPJzMXzFcj8PIrJgP" 

57 posthog.host = "https://eu.posthog.com" 

58 

59 # disable posthog logging 

60 for module_name in ["posthog", "backoff"]: 

61 logging.getLogger(module_name).setLevel(logging.CRITICAL) 

62 # Prevent module from sending errors to stderr when an exception is encountered during an emit() call 

63 logging.getLogger(module_name).addHandler(logging.NullHandler()) 

64 logging.getLogger(module_name).propagate = False 

65 

66 self.user_id = "" 

67 

68 if CONFIG_PATH.exists(): 

69 # Load the config file 

70 try: 

71 with open(CONFIG_PATH, encoding="utf-8") as config_file: 

72 config = yaml.safe_load(config_file) 

73 if "user_id" in config: 

74 self.user_id = config["user_id"] 

75 except Exception as e: 

76 logger.debug( 

77 "Telemetry could not read the config file {config_path}", config_path=CONFIG_PATH, exc_info=e 

78 ) 

79 else: 

80 # Create the config file 

81 logger.info( 

82 "Haystack sends anonymous usage data to understand the actual usage and steer dev efforts " 

83 "towards features that are most meaningful to users. You can opt-out at anytime by manually " 

84 "setting the environment variable HAYSTACK_TELEMETRY_ENABLED as described for different " 

85 "operating systems in the " 

86 "[documentation page](https://docs.haystack.deepset.ai/docs/telemetry#how-can-i-opt-out). " 

87 "More information at [Telemetry](https://docs.haystack.deepset.ai/docs/telemetry)." 

88 ) 

89 CONFIG_PATH.parents[0].mkdir(parents=True, exist_ok=True) 

90 self.user_id = str(uuid.uuid4()) 

91 try: 

92 with open(CONFIG_PATH, "w") as outfile: 

93 yaml.dump({"user_id": self.user_id}, outfile, default_flow_style=False) 

94 except Exception as e: 

95 logger.debug( 

96 "Telemetry could not write config file to {config_path}", config_path=CONFIG_PATH, exc_info=e 

97 ) 

98 

99 self.event_properties = collect_system_specs() 

100 

101 def send_event(self, event_name: str, event_properties: dict[str, Any] | None = None) -> None: 

102 """ 

103 Sends a telemetry event. 

104 

105 :param event_name: The name of the event to show in PostHog. 

106 :param event_properties: Additional event metadata. These are merged with the 

107 system metadata collected in __init__, so take care not to overwrite them. 

108 """ 

109 event_properties = event_properties or {} 

110 try: 

111 posthog.capture( 

112 distinct_id=self.user_id, event=event_name, properties={**self.event_properties, **event_properties} 

113 ) 

114 except Exception as e: 

115 logger.debug("Telemetry couldn't make a POST request to PostHog.", exc_info=e) 

116 

117 

118def send_telemetry(func: Callable[..., Any]) -> Callable[..., None]: 

119 """ 

120 Decorator that sends the output of the wrapped function to PostHog. 

121 

122 The wrapped function is actually called only if telemetry is enabled. 

123 """ 

124 

125 @functools.wraps(func) 

126 def send_telemetry_wrapper(*args: Any, **kwargs: Any) -> None: 

127 try: 

128 if telemetry: 

129 output = func(*args, **kwargs) 

130 if output: 

131 telemetry.send_event(*output) 

132 except Exception as e: 

133 # Never let telemetry break things 

134 logger.debug("There was an issue sending a telemetry event", exc_info=e) 

135 

136 return send_telemetry_wrapper 

137 

138 

139@send_telemetry 

140def pipeline_running(pipeline: "Pipeline") -> tuple[str, dict[str, Any]] | None: 

141 """ 

142 Collects telemetry data for a pipeline run and sends it to Posthog. 

143 

144 Collects name, type and the content of the _telemetry_data attribute, if present, for each component in the 

145 pipeline and sends such data to Posthog. 

146 

147 :param pipeline: the pipeline that is running. 

148 """ 

149 pipeline._telemetry_runs += 1 

150 if ( 

151 pipeline._last_telemetry_sent 

152 and (datetime.datetime.now() - pipeline._last_telemetry_sent).total_seconds() < MIN_SECONDS_BETWEEN_EVENTS 

153 ): 

154 return None 

155 

156 pipeline._last_telemetry_sent = datetime.datetime.now() 

157 

158 # Collect info about components 

159 components: dict[str, list[dict[str, Any]]] = defaultdict(list) 

160 for component_name, instance in pipeline.walk(): 

161 component_qualified_class_name = generate_qualified_class_name(type(instance)) 

162 if hasattr(instance, "_get_telemetry_data"): 

163 telemetry_data = instance._get_telemetry_data() 

164 if not isinstance(telemetry_data, dict): 

165 raise TypeError( 

166 f"Telemetry data for component {component_name} must be a dictionary but is {type(telemetry_data)}." 

167 ) 

168 components[component_qualified_class_name].append({"name": component_name, **telemetry_data}) 

169 else: 

170 components[component_qualified_class_name].append({"name": component_name}) 

171 

172 # Data sent to Posthog 

173 return "Pipeline run (3.x)", { 

174 "pipeline_id": str(id(pipeline)), 

175 "pipeline_type": generate_qualified_class_name(type(pipeline)), 

176 "runs": pipeline._telemetry_runs, 

177 "components": components, 

178 } 

179 

180 

181@send_telemetry 

182def tutorial_running(tutorial_id: str) -> tuple[str, dict[str, Any]]: 

183 """ 

184 Send a telemetry event for a tutorial, if telemetry is enabled. 

185 

186 :param tutorial_id: identifier of the tutorial 

187 """ 

188 return "Tutorial", {"tutorial.id": tutorial_id} 

189 

190 

191telemetry = None 

192if os.getenv("HAYSTACK_TELEMETRY_ENABLED", "true").lower() in ("true", "1"): 

193 telemetry = Telemetry()