[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):
        # type: (lttngctl.TracingDomain, Optional[str]) -> 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}'".format(
                session_name=self.name,
                domain_name=domain_option_name,
                channel_name=channel_name,
            )
        )
        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
ce8470c9 MJ	253	def add_channel(self, domain, channel_name=None):
ce8470c9 MJ	254	# type: (lttngctl.TracingDomain, Optional[str]) -> lttngctl.Channel
ef945e4d JG	255	channel_name = lttngctl.Channel._generate_name()
	256	domain_option_name = _get_domain_option_name(domain)
	257	self._client._run_cmd(
b9780062 JG	258	"enable-channel --session '{session_name}' --{domain_name} '{channel_name}'".format(
	259	session_name=self.name,
	260	domain_name=domain_option_name,
	261	channel_name=channel_name,
ef945e4d JG	262	)
	263	)
	264	return _Channel(self._client, channel_name, domain, self)
	265
ce8470c9 MJ	266	def add_context(self, context_type):
ce8470c9 MJ	267	# type: (lttngctl.ContextType) -> None
ef945e4d JG	268	pass
	269
	270	@property
ce8470c9 MJ	271	def output(self):
	272	# type: () -> "Optional[Type[lttngctl.SessionOutputLocation]]"
	273	return self._output # type: ignore
ef945e4d	274
ce8470c9 MJ	275	def start(self):
ce8470c9 MJ	276	# type: () -> None
b9780062	277	self._client._run_cmd("start '{session_name}'".format(session_name=self.name))
ef945e4d	278
ce8470c9 MJ	279	def stop(self):
ce8470c9 MJ	280	# type: () -> None
b9780062	281	self._client._run_cmd("stop '{session_name}'".format(session_name=self.name))
ef945e4d	282
ce8470c9 MJ	283	def destroy(self):
ce8470c9 MJ	284	# type: () -> None
b9780062 JG	285	self._client._run_cmd("destroy '{session_name}'".format(session_name=self.name))
	286
	287	@property
	288	def is_active(self):
	289	# type: () -> bool
	290	list_session_xml = self._client._run_cmd(
	291	"list '{session_name}'".format(session_name=self.name),
	292	LTTngClient.CommandOutputFormat.MI_XML,
	293	)
	294
	295	root = xml.etree.ElementTree.fromstring(list_session_xml)
	296	command_output = LTTngClient._mi_find_in_element(root, "output")
	297	sessions = LTTngClient._mi_find_in_element(command_output, "sessions")
	298	session_mi = LTTngClient._mi_find_in_element(sessions, "session")
	299
	300	enabled_text = LTTngClient._mi_find_in_element(session_mi, "enabled").text
	301	if enabled_text not in ["true", "false"]:
	302	raise InvalidMI(
	303	"Expected boolean value in element '{}': value='{}'".format(
	304	session_mi.tag, enabled_text
	305	)
	306	)
	307
	308	return enabled_text == "true"
ef945e4d JG	309
ef945e4d JG	310	@property
ce8470c9 MJ	311	def kernel_pid_process_attribute_tracker(self):
ce8470c9 MJ	312	# type: () -> Type[lttngctl.ProcessIDProcessAttributeTracker]
ef945e4d JG	313	return _ProcessAttributeTracker(self._client, _ProcessAttribute.PID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	314
	315	@property
ce8470c9 MJ	316	def kernel_vpid_process_attribute_tracker(self):
ce8470c9 MJ	317	# type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
ef945e4d JG	318	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	319
	320	@property
ce8470c9 MJ	321	def user_vpid_process_attribute_tracker(self):
ce8470c9 MJ	322	# type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
ef945e4d JG	323	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.User, self) # type: ignore
	324
	325	@property
