[lttng-tools.git] / tests / utils / lttngtest / lttng.py

#!/usr/bin/env python3
#
# Copyright (C) 2022 Jérémie Galarneau <jeremie.galarneau@efficios.com>
#
# SPDX-License-Identifier: GPL-2.0-only


from . import lttngctl, logger, environment
import pathlib
import os
from typing import Callable, Optional, Type, Union
import shlex
import subprocess
import enum
import xml.etree.ElementTree

"""
Implementation of the lttngctl interface based on the `lttng` command line client.
"""


class Unsupported(lttngctl.ControlException):
    def __init__(self, msg):
        # type: (str) -> None
        super().__init__(msg)


class InvalidMI(lttngctl.ControlException):
    def __init__(self, msg):
        # type: (str) -> None
        super().__init__(msg)


def _get_domain_option_name(domain):
    # type: (lttngctl.TracingDomain) -> str
    if domain == lttngctl.TracingDomain.User:
        return "userspace"
    elif domain == lttngctl.TracingDomain.Kernel:
        return "kernel"
    elif domain == lttngctl.TracingDomain.Log4j:
        return "log4j"
    elif domain == lttngctl.TracingDomain.JUL:
        return "jul"
    elif domain == lttngctl.TracingDomain.Python:
        return "python"
    else:
        raise Unsupported("Domain `{domain_name}` is not supported by the LTTng client")


def _get_context_type_name(context):
    # type: (lttngctl.ContextType) -> str
    if isinstance(context, lttngctl.VgidContextType):
        return "vgid"
    elif isinstance(context, lttngctl.VuidContextType):
        return "vuid"
    elif isinstance(context, lttngctl.VpidContextType):
        return "vpid"
    elif isinstance(context, lttngctl.JavaApplicationContextType):
        return "$app.{retriever}:{field}".format(
            retriever=context.retriever_name, field=context.field_name
        )
    else:
        raise Unsupported(
            "Context `{context_name}` is not supported by the LTTng client".format(
                type(context).__name__
            )
        )


class _Channel(lttngctl.Channel):
    def __init__(
        self,
        client,  # type: LTTngClient
        name,  # type: str
        domain,  # type: lttngctl.TracingDomain
        session,  # type: _Session
    ):
        self._client = client  # type: LTTngClient
        self._name = name  # type: str
        self._domain = domain  # type: lttngctl.TracingDomain
        self._session = session  # type: _Session

    def add_context(self, context_type):
        # type: (lttngctl.ContextType) -> None
        domain_option_name = _get_domain_option_name(self.domain)
        context_type_name = _get_context_type_name(context_type)
        self._client._run_cmd(
            "add-context --{domain_option_name} --channel '{channel_name}' --type {context_type_name}".format(
                domain_option_name=domain_option_name,
                channel_name=self.name,
                context_type_name=context_type_name,
            )
        )

    def add_recording_rule(self, rule):
        # type: (Type[lttngctl.EventRule]) -> None
        client_args = (
            "enable-event --session {session_name} --channel {channel_name}".format(
                session_name=self._session.name, channel_name=self.name
            )
        )
        if isinstance(rule, lttngctl.TracepointEventRule):
            domain_option_name = (
                "userspace"
                if isinstance(rule, lttngctl.UserTracepointEventRule)
                else "kernel"
            )
            client_args = client_args + " --{domain_option_name}".format(
                domain_option_name=domain_option_name
            )

            if rule.name_pattern:
                client_args = client_args + " " + rule.name_pattern
            else:
                client_args = client_args + " --all"

            if rule.filter_expression:
                client_args = client_args + " " + rule.filter_expression

            if rule.log_level_rule:
                if isinstance(rule.log_level_rule, lttngctl.LogLevelRuleAsSevereAs):
                    client_args = client_args + " --loglevel {log_level}".format(
                        log_level=rule.log_level_rule.level
                    )
                elif isinstance(rule.log_level_rule, lttngctl.LogLevelRuleExactly):
                    client_args = client_args + " --loglevel-only {log_level}".format(
                        log_level=rule.log_level_rule.level
                    )
                else:
                    raise Unsupported(
                        "Unsupported log level rule type `{log_level_rule_type}`".format(
                            log_level_rule_type=type(rule.log_level_rule).__name__
                        )
                    )

            if rule.name_pattern_exclusions:
                client_args = client_args + " --exclude "
                for idx, pattern in enumerate(rule.name_pattern_exclusions):
                    if idx != 0:
                        client_args = client_args + ","
                    client_args = client_args + pattern
        else:
            raise Unsupported(
                "event rule type `{event_rule_type}` is unsupported by LTTng client".format(
                    event_rule_type=type(rule).__name__
                )
            )

        self._client._run_cmd(client_args)

    @property
    def name(self):
        # type: () -> str
        return self._name

    @property
    def domain(self):
        # type: () -> lttngctl.TracingDomain
        return self._domain


