ge/python-compute
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

various updates v.dev3

Browse Source
This commit is contained in:
ge
2023-12-13 01:42:50 +03:00
parent b0fa1b7b25
commit d7a73e9bd1
49 changed files with 1872 additions and 904 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
compute/__init__.py
View File

@@ -15,8 +15,9 @@
"""Compute instances management library."""
__version__ = '0.1.0-dev2'
__version__ = '0.1.0-dev3'
from .instance import Instance, InstanceConfig, InstanceSchema
from .config import Config
from .instance import CloudInit, Instance, InstanceConfig, InstanceSchema
from .session import Session
from .storage import StoragePool, Volume, VolumeConfig

4
compute/__main__.py
View File

@@ -15,7 +15,7 @@
"""Command line interface for compute module."""
from compute.cli import control
from compute.cli import parser
control.cli()
parser.run()

0
compute/common.py → compute/abstract.py
View File

409
compute/cli/commands.py Normal file
View File

@@ -0,0 +1,409 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
"""CLI commands."""
import argparse
import base64
import json
import logging
import pathlib
import re
import shlex
import sys
import uuid
import libvirt
import pydantic
import yaml
from compute import Session
from compute.cli.term import Table, confirm
from compute.exceptions import GuestAgentTimeoutExpired
from compute.instance import CloudInit, GuestAgent, InstanceSchema
from compute.instance.devices import DiskConfig, DiskDriver
from compute.utils import dictutil, diskutils, ids
log = logging.getLogger(__name__)
libvirt.registerErrorHandler(
lambda userdata, err: None, # noqa: ARG005
ctx=None,
)
def init(session: Session, args: argparse.Namespace) -> None:
"""Initialise compute instance using YAML config."""
try:
data = yaml.load(args.file.read(), Loader=yaml.SafeLoader)
log.debug('Read from file: %s', data)
except yaml.YAMLError as e:
sys.exit(f'error: cannot parse YAML: {e}')
capabilities = session.get_capabilities()
node_info = session.get_node_info()
base_instance_config = {
'name': str(uuid.uuid4()),
'title': None,
'description': None,
'arch': capabilities.arch,
'machine': capabilities.machine,
'emulator': capabilities.emulator,
'max_vcpus': node_info.cpus,
'max_memory': node_info.memory,
'cpu': {
'emulation_mode': 'host-passthrough',
'model': None,
'vendor': None,
'topology': None,
'features': None,
},
'network_interfaces': [
{
'source': 'default',
'mac': ids.random_mac(),
},
],
'boot': {'order': ['cdrom', 'hd']},
'cloud_init': None,
}
data = dictutil.override(base_instance_config, data)
volumes = []
targets = []
for volume in data['volumes']:
base_disk_config = {
'bus': 'virtio',
'is_readonly': False,
'driver': {
'name': 'qemu',
'type': 'qcow2',
'cache': 'writethrough',
},
}
base_cdrom_config = {
'bus': 'ide',
'is_readonly': True,
'driver': {
'name': 'qemu',
'type': 'raw',
'cache': 'writethrough',
},
}
if volume.get('device') is None:
volume['device'] = 'disk'
if volume.get('target') is None:
prefix = 'hd' if volume['device'] == 'cdrom' else 'vd'
target = diskutils.get_disk_target(targets, prefix)
volume['target'] = target
targets.append(target)
else:
targets.append(volume['target'])
if volume['device'] == 'disk':
volumes.append(dictutil.override(base_disk_config, volume))
if volume['device'] == 'cdrom':
volumes.append(dictutil.override(base_cdrom_config, volume))
data['volumes'] = volumes
if data['cloud_init'] is not None:
cloud_init_config = {
'user_data': None,
'meta_data': None,
'vendor_data': None,
'network_config': None,
}
data['cloud_init'] = dictutil.override(
cloud_init_config,
data['cloud_init'],
)
for item in data['cloud_init']:
cidata = data['cloud_init'][item]
if cidata is None:
pass
elif isinstance(cidata, str):
if cidata.startswith('base64:'):
data['cloud_init'][item] = base64.b64decode(
cidata.split(':')[1]
).decode('utf-8')
elif re.fullmatch(r'^[^\n]{1,1024}$', cidata, re.I):
data_file = pathlib.Path(cidata)
if data_file.exists():
with data_file.open('r') as f:
data['cloud_init'][item] = f.read()
else:
pass
else:
data['cloud_init'][item] = yaml.dump(cidata)
try:
log.debug('Input data: %s', data)
if args.test:
_ = InstanceSchema(**data)
print(json.dumps(dict(data), indent=4, sort_keys=True))
sys.exit()
instance = session.create_instance(**data)
print(f'Initialised: {instance.name}')
if args.start:
instance.start()
print(f'Started: {instance.name}')
except pydantic.ValidationError as e:
for error in e.errors():
fields = '.'.join([str(lc) for lc in error['loc']])
print(
f"validation error: {fields}: {error['msg']}",
file=sys.stderr,
)
def exec_(session: Session, args: argparse.Namespace) -> None:
"""
Execute command in guest via guest agent.
NOTE: any argument after instance name will be passed into guest's shell
"""
instance = session.get_instance(args.instance)
ga = GuestAgent(instance.domain, timeout=args.timeout)
arguments = args.arguments.copy()
if len(arguments) > 1 and not args.no_join_args:
arguments = [shlex.join(arguments)]
if not args.no_join_args:
arguments.insert(0, '-c')
stdin = None
if not sys.stdin.isatty():
stdin = sys.stdin.read()
try:
output = ga.guest_exec(
path=args.executable,
args=arguments,
env=args.env,
stdin=stdin,
capture_output=True,
decode_output=True,
poll=True,
)
except GuestAgentTimeoutExpired as e:
sys.exit(
f'{e}. NOTE: command may still running in guest, '
f'PID={ga.last_pid}'
)
if output.stderr:
print(output.stderr.strip(), file=sys.stderr)
if output.stdout:
print(output.stdout.strip(), file=sys.stdout)
sys.exit(output.exitcode)
def ls(session: Session, args: argparse.Namespace) -> None: # noqa: ARG001
"""List compute instances."""
table = Table()
table.header = ['NAME', 'STATE', 'NVCPUS', 'MEMORY']
for instance in session.list_instances():
info = instance.get_info()
table.add_row(
[
instance.name,
instance.get_status() + ' ',
info.nproc,
f'{int(info.memory / 1024)} MiB',
]
)
print(table)
def lsdisks(session: Session, args: argparse.Namespace) -> None:
"""List block devices attached to instance."""
instance = session.get_instance(args.instance)
if args.persistent:
disks = instance.list_disks(persistent=True)
else:
disks = instance.list_disks()
table = Table()
table.header = ['TARGET', 'SOURCE']
for disk in disks:
table.add_row([disk.target, disk.source])
print(table)
def start(session: Session, args: argparse.Namespace) -> None:
"""Start instance."""
instance = session.get_instance(args.instance)
instance.start()
def shutdown(session: Session, args: argparse.Namespace) -> None:
"""Shutdown instance."""
instance = session.get_instance(args.instance)
if args.soft:
method = 'SOFT'
elif args.hard:
method = 'HARD'
elif args.unsafe:
method = 'UNSAFE'
else:
method = 'NORMAL'
instance.shutdown(method)
def reboot(session: Session, args: argparse.Namespace) -> None:
"""Reboot instance."""
instance = session.get_instance(args.instance)
instance.reboot()
def reset(session: Session, args: argparse.Namespace) -> None:
"""Reset instance."""
instance = session.get_instance(args.instance)
instance.reset()
def powrst(session: Session, args: argparse.Namespace) -> None:
"""Power reset instance."""
instance = session.get_instance(args.instance)
instance.power_reset()
def pause(session: Session, args: argparse.Namespace) -> None:
"""Pause instance."""
instance = session.get_instance(args.instance)
instance.pause()
def resume(session: Session, args: argparse.Namespace) -> None:
"""Resume instance."""
instance = session.get_instance(args.instance)
instance.resume()
def status(session: Session, args: argparse.Namespace) -> None:
"""Display instance status."""
instance = session.get_instance(args.instance)
print(instance.get_status())
def setvcpus(session: Session, args: argparse.Namespace) -> None:
"""Set instance vCPU number."""
instance = session.get_instance(args.instance)
instance.set_vcpus(args.nvcpus, live=True)
def setmem(session: Session, args: argparse.Namespace) -> None:
"""Set instance memory size."""
instance = session.get_instance(args.instance)
instance.set_memory(args.memory, live=True)
def setpass(session: Session, args: argparse.Namespace) -> None:
"""Set user password in guest."""
instance = session.get_instance(args.instance)
instance.set_user_password(
args.username,
args.password,
encrypted=args.encrypted,
)
def setcdrom(session: Session, args: argparse.Namespace) -> None:
"""Manage CDROM devices."""
instance = session.get_instance(args.instance)
if args.detach:
for disk in instance.list_disks(persistent=True):
if disk.device == 'cdrom' and disk.source == args.source:
instance.detach_disk(disk.target, live=False)
print(
f"disk '{disk.target}' detached, "
'perform power reset to apply changes'
)
return
disks_live = instance.list_disks(persistent=False)
disks_inactive = instance.list_disks(persistent=True)
disks = [d.target for d in disks_inactive if d not in disks_live]
target = diskutils.get_disk_target(disks, 'hd')
cdrom = DiskConfig(
type='file',
device='cdrom',
source=args.source,
target=target,
is_readonly=True,
bus='ide',
driver=DiskDriver('qemu', 'raw', 'writethrough'),
)
instance.attach_device(cdrom, live=False)
print(
f"CDROM attached as disk '{target}', "
'perform power reset to apply changes'
)
def setcloudinit(session: Session, args: argparse.Namespace) -> None:
"""
Set cloud-init configuration.
The cloud-init disk must not be mounted to the host system while making
changes using this command! In this case, data may be damaged when writing
to disk - if the new content of the file is longer than the old one, it
will be truncated.
"""
if (
args.user_data is None
and args.vendor_data is None
and args.network_config is None
and args.meta_data is None
):
sys.exit('nothing to do')
instance = session.get_instance(args.instance)
disks = instance.list_disks()
cloud_init_disk_path = None
cloud_init_disk_target = diskutils.get_disk_target(
[d.target for d in disks], prefix='vd'
)
cloud_init = CloudInit()
if args.user_data:
cloud_init.user_data = args.user_data.read()
if args.vendor_data:
cloud_init.vendor_data = args.vendor_data.read()
if args.network_config:
cloud_init.network_config = args.network_config.read()
if args.meta_data:
cloud_init.meta_data = args.meta_data.read()
for disk in disks:
if disk.source.endswith('cloud-init.img'):
cloud_init_disk_path = disk.source
break
if cloud_init_disk_path is None:
volumes = session.get_storage_pool(session.VOLUMES_POOL)
cloud_init_disk_path = volumes.path.joinpath(
f'{instance.name}-cloud-init.img'
)
cloud_init.create_disk(cloud_init_disk_path)
volumes.refresh()
cloud_init.attach_disk(
cloud_init_disk_path,
cloud_init_disk_target,
instance,
)
else:
cloud_init.update_disk(cloud_init_disk_path)
def delete(session: Session, args: argparse.Namespace) -> None:
"""Delete instance with local storage volumes."""
if args.yes is True or confirm(
'this action is irreversible, continue?',
default=False,
):
instance = session.get_instance(args.instance)
if args.save_volumes is False:
instance.delete(with_volumes=True)
else:
instance.delete()
else:
print('aborted')