ce8470c9 MJ	326	def kernel_gid_process_attribute_tracker(self):
ce8470c9 MJ	327	# type: () -> Type[lttngctl.GroupIDProcessAttributeTracker]
ef945e4d JG	328	return _ProcessAttributeTracker(self._client, _ProcessAttribute.GID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	329
	330	@property
ce8470c9 MJ	331	def kernel_vgid_process_attribute_tracker(self):
ce8470c9 MJ	332	# type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
ef945e4d JG	333	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	334
	335	@property
ce8470c9 MJ	336	def user_vgid_process_attribute_tracker(self):
ce8470c9 MJ	337	# type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
ef945e4d JG	338	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.User, self) # type: ignore
	339
	340	@property
ce8470c9 MJ	341	def kernel_uid_process_attribute_tracker(self):
ce8470c9 MJ	342	# type: () -> Type[lttngctl.UserIDProcessAttributeTracker]
ef945e4d JG	343	return _ProcessAttributeTracker(self._client, _ProcessAttribute.UID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	344
	345	@property
ce8470c9 MJ	346	def kernel_vuid_process_attribute_tracker(self):
ce8470c9 MJ	347	# type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
ef945e4d JG	348	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.Kernel, self) # type: ignore
	349
	350	@property
ce8470c9 MJ	351	def user_vuid_process_attribute_tracker(self):
ce8470c9 MJ	352	# type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
ef945e4d JG	353	return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.User, self) # type: ignore
	354
	355
	356	class LTTngClientError(lttngctl.ControlException):
ce8470c9 MJ	357	def __init__(
	358	self,
	359	command_args, # type: str
	360	error_output, # type: str
	361	):
	362	self._command_args = command_args # type: str
	363	self._output = error_output # type: str
ef945e4d JG	364
	365
	366	class LTTngClient(logger._Logger, lttngctl.Controller):
	367	"""
	368	Implementation of a LTTngCtl Controller that uses the `lttng` client as a back-end.
	369	"""
	370
b9780062 JG	371	class CommandOutputFormat(enum.Enum):
	372	MI_XML = 0
	373	HUMAN = 1
	374
	375	_MI_NS = "{https://lttng.org/xml/ns/lttng-mi}"
	376
ef945e4d JG	377	def __init__(
ef945e4d JG	378	self,
ce8470c9 MJ	379	test_environment, # type: environment._Environment
ce8470c9 MJ	380	log, # type: Optional[Callable[[str], None]]
ef945e4d JG	381	):
ef945e4d JG	382	logger._Logger.__init__(self, log)
ce8470c9	383	self._environment = test_environment # type: environment._Environment
ef945e4d	384
b9780062 JG	385	@staticmethod
	386	def _namespaced_mi_element(property):
	387	# type: (str) -> str
	388	return LTTngClient._MI_NS + property
	389
	390	def _run_cmd(self, command_args, output_format=CommandOutputFormat.MI_XML):
	391	# type: (str, CommandOutputFormat) -> str
ef945e4d JG	392	"""
	393	Invoke the `lttng` client with a set of arguments. The command is
	394	executed in the context of the client's test environment.
	395	"""
ce8470c9	396	args = [str(self._environment.lttng_client_path)] # type: list[str]
b9780062 JG	397	if output_format == LTTngClient.CommandOutputFormat.MI_XML:
	398	args.extend(["--mi", "xml"])
	399
ef945e4d JG	400	args.extend(shlex.split(command_args))
	401
	402	self._log("lttng {command_args}".format(command_args=command_args))
	403
ce8470c9	404	client_env = os.environ.copy() # type: dict[str, str]
ef945e4d JG	405	client_env["LTTNG_HOME"] = str(self._environment.lttng_home_location)
	406
	407	process = subprocess.Popen(
	408	args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=client_env
	409	)
	410
	411	out = process.communicate()[0]
	412
	413	if process.returncode != 0:
	414	decoded_output = out.decode("utf-8")
	415	for error_line in decoded_output.splitlines():
	416	self._log(error_line)
	417	raise LTTngClientError(command_args, decoded_output)