@enum.unique
class _ProcessAttribute(enum.Enum):
    PID = "Process ID"
    VPID = "Virtual Process ID"
    UID = "User ID"
    VUID = "Virtual User ID"
    GID = "Group ID"
    VGID = "Virtual Group ID"

    def __repr__(self):
        return "<%s.%s>" % (self.__class__.__name__, self.name)


def _get_process_attribute_option_name(attribute):
    # type: (_ProcessAttribute) -> str
    return {
        _ProcessAttribute.PID: "pid",
        _ProcessAttribute.VPID: "vpid",
        _ProcessAttribute.UID: "uid",
        _ProcessAttribute.VUID: "vuid",
        _ProcessAttribute.GID: "gid",
        _ProcessAttribute.VGID: "vgid",
    }[attribute]


class _ProcessAttributeTracker(lttngctl.ProcessAttributeTracker):
    def __init__(
        self,
        client,  # type: LTTngClient
        attribute,  # type: _ProcessAttribute
        domain,  # type: lttngctl.TracingDomain
        session,  # type: _Session
    ):
        self._client = client  # type: LTTngClient
        self._tracked_attribute = attribute  # type: _ProcessAttribute
        self._domain = domain  # type: lttngctl.TracingDomain
        self._session = session  # type: _Session
        if attribute == _ProcessAttribute.PID or attribute == _ProcessAttribute.VPID:
            self._allowed_value_types = [int, str]  # type: list[type]
        else:
            self._allowed_value_types = [int]  # type: list[type]

    def _call_client(self, cmd_name, value):
        # type: (str, Union[int, str]) -> None
        if type(value) not in self._allowed_value_types:
            raise TypeError(
                "Value of type `{value_type}` is not allowed for process attribute {attribute_name}".format(
                    value_type=type(value).__name__,
                    attribute_name=self._tracked_attribute.name,
                )
            )

        process_attribute_option_name = _get_process_attribute_option_name(
            self._tracked_attribute
        )
        domain_name = _get_domain_option_name(self._domain)
        self._client._run_cmd(
            "{cmd_name} --session '{session_name}' --{domain_name} --{tracked_attribute_name} {value}".format(
                cmd_name=cmd_name,
                session_name=self._session.name,
                domain_name=domain_name,
                tracked_attribute_name=process_attribute_option_name,
                value=value,
            )
        )

    def track(self, value):
        # type: (Union[int, str]) -> None
        self._call_client("track", value)

    def untrack(self, value):
        # type: (Union[int, str]) -> None
        self._call_client("untrack", value)


class _Session(lttngctl.Session):
    def __init__(
        self,
        client,  # type: LTTngClient
        name,  # type: str
        output,  # type: Optional[lttngctl.SessionOutputLocation]
    ):
        self._client = client  # type: LTTngClient
        self._name = name  # type: str
        self._output = output  # type: Optional[lttngctl.SessionOutputLocation]

    @property
    def name(self):
        # type: () -> str
        return self._name

    def add_channel(
        self,
        domain,
        channel_name=None,
        buffer_sharing_policy=lttngctl.BufferSharingPolicy.PerUID,
    ):
        # type: (lttngctl.TracingDomain, Optional[str], lttngctl.BufferSharingPolicy) -> lttngctl.Channel
        channel_name = lttngctl.Channel._generate_name()
        domain_option_name = _get_domain_option_name(domain)
        self._client._run_cmd(
            "enable-channel --session '{session_name}' --{domain_name} '{channel_name}' {buffer_sharing_policy}".format(
                session_name=self.name,
                domain_name=domain_option_name,
                channel_name=channel_name,
                buffer_sharing_policy="--buffers-uid"
                if buffer_sharing_policy == lttngctl.BufferSharingPolicy.PerUID
                else "--buffers-pid",
            )
        )
        return _Channel(self._client, channel_name, domain, self)

    def add_context(self, context_type):
        # type: (lttngctl.ContextType) -> None
        pass

    @property
    def output(self):
        # type: () -> "Optional[Type[lttngctl.SessionOutputLocation]]"
        return self._output  # type: ignore

    def start(self):
        # type: () -> None
        self._client._run_cmd("start '{session_name}'".format(session_name=self.name))

    def stop(self):
        # type: () -> None
        self._client._run_cmd("stop '{session_name}'".format(session_name=self.name))

    def destroy(self):
        # type: () -> None
        self._client._run_cmd("destroy '{session_name}'".format(session_name=self.name))

    @property
    def is_active(self):
        # type: () -> bool
        list_session_xml = self._client._run_cmd(
            "list '{session_name}'".format(session_name=self.name),
            LTTngClient.CommandOutputFormat.MI_XML,
        )

        root = xml.etree.ElementTree.fromstring(list_session_xml)
        command_output = LTTngClient._mi_find_in_element(root, "output")
        sessions = LTTngClient._mi_find_in_element(command_output, "sessions")
        session_mi = LTTngClient._mi_find_in_element(sessions, "session")

        enabled_text = LTTngClient._mi_find_in_element(session_mi, "enabled").text
        if enabled_text not in ["true", "false"]:
            raise InvalidMI(
                "Expected boolean value in element '{}': value='{}'".format(
                    session_mi.tag, enabled_text
                )
            )

        return enabled_text == "true"

    @property
    def kernel_pid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.ProcessIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.PID, lttngctl.TracingDomain.Kernel, self)  # type: ignore

    @property
    def kernel_vpid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.Kernel, self)  # type: ignore

    @property
    def user_vpid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.User, self)  # type: ignore

    @property
    def kernel_gid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.GroupIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.GID, lttngctl.TracingDomain.Kernel, self)  # type: ignore

    @property
    def kernel_vgid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.Kernel, self)  # type: ignore

    @property
    def user_vgid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.User, self)  # type: ignore

    @property
    def kernel_uid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.UserIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.UID, lttngctl.TracingDomain.Kernel, self)  # type: ignore

    @property
    def kernel_vuid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.Kernel, self)  # type: ignore

    @property
    def user_vuid_process_attribute_tracker(self):
        # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
        return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.User, self)  # type: ignore