624
compute/cli/control.py
View File

@@ -1,624 +0,0 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
"""Command line interface."""
import argparse
import json
import logging
import os
import re
import shlex
import string
import sys
import uuid
import libvirt
import yaml
from pydantic import ValidationError
from compute import __version__
from compute.exceptions import ComputeError, GuestAgentTimeoutError
from compute.instance import GuestAgent, Instance, InstanceSchema
from compute.instance.devices import DiskConfig, DiskDriver
from compute.session import Session
from compute.utils import dictutil, ids
log = logging.getLogger(__name__)
log_levels = [lv.lower() for lv in logging.getLevelNamesMapping()]
libvirt.registerErrorHandler(
lambda userdata, err: None, # noqa: ARG005
ctx=None,
)
class Table:
"""Minimalistic text table constructor."""
def __init__(self, whitespace: str | None = None):
"""Initialise Table."""
self.whitespace = whitespace or '\t'
self.header = []
self.rows = []
self.table = ''
def add_row(self, row: list) -> None:
"""Add table row."""
self.rows.append([str(col) for col in row])
def add_rows(self, rows: list[list]) -> None:
"""Add multiple rows."""
for row in rows:
self.add_row(row)
def __str__(self) -> str:
"""Build table and return."""
widths = [max(map(len, col)) for col in zip(*self.rows, strict=True)]
self.rows.insert(0, [str(h).upper() for h in self.header])
for row in self.rows:
self.table += self.whitespace.join(
(
val.ljust(width)
for val, width in zip(row, widths, strict=True)
)
)
self.table += '\n'
return self.table.strip()
def _list_instances(session: Session) -> None:
table = Table()
table.header = ['NAME', 'STATE']
for instance in session.list_instances():
table.add_row(
[
instance.name,
instance.get_status(),
]
)
print(table)
sys.exit()
def _exec_guest_agent_command(
session: Session, args: argparse.Namespace
) -> None:
instance = session.get_instance(args.instance)
ga = GuestAgent(instance.domain, timeout=args.timeout)
arguments = args.arguments.copy()
if len(arguments) > 1 and not args.no_join_args:
arguments = [shlex.join(arguments)]
if not args.no_join_args:
arguments.insert(0, '-c')
stdin = None
if not sys.stdin.isatty():
stdin = sys.stdin.read()
try:
output = ga.guest_exec(
path=args.executable,
args=arguments,
env=args.env,
stdin=stdin,
capture_output=True,
decode_output=True,
poll=True,
)
except GuestAgentTimeoutError as e:
sys.exit(
f'{e}. NOTE: command may still running in guest, '
f'PID={ga.last_pid}'
)
if output.stderr:
print(output.stderr.strip(), file=sys.stderr)
if output.stdout:
print(output.stdout.strip(), file=sys.stdout)
sys.exit(output.exitcode)
def _init_instance(session: Session, args: argparse.Namespace) -> None:
try:
data = yaml.load(args.file.read(), Loader=yaml.SafeLoader)
log.debug('Read from file: %s', data)
except yaml.YAMLError as e:
sys.exit(f'error: cannot parse YAML: {e}')
capabilities = session.get_capabilities()
node_info = session.get_node_info()
base_instance_config = {
'name': str(uuid.uuid4()),
'title': None,
'description': None,
'arch': capabilities.arch,
'machine': capabilities.machine,
'emulator': capabilities.emulator,
'max_vcpus': node_info.cpus,
'max_memory': node_info.memory,
'cpu': {
'emulation_mode': 'host-passthrough',
'model': None,
'vendor': None,
'topology': None,
'features': None,
},
'network_interfaces': [
{
'source': 'default',
'mac': ids.random_mac(),
},
],
'boot': {'order': ['cdrom', 'hd']},
}
data = dictutil.override(base_instance_config, data)
volumes = []
for volume in data['volumes']:
base_disk_config = {
'bus': 'virtio',
'is_readonly': False,
'driver': {
'name': 'qemu',
'type': 'qcow2',
'cache': 'writethrough',
},
}
base_cdrom_config = {
'bus': 'ide',
'target': 'hda',
'is_readonly': True,
'driver': {
'name': 'qemu',
'type': 'raw',
'cache': 'writethrough',
},
}
if volume.get('device') is None:
volume['device'] = 'disk'
if volume['device'] == 'disk':
volumes.append(dictutil.override(base_disk_config, volume))
if volume['device'] == 'cdrom':
volumes.append(dictutil.override(base_cdrom_config, volume))
data['volumes'] = volumes
try:
log.debug('Input data: %s', data)
if args.test:
_ = InstanceSchema(**data)
print(json.dumps(dict(data), indent=4, sort_keys=True))
sys.exit()
instance = session.create_instance(**data)
print(f'initialised: {instance.name}')
if args.start:
instance.start()
except ValidationError as e:
for error in e.errors():
fields = '.'.join([str(lc) for lc in error['loc']])
print(
f"validation error: {fields}: {error['msg']}",
file=sys.stderr,
)
sys.exit()
def _shutdown_instance(session: Session, args: argparse.Namespace) -> None:
instance = session.get_instance(args.instance)
if args.soft:
method = 'SOFT'
elif args.hard:
method = 'HARD'
elif args.unsafe:
method = 'UNSAFE'
else:
method = 'NORMAL'
instance.shutdown(method)
def _confirm(message: str, *, default: bool | None = None) -> None:
while True:
match default:
case True:
prompt = 'default: yes'
case False:
prompt = 'default: no'
case _:
prompt = 'no default'
try:
answer = input(f'{message} ({prompt}) ')
except KeyboardInterrupt:
sys.exit('aborted')
if not answer and isinstance(default, bool):
return default
if re.match(r'^y(es)?$', answer, re.I):
return True
if re.match(r'^no?$', answer, re.I):
return False
print("Please respond 'yes' or 'no'")
def _delete_instance(session: Session, args: argparse.Namespace) -> None:
if args.yes is True or _confirm(
'this action is irreversible, continue?',
default=False,
):
instance = session.get_instance(args.instance)
if args.save_volumes is False:
instance.delete(with_volumes=True)
else:
instance.delete()
else:
print('aborted')
sys.exit()
def _get_disk_target(instance: Instance, prefix: str = 'hd') -> str:
disks_live = instance.list_disks(persistent=False)
disks_inactive = instance.list_disks(persistent=True)
disks = [d for d in disks_inactive if d not in disks_live]
devs = [d.target[-1] for d in disks if d.target.startswith(prefix)]
return prefix + [x for x in string.ascii_lowercase if x not in devs][0] # noqa: RUF015
def _manage_cdrom(session: Session, args: argparse.Namespace) -> None:
instance = session.get_instance(args.instance)
if args.detach:
for disk in instance.list_disks(persistent=True):
if disk.device == 'cdrom' and disk.source == args.source:
instance.detach_disk(disk.target, live=False)
print(
f"disk '{disk.target}' detached, "
'perform power reset to apply changes'
)
return
target = _get_disk_target(instance, 'hd')
cdrom = DiskConfig(
type='file',
device='cdrom',
source=args.source,
target=target,
is_readonly=True,
bus='ide',
driver=DiskDriver('qemu', 'raw', 'writethrough'),
)
instance.attach_device(cdrom, live=False)
print(
f"CDROM attached as disk '{target}', "
'perform power reset to apply changes'
)
def main(session: Session, args: argparse.Namespace) -> None:
"""Perform actions."""
match args.command:
case 'init':
_init_instance(session, args)
case 'exec':
_exec_guest_agent_command(session, args)
case 'ls':
_list_instances(session)
case 'start':
instance = session.get_instance(args.instance)
instance.start()
case 'shutdown':
_shutdown_instance(session, args)
case 'reboot':
instance = session.get_instance(args.instance)
instance.reboot()
case 'reset':
instance = session.get_instance(args.instance)
instance.reset()
case 'powrst':
instance = session.get_instance(args.instance)
instance.power_reset()
case 'pause':
instance = session.get_instance(args.instance)
instance.pause()
case 'resume':
instance = session.get_instance(args.instance)
instance.resume()
case 'status':
instance = session.get_instance(args.instance)
print(instance.status)
case 'setvcpus':
instance = session.get_instance(args.instance)
instance.set_vcpus(args.nvcpus, live=True)
case 'setmem':
instance = session.get_instance(args.instance)
instance.set_memory(args.memory, live=True)
case 'setpass':
instance = session.get_instance(args.instance)
instance.set_user_password(
args.username,
args.password,
encrypted=args.encrypted,
)
case 'setcdrom':
_manage_cdrom(session, args)
case 'delete':
_delete_instance(session, args)
def get_parser() -> argparse.ArgumentParser: # noqa: PLR0915
"""Return command line arguments parser."""
root = argparse.ArgumentParser(
prog='compute',
description='Manage compute instances.',
formatter_class=argparse.RawTextHelpFormatter,
)
root.add_argument(
'-v',
'--verbose',
action='store_true',
default=False,
help='enable verbose mode',
)
root.add_argument(
'-c',
'--connect',
metavar='URI',
help='libvirt connection URI',
)
root.add_argument(
'-l',
'--log-level',
type=str.lower,
metavar='LEVEL',
choices=log_levels,
help='log level',
)
root.add_argument(
'-V',
'--version',
action='version',
version=__version__,
)
subparsers = root.add_subparsers(dest='command', metavar='COMMAND')
# init command
init = subparsers.add_parser(
'init', help='initialise instance using YAML config file'
)
init.add_argument(
'file',
type=argparse.FileType('r', encoding='UTF-8'),
nargs='?',
default='instance.yaml',
help='instance config [default: instance.yaml]',
)
init.add_argument(
'-s',
'--start',
action='store_true',
default=False,
help='start instance after init',
)
init.add_argument(
'-t',
'--test',
action='store_true',
default=False,
help='just print resulting instance config as JSON and exit',
)
# exec subcommand
execute = subparsers.add_parser(
'exec',
help='execute command in guest via guest agent',
description=(
'Execute command in guest via guest agent. '
'NOTE: any argument after instance name will be passed into '
'guest as shell command.'
),
)
execute.add_argument('instance')
execute.add_argument('arguments', nargs=argparse.REMAINDER)
execute.add_argument(
'-t',
'--timeout',
type=int,
default=60,
help=(
'waiting time in seconds for a command to be executed '
'in guest [default: 60]'
),
)
execute.add_argument(
'-x',
'--executable',
default='/bin/sh',
help='path to executable in guest [default: /bin/sh]',
)
execute.add_argument(
'-e',
'--env',
type=str,
nargs='?',
action='append',
help='environment variables to pass to executable in guest',
)
execute.add_argument(
'-n',
'--no-join-args',
action='store_true',
default=False,
help=(
"do not join arguments list and add '-c' option, suitable "
'for non-shell executables and other specific cases.'
),
)
# ls subcommand
listall = subparsers.add_parser('ls', help='list instances')
listall.add_argument(
'-a',
'--all',
action='store_true',
default=False,
help='list all instances including inactive',
)
# start subcommand
start = subparsers.add_parser('start', help='start instance')
start.add_argument('instance')
# shutdown subcommand
shutdown = subparsers.add_parser('shutdown', help='shutdown instance')
shutdown.add_argument('instance')
shutdown_opts = shutdown.add_mutually_exclusive_group()
shutdown_opts.add_argument(
'-s',
'--soft',
action='store_true',
help='normal guest OS shutdown, guest agent is used',
)
shutdown_opts.add_argument(
'-n',
'--normal',
action='store_true',
help='shutdown with hypervisor selected method [default]',
)
shutdown_opts.add_argument(
'-H',
'--hard',
action='store_true',
help=(
"gracefully destroy instance, it's like long "
'pressing the power button'
),
)
shutdown_opts.add_argument(
'-u',
'--unsafe',
action='store_true',
help=(
'destroy instance, this is similar to a power outage '
'and may result in data loss or corruption'
),
)
# reboot subcommand
reboot = subparsers.add_parser('reboot', help='reboot instance')
reboot.add_argument('instance')
# reset subcommand
reset = subparsers.add_parser('reset', help='reset instance')
reset.add_argument('instance')
# powrst subcommand
powrst = subparsers.add_parser('powrst', help='power reset instance')
powrst.add_argument('instance')
# pause subcommand
pause = subparsers.add_parser('pause', help='pause instance')
pause.add_argument('instance')
# resume subcommand
resume = subparsers.add_parser('resume', help='resume paused instance')
resume.add_argument('instance')
# status subcommand
status = subparsers.add_parser('status', help='display instance status')
status.add_argument('instance')
# setvcpus subcommand
setvcpus = subparsers.add_parser('setvcpus', help='set vCPU number')
setvcpus.add_argument('instance')
setvcpus.add_argument('nvcpus', type=int)
# setmem subcommand
setmem = subparsers.add_parser('setmem', help='set memory size')
setmem.add_argument('instance')
setmem.add_argument('memory', type=int, help='memory in MiB')
# setpass subcommand
setpass = subparsers.add_parser(
'setpass',
help='set user password in guest',
)
setpass.add_argument('instance')
setpass.add_argument('username')
setpass.add_argument('password')
setpass.add_argument(
'-e',
'--encrypted',
action='store_true',
default=False,
help='set it if password is already encrypted',
)
# setcdrom subcommand
setcdrom = subparsers.add_parser('setcdrom', help='manage CDROM devices')
setcdrom.add_argument('instance')
setcdrom.add_argument('source', help='source for CDROM')
setcdrom.add_argument(
'-d',
'--detach',
action='store_true',
default=False,
help='detach CDROM device',
)
# delete subcommand
delete = subparsers.add_parser(
'delete',
help='delete instance',
)
delete.add_argument('instance')
delete.add_argument(
'-y',
'--yes',
action='store_true',
default=False,
help='automatic yes to prompt',
)
delete.add_argument(
'--save-volumes',
action='store_true',
default=False,
help='do not delete local storage volumes',
)
return root
def cli() -> None:
"""Run arguments parser."""
root = get_parser()
args = root.parse_args()
if args.command is None:
root.print_help()
sys.exit()
log_level = args.log_level or os.getenv('CMP_LOG')
if isinstance(log_level, str) and log_level.lower() in log_levels:
logging.basicConfig(
level=logging.getLevelNamesMapping()[log_level.upper()]
)
log.debug('CLI started with args: %s', args)
connect_uri = (
args.connect
or os.getenv('CMP_LIBVIRT_URI')
or os.getenv('LIBVIRT_DEFAULT_URI')
or 'qemu:///system'
)
try:
with Session(connect_uri) as session:
main(session, args)
except ComputeError as e:
sys.exit(f'error: {e}')
except KeyboardInterrupt:
sys.exit()
except SystemExit as e:
sys.exit(e)
if __name__ == '__main__':
cli()

