Add telemetry for RHEL

Resolves: RHEL-219532

Signed-off-by: Dimitri Savineau <dsavinea@redhat.com>

ansible-core.spec

@@ -38,7 +38,7 @@
 Name: ansible-core
 Summary: SSH-based configuration management, deployment, and task execution system
 Version: 2.16.3
-Release: 2%{?dist}
+Release: 3%{?dist}
 ExcludeArch: i686
 
 Group: Development/Libraries
@@ -59,6 +59,11 @@ Source8: https://sourceforge.net/projects/docutils/files/docutils/%{docutils_ver
 
 Patch0: remove-bundled-deps-from-requirements.patch
 
+%if 0%{!?centos:1} && 0%{?rhel}
+Source99: telemetry.py
+Patch99: telemetry.patch
+%endif
+
 URL: http://ansible.com
 
 # We conflict old ansible, and any version of ansible-base.
@@ -123,6 +128,10 @@ developed for ansible.
 # Fix all Python shebangs recursively in ansible-test
 %{py3_shebang_fix} test/lib/ansible_test
 
+%if 0%{!?centos:1} && 0%{?rhel}
+%patch -P99 -p1
+%endif
+
 %build
 %{py3_build}
 
@@ -177,6 +186,12 @@ mkdir -p %{buildroot}%{_sysconfdir}/ansible/roles/
 cp ../ansible-documentation-%{version}/examples/hosts %{buildroot}%{_sysconfdir}/ansible/
 cp ../ansible-documentation-%{version}/examples/ansible.cfg %{buildroot}%{_sysconfdir}/ansible/
 
+%if 0%{!?centos:1} && 0%{?rhel}
+mkdir -p %{buildroot}%{_datadir}/ansible/telemetry
+cp %{SOURCE99} %{buildroot}%{_datadir}/ansible/telemetry/
+%py_byte_compile %{__python3} %{buildroot}%{_datadir}/ansible/telemetry/telemetry.py
+%endif
+
 mkdir -p %{buildroot}/%{_mandir}/man1/
 
 mkdir -p docs/man/man1
@@ -210,6 +225,9 @@ strip --strip-unneeded %{vendor_path}/markupsafe/_speedups%{python3_ext_suffix}
 %{python3_sitelib}/ansible_test
 
 %changelog
+* Wed Jul 29 2026 Dimitri Savineau <dsavinea@redhat.com> - 2.16.3-3
+- Add telemetry for RHEL (RHEL-219532)
+
 * Mon Feb 05 2024 Dimitri Savineau <dsavinea@redhat.com> - 2.16.3-2
 - rebuild with python 3.12 (RHEL-24141)
 

telemetry.patch