class LTTngClientError(lttngctl.ControlException):
    def __init__(
        self,
        command_args,  # type: str
        error_output,  # type: str
    ):
        self._command_args = command_args  # type: str
        self._output = error_output  # type: str


class LTTngClient(logger._Logger, lttngctl.Controller):
    """
    Implementation of a LTTngCtl Controller that uses the `lttng` client as a back-end.
    """

    class CommandOutputFormat(enum.Enum):
        MI_XML = 0
        HUMAN = 1

    _MI_NS = "{https://lttng.org/xml/ns/lttng-mi}"

    def __init__(
        self,
        test_environment,  # type: environment._Environment
        log,  # type: Optional[Callable[[str], None]]
    ):
        logger._Logger.__init__(self, log)
        self._environment = test_environment  # type: environment._Environment

    @staticmethod
    def _namespaced_mi_element(property):
        # type: (str) -> str
        return LTTngClient._MI_NS + property

    def _run_cmd(self, command_args, output_format=CommandOutputFormat.MI_XML):
        # type: (str, CommandOutputFormat) -> str
        """
        Invoke the `lttng` client with a set of arguments. The command is
        executed in the context of the client's test environment.
        """
        args = [str(self._environment.lttng_client_path)]  # type: list[str]
        if output_format == LTTngClient.CommandOutputFormat.MI_XML:
            args.extend(["--mi", "xml"])

        args.extend(shlex.split(command_args))

        self._log("lttng {command_args}".format(command_args=command_args))

        client_env = os.environ.copy()  # type: dict[str, str]
        client_env["LTTNG_HOME"] = str(self._environment.lttng_home_location)

        process = subprocess.Popen(
            args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=client_env
        )

        out = process.communicate()[0]

        if process.returncode != 0:
            decoded_output = out.decode("utf-8")
            for error_line in decoded_output.splitlines():
                self._log(error_line)
            raise LTTngClientError(command_args, decoded_output)
        else:
            return out.decode("utf-8")

    def create_session(self, name=None, output=None):
        # type: (Optional[str], Optional[lttngctl.SessionOutputLocation]) -> lttngctl.Session
        name = name if name else lttngctl.Session._generate_name()

        if isinstance(output, lttngctl.LocalSessionOutputLocation):
            output_option = "--output {output_path}".format(output_path=output.path)
        elif output is None:
            output_option = "--no-output"
        else:
            raise TypeError("LTTngClient only supports local or no output")

        self._run_cmd(
            "create '{session_name}' {output_option}".format(
                session_name=name, output_option=output_option
            )
        )
        return _Session(self, name, output)

    def start_session_by_name(self, name):
        # type: (str) -> None
        self._run_cmd("start '{session_name}'".format(session_name=name))

    def start_session_by_glob_pattern(self, pattern):
        # type: (str) -> None
        self._run_cmd("start --glob '{pattern}'".format(pattern=pattern))

    def start_sessions_all(self):
        # type: () -> None
        self._run_cmd("start --all")

    def stop_session_by_name(self, name):
        # type: (str) -> None
        self._run_cmd("stop '{session_name}'".format(session_name=name))

    def stop_session_by_glob_pattern(self, pattern):
        # type: (str) -> None
        self._run_cmd("stop --glob '{pattern}'".format(pattern=pattern))

    def stop_sessions_all(self):
        # type: () -> None
        self._run_cmd("stop --all")

    def destroy_session_by_name(self, name):
        # type: (str) -> None
        self._run_cmd("destroy '{session_name}'".format(session_name=name))

    def destroy_session_by_glob_pattern(self, pattern):
        # type: (str) -> None
        self._run_cmd("destroy --glob '{pattern}'".format(pattern=pattern))

    def destroy_sessions_all(self):
        # type: () -> None
        self._run_cmd("destroy --all")

    @staticmethod
    def _mi_find_in_element(element, sub_element_name):
        # type: (xml.etree.ElementTree.Element, str) -> xml.etree.ElementTree.Element
        result = element.find(LTTngClient._namespaced_mi_element(sub_element_name))
        if result is None:
            raise InvalidMI(
                "Failed to find element '{}' within command MI element '{}'".format(
                    element.tag, sub_element_name
                )
            )

        return result

    def list_sessions(self):
        # type () -> List[Session]
        list_sessions_xml = self._run_cmd(
            "list", LTTngClient.CommandOutputFormat.MI_XML
        )

        root = xml.etree.ElementTree.fromstring(list_sessions_xml)
        command_output = self._mi_find_in_element(root, "output")
        sessions = self._mi_find_in_element(command_output, "sessions")

        ctl_sessions = []  # type: list[lttngctl.Session]

        for session_mi in sessions:
            name = self._mi_find_in_element(session_mi, "name").text
            path = self._mi_find_in_element(session_mi, "path").text

            if name is None:
                raise InvalidMI(
                    "Invalid empty 'name' element in '{}'".format(session_mi.tag)
                )
            if path is None:
                raise InvalidMI(
                    "Invalid empty 'path' element in '{}'".format(session_mi.tag)
                )
            if not path.startswith("/"):
                raise Unsupported(
                    "{} does not support non-local session outputs".format(type(self))
                )

            ctl_sessions.append(
                _Session(self, name, lttngctl.LocalSessionOutputLocation(path))
            )

        return ctl_sessions