471
compute/cli/parser.py Normal file
View File

@@ -0,0 +1,471 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
"""Command line argument parser."""
import argparse
import logging
import os
import sys
import textwrap
from collections.abc import Callable
from typing import NamedTuple
from compute import Session, __version__
from compute.cli import commands
from compute.exceptions import ComputeError
log = logging.getLogger(__name__)
log_levels = [lv.lower() for lv in logging.getLevelNamesMapping()]
class Doc(NamedTuple):
"""Parsed docstring."""
help: str # noqa: A003
desc: str
def get_doc(func: Callable) -> Doc:
"""Extract help message and description from function docstring."""
doc = func.__doc__
if isinstance(doc, str):
doc = textwrap.dedent(doc).strip().split('\n\n')
return Doc(doc[0][0].lower() + doc[0][1:], '\n\n'.join(doc))
return Doc('', '')
def get_parser() -> argparse.ArgumentParser:
"""Return command line argument parser."""
root = argparse.ArgumentParser(
prog='compute',
description='Manage compute instances.',
)
root.add_argument(
'-V',
'--version',
action='version',
version=__version__,
)
root.add_argument(
'-c',
'--connect',
dest='root_connect',
metavar='URI',
help='libvirt connection URI',
)
root.add_argument(
'-l',
'--log-level',
dest='root_log_level',
type=str.lower,
choices=log_levels,
metavar='LEVEL',
help='log level',
)
# common options
common = argparse.ArgumentParser(add_help=False)
common.add_argument(
'-c',
'--connect',
metavar='URI',
help='libvirt connection URI',
)
common.add_argument(
'-l',
'--log-level',
type=str.lower,
choices=log_levels,
metavar='LEVEL',
help='log level',
)
subparsers = root.add_subparsers(dest='command', metavar='COMMAND')
# init command
init = subparsers.add_parser(
'init',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.init).help,
description=get_doc(commands.init).desc,
)
init.add_argument(
'file',
type=argparse.FileType('r', encoding='UTF-8'),
nargs='?',
default='instance.yaml',
help='instance config [default: instance.yaml]',
)
init.add_argument(
'-s',
'--start',
action='store_true',
default=False,
help='start instance after init',
)
init.add_argument(
'-t',
'--test',
action='store_true',
default=False,
help='just print resulting instance config as JSON and exit',
)
init.set_defaults(func=commands.init)
# exec subcommand
execute = subparsers.add_parser(
'exec',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.exec_).help,
description=get_doc(commands.exec_).desc,
)
execute.add_argument('instance')
execute.add_argument('arguments', nargs=argparse.REMAINDER)
execute.add_argument(
'-t',
'--timeout',
type=int,
default=60,
help=(
'waiting time in seconds for a command to be executed '
'in guest [default: 60]'
),
)
execute.add_argument(
'-x',
'--executable',
default='/bin/sh',
help='path to executable in guest [default: /bin/sh]',
)
execute.add_argument(
'-e',
'--env',
type=str,
nargs='?',
action='append',
help='environment variables to pass to executable in guest',
)
execute.add_argument(
'-n',
'--no-join-args',
action='store_true',
default=False,
help=(
"do not join arguments list and add '-c' option, suitable "
'for non-shell executables and other specific cases.'
),
)
execute.set_defaults(func=commands.exec_)
# ls subcommand
ls = subparsers.add_parser(
'ls',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.ls).help,
description=get_doc(commands.ls).desc,
)
ls.set_defaults(func=commands.ls)
# lsdisks subcommand
lsdisks = subparsers.add_parser(
'lsdisks',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.lsdisks).help,
description=get_doc(commands.lsdisks).desc,
)
lsdisks.add_argument('instance')
lsdisks.add_argument(
'-p',
'--persistent',
action='store_true',
default=False,
help='display only persisnent devices',
)
lsdisks.set_defaults(func=commands.lsdisks)
# start subcommand
start = subparsers.add_parser(
'start',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.start).help,
description=get_doc(commands.start).desc,
)
start.add_argument('instance')
start.set_defaults(func=commands.start)
# shutdown subcommand
shutdown = subparsers.add_parser(
'shutdown',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.shutdown).help,
description=get_doc(commands.shutdown).desc,
)
shutdown.add_argument('instance')
shutdown_opts = shutdown.add_mutually_exclusive_group()
shutdown_opts.add_argument(
'-s',
'--soft',
action='store_true',
help='normal guest OS shutdown, guest agent is used',
)
shutdown_opts.add_argument(
'-n',
'--normal',
action='store_true',
help='shutdown with hypervisor selected method [default]',
)
shutdown_opts.add_argument(
'-H',
'--hard',
action='store_true',
help=(
"gracefully destroy instance, it's like long "
'pressing the power button'
),
)
shutdown_opts.add_argument(
'-u',
'--unsafe',
action='store_true',
help=(
'destroy instance, this is similar to a power outage '
'and may result in data loss or corruption'
),
)
shutdown.set_defaults(func=commands.shutdown)
# reboot subcommand
reboot = subparsers.add_parser(
'reboot',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.reboot).help,
description=get_doc(commands.reboot).desc,
)
reboot.add_argument('instance')
reboot.set_defaults(func=commands.reboot)
# reset subcommand
reset = subparsers.add_parser(
'reset',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.reset).help,
description=get_doc(commands.reset).desc,
)
reset.add_argument('instance')
reset.set_defaults(func=commands.reset)
# powrst subcommand
powrst = subparsers.add_parser(
'powrst',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.powrst).help,
description=get_doc(commands.powrst).desc,
)
powrst.add_argument('instance')
powrst.set_defaults(func=commands.powrst)
# pause subcommand
pause = subparsers.add_parser(
'pause',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.pause).help,
description=get_doc(commands.pause).desc,
)
pause.add_argument('instance')
pause.set_defaults(func=commands.pause)
# resume subcommand
resume = subparsers.add_parser(
'resume',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.resume).help,
description=get_doc(commands.resume).desc,
)
resume.add_argument('instance')
resume.set_defaults(func=commands.resume)
# status subcommand
status = subparsers.add_parser(
'status',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.status).help,
description=get_doc(commands.status).desc,
)
status.add_argument('instance')
status.set_defaults(func=commands.status)
# setvcpus subcommand
setvcpus = subparsers.add_parser(
'setvcpus',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.setvcpus).help,
description=get_doc(commands.setvcpus).desc,
)
setvcpus.add_argument('instance')
setvcpus.add_argument('nvcpus', type=int)
setvcpus.set_defaults(func=commands.setvcpus)
# setmem subcommand
setmem = subparsers.add_parser(
'setmem',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.setmem).help,
description=get_doc(commands.setmem).desc,
)
setmem.add_argument('instance')
setmem.add_argument('memory', type=int, help='memory in MiB')
setmem.set_defaults(func=commands.setmem)
# setpass subcommand
setpass = subparsers.add_parser(
'setpass',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.setpass).help,
description=get_doc(commands.setpass).desc,
)
setpass.add_argument('instance')
setpass.add_argument('username')
setpass.add_argument('password')
setpass.add_argument(
'-e',
'--encrypted',
action='store_true',
default=False,
help='set it if password is already encrypted',
)
setpass.set_defaults(func=commands.setpass)
# setcdrom subcommand
setcdrom = subparsers.add_parser(
'setcdrom',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.setcdrom).help,
description=get_doc(commands.setcdrom).desc,
)
setcdrom.add_argument('instance')
setcdrom.add_argument('source', help='source for CDROM')
setcdrom.add_argument(
'-d',
'--detach',
action='store_true',
default=False,
help='detach CDROM device',
)
setcdrom.set_defaults(func=commands.setcdrom)
# setcloudinit subcommand
setcloudinit = subparsers.add_parser(
'setcloudinit',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.setcloudinit).help,
description=get_doc(commands.setcloudinit).desc,
)
setcloudinit.add_argument('instance')
setcloudinit.add_argument(
'--user-data',
type=argparse.FileType('r'),
help='user-data file',
)
setcloudinit.add_argument(
'--vendor-data',
type=argparse.FileType('r'),
help='vendor-data file',
)
setcloudinit.add_argument(
'--meta-data',
type=argparse.FileType('r'),
help='meta-data file',
)
setcloudinit.add_argument(
'--network-config',
type=argparse.FileType('r'),
help='network-config file',
)
setcloudinit.set_defaults(func=commands.setcloudinit)
# delete subcommand
delete = subparsers.add_parser(
'delete',
parents=[common],
formatter_class=argparse.RawTextHelpFormatter,
help=get_doc(commands.delete).help,
description=get_doc(commands.delete).desc,
)
delete.add_argument('instance')
delete.add_argument(
'-y',
'--yes',
action='store_true',
default=False,
help='automatic yes to prompt',
)
delete.add_argument(
'--save-volumes',
action='store_true',
default=False,
help='do not delete local storage volumes',
)
delete.set_defaults(func=commands.delete)
return root
def run() -> None:
"""Run argument parser."""
parser = get_parser()
args = parser.parse_args()
if args.command is None:
parser.print_help()
sys.exit()
log_level = args.root_log_level or args.log_level or os.getenv('CMP_LOG')
if isinstance(log_level, str) and log_level.lower() in log_levels:
logging.basicConfig(
level=logging.getLevelNamesMapping()[log_level.upper()]
)
log.debug('CLI started with args: %s', args)
connect_uri = (
args.root_connect
or args.connect
or os.getenv('CMP_LIBVIRT_URI')
or os.getenv('LIBVIRT_DEFAULT_URI')
or 'qemu:///system'
)
try:
with Session(connect_uri) as session:
# Invoke command
args.func(session, args)
except ComputeError as e:
sys.exit(f'error: {e}')
except KeyboardInterrupt:
sys.exit()

