from select import select
from ceph_volume import terminal
from ceph_volume.util import as_bytes
+from ceph_volume.util.system import which, run_host_cmd, host_rootfs
import logging
logger = logging.getLogger(__name__)
-host_rootfs = '/rootfs'
-run_host_cmd = [
- 'nsenter',
- '--mount={}/proc/1/ns/mnt'.format(host_rootfs),
- '--ipc={}/proc/1/ns/ipc'.format(host_rootfs),
- '--net={}/proc/1/ns/net'.format(host_rootfs),
- '--uts={}/proc/1/ns/uts'.format(host_rootfs)
-]
-
-def which(executable):
- """
- Proxy function to ceph_volume.util.system.which because the ``system``
- module does import ``process``
- """
- from ceph_volume.util import system
- return system.which(executable)
def log_output(descriptor, message, terminal_logging, logfile_logging):
:param stop_on_error: If a nonzero exit status is return, it raises a ``RuntimeError``
:param fail_msg: If a nonzero exit status is returned this message will be included in the log
"""
- executable = which(command.pop(0))
+ executable = which(command.pop(0), run_on_host)
command.insert(0, executable)
if run_on_host and path.isdir(host_rootfs):
command = run_host_cmd + command
:param verbose_on_failure: On a non-zero exit status, it will forcefully set logging ON for
the terminal. Defaults to True
"""
- executable = which(command.pop(0))
+ executable = which(command.pop(0), run_on_host)
command.insert(0, executable)
if run_on_host and path.isdir(host_rootfs):
command = run_host_cmd + command
from textwrap import dedent
from ceph_volume.util import system
from mock.mock import patch
+from ceph_volume.tests.conftest import Factory
+@pytest.fixture
+def mock_find_executable_on_host(monkeypatch):
+ """
+ Monkeypatches util.system.find_executable_on_host, so that a caller can add behavior to the response
+ """
+ def apply(stdout=None, stderr=None, returncode=0):
+ stdout_stream = Factory(read=lambda: stdout)
+ stderr_stream = Factory(read=lambda: stderr)
+ return_value = Factory(
+ stdout=stdout_stream,
+ stderr=stderr_stream,
+ wait=lambda: returncode,
+ communicate=lambda x: (stdout, stderr, returncode)
+ )
+
+ monkeypatch.setattr(
+ 'ceph_volume.util.system.subprocess.Popen',
+ lambda *a, **kw: return_value)
+
+ return apply
+
class TestMkdirP(object):
def test_existing_dir_does_not_raise_w_chown(self, monkeypatch, tmpdir):
cap = capsys.readouterr()
assert 'Executable exedir not in PATH' in cap.err
+ def test_run_on_host_found(self, mock_find_executable_on_host):
+ mock_find_executable_on_host(stdout="/sbin/lvs\n", stderr="some stderr message\n")
+ assert system.which('lvs', run_on_host=True) == '/sbin/lvs'
+
+ def test_run_on_host_not_found(self, mock_find_executable_on_host):
+ mock_find_executable_on_host(stdout="", stderr="some stderr message\n")
+ assert system.which('lvs', run_on_host=True) == 'lvs'
+
@pytest.fixture
def stub_which(monkeypatch):
def apply(value='/bin/restorecon'):
import platform
import tempfile
import uuid
+import subprocess
from ceph_volume import process, terminal
from . import as_string
BLOCKDIR = '/sys/block'
ROOTGROUP = 'root'
+host_rootfs = '/rootfs'
+run_host_cmd = [
+ 'nsenter',
+ '--mount={}/proc/1/ns/mnt'.format(host_rootfs),
+ '--ipc={}/proc/1/ns/ipc'.format(host_rootfs),
+ '--net={}/proc/1/ns/net'.format(host_rootfs),
+ '--uts={}/proc/1/ns/uts'.format(host_rootfs)
+]
def generate_uuid():
return str(uuid.uuid4())
+def find_executable_on_host(locations=[], executable='', binary_check='/bin/ls'):
+ paths = ['{}/{}'.format(location, executable) for location in locations]
+ command = []
+ command.extend(run_host_cmd + [binary_check] + paths)
+ process = subprocess.Popen(
+ command,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ stdin=subprocess.PIPE,
+ close_fds=True
+ )
+ stdout = as_string(process.stdout.read())
+ if stdout:
+ executable_on_host = stdout.split('\n')[0]
+ mlogger.info('Executable {} found on the host, will use {}'.format(executable, executable_on_host))
+ return executable_on_host
+ else:
+ mlogger.warning('Executable {} not found on the host, will return {} as-is'.format(executable, executable))
+ return executable
-def which(executable):
+def which(executable, run_on_host=False):
"""find the location of an executable"""
def _get_path(executable, locations):
for location in locations:
return executable_path
return None
- path = os.getenv('PATH', '')
- path_locations = path.split(':')
- exec_in_path = _get_path(executable, path_locations)
- if exec_in_path:
- return exec_in_path
- mlogger.warning('Executable {} not in PATH: {}'.format(executable, path))
-
static_locations = (
'/usr/local/bin',
'/bin',
'/usr/sbin',
'/sbin',
)
- exec_in_static_locations = _get_path(executable, static_locations)
- if exec_in_static_locations:
- mlogger.warning('Found executable under {}, please ensure $PATH is set correctly!'.format(exec_in_static_locations))
- return exec_in_static_locations
- # fallback to just returning the argument as-is, to prevent a hard fail,
- # and hoping that the system might have the executable somewhere custom
- return executable
+ if not run_on_host:
+ path = os.getenv('PATH', '')
+ path_locations = path.split(':')
+ exec_in_path = _get_path(executable, path_locations)
+ if exec_in_path:
+ return exec_in_path
+ mlogger.warning('Executable {} not in PATH: {}'.format(executable, path))
+
+ exec_in_static_locations = _get_path(executable, static_locations)
+ if exec_in_static_locations:
+ mlogger.warning('Found executable under {}, please ensure $PATH is set correctly!'.format(exec_in_static_locations))
+ return exec_in_static_locations
+ else:
+ executable = find_executable_on_host(static_locations, executable)
+
+ # At this point, either `find_executable_on_host()` found an executable on the host
+ # or we fallback to just returning the argument as-is, to prevent a hard fail, and
+ # hoping that the system might have the executable somewhere custom
+ return executable
def get_ceph_user_ids():
"""
projects / ceph.git / commitdiff