Commit	Line	Data
ef945e4d JG	1	#!/usr/bin/env python3
	2	#
	3	# Copyright (C) 2022 Jérémie Galarneau <jeremie.galarneau@efficios.com>
	4	#
	5	# SPDX-License-Identifier: GPL-2.0-only
	6
b9780062	7
ef945e4d JG	8	from . import lttngctl, logger, environment
	9	import pathlib
	10	import os
	11	from typing import Callable, Optional, Type, Union
	12	import shlex
	13	import subprocess
	14	import enum
b9780062	15	import xml.etree.ElementTree
ef945e4d JG	16
	17	"""
	18	Implementation of the lttngctl interface based on the `lttng` command line client.
	19	"""
	20
	21
	22	class Unsupported(lttngctl.ControlException):
ce8470c9 MJ	23	def __init__(self, msg):
ce8470c9 MJ	24	# type: (str) -> None
ef945e4d JG	25	super().__init__(msg)
	26
	27
b9780062 JG	28	class InvalidMI(lttngctl.ControlException):
	29	def __init__(self, msg):
	30	# type: (str) -> None
	31	super().__init__(msg)
	32
	33
ce8470c9 MJ	34	def _get_domain_option_name(domain):
ce8470c9 MJ	35	# type: (lttngctl.TracingDomain) -> str
ef945e4d JG	36	if domain == lttngctl.TracingDomain.User:
	37	return "userspace"
	38	elif domain == lttngctl.TracingDomain.Kernel:
	39	return "kernel"
	40	elif domain == lttngctl.TracingDomain.Log4j:
	41	return "log4j"
	42	elif domain == lttngctl.TracingDomain.JUL:
	43	return "jul"
	44	elif domain == lttngctl.TracingDomain.Python:
	45	return "python"
	46	else:
	47	raise Unsupported("Domain `{domain_name}` is not supported by the LTTng client")
	48
	49
ce8470c9 MJ	50	def _get_context_type_name(context):
ce8470c9 MJ	51	# type: (lttngctl.ContextType) -> str
ef945e4d JG	52	if isinstance(context, lttngctl.VgidContextType):
	53	return "vgid"
	54	elif isinstance(context, lttngctl.VuidContextType):
	55	return "vuid"
	56	elif isinstance(context, lttngctl.VpidContextType):
	57	return "vpid"
	58	elif isinstance(context, lttngctl.JavaApplicationContextType):
	59	return "$app.{retriever}:{field}".format(
	60	retriever=context.retriever_name, field=context.field_name
	61	)
	62	else:
	63	raise Unsupported(
	64	"Context `{context_name}` is not supported by the LTTng client".format(
	65	type(context).__name__
	66	)
	67	)
	68
	69
	70	class _Channel(lttngctl.Channel):
	71	def __init__(
	72	self,
ce8470c9 MJ	73	client, # type: LTTngClient
	74	name, # type: str
	75	domain, # type: lttngctl.TracingDomain
	76	session, # type: _Session
ef945e4d	77	):
ce8470c9 MJ	78	self._client = client # type: LTTngClient
	79	self._name = name # type: str
	80	self._domain = domain # type: lttngctl.TracingDomain
	81	self._session = session # type: _Session