77
compute/cli/term.py Normal file
View File

@@ -0,0 +1,77 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
"""Utils for creating terminal output and interface elements."""
import re
import sys
class Table:
"""Minimalistic text table constructor."""
def __init__(self, whitespace: str | None = None):
"""Initialise Table."""
self.whitespace = whitespace or '\t'
self.header = []
self.rows = []
self.table = ''
def add_row(self, row: list) -> None:
"""Add table row."""
self.rows.append([str(col) for col in row])
def add_rows(self, rows: list[list]) -> None:
"""Add multiple rows."""
for row in rows:
self.add_row(row)
def __str__(self) -> str:
"""Return table."""
widths = [max(map(len, col)) for col in zip(*self.rows, strict=True)]
self.rows.insert(0, [str(h).upper() for h in self.header])
for row in self.rows:
widths = widths or [len(i) for i in row]
self.table += self.whitespace.join(
(
val.ljust(width)
for val, width in zip(row, widths, strict=True)
)
)
self.table += '\n'
return self.table.strip()
def confirm(message: str, *, default: bool | None = None) -> None:
"""Start yes/no interactive dialog."""
while True:
match default:
case True:
prompt = 'default: yes'
case False:
prompt = 'default: no'
case _:
prompt = 'no default'
try:
answer = input(f'{message} ({prompt}) ')
except KeyboardInterrupt:
sys.exit('aborted')
if not answer and isinstance(default, bool):
return default
if re.match(r'^y(es)?$', answer, re.I):
return True
if re.match(r'^no?$', answer, re.I):
return False
print("Please respond 'yes' or 'no'")