@@ -0,0 +1,13 @@
+diff --git a/lib/ansible/plugins/loader.py b/lib/ansible/plugins/loader.py
+index f3252d7142..1e4d3bca38 100644
+--- a/lib/ansible/plugins/loader.py
++++ b/lib/ansible/plugins/loader.py
+@@ -1532,7 +1532,7 @@ cache_loader = PluginLoader(
+ callback_loader = PluginLoader(
+     'CallbackModule',
+     'ansible.plugins.callback',
+-    C.DEFAULT_CALLBACK_PLUGIN_PATH,
++    C.DEFAULT_CALLBACK_PLUGIN_PATH + ['/usr/share/ansible/telemetry'],
+     'callback_plugins',
+ )
+ 

telemetry.py

@@ -0,0 +1,302 @@
+#! /usr/libexec/platform-python
+# Copyright: Contributors to the Ansible project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = '''
+    name: telemetry
+    short_description: Track various telemetry related data and store in journald
+    description:
+        - Track various telemetry related data and store in journald
+    type: aggregate
+    options:
+        bypass_insights_check:
+            name: Bypass Insights Check
+            description: Bypass check to see if insights-client is active
+            default: false
+            type: bool
+            env:
+                - name: TELEMETRY_BYPASS_INSIGHTS_CHECK
+    notes:
+        - The file can also be executed directly from the CLI to retrieve
+          the stored telemetry data from journald
+'''
+
+import contextlib
+import ctypes
+import datetime
+import functools
+import importlib.util
+import json
+import os
+import pathlib
+import subprocess
+import sys
+import uuid
+from collections import Counter, defaultdict
+
+if __name__ == '__main__':
+    # When run as a script via insights-client, these are not needed
+    CallbackBase = object
+    Display = None
+else:
+    from ansible.executor.task_result import TaskResult
+    from ansible.module_utils.common.yaml import yaml_load
+    from ansible.plugins.callback import CallbackBase
+    from ansible.plugins.loader import connection_loader
+    from ansible.release import __version__ as ansible_version
+    from ansible.utils.display import Display
+
+journal_sendv = None
+try:
+    from systemd import journal
+    journal_sendv = journal.sendv
+except ImportError:
+    # When invoked using the non-system python
+    # the systemd python bindings are not available
+    with contextlib.suppress(AttributeError, OSError):
+        _libsysetmd = ctypes.CDLL('libsystemd.so.0')
+        sd_journal_send = _libsysetmd.sd_journal_send
+
+        def journal_sendv(*args):
+            b_args = [ctypes.c_char_p(a.encode()) for a in args] + [None]
+            sd_journal_send(*b_args)
+
+if sys.version_info >= (3, 10):
+    from importlib.resources import files
+else:
+    # importlib.resources below Python 3.10 doesn't support
+    # custom loaders, this is not a full featured replacement
+    # but suffices for the needs of this plugin
+    def files(name):
+        spec = importlib.util.find_spec(name)
+        if spec is None:
+            raise ImportError(name)
+        origin = pathlib.Path(spec.origin)
+        if origin.name == '__synthetic__':
+            origin = origin.parent
+        return origin
+
+if Display:
+    display = Display()
+
+_CORE_SYN_COLLECTIONS = frozenset((
+    'ansible.builtin',
+    'ansible.legacy',
+))
+
+_CORE_FACT_MODULES = frozenset((
+    'setup',
+    'ansible.builtin.setup',
+    'ansible.legacy.setup',
+    'gather_facts',
+    'ansible.builtin.gather_facts',
+    'ansible.legacy.gather_facts',
+))
+
+_ANSIBLE_TELEMETRY_MESSAGE_ID = uuid.UUID('73df9e03d8c6473eb811065c4ebc370a')
+
+
+@functools.lru_cache(maxsize=None)
+def _get_connection_fqcn(task, host):
+    connection = host.vars.get('ansible_connection')
+    if not connection:
+        connection = task.connection
+    is_connection_fqcn = '.' in connection
+    if is_connection_fqcn:
+        return connection
+
+    return connection_loader.get_with_context(
+        connection,
+        class_only=True,
+        collection_list=task.collections,
+    ).plugin_load_context.resolved_fqcn
+
+
+class CallbackModule(CallbackBase):
+    CALLBACK_VERSION = 2.0
+    CALLBACK_TYPE = 'aggregate'
+    CALLBACK_NAME = 'telemetry'
+
+    def __init__(self, *args, **kwargs):
+        self.wants_implicit_tasks = False
+        self.disabled = False
+
+        self._telemetry = {
+            'collections': defaultdict(
+                lambda: {
+                    'resources': defaultdict(
+                        lambda: defaultdict(lambda: {
+                            'count': 0,
+                        })
+                    ),
+                    'version': '*',
+                },
+            ),
+        }
+        self._hosts = set()
+        self._os = {}
+        self._conn_cache = {}
+
+    def _is_insights_client_active(self):
+        if self.get_option('bypass_insights_check'):
+            return True
+
+        if not os.path.exists('/etc/insights-client/.registered'):
+            return False
+
+        with contextlib.suppress(subprocess.CalledProcessError):
+            subprocess.run(
+                ['systemctl', 'is-active', '--quiet', 'insights-client.timer'],
+                capture_output=True,
+                check=True,
+            )
+            return True
+
+        return False
+
+    def _set_disabled(self):
+        if not self._is_insights_client_active():
+            return True
+
+        if journal_sendv is None:
+            return True
+
+        return False
+
+    def set_options(self, task_keys=None, var_options=None, direct=None):
+        super().set_options(task_keys=task_keys, var_options=var_options, direct=direct)
+        self.disabled = self._set_disabled()
+
+    def _track_os(self, host_name, result):
+        if host_name in self._os:
+            return
+
+        try:
+            af = result.get('ansible_facts')
+        except AttributeError:
+            return
+
+        if not af:
+            return
+
+        try:
+            os_name = '%(ansible_distribution)s-%(ansible_distribution_version)s' % af
+        except KeyError:
+            return
+        else:
+            self._os[host_name] = os_name
+
+    def _handle_result(self, *args, **kwargs):
+        if not args:
+            return
+
+        result = args[0]
+        if not isinstance(result, TaskResult):
+            return
+
+        host = result._host
+        host_name = host.get_name()
+        task = result._task
+        role = task._role
+
+        self._hosts.add(host_name)
+
+        action = task.resolved_action
+        action_collection = None
+        if action and '.' in action:
+            action_collection = '.'.join(action.split('.', 2)[:2])
+
+        if action in _CORE_FACT_MODULES:
+            self._track_os(host_name, result._result)
+
+        role_collection = None
+        role_name = None if not role else role.get_name()
+        if role and '.' in role_name:
+            role_collection = '.'.join(role_name.split('.', 2)[:2])
+
+        connection_collection = None
+        connection = _get_connection_fqcn(task, host)
+        if '.' in connection:
+            connection_collection = '.'.join(connection.split('.')[:2])
+
+        collections = self._telemetry['collections']
+        for resource, collection, name in (('action', action_collection, action),
+                                           ('role', role_collection, role_name),
+                                           ('connection', connection_collection, connection)):
+            if not collection:
+                continue
+
+            collections[collection]['resources'][resource][name]['count'] += 1
+
+    # v2_on_file_diff = _handle_result
+    v2_runner_item_on_failed = _handle_result
+    v2_runner_item_on_ok = _handle_result
+    v2_runner_item_on_skipped = _handle_result
+    v2_runner_on_async_failed = _handle_result
+    v2_runner_on_async_ok = _handle_result
+    v2_runner_on_async_poll = _handle_result
+    v2_runner_on_failed = _handle_result
+    v2_runner_on_ok = _handle_result
+    v2_runner_on_skipped = _handle_result
+    v2_runner_on_unreachable = _handle_result
+    v2_runner_retry = _handle_result
+
+    def v2_playbook_on_stats(self, stats):
+        self._telemetry['ansible_core'] = {
+            'version': ansible_version,
+        }
+        self._telemetry['hosts'] = {
+            'count': len(self._hosts),
+            'os': Counter(self._os.values()),
+        }
+
+        for collection in self._telemetry['collections']:
+            if collection in _CORE_SYN_COLLECTIONS:
+                continue
+            cpath = files('ansible_collections.%s' % collection)
+            for candidate in ('METADATA.json', 'galaxy.yml'):
+                mdfile = cpath.joinpath(candidate)
+                if mdfile.exists():
+                    self._telemetry['collections'][collection]['version'] = yaml_load(
+                        mdfile.read_text()
+                    ).get('version', '*')
+                    break
+
+        with contextlib.suppress(Exception):
+            self._write_journal()
+
+    def _write_journal(self):
+        journal_sendv(
+            'MESSAGE=%s' % json.dumps(self._telemetry, separators=(',', ':')),
+            'MESSAGE_ID=%s' % _ANSIBLE_TELEMETRY_MESSAGE_ID.hex,
+            'ANSIBLE_TELEMETRY=core',
+            'ANSIBLE_TELEMETRY_ID=%s' % uuid.uuid4().hex,
+        )
+
+
+@contextlib.contextmanager
+def _get_journal_reader(seek=None):
+    with journal.Reader() as j:
+        if seek:
+            j.seek_realtime(seek)
+        j.add_match('MESSAGE_ID=%s' % _ANSIBLE_TELEMETRY_MESSAGE_ID.hex)
+        j.add_match('ANSIBLE_TELEMETRY=core')
+        yield j
+
+
+def main():
+    lastupload_file = pathlib.Path('/etc/insights-client/.lastupload')
+    lastupload = None
+    with contextlib.suppress(ValueError, OSError):
+        lastupload = datetime.datetime.strptime(
+            lastupload_file.read_text(),
+            '%Y-%m-%dT%H:%M:%S.%f'
+        )
+
+    with _get_journal_reader(lastupload) as j:
+        for entry in j:
+            print(entry['MESSAGE'])
+
+
+if __name__ == '__main__':
+    main()