ef945e4d	82
ce8470c9 MJ	83	def add_context(self, context_type):
ce8470c9 MJ	84	# type: (lttngctl.ContextType) -> None
ef945e4d JG	85	domain_option_name = _get_domain_option_name(self.domain)
	86	context_type_name = _get_context_type_name(context_type)
	87	self._client._run_cmd(
b9780062	88	"add-context --{domain_option_name} --channel '{channel_name}' --type {context_type_name}".format(
ef945e4d	89	domain_option_name=domain_option_name,
b9780062	90	channel_name=self.name,
ef945e4d JG	91	context_type_name=context_type_name,
	92	)
	93	)
	94
ce8470c9 MJ	95	def add_recording_rule(self, rule):
ce8470c9 MJ	96	# type: (Type[lttngctl.EventRule]) -> None
ef945e4d JG	97	client_args = (
	98	"enable-event --session {session_name} --channel {channel_name}".format(
	99	session_name=self._session.name, channel_name=self.name
	100	)
	101	)
	102	if isinstance(rule, lttngctl.TracepointEventRule):
	103	domain_option_name = (
	104	"userspace"
	105	if isinstance(rule, lttngctl.UserTracepointEventRule)
	106	else "kernel"
	107	)
	108	client_args = client_args + " --{domain_option_name}".format(
	109	domain_option_name=domain_option_name
	110	)
	111
	112	if rule.name_pattern:
	113	client_args = client_args + " " + rule.name_pattern
	114	else:
	115	client_args = client_args + " --all"
	116
	117	if rule.filter_expression:
	118	client_args = client_args + " " + rule.filter_expression
	119
	120	if rule.log_level_rule:
	121	if isinstance(rule.log_level_rule, lttngctl.LogLevelRuleAsSevereAs):
	122	client_args = client_args + " --loglevel {log_level}".format(
	123	log_level=rule.log_level_rule.level
	124	)
	125	elif isinstance(rule.log_level_rule, lttngctl.LogLevelRuleExactly):
	126	client_args = client_args + " --loglevel-only {log_level}".format(
	127	log_level=rule.log_level_rule.level
	128	)
	129	else:
	130	raise Unsupported(
	131	"Unsupported log level rule type `{log_level_rule_type}`".format(
	132	log_level_rule_type=type(rule.log_level_rule).__name__
	133	)
	134	)
	135
	136	if rule.name_pattern_exclusions:
	137	client_args = client_args + " --exclude "
	138	for idx, pattern in enumerate(rule.name_pattern_exclusions):
	139	if idx != 0:
	140	client_args = client_args + ","
	141	client_args = client_args + pattern
	142	else:
	143	raise Unsupported(
	144	"event rule type `{event_rule_type}` is unsupported by LTTng client".format(
	145	event_rule_type=type(rule).__name__
	146	)
	147	)
	148
	149	self._client._run_cmd(client_args)
	150
	151	@property
ce8470c9 MJ	152	def name(self):
ce8470c9 MJ	153	# type: () -> str
ef945e4d JG	154	return self._name
	155
	156	@property
ce8470c9 MJ	157	def domain(self):
ce8470c9 MJ	158	# type: () -> lttngctl.TracingDomain
ef945e4d JG	159	return self._domain
	160
	161
544d8425	162	@enum.unique
ef945e4d	163	class _ProcessAttribute(enum.Enum):
544d8425 MJ	164	PID = "Process ID"
	165	VPID = "Virtual Process ID"
	166	UID = "User ID"
	167	VUID = "Virtual User ID"
	168	GID = "Group ID"
	169	VGID = "Virtual Group ID"
	170
	171	def __repr__(self):
	172	return "<%s.%s>" % (self.__class__.__name__, self.name)
ef945e4d JG	173
ef945e4d JG	174
ce8470c9 MJ	175	def _get_process_attribute_option_name(attribute):
ce8470c9 MJ	176	# type: (_ProcessAttribute) -> str
ef945e4d JG	177	return {
	178	_ProcessAttribute.PID: "pid",
	179	_ProcessAttribute.VPID: "vpid",
	180	_ProcessAttribute.UID: "uid",
	181	_ProcessAttribute.VUID: "vuid",
	182	_ProcessAttribute.GID: "gid",
	183	_ProcessAttribute.VGID: "vgid",
	184	}[attribute]
	185
	186
	187	class _ProcessAttributeTracker(lttngctl.ProcessAttributeTracker):
	188	def __init__(
	189	self,
ce8470c9 MJ	190	client, # type: LTTngClient
	191	attribute, # type: _ProcessAttribute
	192	domain, # type: lttngctl.TracingDomain
	193	session, # type: _Session
ef945e4d	194	):
ce8470c9 MJ	195	self._client = client # type: LTTngClient
	196	self._tracked_attribute = attribute # type: _ProcessAttribute
	197	self._domain = domain # type: lttngctl.TracingDomain
	198	self._session = session # type: _Session