121
compute/config.py Normal file
View File

@@ -0,0 +1,121 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
"""Configuration loader."""
__all__ = ['Config', 'ConfigSchema']
import os
import tomllib
from collections import UserDict
from pathlib import Path
from typing import ClassVar
from .abstract import EntityModel
from .exceptions import ConfigLoaderError
from .utils import dictutil
class LibvirtConfigSchema(EntityModel):
"""Schema for libvirt config."""
uri: str
class LogConfigSchema(EntityModel):
"""Logger congif schema."""
level: str | None = None
file: str | None = None
class StorageConfigSchema(EntityModel):
"""Storage config schema."""
volumes: str
images: str
class ConfigSchema(EntityModel):
"""Configuration file schema."""
libvirt: LibvirtConfigSchema | None
log: LogConfigSchema | None
storage: StorageConfigSchema
class Config(UserDict):
"""
UserDict for storing configuration.
Environment variables prefix is ``CMP_``. Environment variables
have higher proirity then configuration file.
:cvar str IMAGES_POOL: images storage pool name taken from env
:cvar str VOLUMES_POOL: volumes storage pool name taken from env
:cvar Path DEFAULT_CONFIG_FILE: :file:`/etc/computed/computed.toml`
:cvar dict DEFAULT_CONFIGURATION:
"""
LIBVIRT_URI = os.getenv('CMP_LIBVIRT_URI')
IMAGES_POOL = os.getenv('CMP_IMAGES_POOL')
VOLUMES_POOL = os.getenv('CMP_VOLUMES_POOL')
DEFAULT_CONFIG_FILE = Path('/etc/compute/computed.toml')
DEFAULT_CONFIGURATION: ClassVar[dict] = {
'libvirt': {
'uri': 'qemu:///system',
},
'log': {
'level': None,
'file': None,
},
'storage': {
'images': 'images',
'volumes': 'volumes',
},
}
def __init__(self, file: Path | None = None):
"""
Initialise Config.
:param file: Path to configuration file. If `file` is None
use default path from :var:`Config.DEFAULT_CONFIG_FILE`.
"""
self.file = Path(file) if file else self.DEFAULT_CONFIG_FILE
try:
if self.file.exists():
with self.file.open('rb') as configfile:
loaded = tomllib.load(configfile)
else:
loaded = {}
except tomllib.TOMLDecodeError as etoml:
raise ConfigLoaderError(
f'Bad TOML syntax: {self.file}: {etoml}'
) from etoml
except (OSError, ValueError) as eread:
raise ConfigLoaderError(
f'Config read error: {self.file}: {eread}'
) from eread
config = dictutil.override(self.DEFAULT_CONFIGURATION, loaded)
if self.LIBVIRT_URI:
config['libvirt']['uri'] = self.LIBVIRT_URI
if self.VOLUMES_POOL:
config['storage']['volumes'] = self.VOLUMES_POOL
if self.IMAGES_POOL:
config['storage']['images'] = self.IMAGES_POOL
ConfigSchema(**config)
super().__init__(config)

