+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (C) 2015 - Philippe Proulx <pproulx@efficios.com>
-# 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
-from __future__ import print_function
-from __future__ import division
-import lttngust.debug as dbg
-import lttngust.loghandler
-import lttngust.cmd
-from io import open
-import threading
-import logging
-import socket
-import time
-import sys
-import os
-
-
-try:
- # Python 2
- import Queue as queue
-except ImportError:
- # Python 3
- import queue
-
-
-_PROTO_DOMAIN = 5
-_PROTO_MAJOR = 2
-_PROTO_MINOR = 0
-
-
-def _get_env_value_ms(key, default_s):
- try:
- val = int(os.getenv(key, default_s * 1000)) / 1000
- except:
- val = -1
-
- if val < 0:
- fmt = 'invalid ${} value; {} seconds will be used'
- dbg._pwarning(fmt.format(key, default_s))
- val = default_s
-
- return val
-
-
-_REG_TIMEOUT = _get_env_value_ms('LTTNG_UST_PYTHON_REGISTER_TIMEOUT', 5)
-_RETRY_REG_DELAY = _get_env_value_ms('LTTNG_UST_PYTHON_REGISTER_RETRY_DELAY', 3)
-
-
-class _TcpClient(object):
- def __init__(self, name, host, port, reg_queue):
- super(self.__class__, self).__init__()
- self._name = name
- self._host = host
- self._port = port
-
- try:
- self._log_handler = lttngust.loghandler._Handler()
- except (OSError) as e:
- dbg._pwarning('cannot load library: {}'.format(e))
- raise e
-
- self._root_logger = logging.getLogger()
- self._root_logger.setLevel(logging.NOTSET)
- self._ref_count = 0
- self._sessiond_sock = None
- self._reg_queue = reg_queue
- self._server_cmd_handlers = {
- lttngust.cmd._ServerCmdRegistrationDone: self._handle_server_cmd_reg_done,
- lttngust.cmd._ServerCmdEnable: self._handle_server_cmd_enable,
- lttngust.cmd._ServerCmdDisable: self._handle_server_cmd_disable,
- lttngust.cmd._ServerCmdList: self._handle_server_cmd_list,
- }
-
- def _debug(self, msg):
- return 'client "{}": {}'.format(self._name, msg)
-
- def run(self):
- while True:
- try:
- # connect to the session daemon
- dbg._pdebug(self._debug('connecting to session daemon'))
- self._connect_to_sessiond()
-
- # register to the session daemon after a successful connection
- dbg._pdebug(self._debug('registering to session daemon'))
- self._register()
-
- # wait for commands from the session daemon
- self._wait_server_cmd()
- except (Exception) as e:
- # Whatever happens here, we have to close the socket and
- # retry to connect to the session daemon since either
- # the socket was closed, a network timeout occured, or
- # invalid data was received.
- dbg._pdebug(self._debug('got exception: {}'.format(e)))
- self._cleanup_socket()
- dbg._pdebug(self._debug('sleeping for {} s'.format(_RETRY_REG_DELAY)))
- time.sleep(_RETRY_REG_DELAY)
-
- def _recv_server_cmd_header(self):
- data = self._sessiond_sock.recv(lttngust.cmd._SERVER_CMD_HEADER_SIZE)
-
- if not data:
- dbg._pdebug(self._debug('received empty server command header'))
- return None
-
- assert(len(data) == lttngust.cmd._SERVER_CMD_HEADER_SIZE)
- dbg._pdebug(self._debug('received server command header ({} bytes)'.format(len(data))))
-
- return lttngust.cmd._server_cmd_header_from_data(data)
-
- def _recv_server_cmd(self):
- server_cmd_header = self._recv_server_cmd_header()
-
- if server_cmd_header is None:
- return None
-
- dbg._pdebug(self._debug('server command header: data size: {} bytes'.format(server_cmd_header.data_size)))
- dbg._pdebug(self._debug('server command header: command ID: {}'.format(server_cmd_header.cmd_id)))
- dbg._pdebug(self._debug('server command header: command version: {}'.format(server_cmd_header.cmd_version)))
- data = bytes()
-
- if server_cmd_header.data_size > 0:
- data = self._sessiond_sock.recv(server_cmd_header.data_size)
- assert(len(data) == server_cmd_header.data_size)
-
- return lttngust.cmd._server_cmd_from_data(server_cmd_header, data)
-
- def _send_cmd_reply(self, cmd_reply):
- data = cmd_reply.get_data()
- dbg._pdebug(self._debug('sending command reply ({} bytes)'.format(len(data))))
- self._sessiond_sock.sendall(data)
-
- def _handle_server_cmd_reg_done(self, server_cmd):
- dbg._pdebug(self._debug('got "registration done" server command'))
-
- if self._reg_queue is not None:
- dbg._pdebug(self._debug('notifying _init_threads()'))
-
- try:
- self._reg_queue.put(True)
- except (Exception) as e:
- # read side could be closed by now; ignore it
- pass
-
- self._reg_queue = None
-
- def _handle_server_cmd_enable(self, server_cmd):
- dbg._pdebug(self._debug('got "enable" server command'))
- self._ref_count += 1
-
- if self._ref_count == 1:
- dbg._pdebug(self._debug('adding our handler to the root logger'))
- self._root_logger.addHandler(self._log_handler)
-
- dbg._pdebug(self._debug('ref count is {}'.format(self._ref_count)))
-
- return lttngust.cmd._ClientCmdReplyEnable()
-
- def _handle_server_cmd_disable(self, server_cmd):
- dbg._pdebug(self._debug('got "disable" server command'))
- self._ref_count -= 1
-
- if self._ref_count < 0:
- # disable command could be sent again when a session is destroyed
- self._ref_count = 0
-
- if self._ref_count == 0:
- dbg._pdebug(self._debug('removing our handler from the root logger'))
- self._root_logger.removeHandler(self._log_handler)
-
- dbg._pdebug(self._debug('ref count is {}'.format(self._ref_count)))
-
- return lttngust.cmd._ClientCmdReplyDisable()
-
- def _handle_server_cmd_list(self, server_cmd):
- dbg._pdebug(self._debug('got "list" server command'))
- names = logging.Logger.manager.loggerDict.keys()
- dbg._pdebug(self._debug('found {} loggers'.format(len(names))))
- cmd_reply = lttngust.cmd._ClientCmdReplyList(names=names)
-
- return cmd_reply
-
- def _handle_server_cmd(self, server_cmd):
- cmd_reply = None
-
- if server_cmd is None:
- dbg._pdebug(self._debug('bad server command'))
- status = lttngust.cmd._CLIENT_CMD_REPLY_STATUS_INVALID_CMD
- cmd_reply = lttngust.cmd._ClientCmdReply(status)
- elif type(server_cmd) in self._server_cmd_handlers:
- cmd_reply = self._server_cmd_handlers[type(server_cmd)](server_cmd)
- else:
- dbg._pdebug(self._debug('unknown server command'))
- status = lttngust.cmd._CLIENT_CMD_REPLY_STATUS_INVALID_CMD
- cmd_reply = lttngust.cmd._ClientCmdReply(status)
-
- if cmd_reply is not None:
- self._send_cmd_reply(cmd_reply)
-
- def _wait_server_cmd(self):
- while True:
- try:
- server_cmd = self._recv_server_cmd()
- except socket.timeout:
- # simply retry here; the protocol has no KA and we could
- # wait for hours
- continue
-
- self._handle_server_cmd(server_cmd)
-
- def _cleanup_socket(self):
- try:
- self._sessiond_sock.shutdown(socket.SHUT_RDWR)
- self._sessiond_sock.close()
- except:
- pass
-
- self._sessiond_sock = None
-
- def _connect_to_sessiond(self):
- # create session daemon TCP socket
- if self._sessiond_sock is None:
- self._sessiond_sock = socket.socket(socket.AF_INET,
- socket.SOCK_STREAM)
-
- # Use str(self._host) here. Since this host could be a string
- # literal, and since we're importing __future__.unicode_literals,
- # we want to make sure the host is a native string in Python 2.
- # This avoids an indirect module import (unicode module to
- # decode the unicode string, eventually imported by the
- # socket module if needed), which is not allowed in a thread
- # directly created by a module in Python 2 (our case).
- #
- # tl;dr: Do NOT remove str() here, or this call in Python 2
- # _will_ block on an interpreter's mutex until the waiting
- # register queue timeouts.
- self._sessiond_sock.connect((str(self._host), self._port))
-
- def _register(self):
- cmd = lttngust.cmd._ClientRegisterCmd(_PROTO_DOMAIN, os.getpid(),
- _PROTO_MAJOR, _PROTO_MINOR)
- data = cmd.get_data()
- self._sessiond_sock.sendall(data)
-
-
-def _get_port_from_file(path):
- port = None
- dbg._pdebug('reading port from file "{}"'.format(path))
-
- try:
- f = open(path)
- r_port = int(f.readline())
- f.close()
-
- if r_port > 0 or r_port <= 65535:
- port = r_port
- except:
- pass
-
- return port
-
-
-def _get_user_home_path():
- # $LTTNG_HOME overrides $HOME if it exists
- return os.getenv('LTTNG_HOME', os.path.expanduser('~'))
-
-
-_initialized = False
-_SESSIOND_HOST = '127.0.0.1'
-
-
-def _client_thread_target(name, port, reg_queue):
- dbg._pdebug('creating client "{}" using TCP port {}'.format(name, port))
- client = _TcpClient(name, _SESSIOND_HOST, port, reg_queue)
- dbg._pdebug('starting client "{}"'.format(name))
- client.run()
-
-
-def _init_threads():
- global _initialized
-
- dbg._pdebug('entering')
-
- if _initialized:
- dbg._pdebug('agent is already initialized')
- return
-
- # This makes sure that the appropriate modules for encoding and
- # decoding strings/bytes are imported now, since no import should
- # happen within a thread at import time (our case).
- 'lttng'.encode().decode()
-
- _initialized = True
- sys_port = _get_port_from_file('/var/run/lttng/agent.port')
- user_port_file = os.path.join(_get_user_home_path(), '.lttng', 'agent.port')
- user_port = _get_port_from_file(user_port_file)
- reg_queue = queue.Queue()
- reg_expecting = 0
-
- dbg._pdebug('system session daemon port: {}'.format(sys_port))
- dbg._pdebug('user session daemon port: {}'.format(user_port))
-
- if sys_port == user_port and sys_port is not None:
- # The two session daemon ports are the same. This is not normal.
- # Connect to only one.
- dbg._pdebug('both user and system session daemon have the same port')
- sys_port = None
-
- try:
- if sys_port is not None:
- dbg._pdebug('creating system client thread')
- t = threading.Thread(target=_client_thread_target,
- args=('system', sys_port, reg_queue))
- t.name = 'system'
- t.daemon = True
- t.start()
- dbg._pdebug('created and started system client thread')
- reg_expecting += 1
-
- if user_port is not None:
- dbg._pdebug('creating user client thread')
- t = threading.Thread(target=_client_thread_target,
- args=('user', user_port, reg_queue))
- t.name = 'user'
- t.daemon = True
- t.start()
- dbg._pdebug('created and started user client thread')
- reg_expecting += 1
- except:
- # cannot create threads for some reason; stop this initialization
- dbg._pwarning('cannot create client threads')
- return
-
- if reg_expecting == 0:
- # early exit: looks like there's not even one valid port
- dbg._pwarning('no valid LTTng session daemon port found (is the session daemon started?)')
- return
-
- cur_timeout = _REG_TIMEOUT
-
- # We block here to make sure the agent is properly registered to
- # the session daemon. If we timeout, the client threads will still
- # continue to try to connect and register to the session daemon,
- # but there is no guarantee that all following logging statements
- # will make it to LTTng-UST.
- #
- # When a client thread receives a "registration done" confirmation
- # from the session daemon it's connected to, it puts True in
- # reg_queue.
- while True:
- try:
- dbg._pdebug('waiting for registration done (expecting {}, timeout is {} s)'.format(reg_expecting,
- cur_timeout))
- t1 = time.clock()
- reg_queue.get(timeout=cur_timeout)
- t2 = time.clock()
- reg_expecting -= 1
- dbg._pdebug('unblocked')
-
- if reg_expecting == 0:
- # done!
- dbg._pdebug('successfully registered to session daemon(s)')
- break
-
- cur_timeout -= (t2 - t1)
-
- if cur_timeout <= 0:
- # timeout
- dbg._pdebug('ran out of time')
- break
- except queue.Empty:
- dbg._pdebug('ran out of time')
- break
-
- dbg._pdebug('leaving')
-
-
-_init_threads()
projects / deliverable / lttng-ust.git / blobdiff