python-manilaclient/manilaclient/osc/v2/share_backups.py

# Copyright 2023 Cloudification GmbH.
# All Rights Reserved.
#
#   Licensed under the Apache License, Version 2.0 (the "License"); you may
#   not use this file except in compliance with the License. You may obtain
#   a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#   License for the specific language governing permissions and limitations
#   under the License.

import logging

from osc_lib.cli import parseractions
from osc_lib.command import command
from osc_lib import exceptions
from osc_lib import utils as osc_utils

from manilaclient import api_versions
from manilaclient.common._i18n import _
from manilaclient.common import constants
from manilaclient.osc import utils


LOG = logging.getLogger(__name__)


class CreateShareBackup(command.ShowOne):
    """Create a share backup."""

    _description = _("Create a backup of the given share")

    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument(
            "share",
            metavar="<share>",
            help=_("Name or ID of the share to backup."),
        )
        parser.add_argument(
            '--name',
            metavar='<name>',
            default=None,
            help=_('Optional share backup name. (Default=None).'),
        )
        parser.add_argument(
            '--description',
            metavar='<description>',
            default=None,
            help=_('Optional share backup description. (Default=None).'),
        )
        parser.add_argument(
            "--backup-options",
            metavar="<key=value>",
            default={},
            action=parseractions.KeyValueAction,
            help=_(
                "Backup driver option key=value pairs (Optional, "
                "Default=None)."
            ),
        )
        return parser

    def take_action(self, parsed_args):
        share_client = self.app.client_manager.share
        share = osc_utils.find_resource(share_client.shares, parsed_args.share)

        body = {}
        if parsed_args.backup_options:
            body['backup_options'] = utils.extract_key_value_options(
                parsed_args.backup_options
            )
        if parsed_args.description:
            body['description'] = parsed_args.description
        if parsed_args.name:
            body['name'] = parsed_args.name

        share_backup = share_client.share_backups.create(share, **body)
        share_backup._info.pop('links', None)
        return self.dict2columns(share_backup._info)


class DeleteShareBackup(command.Command):
    """Delete one or more share backups."""

    _description = _("Delete one or more share backups")

    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument(
            "backup",
            metavar="<backup>",
            nargs="+",
            help=_("Name or ID of the backup(s) to delete"),
        )
        parser.add_argument(
            "--wait",
            action='store_true',
            default=False,
            help=_("Wait for share backup deletion"),
        )
        return parser

    def take_action(self, parsed_args):
        share_client = self.app.client_manager.share
        result = 0

        for backup in parsed_args.backup:
            try:
                share_backup_obj = osc_utils.find_resource(
                    share_client.share_backups, backup
                )
                share_client.share_backups.delete(share_backup_obj)

                if parsed_args.wait:
                    if not osc_utils.wait_for_delete(
                        manager=share_client.share_backups,
                        res_id=share_backup_obj.id,
                    ):
                        result += 1

            except Exception as e:
                result += 1
                LOG.error(
                    _(
                        "Failed to delete a share backup with "
                        "name or ID '%(backup)s': %(e)s"
                    ),
                    {'backup': backup, 'e': e},
                )

        if result > 0:
            total = len(parsed_args.backup)
            msg = _("%(result)s of %(total)s backups failed to delete.") % {
                'result': result,
                'total': total,
            }
            raise exceptions.CommandError(msg)