8
compute/exceptions.py
View File

@@ -36,12 +36,12 @@ class GuestAgentUnavailableError(GuestAgentError):
"""Guest agent is not connected or is unavailable."""
class GuestAgentTimeoutError(GuestAgentError):
"""QEMU timeout exceeded."""
class GuestAgentTimeoutExpired(GuestAgentError): # noqa: N818
"""QEMU timeout expired."""
def __init__(self, seconds: int):
"""Initialise GuestAgentTimeoutExceededError."""
super().__init__(f'QEMU timeout ({seconds} sec) exceeded')
"""Initialise GuestAgentTimeoutExpired."""
super().__init__(f'QEMU timeout ({seconds} sec) expired')
class GuestAgentCommandNotSupportedError(GuestAgentError):

1
compute/instance/__init__.py
View File

@@ -13,6 +13,7 @@
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
from .cloud_init import CloudInit
from .guest_agent import GuestAgent
from .instance import Instance, InstanceConfig
from .schemas import InstanceSchema

221
compute/instance/cloud_init.py Normal file
View File

@@ -0,0 +1,221 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
# ruff: noqa: S603
"""
`Cloud-init`_ integration for bootstraping compute instances.
.. _Cloud-init: https://cloudinit.readthedocs.io
"""
import logging
import subprocess
import tempfile
from pathlib import Path
from compute.exceptions import InstanceError
from .devices import DiskConfig, DiskDriver
from .instance import Instance
log = logging.getLogger(__name__)
class CloudInit:
"""
Cloud-init integration.
:ivar str user_data: user-data.
:ivar str vendor_data: vendor-data.
:ivar str network_config: network-config.
:ivar str meta_data: meta-data.
"""
def __init__(self):
"""Initialise :class:`CloudInit`."""
self.user_data = None
self.vendor_data = None
self.network_config = None
self.meta_data = None
def __repr__(self) -> str:
"""Represent :class:`CloudInit` object."""
return (
self.__class__.__name__
+ '('
+ ', '.join(
[
f'{self.user_data=}',
f'{self.vendor_data=}',
f'{self.network_config=}',
f'{self.meta_data=}',
]
)
+ ')'
).replace('self.', '')
def _write_to_disk(
self,
disk: str,
filename: str,
data: str | None,
*,
force_file_create: bool = False,
delete_existing_file: bool = False,
default_data: str | None = None,
) -> None:
data = data or default_data
log.debug('Input data %s: %r', filename, data)
if isinstance(data, str):
data = data.encode()
if data is None and force_file_create is False:
return
with tempfile.NamedTemporaryFile() as data_file:
if data is not None:
data_file.write(data)
data_file.flush()
if delete_existing_file:
log.debug('Deleting existing file')
filelist = subprocess.run(
['/usr/bin/mdir', '-i', disk, '-b'],
capture_output=True,
check=True,
)
files = [
f.replace('::/', '')
for f in filelist.stdout.decode().splitlines()
]
log.debug('Files on disk: %s', files)
log.debug("Removing '%s'", filename)
if filename in files:
subprocess.run(
['/usr/bin/mdel', '-i', disk, f'::{filename}'],
check=True,
)
log.debug("Writing file '%s'", filename)
subprocess.run(
[
'/usr/bin/mcopy',
'-i',
disk,
data_file.name,
f'::{filename}',
],
check=True,
)
def create_disk(self, disk: Path, *, force: bool = False) -> None:
"""
Create disk with cloud-init config files.
:param path: Disk path.
:param force: Replace existing disk.
"""
if not isinstance(disk, Path):
disk = Path(disk)
if disk.exists():
if disk.is_file() is False:
raise InstanceError('Cloud-init disk must be regular file')
if force:
disk.unlink()
else:
raise InstanceError('File already exists')
subprocess.run(
['/usr/sbin/mkfs.vfat', '-n', 'CIDATA', '-C', str(disk), '1024'],
check=True,
stderr=subprocess.DEVNULL,
)
self._write_to_disk(
disk=disk,
filename='user-data',
data=self.user_data,
force_file_create=True,
default_data='#cloud-config',
)
self._write_to_disk(
disk=disk,
filename='vendor-data',
data=self.vendor_data,
)
self._write_to_disk(
disk=disk,
filename='network-config',
data=self.network_config,
)
self._write_to_disk(
disk=disk,
filename='meta-data',
data=self.meta_data,
force_file_create=True,
)
def update_disk(self, disk: Path) -> None:
"""Update files on existing disk."""
if not isinstance(disk, Path):
disk = Path(disk)
if not disk.exists():
raise InstanceError(f"File '{disk}' does not exists")
if self.user_data:
self._write_to_disk(
disk=disk,
filename='user-data',
data=self.user_data,
force_file_create=True,
default_data='#cloud-config',
delete_existing_file=True,
)
if self.vendor_data:
self._write_to_disk(
disk=disk,
filename='vendor-data',
data=self.vendor_data,
delete_existing_file=True,
)
if self.network_config:
self._write_to_disk(
disk=disk,
filename='network-config',
data=self.network_config,
delete_existing_file=True,
)
if self.meta_data:
self._write_to_disk(
disk=disk,
filename='meta-data',
data=self.meta_data,
force_file_create=True,
delete_existing_file=True,
)
def attach_disk(self, disk: Path, target: str, instance: Instance) -> None:
"""
Attach cloud-init disk to instance.
:param disk: Path to disk.
:param instance: Compute instance object.
"""
instance.attach_device(
DiskConfig(
type='file',
device='disk',
source=disk,
target=target,
is_readonly=True,
bus='virtio',
driver=DiskDriver('qemu', 'raw'),
)
)

21
compute/instance/devices.py
View File

