Replace python3-lttng with lttngpy

Signed-off-by: Christophe Bedard <christophe.bedard@apex.ai>

test_ros2trace/package.xml

@@ -18,6 +18,7 @@
   <test_depend>ament_xmllint</test_depend>
   <test_depend>launch</test_depend>
   <test_depend>launch_ros</test_depend>
+  <test_depend>lttngpy</test_depend>
   <test_depend>python3-pytest</test_depend>
   <test_depend>ros2run</test_depend>
   <test_depend>ros2trace</test_depend>

test_ros2trace/test/test_ros2trace/test_trace.py

@@ -25,6 +25,7 @@
 from launch import LaunchDescription
 from launch import LaunchService
 from launch_ros.actions import Node
+from lttngpy import impl as lttngpy
 from tracetools_trace.tools import tracepoints
 from tracetools_trace.tools.lttng import is_lttng_installed
 
@@ -63,30 +64,26 @@ def tearDown(self) -> None:
         # Even if running 'ros2 trace' fails, we do not want any lingering tracing session
         self.assertNoTracingSession()
 
-    def get_tracing_sessions(self) -> Tuple[bool, str]:
-        output = self.run_lttng_list()
-        # If there is no session daemon, then there are no tracing sessions
-        no_session_daemon_available = 'No session daemon is available' in output
-        if no_session_daemon_available:
-            return False, output
-        # Starting from LTTng 2.13, 'tracing session' was replaced with 'recording session'
-        # (see lttng-tools e971184)
-        has_tracing_sessions = not any(
-            f'Currently no available {name} session' in output for name in ('tracing', 'recording')
-        )
-        return has_tracing_sessions, output
-
     def assertTracingSession(self) -> None:
-        has_tracing_sessions, output = self.get_tracing_sessions()
-        self.assertTrue(has_tracing_sessions, 'no tracing sessions exist:\n' + output)
+        self.assertTrue(
+            lttngpy.is_lttng_session_daemon_alive(),
+            'no tracing sessions exist because there is no daemon',
+        )
+        session_names = lttngpy.get_session_names()
+        has_tracing_sessions = session_names is not None and 0 < len(session_names)
+        self.assertTrue(has_tracing_sessions, 'no tracing sessions exist')
 
     def assertNoTracingSession(self) -> None:
-        has_tracing_sessions, output = self.get_tracing_sessions()
-        if has_tracing_sessions:
+        # If there is no session daemon, then there are no tracing sessions
+        if not lttngpy.is_lttng_session_daemon_alive():
+            return
+        session_names = lttngpy.get_session_names()
+        no_tracing_sessions = 0 == len(session_names)
+        if not no_tracing_sessions:
             # Destroy tracing sessions if there are any, this way we can continue running tests and
             # avoid possible interference between them
-            self.run_lttng_destroy_all()
-        self.assertFalse(has_tracing_sessions, 'tracing session(s) exist:\n' + output)
+            self.assertEqual(0, lttngpy.destroy_all_sessions())
+        self.assertTrue(no_tracing_sessions, f'tracing session(s) exist: {session_names}')
 
     def assertTraceExist(self, trace_dir: str) -> None:
         self.assertTrue(os.path.isdir(trace_dir), f'trace directory does not exist: {trace_dir}')
@@ -116,25 +113,6 @@ def create_test_tmpdir(self, test_name: str) -> str:
     def get_subdirectories(self, directory: str) -> List[str]:
         return [f.name for f in os.scandir(directory) if f.is_dir()]
 