class ListShareBackup(command.Lister):
    """List share backups."""

    _description = _("List share backups")

    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument(
            "--share",
            metavar="<share>",
            default=None,
            help=_("Name or ID of the share to list backups for."),
        )
        parser.add_argument(
            "--name",
            metavar="<name>",
            default=None,
            help=_("Filter results by name. Default=None."),
        )
        parser.add_argument(
            '--description',
            metavar="<description>",
            default=None,
            help=_("Filter results by description. Default=None."),
        )
        parser.add_argument(
            "--name~",
            metavar="<name~>",
            default=None,
            help=_("Filter results matching a share backup name pattern. "),
        )
        parser.add_argument(
            '--description~',
            metavar="<description~>",
            default=None,
            help=_("Filter results matching a share backup description "),
        )
        parser.add_argument(
            '--status',
            metavar="<status>",
            default=None,
            help=_('Filter results by status. Default=None.'),
        )
        parser.add_argument(
            "--limit",
            metavar="<num-backups>",
            type=int,
            default=None,
            action=parseractions.NonNegativeAction,
            help=_("Limit the number of backups returned. Default=None."),
        )
        parser.add_argument(
            '--offset',
            metavar="<offset>",
            default=None,
            help='Start position of backup records listing.',
        )
        parser.add_argument(
            '--sort-key',
            '--sort_key',
            metavar='<sort_key>',
            type=str,
            default=None,
            help=(
                f'Key to be sorted, available keys are '
                f'{constants.BACKUP_SORT_KEY_VALUES}. Default=None.'
            ),
        )
        parser.add_argument(
            '--sort-dir',
            '--sort_dir',
            metavar='<sort_dir>',
            type=str,
            default=None,
            help=(
                f'Sort direction, available values are '
                f'{constants.SORT_DIR_VALUES}. OPTIONAL: Default=None.'
            ),
        )
        parser.add_argument(
            '--detail',
            dest='detail',
            metavar='<0|1>',
            nargs='?',
            type=int,
            const=1,
            default=0,
            help="Show detailed information about share backups.",
        )
        return parser

    def take_action(self, parsed_args):
        share_client = self.app.client_manager.share

        share_id = None
        if parsed_args.share:
            share_id = osc_utils.find_resource(
                share_client.shares, parsed_args.share
            ).id
        columns = ['ID', 'Name', 'Share ID', 'Status']

        if parsed_args.detail:
            columns.extend(
                [
                    'Description',
                    'Size',
                    'Created At',
                    'Updated At',
                    'Availability Zone',
                    'Progress',
                    'Restore Progress',
                    'Host',
                    'Topic',
                ]
            )

        search_opts = {
            'limit': parsed_args.limit,
            'offset': parsed_args.offset,
            'name': parsed_args.name,
            'description': parsed_args.description,
            'status': parsed_args.status,
            'share_id': share_id,
        }

        search_opts['name~'] = getattr(parsed_args, 'name~')
        search_opts['description~'] = getattr(parsed_args, 'description~')

        backups = share_client.share_backups.list(
            detailed=parsed_args.detail,
            search_opts=search_opts,
            sort_key=parsed_args.sort_key,
            sort_dir=parsed_args.sort_dir,
        )

        return (
            columns,
            (osc_utils.get_item_properties(b, columns) for b in backups),
        )


class ShowShareBackup(command.ShowOne):
    """Show share backup."""

    _description = _("Show details of a backup")

    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument(
            "backup", metavar="<backup>", help=_("ID of the share backup. ")
        )
        return parser

    def take_action(self, parsed_args):
        share_client = self.app.client_manager.share
        backup = osc_utils.find_resource(
            share_client.share_backups, parsed_args.backup
        )
        backup._info.pop('links', None)
        return self.dict2columns(backup._info)


