Coverage for .tox/coverage/lib/python3.13/site-packages/wuttatell/telemetry.py: 100%
163 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-21 19:19 -0500
1# -*- coding: utf-8; -*-
2################################################################################
3#
4# WuttaTell -- Telemetry submission for Wutta Framework
5# Copyright © 2025-2026 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 logging
28import os
29import re
30import shutil
31import subprocess
33from wuttjamaican.app import GenericHandler
34from wuttjamaican.conf import WuttaConfigProfile
36from wuttatell.client import SimpleAPIClient
38log = logging.getLogger(__name__)
41class TelemetryHandler(GenericHandler):
42 """
43 Handler for submission of telemetry data
45 The primary caller interface involves just two methods:
47 * :meth:`collect_all_data()`
48 * :meth:`submit_all_data()`
49 """
51 def get_all_profile_keys(self):
52 """
53 Retrieve list of all configured profile keys.
55 This will iterate over defined config values **from file(s)
56 only** (will not read DB settings), and parse each option name
57 to determine the set of unique profile names.
59 :returns: Sorted list of profile keys. If ``default`` is
60 included, it will always be first in the list.
61 """
62 keys = set()
64 defined = self.config.get_dict(f"{self.config.appname}.telemetry")
65 for key in defined:
66 if key.startswith("_"):
67 continue
68 if "." not in key:
69 continue
70 key = key.split(".")[0]
71 keys.add(key)
73 keys = sorted(keys)
74 if "default" in keys and keys[0] != "default":
75 keys.remove("default")
76 keys.insert(0, "default")
78 return keys
80 def get_profile(self, profile): # pylint: disable=empty-docstring
81 """ """
82 if isinstance(profile, TelemetryProfile):
83 return profile
85 return TelemetryProfile(self.config, profile or "default")
87 def collect_all_data(self, profile=None):
88 """
89 Collect and return all data pertaining to the given profile.
91 The profile will determine which types of data to collect,
92 e.g. ``('os', 'python')``. Corresponding handler methods
93 are then called to collect each type; for instance:
95 * :meth:`collect_data_os()`
96 * :meth:`collect_data_python()`
98 Once all data has been collected, errors are grouped to the
99 top level of the structure.
101 :param profile: :class:`TelemetryProfile` instance, or key
102 thereof. If not specified, ``'default'`` is assumed.
104 :returns: A dict of data, keyed by collection type. If any
105 errors were encountered during collection, the dict will
106 also have an ``'errors'`` key.
107 """
108 data = {}
109 profile = self.get_profile(profile)
111 log.debug(
112 "collecting data for '%s' profile: %s", profile.key, profile.collect_keys
113 )
114 for key in profile.collect_keys:
115 collector = getattr(self, f"collect_data_{key}")
116 data[key] = collector(profile=profile)
118 self.normalize_errors(data)
119 log.debug(data)
120 return data
122 def normalize_errors(self, data): # pylint: disable=empty-docstring
123 """ """
124 all_errors = []
125 for value in data.values():
126 if value:
127 errors = value.pop("errors", None)
128 if errors:
129 all_errors.extend(errors)
130 if all_errors:
131 data["errors"] = all_errors
133 def collect_data_os(self, profile, **kwargs): # pylint: disable=unused-argument
134 """
135 Collect basic data about the operating system.
137 This parses ``/etc/os-release`` for basic OS info, and
138 ``/etc/timezone`` for the timezone.
140 If all goes well the result looks like::
142 {
143 "release_id": "debian",
144 "release_version": "12",
145 "release_full": "Debian GNU/Linux 12 (bookworm)",
146 "timezone": "America/Chicago",
147 }
149 :param profile: :class:`TelemetryProfile` instance. Note that
150 the default logic here ignores the profile.
152 :returns: Data dict similar to the above. May have an
153 ``'errors'`` key if anything goes wrong.
154 """
155 data = {}
156 errors = []
158 # release
159 release_path = kwargs.get("release_path", "/etc/os-release")
160 try:
161 with open(release_path, "rt", encoding="utf_8") as f:
162 output = f.read()
163 except Exception: # pylint: disable=broad-exception-caught
164 errors.append(f"Failed to read {release_path}")
165 else:
166 release = {}
167 pattern = re.compile(r"^([^=]+)=(.*)$")
168 for line in output.strip().split("\n"):
169 if match := pattern.match(line):
170 key, val = match.groups()
171 if val.startswith('"') and val.endswith('"'):
172 val = val.strip('"')
173 release[key] = val
174 try:
175 data["release_id"] = release["ID"]
176 data["release_version"] = release["VERSION_ID"]
177 data["release_full"] = release["PRETTY_NAME"]
178 except KeyError:
179 errors.append(f"Failed to parse {release_path}")
181 # timezone
182 timezone_path = kwargs.get("timezone_path", "/etc/timezone")
183 try:
184 with open(timezone_path, "rt", encoding="utf_8") as f:
185 output = f.read()
186 except Exception: # pylint: disable=broad-exception-caught
187 errors.append(f"Failed to read {timezone_path}")
188 else:
189 data["timezone"] = output.strip()
191 if errors:
192 data["errors"] = errors
193 return data
195 def collect_data_python(self, profile):
196 """
197 Collect basic data about the Python environment.
199 This primarily runs ``python --version`` for the desired
200 environment. Note that the profile will determine which
201 environment to inspect, e.g. system-wide or a specific virtual
202 environment.
204 If all goes well the system-wide result looks like::
206 {
207 "executable": "/usr/bin/python3",
208 "release_full": "Python 3.11.2",
209 "release_version": "3.11.2",
210 }
212 If a virtual environment is involved the result will include
213 its root path::
215 {
216 "envroot": "/srv/envs/poser",
217 "executable": "/srv/envs/poser/bin/python",
218 "release_full": "Python 3.11.2",
219 "release_version": "3.11.2",
220 }
222 :param profile: :class:`TelemetryProfile` instance.
224 :returns: Data dict similar to the above. May have an
225 ``'errors'`` key if anything goes wrong.
226 """
227 data = {}
228 errors = []
230 # envroot determines python executable
231 envroot = profile.get_str("collect.python.envroot")
232 if envroot:
233 data["envroot"] = envroot
234 python = os.path.join(envroot, "bin/python")
235 else:
236 python = profile.get_str(
237 "collect.python.executable", default="/usr/bin/python3"
238 )
240 # python version
241 data["executable"] = python
242 try:
243 # nb. must capture stderr also, for sake of python 2.7
244 output = subprocess.check_output(
245 [python, "--version"], stderr=subprocess.STDOUT
246 )
247 except (subprocess.CalledProcessError, FileNotFoundError) as err:
248 errors.append("Failed to execute `python --version`")
249 errors.append(str(err))
250 else:
251 output = output.decode("utf_8").strip()
252 data["release_full"] = output
253 if match := re.match(r"^Python (\d+\.\d+\.\d+)", output):
254 data["release_version"] = match.group(1)
255 else:
256 errors.append("Failed to parse Python version")
258 if errors:
259 data["errors"] = errors
260 return data
262 def collect_data_black(self, profile):
263 """
264 Collect basic data about the `black`_ command, if present.
266 .. _black: https://black.readthedocs.io/en/stable/
268 The ``black`` executable is assumed to be found in ``PATH``; if
269 so then ``black --version`` is called to get the result::
271 {
272 "executable": "/usr/local/bin/black",
273 "release_full": ("black, 25.1.0 (compiled: yes)\\n"
274 "Python (CPython) 3.13.5"),
275 "release_version": "25.1.0",
276 }
278 :param profile: :class:`TelemetryProfile` instance.
280 :returns: Data dict similar to the above. May have an
281 ``'errors'`` key if anything goes wrong.
282 """
283 data = {}
284 errors = []
286 black_path = profile.get_str("collect.black.executable")
287 if not black_path:
288 black_path = shutil.which("black")
289 if not black_path:
290 errors.append("Failed to locate black executable")
292 data["executable"] = black_path
293 if black_path:
294 try:
295 output = subprocess.check_output([black_path, "--version"])
296 except (subprocess.CalledProcessError, FileNotFoundError) as err:
297 errors.append("Failed to execute `black --version`")
298 errors.append(str(err))
299 else:
300 output = output.decode("utf_8").strip()
301 lines = output.split("\n")
302 if match := re.match(r"^black\, (\d+\.\d+\.\d+) ", lines[0]):
303 data["release_full"] = output
304 data["release_version"] = match.group(1)
305 else:
306 errors.append(f"Failed to parse Black version: {output}")
308 if errors:
309 data["errors"] = errors
310 return data
312 def collect_data_borg(self, profile):
313 """
314 Collect basic data about the Borg backup command, if present.
316 The ``borg`` executable is assumed to be found in ``PATH``; if
317 so then ``borg --version`` is called to get the result::
319 {
320 "executable": "/usr/local/bin/borg",
321 "release_version": "1.4.5",
322 }
324 :param profile: :class:`TelemetryProfile` instance.
326 :returns: Data dict similar to the above. May have an
327 ``'errors'`` key if anything goes wrong.
328 """
329 data = {}
330 errors = []
332 borg_path = profile.get_str("collect.borg.executable")
333 if not borg_path:
334 borg_path = shutil.which("borg")
335 if not borg_path:
336 errors.append("Failed to locate borg executable")
338 data["executable"] = borg_path
339 if borg_path:
340 try:
341 output = subprocess.check_output([borg_path, "--version"])
342 except (subprocess.CalledProcessError, FileNotFoundError) as err:
343 errors.append("Failed to execute `borg --version`")
344 errors.append(str(err))
345 else:
346 output = output.decode("utf_8").strip()
347 if match := re.match(r"^borg (\d+\.\d+\.\d+)", output):
348 data["release_version"] = match.group(1)
349 else:
350 errors.append(f"Failed to parse Borg version: {output}")
352 if errors:
353 data["errors"] = errors
354 return data
356 def submit_all_data(self, profile=None, data=None):
357 """
358 Submit telemetry data to the configured collection service.
360 Default logic will use
361 :class:`~wuttatell.client.SimpleAPIClient` and submit all
362 collected data to the configured API endpoint.
364 :param profile: :class:`TelemetryProfile` instance.
366 :param data: Data dict as obtained by
367 :meth:`collect_all_data()`.
368 """
369 profile = self.get_profile(profile)
370 if data is None:
371 data = self.collect_all_data(profile)
373 client = SimpleAPIClient(self.config)
374 client.post(profile.submit_url, data=data)
377class TelemetryProfile(WuttaConfigProfile):
378 """
379 Represents a configured profile for telemetry submission.
381 This is a subclass of
382 :class:`~wuttjamaican:wuttjamaican.conf.WuttaConfigProfile`, and
383 similarly works off the
384 :attr:`~wuttjamaican:wuttjamaican.conf.WuttaConfigProfile.key` to
385 identify each configured profile.
387 Upon construction each profile instance will have the following
388 attributes, determined by config:
390 .. attribute:: collect_keys
392 List of keys identifying the types of data to collect,
393 e.g. ``["os", "python", "borg"]``.
395 .. attribute:: submit_url
397 URL to which collected telemetry data should be submitted.
398 """
400 @property
401 def section(self): # pylint: disable=empty-docstring
402 """ """
403 return f"{self.config.appname}.telemetry"
405 def load(self): # pylint: disable=empty-docstring
406 """ """
407 keys = self.get_str("collect.keys", default="os,python,black,borg")
408 self.collect_keys = self.config.parse_list(keys)
409 self.submit_url = self.get_str("submit.url")