-    def run_lttng_list(self) -> str:
-        process = subprocess.run(
-            ['lttng', 'list'],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            encoding='utf-8',
-        )
-        return process.stdout + process.stderr
-
-    def run_lttng_destroy_all(self):
-        process = subprocess.run(
-            ['lttng', 'destroy', '--all'],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            encoding='utf-8',
-        )
-        output = process.stdout + process.stderr
-        self.assertEqual(0, process.returncode, f"'lttng destroy' command failed: {output}")
-
     def run_command(
         self,
         args: List[str],
@@ -370,6 +348,17 @@ def test_base_path_not_exist(self) -> None:
 
         shutil.rmtree(tmpdir)
 
+    def test_no_events(self) -> None:
+        tmpdir = self.create_test_tmpdir('test_no_events')
+
+        # Enabling no events should result in an error
+        ret = self.run_trace_command(
+            ['--path', tmpdir, '--ust', '--kernel'],
+        )
+        self.assertEqual(1, ret)
+
+        shutil.rmtree(tmpdir)
+
     def test_unknown_context_field(self) -> None:
         tmpdir = self.create_test_tmpdir('test_unknown_context_field')
 

tracetools_trace/package.xml

@@ -12,8 +12,7 @@
   <url type="bugtracker">https://github.com/ros2/ros2_tracing/issues</url>
   <author email="fixed-term.christophe.bourquebedard@de.bosch.com">Christophe Bedard</author>
 
-  <exec_depend>lttng-tools</exec_depend>
-  <exec_depend>python3-lttng</exec_depend>
+  <exec_depend>lttngpy</exec_depend>
 
   <test_depend>ament_copyright</test_depend>
   <test_depend>ament_flake8</test_depend>

tracetools_trace/test/tracetools_trace/test_lttng_tracing.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 
 import os
-import subprocess
 import tempfile
 import unittest
 from unittest import mock
@@ -35,32 +34,7 @@ def test_is_lttng_installed(self):
         with mock.patch('platform.system', return_value='Windows'):
             self.assertFalse(is_lttng_installed())
 
-        # LTTng command not found
-        class PopenFileNotFound:
-
-            def __init__(self, *args, **kwargs):
-                raise FileNotFoundError('file not found')
-
-        with mock.patch.object(subprocess, 'Popen', PopenFileNotFound):
-            self.assertFalse(is_lttng_installed())
-
-        # Other error when running LTTng command
-        class PopenReturnCodeError:
-
-            def __init__(self, *args, **kwargs):
-                pass
-
-            def communicate(self):
-                return 'stdout'.encode(), 'stderr'.encode()
-
-            @property
-            def returncode(self):
-                return 1
-
-        with mock.patch.object(subprocess, 'Popen', PopenReturnCodeError):
-            self.assertFalse(is_lttng_installed())
-
-        # lttng Python package or version not found
+        # lttng-ctl or version not found
         with mock.patch('tracetools_trace.tools.lttng.get_lttng_version', return_value=None):
             self.assertFalse(is_lttng_installed())
 
@@ -81,11 +55,18 @@ def test_lttng_not_installed(self):
     def test_no_kernel_tracer(self):
         from tracetools_trace.tools.lttng_impl import setup
         with (
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_not_alive',
+                return_value=False,
+            ),
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_unreachable',
+                return_value=False,
+            ),
             mock.patch(
                 'tracetools_trace.tools.lttng_impl.is_kernel_tracer_available',
-                return_value=(False, 'some error message'),
+                return_value=False,
             ),
-            mock.patch('lttng.session_daemon_alive', return_value=1),
         ):
             with self.assertRaises(RuntimeError):
                 setup(
@@ -166,9 +147,15 @@ def test_is_session_daemon_unreachable(self):
 
     def test_unreachable_session_daemon(self):
         from tracetools_trace.tools.lttng_impl import setup
-        with mock.patch(
-            'tracetools_trace.tools.lttng_impl.is_session_daemon_unreachable',
-            return_value=True,
+        with (
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_not_alive',
+                return_value=False,
+            ),
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_unreachable',
+                return_value=True,
+            ),
         ):
             with self.assertRaises(RuntimeError):
                 setup(session_name='test-session', base_path='/tmp')