ef945e4d	199	if attribute == _ProcessAttribute.PID or attribute == _ProcessAttribute.VPID:
ce8470c9	200	self._allowed_value_types = [int, str] # type: list[type]
ef945e4d	201	else:
ce8470c9	202	self._allowed_value_types = [int] # type: list[type]
ef945e4d	203
ce8470c9 MJ	204	def _call_client(self, cmd_name, value):
ce8470c9 MJ	205	# type: (str, Union[int, str]) -> None
ef945e4d JG	206	if type(value) not in self._allowed_value_types:
	207	raise TypeError(
	208	"Value of type `{value_type}` is not allowed for process attribute {attribute_name}".format(
	209	value_type=type(value).__name__,
	210	attribute_name=self._tracked_attribute.name,
	211	)
	212	)
	213
	214	process_attribute_option_name = _get_process_attribute_option_name(
	215	self._tracked_attribute
	216	)
	217	domain_name = _get_domain_option_name(self._domain)
	218	self._client._run_cmd(
b9780062	219	"{cmd_name} --session '{session_name}' --{domain_name} --{tracked_attribute_name} {value}".format(
ef945e4d JG	220	cmd_name=cmd_name,
	221	session_name=self._session.name,
	222	domain_name=domain_name,
	223	tracked_attribute_name=process_attribute_option_name,
	224	value=value,
	225	)
	226	)
	227
ce8470c9 MJ	228	def track(self, value):
ce8470c9 MJ	229	# type: (Union[int, str]) -> None
ef945e4d JG	230	self._call_client("track", value)
ef945e4d JG	231
ce8470c9 MJ	232	def untrack(self, value):
ce8470c9 MJ	233	# type: (Union[int, str]) -> None
ef945e4d JG	234	self._call_client("untrack", value)
	235
	236
	237	class _Session(lttngctl.Session):
	238	def __init__(
	239	self,
ce8470c9 MJ	240	client, # type: LTTngClient
	241	name, # type: str
	242	output, # type: Optional[lttngctl.SessionOutputLocation]
ef945e4d	243	):
ce8470c9 MJ	244	self._client = client # type: LTTngClient
	245	self._name = name # type: str
	246	self._output = output # type: Optional[lttngctl.SessionOutputLocation]
ef945e4d JG	247
ef945e4d JG	248	@property
ce8470c9 MJ	249	def name(self):
ce8470c9 MJ	250	# type: () -> str
ef945e4d JG	251	return self._name
ef945e4d JG	252
a631186c JG	253	def add_channel(
	254	self,
	255	domain,
	256	channel_name=None,
	257	buffer_sharing_policy=lttngctl.BufferSharingPolicy.PerUID,
	258	):
	259	# type: (lttngctl.TracingDomain, Optional[str], lttngctl.BufferSharingPolicy) -> lttngctl.Channel
ef945e4d JG	260	channel_name = lttngctl.Channel._generate_name()
	261	domain_option_name = _get_domain_option_name(domain)
	262	self._client._run_cmd(
a631186c	263	"enable-channel --session '{session_name}' --{domain_name} '{channel_name}' {buffer_sharing_policy}".format(
b9780062 JG	264	session_name=self.name,
	265	domain_name=domain_option_name,
	266	channel_name=channel_name,
a631186c JG	267	buffer_sharing_policy="--buffers-uid"
	268	if buffer_sharing_policy == lttngctl.BufferSharingPolicy.PerUID
	269	else "--buffers-pid",
ef945e4d JG	270	)
	271	)
	272	return _Channel(self._client, channel_name, domain, self)
	273
ce8470c9 MJ	274	def add_context(self, context_type):
ce8470c9 MJ	275	# type: (lttngctl.ContextType) -> None
ef945e4d JG	276	pass
	277
	278	@property
ce8470c9 MJ	279	def output(self):
	280	# type: () -> "Optional[Type[lttngctl.SessionOutputLocation]]"
	281	return self._output # type: ignore