@@ -24,7 +24,7 @@ from typing import Union
from lxml import etree
from lxml.builder import E
from compute.common import DeviceConfig
from compute.abstract import DeviceConfig
from compute.exceptions import InvalidDeviceConfigError
@@ -32,9 +32,9 @@ from compute.exceptions import InvalidDeviceConfigError
class DiskDriver:
"""Disk driver description for libvirt."""
name: str
type: str
cache: str
name: str = 'qemu'
type: str = 'qcow2'
cache: str = 'default'
def __call__(self):
"""Return self."""
@@ -56,13 +56,7 @@ class DiskConfig(DeviceConfig):
is_readonly: bool = False
device: str = 'disk'
bus: str = 'virtio'
driver: DiskDriver = field(
default_factory=DiskDriver(
name='qemu',
type='qcow2',
cache='writethrough',
)
)
driver: DiskDriver = field(default_factory=DiskDriver())
def to_xml(self) -> str:
"""Return XML config for libvirt."""
@@ -99,13 +93,14 @@ class DiskConfig(DeviceConfig):
pretty_print=True,
).strip()
driver = xml.find('driver')
cachetype = driver.get('cache')
disk_params = {
'type': xml.get('type'),
'device': xml.get('device'),
'driver': DiskDriver(
name=driver.get('name'),
type=driver.get('type'),
cache=driver.get('cache'),
**({'cache': cachetype} if cachetype else {}),
),
'source': xml.find('source').get('file'),
'target': xml.find('target').get('dev'),
@@ -122,7 +117,7 @@ class DiskConfig(DeviceConfig):
if driver_param is None:
msg = (
"'driver' tag must have "
"'name', 'type' and 'cache' attributes"
"'name' and 'type' attributes"
)
raise InvalidDeviceConfigError(msg, xml_str)
return cls(**disk_params)

6
compute/instance/guest_agent.py
View File

@@ -27,7 +27,7 @@ import libvirt_qemu
from compute.exceptions import (
GuestAgentCommandNotSupportedError,
GuestAgentError,
GuestAgentTimeoutError,
GuestAgentTimeoutExpired,
GuestAgentUnavailableError,
)
@@ -114,7 +114,7 @@ class GuestAgent:
if command not in supported:
raise GuestAgentCommandNotSupportedError(command)
def guest_exec( # noqa: PLR0913
def guest_exec(
self,
path: str,
args: list[str] | None = None,
@@ -199,7 +199,7 @@ class GuestAgent:
sleep(poll_interval)
now = time()
if now - start_time > self.timeout:
raise GuestAgentTimeoutError(self.timeout)
raise GuestAgentTimeoutExpired(self.timeout)
log.debug(
'Polling command pid=%s finished, time taken: %s seconds',
pid,

62
compute/instance/instance.py
View File

@@ -25,7 +25,7 @@ import libvirt
from lxml import etree
from lxml.builder import E
from compute.common import DeviceConfig, EntityConfig
from compute.abstract import DeviceConfig, EntityConfig
from compute.exceptions import (
GuestAgentCommandNotSupportedError,
InstanceError,
@@ -141,6 +141,14 @@ class InstanceConfig(EntityConfig):
)
)
xml.append(self._gen_cpu_xml(self.cpu))
xml.append(
E.clock(
E.timer(name='rtc', tickpolicy='catchup'),
E.timer(name='pit', tickpolicy='delay'),
E.timer(name='hpet', present='no'),
offset='utc',
)
)
os = E.os(E.type('hvm', machine=self.machine, arch=self.arch))
for dev in self.boot.order:
os.append(E.boot(dev=dev))
@@ -159,7 +167,7 @@ class InstanceConfig(EntityConfig):
devices.append(E.emulator(str(self.emulator)))
for interface in self.network_interfaces:
devices.append(self._gen_network_interface_xml(interface))
devices.append(E.graphics(type='vnc', port='-1', autoport='yes'))
devices.append(E.graphics(type='vnc', autoport='yes'))
devices.append(E.input(type='tablet', bus='usb'))
devices.append(
E.channel(
@@ -171,6 +179,7 @@ class InstanceConfig(EntityConfig):
type='unix',
)
)
devices.append(E.serial(E.target(port='0'), type='pty'))
devices.append(
E.console(E.target(type='serial', port='0'), type='pty')
)
@@ -212,10 +221,30 @@ class Instance:
:param domain: libvirt domain object
"""
self.domain = domain
self.connection = domain.connect()
self.name = domain.name()
self.guest_agent = GuestAgent(domain)
self._domain = domain
self._connection = domain.connect()
self._name = domain.name()
self._guest_agent = GuestAgent(domain)
@property
def connection(self) -> libvirt.virConnect:
"""Libvirt connection object."""
return self._connection
@property
def domain(self) -> libvirt.virDomain:
"""Libvirt domain object."""
return self._domain
@property
def name(self) -> str:
"""Instance name."""
return self._name
@property
def guest_agent(self) -> GuestAgent:
""":class:`GuestAgent` object."""
return self._guest_agent
def _expand_instance_state(self, state: int) -> str:
states = {
@@ -279,6 +308,9 @@ class Instance:
def get_max_vcpus(self) -> int:
"""Maximum vCPUs number for domain."""
if not self.is_running():
xml = etree.fromstring(self.dump_xml(inactive=True))
return int(xml.xpath('/domain/vcpu/text()')[0])
return self.domain.maxVcpus()
def start(self) -> None:
@@ -324,7 +356,6 @@ class Instance:
:param method: Method used to shutdown instance
"""
if not self.is_running():
log.warning('Instance is not running, nothing to do')
return
methods = {
'SOFT': libvirt.VIR_DOMAIN_SHUTDOWN_GUEST_AGENT,
@@ -737,13 +768,24 @@ class Instance:
:param with_volumes: If True delete local volumes with instance.
"""
log.info("Shutdown instance '%s'", self.name)
self.shutdown(method='HARD')
disks = self.list_disks(persistent=True)
log.debug('Disks list: %s', disks)
for disk in disks:
if with_volumes and disk.type == 'file':
volume = self.connection.storageVolLookupByPath(disk.source)
log.debug('Delete volume: %s', volume.path())
try:
volume = self.connection.storageVolLookupByPath(
disk.source
)
except libvirt.libvirtError as e:
if e.get_error_code() == libvirt.VIR_ERR_NO_STORAGE_VOL:
log.warning(
"Volume '%s' not found, skipped",
disk.source,
)
continue
log.info('Delete volume: %s', volume.path())
volume.delete()
log.debug('Undefine instance')
log.info('Undefine instance')
self.domain.undefine()

12
compute/instance/schemas.py
View File

@@ -22,7 +22,7 @@ from pathlib import Path
from pydantic import ValidationError, validator
from pydantic.error_wrappers import ErrorWrapper
from compute.common import EntityModel
from compute.abstract import EntityModel
from compute.utils.units import DataUnit
@@ -109,6 +109,15 @@ class BootOptionsSchema(EntityModel):
order: tuple
class CloudInitSchema(EntityModel):
"""Cloud-init config model."""
user_data: str | None = None
meta_data: str | None = None
vendor_data: str | None = None
network_config: str | None = None
class InstanceSchema(EntityModel):
"""Compute instance model."""
@@ -127,6 +136,7 @@ class InstanceSchema(EntityModel):
volumes: list[VolumeSchema]
network_interfaces: list[NetworkInterfaceSchema]
image: str | None = None
cloud_init: CloudInitSchema | None = None
@validator('name')
def _check_name(cls, value: str) -> str: # noqa: N805

116
compute/session.py
View File

@@ -16,7 +16,6 @@
"""Hypervisor session manager."""
import logging
import os
from contextlib import AbstractContextManager
from types import TracebackType
from typing import Any, NamedTuple
@@ -25,19 +24,23 @@ from uuid import uuid4
import libvirt
from lxml import etree
from .config import Config
from .exceptions import (
InstanceNotFoundError,
SessionError,
StoragePoolNotFoundError,
)
from .instance import Instance, InstanceConfig, InstanceSchema
from .instance.cloud_init import CloudInit
from .instance.devices import DiskConfig, DiskDriver
from .storage import StoragePool, VolumeConfig
from .utils import units
from .utils import diskutils, units
log = logging.getLogger(__name__)
config = Config()
class Capabilities(NamedTuple):
"""Store domain capabilities info."""
@@ -72,27 +75,20 @@ class NodeInfo(NamedTuple):
class Session(AbstractContextManager):
"""
Hypervisor session context manager.
:cvar IMAGES_POOL: images storage pool name taken from env
:cvar VOLUMES_POOL: volumes storage pool name taken from env
"""
IMAGES_POOL = os.getenv('CMP_IMAGES_POOL')
VOLUMES_POOL = os.getenv('CMP_VOLUMES_POOL')
"""Hypervisor session context manager."""
def __init__(self, uri: str | None = None):
"""
Initialise session with hypervisor.
:ivar str uri: libvirt connection URI.
:ivar libvirt.virConnect connection: libvirt connection object.
:param uri: libvirt connection URI.
"""
self.uri = uri or 'qemu:///system'
self.connection = libvirt.open(self.uri)
log.debug('Config=%s', config)
self.LIBVIRT_URI = config['libvirt']['uri']
self.IMAGES_POOL = config['storage']['images']
self.VOLUMES_POOL = config['storage']['volumes']
self._uri = uri or self.LIBVIRT_URI
self._connection = libvirt.open(self._uri)
def __enter__(self):
"""Return Session object."""
@@ -107,6 +103,16 @@ class Session(AbstractContextManager):
"""Close the connection when leaving the context."""
self.close()
@property
def uri(self) -> str:
"""Libvirt connection URI."""
return self._uri
@property
def connection(self) -> libvirt.virConnect:
"""Libvirt connection object."""
return self._connection
def close(self) -> None:
"""Close connection to libvirt daemon."""
self.connection.close()
@@ -207,38 +213,51 @@ class Session(AbstractContextManager):
"""
data = InstanceSchema(**kwargs)
config = InstanceConfig(data)
log.info('Define XML...')
log.info(config.to_xml())
log.info('Define instance XML')
log.debug(config.to_xml())
try:
self.connection.defineXML(config.to_xml())
except libvirt.libvirtError as e:
raise SessionError(f'Error defining instance: {e}') from e
log.info('Getting instance...')
log.info('Getting instance object...')
instance = self.get_instance(config.name)
log.info('Start processing volumes...')
log.info('Connecting to images pool...')
images_pool = self.get_storage_pool(self.IMAGES_POOL)
images_pool.refresh()
log.info('Connecting to volumes pool...')
volumes_pool = self.get_storage_pool(self.VOLUMES_POOL)
volumes_pool.refresh()
disk_targets = []
for volume in data.volumes:
log.info('Processing volume=%s', volume)
log.info('Connecting to images pool...')
images_pool = self.get_storage_pool(self.IMAGES_POOL)
log.info('Connecting to volumes pool...')
volumes_pool = self.get_storage_pool(self.VOLUMES_POOL)
log.info('Building volume configuration...')
capacity = None
disk_targets.append(volume.target)
if not volume.source:
vol_name = f'{uuid4()}.qcow2'
volume_name = f'{uuid4()}.qcow2'
else:
vol_name = volume.source
volume_name = volume.source
if volume.device == 'cdrom':
log.debug('Volume %s is CDROM device', vol_name)
log.info('Volume %s is CDROM device', volume_name)
elif volume.source is not None:
log.info('Using volume %s as source', volume_name)
volume_source = volume.source
if volume.capacity:
capacity = units.to_bytes(
volume.capacity.value, volume.capacity.unit
)
else:
capacity = units.to_bytes(
volume.capacity.value, volume.capacity.unit
)
vol_conf = VolumeConfig(
name=vol_name,
path=str(volumes_pool.path.joinpath(vol_name)),
volume_config = VolumeConfig(
name=volume_name,
path=str(volumes_pool.path.joinpath(volume_name)),
capacity=capacity,
)
log.info('Volume configuration is:\n %s', vol_conf.to_xml())
volume_source = volume_config.path
log.debug('Volume config: %s', volume_config)
if volume.is_system is True and data.image:
log.info(
"Volume is marked as 'system', start cloning image..."
@@ -246,21 +265,22 @@ class Session(AbstractContextManager):
log.info('Get image %s', data.image)
image = images_pool.get_volume(data.image)
log.info('Cloning image into volumes pool...')
vol = volumes_pool.clone_volume(image, vol_conf)
log.info(
'Resize cloned volume to specified size: %s',
capacity,
)
vol.resize(capacity, unit=units.DataUnit.BYTES)
vol = volumes_pool.clone_volume(image, volume_config)
else:
log.info('Create volume...')
volumes_pool.create_volume(vol_conf)
log.info('Create volume %s', volume_config.name)
volumes_pool.create_volume(volume_config)
if capacity is not None:
log.info(
'Resize cloned volume to specified size: %s',
capacity,
)
vol.resize(capacity, unit=units.DataUnit.BYTES)
log.info('Attaching volume to instance...')
instance.attach_device(
DiskConfig(
type=volume.type,
device=volume.device,
source=vol_conf.path,
source=volume_source,
target=volume.target,
is_readonly=volume.is_readonly,
bus=volume.bus,
@@ -271,6 +291,24 @@ class Session(AbstractContextManager):
),
)
)
if data.cloud_init:
log.info('Crating disk for cloud-init...')
cloud_init = CloudInit()
cloud_init.user_data = data.cloud_init.user_data
cloud_init.vendor_data = data.cloud_init.vendor_data
cloud_init.network_config = data.cloud_init.network_config
cloud_init.meta_data = data.cloud_init.meta_data
cloud_init_disk_path = volumes_pool.path.joinpath(
f'{instance.name}-cloud-init.img'
)
cloud_init.create_disk(cloud_init_disk_path)
log.info('Attaching cloud-init disk to instance...')
volumes_pool.refresh()
cloud_init.attach_disk(
cloud_init_disk_path,
diskutils.get_disk_target(disk_targets, prefix='vd'),
instance,
)
return instance
def get_instance(self, name: str) -> Instance:

34
compute/storage/pool.py
View File

@@ -15,7 +15,11 @@
"""Manage storage pools."""
import datetime
import logging
import time
from datetime import datetime as dt
from datetime import timedelta
from pathlib import Path
from typing import NamedTuple
@@ -65,10 +69,32 @@ class StoragePool:
"""Return storage pool XML description as string."""
return self.pool.XMLDesc()
def refresh(self) -> None:
"""Refresh storage pool."""
# TODO @ge: handle libvirt asynchronous job related exceptions
self.pool.refresh()
def refresh(self, *, retry: bool = True, timeout: int = 30) -> None:
"""
Refresh storage pool.
:param retry: If True retry pool refresh on :class:`libvirtError`
with running asynchronous jobs.
:param timeout: Retry timeout in secodns. Affets only if `retry`
is True.
"""
retry_timeout = dt.now(tz=datetime.UTC) + timedelta(seconds=timeout)
while dt.now(tz=datetime.UTC) < retry_timeout:
try:
self.pool.refresh()
except libvirt.libvirtError as e:
if 'asynchronous jobs running' in e.get_error_message():
if retry is False:
raise StoragePoolError(e) from e
log.debug(
'An error ocurred when refreshing storage pool '
'retrying after 1 sec...'
)
time.sleep(1)
else:
raise StoragePoolError(e) from e
else:
return
def create_volume(self, vol_conf: VolumeConfig) -> Volume:
"""Create storage volume and return Volume instance."""

2
compute/storage/volume.py
View File

@@ -23,7 +23,7 @@ import libvirt
from lxml import etree
from lxml.builder import E
from compute.common import EntityConfig
from compute.abstract import EntityConfig
from compute.utils import units

56
compute/utils/config_loader.py
View File

@@ -1,56 +0,0 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
"""Configuration loader."""
import tomllib
from collections import UserDict
from pathlib import Path
from compute.exceptions import ConfigLoaderError
DEFAULT_CONFIGURATION = {}
DEFAULT_CONFIG_FILE = '/etc/computed/computed.toml'
class ConfigLoader(UserDict):
"""UserDict for storing configuration."""
def __init__(self, file: Path | None = None):
"""
Initialise ConfigLoader.
:param file: Path to configuration file. If `file` is None
use default path from DEFAULT_CONFIG_FILE constant.
"""
# TODO @ge: load deafult configuration
self.file = Path(file) if file else Path(DEFAULT_CONFIG_FILE)
super().__init__(self.load())
def load(self) -> dict:
"""Load confguration object from TOML file."""
try:
with Path(self.file).open('rb') as configfile:
return tomllib.load(configfile)
# TODO @ge: add config schema validation
except tomllib.TOMLDecodeError as tomlerr:
raise ConfigLoaderError(
f'Bad TOML syntax in config file: {self.file}: {tomlerr}'
) from tomlerr
except (OSError, ValueError) as readerr:
raise ConfigLoaderError(
f'Cannot read config file: {self.file}: {readerr}'
) from readerr

45
compute/utils/diskutils.py Normal file
View File

@@ -0,0 +1,45 @@
# This file is part of Compute
#
# Compute is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Compute 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Compute. If not, see <http://www.gnu.org/licenses/>.
"""Auxiliary functions for working with disks."""
import string
def get_disk_target(
disks: list[str], prefix: str, *, from_end: bool = False
) -> str:
"""
Return free disk name.
.. code-block:: shell-session
>>> get_disk_target(['vda', 'vdb'], 'vd')
'vdc'
>>> get_disk_target(['vda', 'vdc'], 'vd')
'vdb'
>>> get_disk_target(['vda', 'vdd'], 'vd', from_end=True)
'vdz'
>>> get_disk_target(['vda', 'hda'], 'hd')
'hdb'
:param disks: List of attached disk names.
:param prefix: Disk name prefix.
:param from_end: If True select a drive letter starting from the
end of the alphabet.
"""
index = -1 if from_end else 0
devs = [d[-1] for d in disks if d.startswith(prefix)]
return prefix + [x for x in string.ascii_lowercase if x not in devs][index]