ceph-devstack/ceph_devstack/host.py at main · MightyShashank/ceph-devstack · GitHub

Repository navigation

History

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

import logging

import os

import pathlib

import socket

import sys

import yaml

from packaging.version import parse as parse_version, Version

from typing import Dict, List, Optional, Union

from .exec import Command

logger = logging.getLogger(__name__)

class Host:

type = "local"

def cmd(

self,

args: List[str],

cwd: Optional[pathlib.Path] = None,

env: Optional[Dict] = None,

stream_output: bool = False,

) -> Command:

return Command(

args,

cwd=cwd,

env=env,

stream_output=stream_output,

)

def run(

self,

args: List[str],

cwd: Optional[pathlib.Path] = None,

env: Optional[Dict] = None,

return self.cmd(args, cwd=cwd, env=env).run()

async def arun(

self,

args: List[str],

cwd: Optional[pathlib.Path] = None,

env: Optional[Dict] = None,

stream_output: bool = False,

return await self.cmd(

args, cwd=cwd, env=env, stream_output=stream_output

).arun()

def path_exists(self, path: Union[str, pathlib.Path]):

if isinstance(path, pathlib.Path):

return path.exists()

return os.path.exists(path)

def hostname(self) -> str:

name = socket.getfqdn()

try:

socket.gethostbyname(name)

return name

except socket.gaierror:

return "localhost"

def kernel_version(self) -> Version:

if not hasattr(self, "_kernel_version"):

proc = self.run(["uname", "-r"])

assert proc.stdout is not None

assert proc.wait() == 0, "`uname -r` failed?!"

raw_version = proc.stdout.read().decode().strip()

self._kernel_version = parse_version(raw_version.split("-")[0])

return self._kernel_version

def os_type(self) -> str:

if not hasattr(self, "_os_type"):

proc = self.run(["bash", "-c", ". /etc/os-release && echo $ID"])

assert proc.stdout is not None

assert proc.wait() == 0, "is /etc/os-release missing?"

self._os_type = proc.stdout.read().decode().strip().lower()

return self._os_type

async def podman_info(self, force: bool = False) -> Dict:

if force or not hasattr(self, "_podman_info"):

proc = await self.arun(["podman", "info"])

assert proc.stdout is not None

await proc.wait()

stdout = await proc.stdout.read()

self._podman_info = yaml.safe_load(stdout.decode().strip())

return self._podman_info

async def selinux_enforcing(self) -> bool:

proc = await host.arun(["cat", "/sys/fs/selinux/enforce"])

assert proc.stdout is not None

await proc.wait()

out = (await proc.stdout.read()).decode()

return proc.returncode == 0 and out == "1"

async def check_selinux_bool(self, name: str):

proc = await host.arun(["getsebool", name])

assert proc.stdout is not None

out = await proc.stdout.read()

if out.decode().strip() != f"{name} --> on":

return False

return True

async def get_sysctl_value(self, name: str) -> int:

proc = await host.arun(["sysctl", "-b", name])

assert proc.stdout is not None

out = await proc.stdout.read()

return int(out.decode().strip())

class LocalHost(Host):

pass

class RemoteHost(Host):

type = "remote"

base_args = ["podman", "machine", "ssh", "--"]

def cmd(

self,

args: List[str],

cwd: Optional[pathlib.Path] = None,

env: Optional[Dict] = None,

stream_output: bool = False,

if args[0] != "podman":

args = self.base_args + args

return super().cmd(args, cwd=cwd, env=env, stream_output=stream_output)

def path_exists(self, path: Union[str, pathlib.Path]):

path = os.path.expanduser(path)

proc = host.run(["ls", path])

return proc.returncode == 0

def hostname(self) -> str:

proc = self.run(["hostname"])

assert proc.stdout is not None

return proc.stdout.read().decode().strip()

local_host = LocalHost()

if sys.platform == "darwin":

host = RemoteHost()

else:

host = local_host

ceph-devstack/ceph_devstack/host.py at main · MightyShashank/ceph-devstack · GitHub

Repository navigation

Expand file tree

Breadcrumbs

host.py

Latest commit

History

Breadcrumbs

host.py

File metadata and controls