class RestoreShareBackup(command.Command):
    """Restore share backup to share"""

    _description = _("Attempt to restore share backup")

    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument(
            "backup", metavar="<backup>", help=_('ID of backup to restore.')
        )
        parser.add_argument(
            "--target-share",
            metavar="<target-share>",
            default=None,
            help=_(
                'share to restore backup to. Source share if none supplied'
            ),
        )
        parser.add_argument(
            '--wait',
            action='store_true',
            default=False,
            help=_('Wait for restore conclusion'),
        )
        return parser

    def take_action(self, parsed_args):
        share_client = self.app.client_manager.share
        kwargs = {}

        share_backup = osc_utils.find_resource(
            share_client.share_backups, parsed_args.backup
        )
        target_share_id = None
        if parsed_args.target_share is not None:
            if share_client.api_version < api_versions.APIVersion('2.91'):
                raise exceptions.CommandError(
                    'performing targeted restores is only available '
                    'for API microversion >= 2.91'
                )
            else:
                target_share_id = osc_utils.find_resource(
                    share_client.shares, parsed_args.target_share
                ).id
                kwargs['target_share_id'] = target_share_id

        share_client.share_backups.restore(share_backup.id, **kwargs)

        if parsed_args.wait:
            if not osc_utils.wait_for_status(
                status_f=share_client.shares.get,
                res_id=(target_share_id or share_backup.share_id),
                success_status=['available'],
                error_status=['error', 'backup_restoring_error'],
            ):
                LOG.error(_("ERROR: share is in error state."))


class SetShareBackup(command.Command):
    """Set share backup properties."""

    _description = _("Set share backup properties")

    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument(
            "backup",
            metavar="<backup>",
            help=_('Name or ID of the backup to set a property for'),
        )
        parser.add_argument(
            "--name",
            metavar="<name>",
            default=None,
            help=_("Set a name to the backup."),
        )
        parser.add_argument(
            "--description",
            metavar="<description>",
            default=None,
            help=_("Set a description to the backup."),
        )
        parser.add_argument(
            "--status",
            metavar="<status>",
            choices=[
                'available',
                'error',
                'creating',
                'deleting',
                'restoring',
            ],
            help=_(
                "Assign a status to the backup(Admin only). "
                "Options include : available, error, creating, "
                "deleting, restoring."
            ),
        )
        return parser

    def take_action(self, parsed_args):
        share_client = self.app.client_manager.share
        result = 0

        share_backup = osc_utils.find_resource(
            share_client.share_backups, parsed_args.backup
        )

        kwargs = {}

        if parsed_args.name is not None:
            kwargs['name'] = parsed_args.name
        if parsed_args.description is not None:
            kwargs['description'] = parsed_args.description

        try:
            share_client.share_backups.update(share_backup, **kwargs)
        except Exception as e:
            result += 1
            LOG.error(
                _(
                    "Failed to set share backup properties "
                    "'%(properties)s': %(exception)s"
                ),
                {'properties': kwargs, 'exception': e},
            )

        if parsed_args.status:
            try:
                share_client.share_backups.reset_status(
                    share_backup, parsed_args.status
                )
            except Exception as e:
                result += 1
                LOG.error(
                    _("Failed to update backup status to '%(status)s': %(e)s"),
                    {'status': parsed_args.status, 'e': e},
                )
        if result > 0:
            raise exceptions.CommandError(
                _("One or more of the set operations failed")
            )


class UnsetShareBackup(command.Command):
    """Unset share backup properties."""

    _description = _("Unset share backup properties")

    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument(
            "backup",
            metavar="<backup>",
            help=_('Name or ID of the backup to unset a property for'),
        )
        parser.add_argument(
            "--name",
            action='store_true',
            help=_("Unset a name to the backup."),
        )
        parser.add_argument(
            "--description",
            action='store_true',
            help=_("Unset a description to the backup."),
        )
        return parser

    def take_action(self, parsed_args):
        share_client = self.app.client_manager.share
        share_backup = osc_utils.find_resource(
            share_client.share_backups, parsed_args.backup
        )

        kwargs = {}
        if parsed_args.name:
            kwargs['name'] = None
        if parsed_args.description:
            kwargs['description'] = None
        if not kwargs:
            msg = "Either name or description must be provided."
            raise exceptions.CommandError(msg)

        try:
            share_client.share_backups.update(share_backup, **kwargs)
        except Exception as e:
            LOG.error(
                _(
                    "Failed to unset share backup properties "
                    "'%(properties)s': %(exception)s"
                ),
                {'properties': kwargs, 'exception': e},
            )