ef945e4d	282
ce8470c9 MJ	283	def start(self):
ce8470c9 MJ	284	# type: () -> None
b9780062	285	self._client._run_cmd("start '{session_name}'".format(session_name=self.name))
ef945e4d	286
ce8470c9 MJ	287	def stop(self):
ce8470c9 MJ	288	# type: () -> None
b9780062	289	self._client._run_cmd("stop '{session_name}'".format(session_name=self.name))
ef945e4d	290
ce8470c9 MJ	291	def destroy(self):
ce8470c9 MJ	292	# type: () -> None
b9780062 JG	293	self._client._run_cmd("destroy '{session_name}'".format(session_name=self.name))
	294
	295	@property
	296	def is_active(self):
	297	# type: () -> bool
	298	list_session_xml = self._client._run_cmd(
	299	"list '{session_name}'".format(session_name=self.name),
	300	LTTngClient.CommandOutputFormat.MI_XML,
	301	)
	302
	303	root = xml.etree.ElementTree.fromstring(list_session_xml)
	304	command_output = LTTngClient._mi_find_in_element(root, "output")
	305	sessions = LTTngClient._mi_find_in_element(command_output, "sessions")
	306	session_mi = LTTngClient._mi_find_in_element(sessions, "session")
	307
	308	enabled_text = LTTngClient._mi_find_in_element(session_mi, "enabled").text
	309	if enabled_text not in ["true", "false"]:
	310	raise InvalidMI(
	311	"Expected boolean value in element '{}': value='{}'".format(
	312	session_mi.tag, enabled_text
	313	)
	314	)
	315
	316	return enabled_text == "true"
ef945e4d JG	317
ef945e4d JG	318	@property
ce8470c9 MJ	319	def kernel_pid_process_attribute_tracker(self):
ce8470c9 MJ	320	# type: () -> Type[lttngctl.ProcessIDProcessAttributeTracker]
ef945e4d JG	321	return _ProcessAttributeTracker(self._client, _ProcessAttribute.PID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	322
	323	@property
ce8470c9 MJ	324	def kernel_vpid_process_attribute_tracker(self):
ce8470c9 MJ	325	# type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
ef945e4d JG	326	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	327
	328	@property
ce8470c9 MJ	329	def user_vpid_process_attribute_tracker(self):
ce8470c9 MJ	330	# type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
ef945e4d JG	331	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.User, self) # type: ignore
	332
	333	@property
ce8470c9 MJ	334	def kernel_gid_process_attribute_tracker(self):
ce8470c9 MJ	335	# type: () -> Type[lttngctl.GroupIDProcessAttributeTracker]
ef945e4d JG	336	return _ProcessAttributeTracker(self._client, _ProcessAttribute.GID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	337
	338	@property
ce8470c9 MJ	339	def kernel_vgid_process_attribute_tracker(self):
ce8470c9 MJ	340	# type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
ef945e4d JG	341	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	342
	343	@property
ce8470c9 MJ	344	def user_vgid_process_attribute_tracker(self):
ce8470c9 MJ	345	# type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
ef945e4d JG	346	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.User, self) # type: ignore
	347
	348	@property
ce8470c9 MJ	349	def kernel_uid_process_attribute_tracker(self):
ce8470c9 MJ	350	# type: () -> Type[lttngctl.UserIDProcessAttributeTracker]
ef945e4d JG	351	return _ProcessAttributeTracker(self._client, _ProcessAttribute.UID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	352
	353	@property
ce8470c9 MJ	354	def kernel_vuid_process_attribute_tracker(self):
ce8470c9 MJ	355	# type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
ef945e4d JG	356	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	357
	358	@property
ce8470c9 MJ	359	def user_vuid_process_attribute_tracker(self):
ce8470c9 MJ	360	# type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
ef945e4d JG	361	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.User, self) # type: ignore
	362
	363
	364	class LTTngClientError(lttngctl.ControlException):
ce8470c9 MJ	365	def __init__(
	366	self,
	367	command_args, # type: str
	368	error_output, # type: str
	369	):
	370	self._command_args = command_args # type: str
	371	self._output = error_output # type: str
ef945e4d JG	372
	373
	374	class LTTngClient(logger._Logger, lttngctl.Controller):
	375	"""
	376	Implementation of a LTTngCtl Controller that uses the `lttng` client as a back-end.
	377	"""
	378
b9780062 JG	379	class CommandOutputFormat(enum.Enum):
	380	MI_XML = 0
	381	HUMAN = 1
	382
	383	_MI_NS = "{https://lttng.org/xml/ns/lttng-mi}"
	384
ef945e4d JG	385	def __init__(
ef945e4d JG	386	self,
ce8470c9 MJ	387	test_environment, # type: environment._Environment
ce8470c9 MJ	388	log, # type: Optional[Callable[[str], None]]
ef945e4d JG	389	):
ef945e4d JG	390	logger._Logger.__init__(self, log)
ce8470c9	391	self._environment = test_environment # type: environment._Environment
ef945e4d	392
b9780062 JG	393	@staticmethod
	394	def _namespaced_mi_element(property):
	395	# type: (str) -> str
	396	return LTTngClient._MI_NS + property
	397
	398	def _run_cmd(self, command_args, output_format=CommandOutputFormat.MI_XML):
	399	# type: (str, CommandOutputFormat) -> str
ef945e4d JG	400	"""
	401	Invoke the `lttng` client with a set of arguments. The command is
	402	executed in the context of the client's test environment.
	403	"""
ce8470c9	404	args = [str(self._environment.lttng_client_path)] # type: list[str]
b9780062 JG	405	if output_format == LTTngClient.CommandOutputFormat.MI_XML:
	406	args.extend(["--mi", "xml"])
	407
