+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (C) 2014 - David Goulet <dgoulet@efficios.com>
-#
-# This library is free software; you can redistribute it and/or modify it under
-# the terms of the GNU Lesser General Public License as published by the Free
-# Software Foundation; version 2.1 of the License.
-#
-# This library is distributed in the hope that it will be useful, but WITHOUT
-# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
-# details.
-#
-# You should have received a copy of the GNU Lesser General Public License
-# along with this library; if not, write to the Free Software Foundation, Inc.,
-# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
-
-from __future__ import unicode_literals
-
-import ctypes
-import errno
-import logging
-import os
-import sys
-import threading
-import struct
-import select
-
-from select import epoll, EPOLLIN, EPOLLERR, EPOLLHUP
-from socket import *
-from time import sleep
-
-__all__ = ["lttng-agent"]
-__author__ = "David Goulet <dgoulet@efficios.com>"
-
-class LTTngAgent():
- """
- LTTng agent python code. A LTTng Agent is responsible to spawn two threads,
- the current UID and root session daemon. Those two threads register to the
- right daemon and handle the tracing.
-
- This class needs to be instantiate once and once the init returns, tracing
- is ready to happen.
- """
-
- SESSIOND_ADDR = "127.0.0.1"
- SEM_COUNT = 2
- # Timeout for the sempahore in seconds.
- SEM_TIMEOUT = 5
- SEM_WAIT_PERIOD = 0.2
-
- def __init__(self):
- # Session daemon register semaphore.
- self.register_sem = threading.Semaphore(LTTngAgent.SEM_COUNT);
-
- self.client_user = LTTngTCPClient(LTTngAgent.SESSIOND_ADDR, self.register_sem)
- self.client_user.start()
-
- self.client_root = LTTngTCPClient(LTTngAgent.SESSIOND_ADDR, self.register_sem)
- self.client_root.log_handler.is_root = True
- self.client_root.start()
-
- acquire = 0
- timeout = LTTngAgent.SEM_TIMEOUT
- while True:
- # Quit if timeout has reached 0 or below.
- if acquire == LTTngAgent.SEM_COUNT or timeout <= 0:
- break;
-
- # Acquire semaphore for *user* thread.
- if not self.register_sem.acquire(False):
- sleep(LTTngAgent.SEM_WAIT_PERIOD)
- timeout -= LTTngAgent.SEM_WAIT_PERIOD
- else:
- acquire += 1
-
- def __del__(self):
- self.destroy()
-
- def destroy(self):
- self.client_user.destroy()
- self.client_user.join()
-
- self.client_root.destroy()
- self.client_root.join()
-
-class LTTngCmdError(RuntimeError):
- """
- Command error thrown if an error is encountered in a command from the
- session daemon.
- """
-
- def __init__(self, code):
- super().__init__('LTTng command error: code {}'.format(code))
- self._code = code
-
- def get_code(self):
- return self._code
-
-class LTTngUnknownCmdError(RuntimeError):
- pass
-
-class LTTngLoggingHandler(logging.Handler):
- """
- Class handler for the Python logging API.
- """
-
- def __init__(self):
- logging.Handler.__init__(self, level = logging.NOTSET)
-
- # Refcount tracking how many events have been enabled. This value above
- # 0 means that the handler is attached to the root logger.
- self.refcount = 0
-
- # Dict of enabled event. We track them so we know if it's ok to disable
- # the received event.
- self.enabled_events = {}
-
- # Am I root ?
- self.is_root = False
-
- # Using the logging formatter to extract the asctime only.
- self.log_fmt = logging.Formatter("%(asctime)s")
- self.setFormatter(self.log_fmt)
-
- # ctypes lib for lttng-ust
- try:
- self.lttng_ust = ctypes.cdll.LoadLibrary("LIBDIR_STR/liblttng-ust-python-agent.so")
- except OSError as e:
- print("Unable to find libust for Python.")
-
- def emit(self, record):
- """
- Fire LTTng UST tracepoint with the given record.
- """
- asctime = self.format(record)
-
- self.lttng_ust.py_tracepoint(asctime.encode(),
- record.getMessage().encode(), record.name.encode(),
- record.funcName.encode(), record.lineno, record.levelno,
- record.thread, record.threadName.encode())
-
- def enable_event(self, name):
- """
- Enable an event name which will ultimately add an handler to the root
- logger if none is present.
- """
- # Don't update the refcount if the event has been enabled prior.
- if name in self.enabled_events:
- return
-
- # Get the root logger and attach our handler.
- root_logger = logging.getLogger()
- # First thing first, we need to set the root logger to the loglevel
- # NOTSET so we can catch everything. The default is 30.
- root_logger.setLevel(logging.NOTSET)
-
- self.refcount += 1
- if self.refcount == 1:
- root_logger.addHandler(self)
-
- self.enabled_events[name] = True
-
- def disable_event(self, name):
- """
- Disable an event name which will ultimately add an handler to the root
- logger if none is present.
- """
-
- if name not in self.enabled_events:
- # Event was not enabled prior, do nothing.
- return
-
- # Get the root logger and attach our handler.
- root_logger = logging.getLogger()
-
- self.refcount -= 1
- if self.refcount == 0:
- root_logger.removeHandler(self)
- del self.enabled_events[name]
-
- def list_logger(self):
- """
- Return a list of logger name.
- """
- return logging.Logger.manager.loggerDict.keys()
-
-class LTTngSessiondCmd():
- """
- Class handling session daemon command.
- """
-
- # Command values from the agent protocol
- CMD_LIST = 1
- CMD_ENABLE = 2
- CMD_DISABLE = 3
- CMD_REG_DONE = 4
-
- # Return code
- CODE_SUCCESS = 1
- CODE_INVALID_CMD = 2
-
- # Python Logger LTTng domain value taken from lttng/domain.h
- DOMAIN = 5
-
- # Protocol version
- MAJOR_VERSION = 1
- MINOR_VERSION = 0
-
- def execute(self):
- """
- This is part of the command interface. Must be implemented.
- """
- raise NotImplementedError
-
-class LTTngCommandReply():
- """
- Object that contains the information that should be replied to the session
- daemon after a command execution.
- """
-
- def __init__(self, payload = None, reply = True):
- self.payload = payload
- self.reply = reply
-
-class LTTngCommandEnable(LTTngSessiondCmd):
- """
- Handle the enable event command from the session daemon.
- """
-
- def __init__(self, log_handler, data):
- self.log_handler = log_handler
- # 4 bytes for loglevel and 4 bytes for loglevel_type thus 8.
- name_offset = 8;
-
- data_size = len(data)
- if data_size == 0:
- raise LTTngCmdError(LTTngSessiondCmd.CODE_INVALID_CMD)
-
- try:
- self.loglevel, self.loglevel_type, self.name = \
- struct.unpack('>II%us' % (data_size - name_offset), data)
- # Remove trailing NULL bytes from name.
- self.name = self.name.decode().rstrip('\x00')
- except struct.error:
- raise LTTngCmdError(LTTngSessiondCmd.CODE_INVALID_CMD)
-
- def execute(self):
- self.log_handler.enable_event(self.name)
- return LTTngCommandReply()
-
-class LTTngCommandDisable(LTTngSessiondCmd):
- """
- Handle the disable event command from the session daemon.
- """
-
- def __init__(self, log_handler, data):
- self.log_handler = log_handler
-
- data_size = len(data)
- if data_size == 0:
- raise LTTngCmdError(LTTngSessiondCmd.CODE_INVALID_CMD)
-
- try:
- self.name = struct.unpack('>%us' % (data_size), data)[0]
- # Remove trailing NULL bytes from name.
- self.name = self.name.decode().rstrip('\x00')
- except struct.error:
- raise LTTngCmdError(LTTngSessiondCmd.CODE_INVALID_CMD)
-
- def execute(self):
- self.log_handler.disable_event(self.name)
- return LTTngCommandReply()
-
-class LTTngCommandRegDone(LTTngSessiondCmd):
- """
- Handle register done command. This is sent back after a successful
- registration from the session daemon. We basically release the given
- semaphore so the agent can return to the caller.
- """
-
- def __init__(self, sem):
- self.sem = sem
-
- def execute(self):
- self.sem.release()
- return LTTngCommandReply(reply = False)
-
-class LTTngCommandList(LTTngSessiondCmd):
- """
- Handle the list command from the session daemon on the given socket.
- """
-
- def __init__(self, log_handler):
- self.log_handler = log_handler
-
- def execute(self):
- data_size = 0
- data = logger_data = bytearray()
-
- loggers = self.log_handler.list_logger()
- # First, pack nb_event that must preceed the data.
- logger_data += struct.pack('>I', len(loggers))
-
- # Populate payload with logger name.
- for logger in loggers:
- # Increment data size plus the NULL byte at the end of the name.
- data_size += len(logger) + 1
- # Pack logger name and NULL byte.
- logger_data += struct.pack('>%usB' % (len(logger)), \
- bytes(bytearray(str.encode(logger))), 0)
-
- # Pack uint32_t data_size followed by nb event (number of logger)
- data = struct.pack('>I', data_size)
- data += logger_data
- return LTTngCommandReply(payload = data)
-
-class LTTngTCPClient(threading.Thread):
- """
- TCP client that register and receives command from the session daemon.
- """
-
- SYSTEM_PORT_FILE = "/var/run/lttng/agent.port"
- USER_PORT_FILE = os.path.join(os.path.expanduser("~"), ".lttng/agent.port")
-
- # The time in seconds this client should wait before trying again to
- # register back to the session daemon.
- WAIT_TIME = 3
-
- def __init__(self, host, sem):
- threading.Thread.__init__(self)
-
- # Which host to connect to. The port is fetch dynamically.
- self.sessiond_host = host
-
- # The session daemon register done semaphore. Needs to be released when
- # receiving a CMD_REG_DONE command.
- self.register_sem = sem
- self.register_sem.acquire()
-
- # Indicate that we have to quit thus stop the main loop.
- self.quit_flag = False
- # Quit pipe. The thread poll on it to know when to quit.
- self.quit_pipe = os.pipe()
-
- # Socket on which we communicate with the session daemon.
- self.sessiond_sock = None
- # LTTng Logging Handler
- self.log_handler = LTTngLoggingHandler()
-
- def cleanup_socket(self, epfd = None):
- # Ease our life a bit.
- sock = self.sessiond_sock
- if not sock:
- return
-
- try:
- if epfd is not None:
- epfd.unregister(sock)
- sock.shutdown(SHUT_RDWR)
- sock.close()
- except select.error:
- # Cleanup fail, we can't do anything much...
- pass
- except IOError:
- pass
-
- self.sessiond_sock = None
-
- def destroy(self):
- self.quit_flag = True
- try:
- fp = os.fdopen(self.quit_pipe[1], 'w')
- fp.write("42")
- fp.close()
- except OSError as e:
- pass
-
- def register(self):
- """
- Register to session daemon using the previously connected socket of the
- class.
-
- Command ABI:
- uint32 domain
- uint32 pid
- """
- data = struct.pack('>IIII', LTTngSessiondCmd.DOMAIN, os.getpid(), \
- LTTngSessiondCmd.MAJOR_VERSION, LTTngSessiondCmd.MINOR_VERSION)
- self.sessiond_sock.send(data)
-
- def run(self):
- """
- Start the TCP client thread by registering to the session daemon and polling
- on that socket for commands.
- """
-
- epfd = epoll()
- epfd.register(self.quit_pipe[0], EPOLLIN)
-
- # Main loop to handle session daemon command and disconnection.
- while not self.quit_flag:
- try:
- # First, connect to the session daemon.
- self.connect_sessiond()
-
- # Register to session daemon after a successful connection.
- self.register()
- # Add registered socket to poll set.
- epfd.register(self.sessiond_sock, EPOLLIN | EPOLLERR | EPOLLHUP)
-
- self.quit_flag = self.wait_cmd(epfd)
- except IOError as e:
- # Whatever happens here, we have to close down everything and
- # retry to connect to the session daemon since either the
- # socket is closed or invalid data was sent.
- self.cleanup_socket(epfd)
- self.register_sem.release()
- sleep(LTTngTCPClient.WAIT_TIME)
- continue
-
- self.cleanup_socket(epfd)
- os.close(self.quit_pipe[0])
- epfd.close()
-
- def recv_header(self, sock):
- """
- Receive the command header from the given socket. Set the internal
- state of this object with the header data.
-
- Header ABI is defined like this:
- uint64 data_size
- uint32 cmd
- uint32 cmd_version
- """
- s_pack = struct.Struct('>QII')
-
- pack_data = sock.recv(s_pack.size)
- data_received = len(pack_data)
- if data_received == 0:
- raise IOError(errno.ESHUTDOWN)
-
- try:
- return s_pack.unpack(pack_data)
- except struct.error:
- raise IOError(errno.EINVAL)
-
- def create_command(self, cmd_type, version, data):
- """
- Return the right command object using the given command type. The
- command version is unused since we only have once for now.
- """
-
- cmd_dict = {
- LTTngSessiondCmd.CMD_LIST: \
- lambda: LTTngCommandList(self.log_handler),
- LTTngSessiondCmd.CMD_ENABLE: \
- lambda: LTTngCommandEnable(self.log_handler, data),
- LTTngSessiondCmd.CMD_DISABLE: \
- lambda: LTTngCommandDisable(self.log_handler, data),
- LTTngSessiondCmd.CMD_REG_DONE: \
- lambda: LTTngCommandRegDone(self.register_sem),
- }
-
- if cmd_type in cmd_dict:
- return cmd_dict[cmd_type]()
- else:
- raise LTTngUnknownCmdError()
-
- def pack_code(self, code):
- return struct.pack('>I', code)
-
- def handle_command(self, data, cmd_type, cmd_version):
- """
- Handle the given command type with the received payload. This function
- sends back data to the session daemon using to the return value of the
- command.
- """
- payload = bytearray()
-
- try:
- cmd = self.create_command(cmd_type, cmd_version, data)
- cmd_reply = cmd.execute()
- # Set success code in data
- payload += self.pack_code(LTTngSessiondCmd.CODE_SUCCESS)
- if cmd_reply.payload is not None:
- payload += cmd_reply.payload
- except LTTngCmdError as e:
- # Set error code in payload
- payload += self.pack_code(e.get_code())
- except LTTngUnknownCmdError:
- # Set error code in payload
- payload += self.pack_code(LTTngSessiondCmd.CODE_INVALID_CMD)
-
- # Send response only if asked for.
- if cmd_reply.reply:
- self.sessiond_sock.send(payload)
-
- def wait_cmd(self, epfd):
- """
- """
-
- while True:
- try:
- # Poll on socket for command.
- events = epfd.poll()
- except select.error as e:
- raise IOError(e.errno, e.message)
-
- for fileno, event in events:
- if fileno == self.quit_pipe[0]:
- return True
- elif event & (EPOLLERR | EPOLLHUP):
- raise IOError(errno.ESHUTDOWN)
- elif event & EPOLLIN:
- data = bytearray()
-
- data_size, cmd, cmd_version = self.recv_header(self.sessiond_sock)
- if data_size:
- data += self.sessiond_sock.recv(data_size)
-
- self.handle_command(data, cmd, cmd_version)
- else:
- raise IOError(errno.ESHUTDOWN)
-
- def get_port_from_file(self, path):
- """
- Open the session daemon agent port file and returns the value. If none
- found, 0 is returned.
- """
-
- # By default, the port is set to 0 so if we can not find the agent port
- # file simply don't try to connect. A value set to 0 indicates that.
- port = 0
-
- try:
- f = open(path, "r")
- r_port = int(f.readline())
- if r_port > 0 or r_port <= 65535:
- port = r_port
- f.close()
- except IOError as e:
- pass
- except ValueError as e:
- pass
-
- return port
-
- def connect_sessiond(self):
- """
- Connect sessiond_sock to running session daemon using the port file.
- """
- # Create session daemon TCP socket
- if not self.sessiond_sock:
- self.sessiond_sock = socket(AF_INET, SOCK_STREAM)
-
- if self.log_handler.is_root:
- port = self.get_port_from_file(LTTngTCPClient.SYSTEM_PORT_FILE)
- else:
- port = self.get_port_from_file(LTTngTCPClient.USER_PORT_FILE)
-
- # No session daemon available
- if port == 0:
- raise IOError(errno.ECONNREFUSED)
-
- # Can raise an IOError so caller must catch it.
- self.sessiond_sock.connect((self.sessiond_host, port))
projects / lttng-ust.git / blobdiff