b9780062 JG	418	else:
b9780062 JG	419	return out.decode("utf-8")
ef945e4d	420
ce8470c9 MJ	421	def create_session(self, name=None, output=None):
ce8470c9 MJ	422	# type: (Optional[str], Optional[lttngctl.SessionOutputLocation]) -> lttngctl.Session
ef945e4d JG	423	name = name if name else lttngctl.Session._generate_name()
	424
	425	if isinstance(output, lttngctl.LocalSessionOutputLocation):
	426	output_option = "--output {output_path}".format(output_path=output.path)
	427	elif output is None:
	428	output_option = "--no-output"
	429	else:
	430	raise TypeError("LTTngClient only supports local or no output")
	431
	432	self._run_cmd(
b9780062	433	"create '{session_name}' {output_option}".format(
ef945e4d JG	434	session_name=name, output_option=output_option
	435	)
	436	)
	437	return _Session(self, name, output)
b9780062 JG	438
	439	def start_session_by_name(self, name):
	440	# type: (str) -> None
	441	self._run_cmd("start '{session_name}'".format(session_name=name))
	442
	443	def start_session_by_glob_pattern(self, pattern):
	444	# type: (str) -> None
	445	self._run_cmd("start --glob '{pattern}'".format(pattern=pattern))
	446
	447	def start_sessions_all(self):
	448	# type: () -> None
	449	self._run_cmd("start --all")
	450
	451	def stop_session_by_name(self, name):
	452	# type: (str) -> None
	453	self._run_cmd("stop '{session_name}'".format(session_name=name))
	454
	455	def stop_session_by_glob_pattern(self, pattern):
	456	# type: (str) -> None
	457	self._run_cmd("stop --glob '{pattern}'".format(pattern=pattern))
	458
	459	def stop_sessions_all(self):
	460	# type: () -> None
	461	self._run_cmd("stop --all")
	462
	463	def destroy_session_by_name(self, name):
	464	# type: (str) -> None
	465	self._run_cmd("destroy '{session_name}'".format(session_name=name))
	466
	467	def destroy_session_by_glob_pattern(self, pattern):
	468	# type: (str) -> None
	469	self._run_cmd("destroy --glob '{pattern}'".format(pattern=pattern))
	470
	471	def destroy_sessions_all(self):
	472	# type: () -> None
	473	self._run_cmd("destroy --all")
	474
	475	@staticmethod
	476	def _mi_find_in_element(element, sub_element_name):
	477	# type: (xml.etree.ElementTree.Element, str) -> xml.etree.ElementTree.Element
	478	result = element.find(LTTngClient._namespaced_mi_element(sub_element_name))
	479	if result is None:
	480	raise InvalidMI(
	481	"Failed to find element '{}' within command MI element '{}'".format(
	482	element.tag, sub_element_name
	483	)
	484	)
	485
	486	return result
	487
	488	def list_sessions(self):
	489	# type () -> List[Session]
	490	list_sessions_xml = self._run_cmd(
	491	"list", LTTngClient.CommandOutputFormat.MI_XML
	492	)
	493
	494	root = xml.etree.ElementTree.fromstring(list_sessions_xml)
	495	command_output = self._mi_find_in_element(root, "output")
	496	sessions = self._mi_find_in_element(command_output, "sessions")
	497
	498	ctl_sessions = [] # type: list[lttngctl.Session]
	499
	500	for session_mi in sessions:
	501	name = self._mi_find_in_element(session_mi, "name").text
502	path = self._mi_find_in_element(session_mi, "path").text
503
504	if name is None:
505	raise InvalidMI(
506	"Invalid empty 'name' element in '{}'".format(session_mi.tag)
507	)
508	if path is None:
509	raise InvalidMI(
510	"Invalid empty 'path' element in '{}'".format(session_mi.tag)
511	)
512	if not path.startswith("/"):
513	raise Unsupported(
514	"{} does not support non-local session outputs".format(type(self))
515	)
516
517	ctl_sessions.append(
518	_Session(self, name, lttngctl.LocalSessionOutputLocation(path))
519	)
520
521	return ctl_sessions