ef945e4d JG	408	args.extend(shlex.split(command_args))
	409
	410	self._log("lttng {command_args}".format(command_args=command_args))
	411
ce8470c9	412	client_env = os.environ.copy() # type: dict[str, str]
ef945e4d JG	413	client_env["LTTNG_HOME"] = str(self._environment.lttng_home_location)
	414
	415	process = subprocess.Popen(
	416	args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=client_env
	417	)
	418
	419	out = process.communicate()[0]
	420
	421	if process.returncode != 0:
	422	decoded_output = out.decode("utf-8")
	423	for error_line in decoded_output.splitlines():
	424	self._log(error_line)
	425	raise LTTngClientError(command_args, decoded_output)
b9780062 JG	426	else:
b9780062 JG	427	return out.decode("utf-8")
ef945e4d	428
ce8470c9 MJ	429	def create_session(self, name=None, output=None):
ce8470c9 MJ	430	# type: (Optional[str], Optional[lttngctl.SessionOutputLocation]) -> lttngctl.Session
ef945e4d JG	431	name = name if name else lttngctl.Session._generate_name()
	432
	433	if isinstance(output, lttngctl.LocalSessionOutputLocation):
	434	output_option = "--output {output_path}".format(output_path=output.path)
	435	elif output is None:
	436	output_option = "--no-output"
	437	else:
	438	raise TypeError("LTTngClient only supports local or no output")
	439
	440	self._run_cmd(
b9780062	441	"create '{session_name}' {output_option}".format(
ef945e4d JG	442	session_name=name, output_option=output_option
	443	)
	444	)
	445	return _Session(self, name, output)
b9780062 JG	446
	447	def start_session_by_name(self, name):
	448	# type: (str) -> None
	449	self._run_cmd("start '{session_name}'".format(session_name=name))
	450
	451	def start_session_by_glob_pattern(self, pattern):
	452	# type: (str) -> None
	453	self._run_cmd("start --glob '{pattern}'".format(pattern=pattern))
	454
	455	def start_sessions_all(self):
	456	# type: () -> None
	457	self._run_cmd("start --all")
	458
	459	def stop_session_by_name(self, name):
	460	# type: (str) -> None
	461	self._run_cmd("stop '{session_name}'".format(session_name=name))
	462
	463	def stop_session_by_glob_pattern(self, pattern):
	464	# type: (str) -> None
	465	self._run_cmd("stop --glob '{pattern}'".format(pattern=pattern))
	466
	467	def stop_sessions_all(self):
	468	# type: () -> None
	469	self._run_cmd("stop --all")
	470
	471	def destroy_session_by_name(self, name):
	472	# type: (str) -> None
	473	self._run_cmd("destroy '{session_name}'".format(session_name=name))
	474
	475	def destroy_session_by_glob_pattern(self, pattern):
	476	# type: (str) -> None
	477	self._run_cmd("destroy --glob '{pattern}'".format(pattern=pattern))
	478
	479	def destroy_sessions_all(self):
	480	# type: () -> None
	481	self._run_cmd("destroy --all")
	482
	483	@staticmethod
	484	def _mi_find_in_element(element, sub_element_name):
	485	# type: (xml.etree.ElementTree.Element, str) -> xml.etree.ElementTree.Element
	486	result = element.find(LTTngClient._namespaced_mi_element(sub_element_name))
	487	if result is None:
	488	raise InvalidMI(
	489	"Failed to find element '{}' within command MI element '{}'".format(
	490	element.tag, sub_element_name
	491	)
	492	)
	493
	494	return result
	495
	496	def list_sessions(self):
	497	# type () -> List[Session]
	498	list_sessions_xml = self._run_cmd(
	499	"list", LTTngClient.CommandOutputFormat.MI_XML
	500	)
	501
	502	root = xml.etree.ElementTree.fromstring(list_sessions_xml)
	503	command_output = self._mi_find_in_element(root, "output")
	504	sessions = self._mi_find_in_element(command_output, "sessions")
	505
	506	ctl_sessions = [] # type: list[lttngctl.Session]
	507
	508	for session_mi in sessions:
	509	name = self._mi_find_in_element(session_mi, "name").text
510	path = self._mi_find_in_element(session_mi, "path").text
511
512	if name is None:
513	raise InvalidMI(
514	"Invalid empty 'name' element in '{}'".format(session_mi.tag)
515	)
516	if path is None:
517	raise InvalidMI(
518	"Invalid empty 'path' element in '{}'".format(session_mi.tag)
519	)
520	if not path.startswith("/"):
521	raise Unsupported(
522	"{} does not support non-local session outputs".format(type(self))
523	)
524
525	ctl_sessions.append(
526	_Session(self, name, lttngctl.LocalSessionOutputLocation(path))
527	)
528
529	return ctl_sessions