neutron/agent/linux/utils.py

   1 # Copyright 2012 Locaweb.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import fcntl
  17 import glob
  18 import grp
  19 import os
  20 import pwd
  21 import shlex
  22 import socket
  23 import struct
  24 import tempfile
  25 import threading
  26
  27 import debtcollector
  28 import eventlet
  29 from eventlet.green import subprocess
  30 from eventlet import greenthread
  31 from oslo_config import cfg
  32 from oslo_log import log as logging
  33 from oslo_rootwrap import client
  34 from oslo_utils import excutils
  35 import six
  36 from six.moves import http_client as httplib
  37
  38 from neutron._i18n import _, _LE
  39 from neutron.agent.common import config
  40 from neutron.common import constants
  41 from neutron.common import utils
  42 from neutron import wsgi
  43
  44
  45 LOG = logging.getLogger(__name__)
  46
  47
  48 class RootwrapDaemonHelper(object):
  49     __client = None
  50     __lock = threading.Lock()
  51
  52     def __new__(cls):
  53         """There is no reason to instantiate this class"""
  54         raise NotImplementedError()
  55
  56     @classmethod
  57     def get_client(cls):
  58         with cls.__lock:
  59             if cls.__client is None:
  60                 cls.__client = client.Client(
  61                     shlex.split(cfg.CONF.AGENT.root_helper_daemon))
  62             return cls.__client
  63
  64
  65 def addl_env_args(addl_env):
  66     """Build arugments for adding additional environment vars with env"""
  67
  68     # NOTE (twilson) If using rootwrap, an EnvFilter should be set up for the
  69     # command instead of a CommandFilter.
  70     if addl_env is None:
  71         return []
  72     return ['env'] + ['%s=%s' % pair for pair in addl_env.items()]
  73
  74
  75 def create_process(cmd, run_as_root=False, addl_env=None):
  76     """Create a process object for the given command.
  77
  78     The return value will be a tuple of the process object and the
  79     list of command arguments used to create it.
  80     """
  81     cmd = list(map(str, addl_env_args(addl_env) + cmd))
  82     if run_as_root:
  83         cmd = shlex.split(config.get_root_helper(cfg.CONF)) + cmd
  84     LOG.debug("Running command: %s", cmd)
  85     obj = utils.subprocess_popen(cmd, shell=False,
  86                                  stdin=subprocess.PIPE,
  87                                  stdout=subprocess.PIPE,
  88                                  stderr=subprocess.PIPE)
  89
  90     return obj, cmd
  91
  92
  93 def execute_rootwrap_daemon(cmd, process_input, addl_env):
  94     cmd = list(map(str, addl_env_args(addl_env) + cmd))
  95     # NOTE(twilson) oslo_rootwrap.daemon will raise on filter match
  96     # errors, whereas oslo_rootwrap.cmd converts them to return codes.
  97     # In practice, no neutron code should be trying to execute something that
  98     # would throw those errors, and if it does it should be fixed as opposed to
  99     # just logging the execution error.
 100     LOG.debug("Running command (rootwrap daemon): %s", cmd)
 101     client = RootwrapDaemonHelper.get_client()
 102     return client.execute(cmd, process_input)
 103
 104
 105 def execute(cmd, process_input=None, addl_env=None,
 106             check_exit_code=True, return_stderr=False, log_fail_as_error=True,
 107             extra_ok_codes=None, run_as_root=False):
 108     try:
 109         if (process_input is None or
 110             isinstance(process_input, six.binary_type)):
 111             _process_input = process_input
 112         else:
 113             _process_input = process_input.encode('utf-8')
 114         if run_as_root and cfg.CONF.AGENT.root_helper_daemon:
 115             returncode, _stdout, _stderr = (
 116                 execute_rootwrap_daemon(cmd, process_input, addl_env))
 117         else:
 118             obj, cmd = create_process(cmd, run_as_root=run_as_root,
 119                                       addl_env=addl_env)
 120             _stdout, _stderr = obj.communicate(_process_input)
 121             returncode = obj.returncode
 122             obj.stdin.close()
 123         _stdout = utils.safe_decode_utf8(_stdout)
 124         _stderr = utils.safe_decode_utf8(_stderr)
 125
 126         extra_ok_codes = extra_ok_codes or []
 127         if returncode and returncode not in extra_ok_codes:
 128             msg = _("Exit code: %(returncode)d; "
 129                     "Stdin: %(stdin)s; "
 130                     "Stdout: %(stdout)s; "
 131                     "Stderr: %(stderr)s") % {
 132                         'returncode': returncode,
 133                         'stdin': process_input or '',
 134                         'stdout': _stdout,
 135                         'stderr': _stderr}
 136
 137             if log_fail_as_error:
 138                 LOG.error(msg)
 139             if check_exit_code:
 140                 raise RuntimeError(msg)
 141         else:
 142             LOG.debug("Exit code: %d", returncode)
 143
 144     finally:
 145         # NOTE(termie): this appears to be necessary to let the subprocess
 146         #               call clean something up in between calls, without
 147         #               it two execute calls in a row hangs the second one
 148         greenthread.sleep(0)
 149
 150     return (_stdout, _stderr) if return_stderr else _stdout
 151
 152
 153 def get_interface_mac(interface):
 154     MAC_START = 18
 155     MAC_END = 24
 156     s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 157     dev = interface[:constants.DEVICE_NAME_MAX_LEN]
 158     if isinstance(dev, six.text_type):
 159         dev = dev.encode('utf-8')
 160     info = fcntl.ioctl(s.fileno(), 0x8927, struct.pack('256s', dev))
 161     return ''.join(['%02x:' % ord(char)
 162                     for char in info[MAC_START:MAC_END]])[:-1]
 163
 164
 165 @debtcollector.removals.remove(message="Redundant in Mitaka release.")
 166 def replace_file(file_name, data, file_mode=0o644):
 167     """Replaces the contents of file_name with data in a safe manner.
 168
 169     First write to a temp file and then rename. Since POSIX renames are
 170     atomic, the file is unlikely to be corrupted by competing writes.
 171
 172     We create the tempfile on the same device to ensure that it can be renamed.
 173     """
 174
 175     base_dir = os.path.dirname(os.path.abspath(file_name))
 176     tmp_file = tempfile.NamedTemporaryFile('w+', dir=base_dir, delete=False)
 177     tmp_file.write(data)
 178     tmp_file.close()
 179     os.chmod(tmp_file.name, file_mode)
 180     os.rename(tmp_file.name, file_name)
 181
 182
 183 def find_child_pids(pid):
 184     """Retrieve a list of the pids of child processes of the given pid."""
 185
 186     try:
 187         raw_pids = execute(['ps', '--ppid', pid, '-o', 'pid='],
 188                            log_fail_as_error=False)
 189     except RuntimeError as e:
 190         # Unexpected errors are the responsibility of the caller
 191         with excutils.save_and_reraise_exception() as ctxt:
 192             # Exception has already been logged by execute
 193             no_children_found = 'Exit code: 1' in str(e)
 194             if no_children_found:
 195                 ctxt.reraise = False
 196                 return []
 197     return [x.strip() for x in raw_pids.split('\n') if x.strip()]
 198
 199
 200 def _get_conf_base(cfg_root, uuid, ensure_conf_dir):
 201     #TODO(mangelajo): separate responsibilities here, ensure_conf_dir
 202     #                 should be a separate function
 203     conf_dir = os.path.abspath(os.path.normpath(cfg_root))
 204     conf_base = os.path.join(conf_dir, uuid)
 205     if ensure_conf_dir:
 206         utils.ensure_dir(conf_dir)
 207     return conf_base
 208
 209
 210 def get_conf_file_name(cfg_root, uuid, cfg_file, ensure_conf_dir=False):
 211     """Returns the file name for a given kind of config file."""
 212     conf_base = _get_conf_base(cfg_root, uuid, ensure_conf_dir)
 213     return "%s.%s" % (conf_base, cfg_file)
 214
 215
 216 def get_value_from_file(filename, converter=None):
 217
 218     try:
 219         with open(filename, 'r') as f:
 220             try:
 221                 return converter(f.read()) if converter else f.read()
 222             except ValueError:
 223                 LOG.error(_LE('Unable to convert value in %s'), filename)
 224     except IOError:
 225         LOG.debug('Unable to access %s', filename)
 226
 227
 228 def get_value_from_conf_file(cfg_root, uuid, cfg_file, converter=None):
 229     """A helper function to read a value from one of a config file."""
 230     file_name = get_conf_file_name(cfg_root, uuid, cfg_file)
 231     return get_value_from_file(file_name, converter)
 232
 233
 234 def remove_conf_files(cfg_root, uuid):
 235     conf_base = _get_conf_base(cfg_root, uuid, False)
 236     for file_path in glob.iglob("%s.*" % conf_base):
 237         os.unlink(file_path)
 238
 239
 240 def get_root_helper_child_pid(pid, run_as_root=False):
 241     """
 242     Get the lowest child pid in the process hierarchy
 243
 244     If root helper was used, two or more processes would be created:
 245
 246      - a root helper process (e.g. sudo myscript)
 247      - possibly a rootwrap script (e.g. neutron-rootwrap)
 248      - a child process (e.g. myscript)
 249
 250     Killing the root helper process will leave the child process
 251     running, re-parented to init, so the only way to ensure that both
 252     die is to target the child process directly.
 253     """
 254     pid = str(pid)
 255     if run_as_root:
 256         try:
 257             pid = find_child_pids(pid)[0]
 258         except IndexError:
 259             # Process is already dead
 260             return None
 261         while True:
 262             try:
 263                 # We shouldn't have more than one child per process
 264                 # so keep getting the children of the first one
 265                 pid = find_child_pids(pid)[0]
 266             except IndexError:
 267                 # Last process in the tree, return it
 268                 break
 269     return pid
 270
 271
 272 def remove_abs_path(cmd):
 273     """Remove absolute path of executable in cmd
 274
 275     Note: New instance of list is returned
 276
 277     :param cmd: parsed shlex command (e.g. ['/bin/foo', 'param1', 'param two'])
 278
 279     """
 280     if cmd and os.path.isabs(cmd[0]):
 281         cmd = list(cmd)
 282         cmd[0] = os.path.basename(cmd[0])
 283
 284     return cmd
 285
 286
 287 def get_cmdline_from_pid(pid):
 288     if pid is None or not os.path.exists('/proc/%s' % pid):
 289         return []
 290     with open('/proc/%s/cmdline' % pid, 'r') as f:
 291         return f.readline().split('\0')[:-1]
 292
 293
 294 def cmd_matches_expected(cmd, expected_cmd):
 295     abs_cmd = remove_abs_path(cmd)
 296     abs_expected_cmd = remove_abs_path(expected_cmd)
 297     if abs_cmd != abs_expected_cmd:
 298         # Commands executed with #! are prefixed with the script
 299         # executable. Check for the expected cmd being a subset of the
 300         # actual cmd to cover this possibility.
 301         abs_cmd = remove_abs_path(abs_cmd[1:])
 302     return abs_cmd == abs_expected_cmd
 303
 304
 305 def pid_invoked_with_cmdline(pid, expected_cmd):
 306     """Validate process with given pid is running with provided parameters
 307
 308     """
 309     cmd = get_cmdline_from_pid(pid)
 310     return cmd_matches_expected(cmd, expected_cmd)
 311
 312
 313 def wait_until_true(predicate, timeout=60, sleep=1, exception=None):
 314     """
 315     Wait until callable predicate is evaluated as True
 316
 317     :param predicate: Callable deciding whether waiting should continue.
 318     Best practice is to instantiate predicate with functools.partial()
 319     :param timeout: Timeout in seconds how long should function wait.
 320     :param sleep: Polling interval for results in seconds.
 321     :param exception: Exception class for eventlet.Timeout.
 322     (see doc for eventlet.Timeout for more information)
 323     """
 324     with eventlet.timeout.Timeout(timeout, exception):
 325         while not predicate():
 326             eventlet.sleep(sleep)
 327
 328
 329 def ensure_directory_exists_without_file(path):
 330     dirname = os.path.dirname(path)
 331     if os.path.isdir(dirname):
 332         try:
 333             os.unlink(path)
 334         except OSError:
 335             with excutils.save_and_reraise_exception() as ctxt:
 336                 if not os.path.exists(path):
 337                     ctxt.reraise = False
 338     else:
 339         utils.ensure_dir(dirname)
 340
 341
 342 def is_effective_user(user_id_or_name):
 343     """Returns True if user_id_or_name is effective user (id/name)."""
 344     euid = os.geteuid()
 345     if str(user_id_or_name) == str(euid):
 346         return True
 347     effective_user_name = pwd.getpwuid(euid).pw_name
 348     return user_id_or_name == effective_user_name
 349
 350
 351 def is_effective_group(group_id_or_name):
 352     """Returns True if group_id_or_name is effective group (id/name)."""
 353     egid = os.getegid()
 354     if str(group_id_or_name) == str(egid):
 355         return True
 356     effective_group_name = grp.getgrgid(egid).gr_name
 357     return group_id_or_name == effective_group_name
 358
 359
 360 class UnixDomainHTTPConnection(httplib.HTTPConnection):
 361     """Connection class for HTTP over UNIX domain socket."""
 362     def __init__(self, host, port=None, strict=None, timeout=None,
 363                  proxy_info=None):
 364         httplib.HTTPConnection.__init__(self, host, port, strict)
 365         self.timeout = timeout
 366         self.socket_path = cfg.CONF.metadata_proxy_socket
 367
 368     def connect(self):
 369         self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
 370         if self.timeout:
 371             self.sock.settimeout(self.timeout)
 372         self.sock.connect(self.socket_path)
 373
 374
 375 class UnixDomainHttpProtocol(eventlet.wsgi.HttpProtocol):
 376     def __init__(self, request, client_address, server):
 377         if client_address == '':
 378             client_address = ('<local>', 0)
 379         # base class is old-style, so super does not work properly
 380         eventlet.wsgi.HttpProtocol.__init__(self, request, client_address,
 381                                             server)
 382
 383
 384 class UnixDomainWSGIServer(wsgi.Server):
 385     def __init__(self, name):
 386         self._socket = None
 387         self._launcher = None
 388         self._server = None
 389         super(UnixDomainWSGIServer, self).__init__(name)
 390
 391     def start(self, application, file_socket, workers, backlog, mode=None):
 392         self._socket = eventlet.listen(file_socket,
 393                                        family=socket.AF_UNIX,
 394                                        backlog=backlog)
 395         if mode is not None:
 396             os.chmod(file_socket, mode)
 397
 398         self._launch(application, workers=workers)
 399
 400     def _run(self, application, socket):
 401         """Start a WSGI service in a new green thread."""
 402         logger = logging.getLogger('eventlet.wsgi.server')
 403         eventlet.wsgi.server(socket,
 404                              application,
 405                              max_size=self.num_threads,
 406                              protocol=UnixDomainHttpProtocol,
 407                              log=logger)