Coverage for .tox/coverage/lib/python3.11/site-packages/wuttatell/telemetry.py: 100%
97 statements
« prev ^ index » next coverage.py v7.10.2, created at 2025-08-31 18:58 -0500
1# -*- coding: utf-8; -*-
2################################################################################
3#
4# WuttaTell -- Telemetry submission for Wutta Framework
5# Copyright © 2025 Lance Edgar
6#
7# This file is part of Wutta Framework.
8#
9# Wutta Framework is free software: you can redistribute it and/or modify it
10# under the terms of the GNU General Public License as published by the Free
11# Software Foundation, either version 3 of the License, or (at your option) any
12# later version.
13#
14# Wutta Framework is distributed in the hope that it will be useful, but
15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
17# more details.
18#
19# You should have received a copy of the GNU General Public License along with
20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.
21#
22################################################################################
23"""
24Telemetry submission handler
25"""
27import os
28import re
29import subprocess
31from wuttjamaican.app import GenericHandler
32from wuttjamaican.conf import WuttaConfigProfile
34from wuttatell.client import SimpleAPIClient
37class TelemetryHandler(GenericHandler):
38 """
39 Handler for submission of telemetry data
41 The primary caller interface involves just two methods:
43 * :meth:`collect_all_data()`
44 * :meth:`submit_all_data()`
45 """
47 def get_profile(self, profile): # pylint: disable=empty-docstring
48 """ """
49 if isinstance(profile, TelemetryProfile):
50 return profile
52 return TelemetryProfile(self.config, profile or "default")
54 def collect_all_data(self, profile=None):
55 """
56 Collect and return all data pertaining to the given profile.
58 The profile will determine which types of data to collect,
59 e.g. ``('os', 'python')``. Corresponding handler methods
60 are then called to collect each type; for instance:
62 * :meth:`collect_data_os()`
63 * :meth:`collect_data_python()`
65 Once all data has been collected, errors are grouped to the
66 top level of the structure.
68 :param profile: :class:`TelemetryProfile` instance, or key
69 thereof. If not specified, ``'default'`` is assumed.
71 :returns: A dict of data, keyed by collection type. If any
72 errors were encountered during collection, the dict will
73 also have an ``'errors'`` key.
74 """
75 data = {}
76 profile = self.get_profile(profile)
78 for key in profile.collect_keys:
79 collector = getattr(self, f"collect_data_{key}")
80 data[key] = collector(profile=profile)
82 self.normalize_errors(data)
83 return data
85 def normalize_errors(self, data): # pylint: disable=empty-docstring
86 """ """
87 all_errors = []
88 for value in data.values():
89 if value:
90 errors = value.pop("errors", None)
91 if errors:
92 all_errors.extend(errors)
93 if all_errors:
94 data["errors"] = all_errors
96 def collect_data_os(self, profile, **kwargs): # pylint: disable=unused-argument
97 """
98 Collect basic data about the operating system.
100 This parses ``/etc/os-release`` for basic OS info, and
101 ``/etc/timezone`` for the timezone.
103 If all goes well the result looks like::
105 {
106 "release_id": "debian",
107 "release_version": "12",
108 "release_full": "Debian GNU/Linux 12 (bookworm)",
109 "timezone": "America/Chicago",
110 }
112 :param profile: :class:`TelemetryProfile` instance. Note that
113 the default logic here ignores the profile.
115 :returns: Data dict similar to the above. May have an
116 ``'errors'`` key if anything goes wrong.
117 """
118 data = {}
119 errors = []
121 # release
122 release_path = kwargs.get("release_path", "/etc/os-release")
123 try:
124 with open(release_path, "rt", encoding="utf_8") as f:
125 output = f.read()
126 except Exception: # pylint: disable=broad-exception-caught
127 errors.append(f"Failed to read {release_path}")
128 else:
129 release = {}
130 pattern = re.compile(r"^([^=]+)=(.*)$")
131 for line in output.strip().split("\n"):
132 if match := pattern.match(line):
133 key, val = match.groups()
134 if val.startswith('"') and val.endswith('"'):
135 val = val.strip('"')
136 release[key] = val
137 try:
138 data["release_id"] = release["ID"]
139 data["release_version"] = release["VERSION_ID"]
140 data["release_full"] = release["PRETTY_NAME"]
141 except KeyError:
142 errors.append(f"Failed to parse {release_path}")
144 # timezone
145 timezone_path = kwargs.get("timezone_path", "/etc/timezone")
146 try:
147 with open(timezone_path, "rt", encoding="utf_8") as f:
148 output = f.read()
149 except Exception: # pylint: disable=broad-exception-caught
150 errors.append(f"Failed to read {timezone_path}")
151 else:
152 data["timezone"] = output.strip()
154 if errors:
155 data["errors"] = errors
156 return data
158 def collect_data_python(self, profile):
159 """
160 Collect basic data about the Python environment.
162 This primarily runs ``python --version`` for the desired
163 environment. Note that the profile will determine which
164 environment to inspect, e.g. system-wide or a specific virtual
165 environment.
167 If all goes well the system-wide result looks like::
169 {
170 "executable": "/usr/bin/python3",
171 "release_full": "Python 3.11.2",
172 "release_version": "3.11.2",
173 }
175 If a virtual environment is involved the result will include
176 its root path::
178 {
179 "envroot": "/srv/envs/poser",
180 "executable": "/srv/envs/poser/bin/python",
181 "release_full": "Python 3.11.2",
182 "release_version": "3.11.2",
183 }
185 :param profile: :class:`TelemetryProfile` instance.
187 :returns: Data dict similar to the above. May have an
188 ``'errors'`` key if anything goes wrong.
189 """
190 data = {}
191 errors = []
193 # envroot determines python executable
194 envroot = profile.get_str("collect.python.envroot")
195 if envroot:
196 data["envroot"] = envroot
197 python = os.path.join(envroot, "bin/python")
198 else:
199 python = profile.get_str(
200 "collect.python.executable", default="/usr/bin/python3"
201 )
203 # python version
204 data["executable"] = python
205 try:
206 output = subprocess.check_output([python, "--version"])
207 except (subprocess.CalledProcessError, FileNotFoundError) as err:
208 errors.append("Failed to execute `python --version`")
209 errors.append(str(err))
210 else:
211 output = output.decode("utf_8").strip()
212 data["release_full"] = output
213 if match := re.match(r"^Python (\d+\.\d+\.\d+)", output):
214 data["release_version"] = match.group(1)
215 else:
216 errors.append("Failed to parse Python version")
218 if errors:
219 data["errors"] = errors
220 return data
222 def submit_all_data(self, profile=None, data=None):
223 """
224 Submit telemetry data to the configured collection service.
226 Default logic will use
227 :class:`~wuttatell.client.SimpleAPIClient` and submit all
228 collected data to the configured API endpoint.
230 :param profile: :class:`TelemetryProfile` instance.
232 :param data: Data dict as obtained by
233 :meth:`collect_all_data()`.
234 """
235 profile = self.get_profile(profile)
236 if data is None:
237 data = self.collect_all_data(profile)
239 client = SimpleAPIClient(self.config)
240 client.post(profile.submit_url, data=data)
243class TelemetryProfile(WuttaConfigProfile):
244 """
245 Represents a configured profile for telemetry submission.
247 This is a subclass of
248 :class:`~wuttjamaican:wuttjamaican.conf.WuttaConfigProfile`, and
249 similarly works off the
250 :attr:`~wuttjamaican:wuttjamaican.conf.WuttaConfigProfile.key` to
251 identify each configured profile.
253 Upon construction each profile instance will have the following
254 attributes, determined by config:
256 .. attribute:: collect_keys
258 List of keys identifying the types of data to collect,
259 e.g. ``['os', 'python']``.
261 .. attribute:: submit_url
263 URL to which collected telemetry data should be submitted.
264 """
266 @property
267 def section(self): # pylint: disable=empty-docstring
268 """ """
269 return f"{self.config.appname}.telemetry"
271 def load(self): # pylint: disable=empty-docstring
272 """ """
273 keys = self.get_str("collect.keys", default="os,python")
274 self.collect_keys = self.config.parse_list(keys)
275 self.submit_url = self.get_str("submit.url")