tracetools_trace/tracetools_trace/tools/lttng.py

@@ -16,42 +16,30 @@
 """Interface for tracing with LTTng."""
 
 import platform
-import subprocess
 import sys
 from typing import Optional
 
 from packaging.version import Version
 
 try:
-    from . import lttng_impl
-    _lttng = lttng_impl  # type: ignore
+    from . import lttng_impl as _lttng
 except ImportError:
     # Fall back on stub functions so that this still passes linter checks
-    from . import lttng_stub
-    _lttng = lttng_stub  # type: ignore
+    # This will happen if lttngpy isn't found, in which case importing lttng_impl will fail
+    from . import lttng_stub as _lttng  # type: ignore
 
 
 def get_lttng_version() -> Optional[Version]:
     """
-    Get version of lttng Python package.
+    Get version of lttng-ctl.
 
-    :return: the version of the lttng Python package, or `None` if it is not available
+    :return: the version of lttng-ctl, or `None` if it is not available
     """
     if not hasattr(_lttng, 'get_version') or not callable(_lttng.get_version):
         return None
     return _lttng.get_version()
 
 
-# Check lttng module version
-current_version = get_lttng_version()
-LTTNG_MIN_VERSION = '2.10.7'
-if current_version is None or current_version < Version(LTTNG_MIN_VERSION):
-    print(
-        f'lttng module version >={LTTNG_MIN_VERSION} required, found {str(current_version)}',
-        file=sys.stderr,
-    )
-
-
 def lttng_init(**kwargs) -> Optional[str]:
     """
     Set up and start LTTng session.
@@ -114,14 +102,13 @@ def is_lttng_installed(
     Check if LTTng is installed.
 
     It first checks if the OS can support LTTng.
-    If so, it then simply checks if LTTng is installed using the 'lttng' command, and checks if the
-    lttng Python package is installed (python3-lttng).
+    If so, it then checks if lttng-ctl is installed.
 
-    Optionally, a minimum version can also be specified for the lttng Python package.
+    Optionally, a minimum version can also be specified for lttng-ctl.
 
-    :param minimum_version: the minimum required lttng Python package version
-    :return: True if LTTng and the lttng Python package are installed, and optionally if the
-        version of lttng Python package is sufficient, False otherwise
+    :param minimum_version: the minimum required lttng-ctl version
+    :return: True if lttng-ctl is installed, and optionally if the version of lttng-ctl is
+        sufficient, False otherwise
     """
     # Check system
     message_doc = (
@@ -132,32 +119,19 @@ def is_lttng_installed(
     if 'Linux' != system:
         print(f"System '{system}' does not support LTTng.\n{message_doc}", file=sys.stderr)
         return False
-    # Check if LTTng (CLI) is installed
-    try:
-        process = subprocess.Popen(
-            ['lttng', '--version'],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
-        _, stderr = process.communicate()
-        if 0 != process.returncode:
-            raise RuntimeError(stderr.decode())
-    except (RuntimeError, FileNotFoundError) as e:
-        print(f'LTTng not found: {e}\n{message_doc}', file=sys.stderr)
-        return False
-    # Check if lttng Python package is installed
+    # Check if lttng-ctl is installed
     lttng_version = get_lttng_version()
     if not lttng_version:
         print(
-            f'lttng Python package (python3-lttng) not installed\n{message_doc}',
+            f'lttng-ctl (liblttng-ctl-dev) not installed\n{message_doc}',
             file=sys.stderr,
         )
         return False
-    # Check if lttng Python package version is sufficient
+    # Check if lttng-ctl version is sufficient
     if minimum_version and lttng_version < Version(minimum_version):
         print(
             (
-                f'lttng Python package (python3-lttng) version >={minimum_version} required, '
+                f'lttng-ctl (liblttng-ctl-dev) version >={minimum_version} required, '
                 f'found {str(lttng_version)}'
             ),
             file=sys.stderr,