ceph-devstack/ceph_devstack/cli.py at deploy-config · ceph/ceph-devstack · GitHub

Repository navigation

History

import asyncio

import logging

import sys

from argparse import Namespace

from subprocess import CalledProcessError

from ceph_devstack import config, logger, parse_args, VERBOSE

from ceph_devstack.requirements import check_requirements

from ceph_devstack.resources.ceph import CephDevStack

CONFIG_HANDLERS = {

"dump": lambda config, args: print(config.dump()),

"get": lambda config, args: print(config.get_value(args.name)),

"set": lambda config, args: print(config.set_value(args.name, args.value)),

"unset": lambda config, args: config.unset_value(args.name),

}

def _configure_console_logging(verbose: bool) -> None:

if not verbose:

return

for handler in logging.getLogger("root").handlers:

if not isinstance(handler, logging.FileHandler):

handler.setLevel(VERBOSE)

def _action_kwargs(args: Namespace) -> dict:

match args.command:

case "wait":

return {"container_name": args.container}

case "logs":

return {

"run_name": args.run_name,

"job_id": args.job_id,

"locate": args.locate,

}

case _:

return {}

async def _apply(stack: CephDevStack, action: str, **kwargs) -> int:

"""Map a CLI command to a stack action and apply it."""

if not all(

await asyncio.gather(

check_requirements(),

stack.check_requirements(),

)

logger.error("Requirements not met!")

return 1

if action == "doctor":

return 0

result = await stack.apply(action, **kwargs)

return result if result is not None else 0

def main() -> int:

args = parse_args(sys.argv[1:])

config.load(args.config_file)

_configure_console_logging(args.verbose)

if args.command == "config":

CONFIG_HANDLERS[args.config_op](config, args)

return 0

if args.command == "block-pool":

from ceph_devstack.block_pool import BlockPool

if args.block_pool_op != "status":

logger.error("Usage: ceph-devstack block-pool status")

return 2

return BlockPool.status_from_config(config)

config["args"] = vars(args)

try:

stack = CephDevStack(stack_name=args.stack)

except ValueError as exc:

logger.error(str(exc))

return 1

stack.data_dir.mkdir(parents=True, exist_ok=True)

action = args.command

try:

return asyncio.run(_apply(stack, action, **_action_kwargs(args)))

except CalledProcessError as exc:

return exc.returncode if exc.returncode is not None else 1

except KeyboardInterrupt:

logger.debug("Exiting!")

return 130 # 128 + SIGINT

ceph-devstack/ceph_devstack/cli.py at deploy-config · ceph/ceph-devstack · GitHub

Repository navigation

Expand file tree

Breadcrumbs

cli.py

Latest commit

History

Breadcrumbs

cli.py

File metadata and controls