"Fossies" - the Fresh Open Source Software Archive

Member "karbor-1.3.0/karbor/api/v1/operation_logs.py" (26 Mar 2019, 7974 Bytes) of package /linux/misc/openstack/karbor-1.3.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. For more information about "operation_logs.py" see the Fossies "Dox" file reference documentation and the latest Fossies "Diffs" side-by-side code changes report: 1.1.0_vs_1.3.0.

    1 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 #    not use this file except in compliance with the License. You may obtain
    3 #    a copy of the License at
    4 #
    5 #         http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 #    Unless required by applicable law or agreed to in writing, software
    8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 #    License for the specific language governing permissions and limitations
   11 #    under the License.
   12 
   13 """The operation_logs api."""
   14 
   15 from oslo_config import cfg
   16 from oslo_log import log as logging
   17 from oslo_utils import uuidutils
   18 
   19 from webob import exc
   20 
   21 from karbor.api import common
   22 from karbor.api.openstack import wsgi
   23 from karbor import exception
   24 from karbor.i18n import _
   25 
   26 from karbor import objects
   27 from karbor.policies import operation_logs as operation_log_policy
   28 from karbor.services.operationengine import api as operationengine_api
   29 from karbor.services.protection import api as protection_api
   30 from karbor import utils
   31 
   32 import six
   33 
   34 query_operation_log_filters_opt = cfg.ListOpt(
   35     'query_operation_log_filters',
   36     default=['checkpoint_id', 'plan_id', 'restore_id', 'status'],
   37     help="Operation log filter options which "
   38     "non-admin user could use to "
   39     "query operation_logs. Default values "
   40     "are: ['checkpoint_id', 'plan_id', 'restore_id', 'status']")
   41 
   42 CONF = cfg.CONF
   43 CONF.register_opt(query_operation_log_filters_opt)
   44 
   45 LOG = logging.getLogger(__name__)
   46 
   47 
   48 class OperationLogViewBuilder(common.ViewBuilder):
   49     """Model a server API response as a python dictionary."""
   50 
   51     _collection_name = "operation_logs"
   52 
   53     def detail(self, request, operation_log):
   54         """Detailed view of a single operation_log."""
   55 
   56         operation_log_ref = {
   57             'operation_log': {
   58                 'id': operation_log.get('id'),
   59                 'operation_type': operation_log.get('operation_type'),
   60                 'checkpoint_id': operation_log.get('checkpoint_id'),
   61                 'plan_id': operation_log.get('plan_id'),
   62                 'provider_id': operation_log.get('provider_id'),
   63                 'restore_id': operation_log.get('restore_id'),
   64                 'scheduled_operation_id': operation_log.get(
   65                     'scheduled_operation_id'),
   66                 'status': operation_log.get('status'),
   67                 'started_at': operation_log.get('started_at'),
   68                 'ended_at': operation_log.get('ended_at'),
   69                 'error_info': operation_log.get('error_info'),
   70                 'extra_info': operation_log.get('extra_info'),
   71             }
   72         }
   73         return operation_log_ref
   74 
   75     def detail_list(self, request, operation_logs,
   76                     operation_log_count=None):
   77         """Detailed view of a list of operation_logs."""
   78         return self._list_view(self.detail, request, operation_logs,
   79                                operation_log_count,
   80                                self._collection_name)
   81 
   82     def _list_view(self, func, request, operation_logs,
   83                    operation_log_count,
   84                    coll_name=_collection_name):
   85         """Provide a view for a list of operation_logs.
   86 
   87         """
   88         operation_logs_list = [func(
   89             request, operation_log)['operation_log']
   90             for operation_log in operation_logs]
   91         operation_logs_links = self._get_collection_links(
   92             request, operation_logs, coll_name, operation_log_count)
   93         operation_logs_dict = {}
   94         operation_logs_dict['operation_logs'] = operation_logs_list
   95         if operation_logs_links:
   96             operation_logs_dict['operation_logs_links'] = (
   97                 operation_logs_links)
   98 
   99         return operation_logs_dict
  100 
  101 
  102 class OperationLogsController(wsgi.Controller):
  103     """The operation_log API controller for the OpenStack API."""
  104 
  105     _view_builder_class = OperationLogViewBuilder
  106 
  107     def __init__(self):
  108         self.operationengine_api = operationengine_api.API()
  109         self.protection_api = protection_api.API()
  110         super(OperationLogsController, self).__init__()
  111 
  112     def show(self, req, id):
  113         """Return data about the given OperationLogs."""
  114         context = req.environ['karbor.context']
  115 
  116         LOG.info("Show operation log with id: %s", id, context=context)
  117 
  118         if not uuidutils.is_uuid_like(id):
  119             msg = _("Invalid operation log id provided.")
  120             raise exc.HTTPBadRequest(explanation=msg)
  121 
  122         try:
  123             operation_log = self._operation_log_get(context, id)
  124         except exception.OperationLogFound as error:
  125             raise exc.HTTPNotFound(explanation=error.msg)
  126 
  127         LOG.info("Show operation log request issued successfully.")
  128         return self._view_builder.detail(req, operation_log)
  129 
  130     def index(self, req):
  131         """Returns a list of operation_logs.
  132 
  133         """
  134         context = req.environ['karbor.context']
  135 
  136         LOG.info("Show operation log list", context=context)
  137 
  138         params = req.params.copy()
  139         marker, limit, offset = common.get_pagination_params(params)
  140         sort_keys, sort_dirs = common.get_sort_params(params)
  141         filters = params
  142 
  143         utils.remove_invalid_filter_options(
  144             context,
  145             filters,
  146             self._get_operation_log_filter_options())
  147 
  148         utils.check_filters(filters)
  149         operation_logs = self._get_all(context, marker, limit,
  150                                        sort_keys=sort_keys,
  151                                        sort_dirs=sort_dirs,
  152                                        filters=filters,
  153                                        offset=offset)
  154 
  155         retval_operation_logs = self._view_builder.detail_list(
  156             req, operation_logs)
  157 
  158         LOG.info("Show operation_log list request issued "
  159                  "successfully.")
  160 
  161         return retval_operation_logs
  162 
  163     def _get_all(self, context, marker=None, limit=None, sort_keys=None,
  164                  sort_dirs=None, filters=None, offset=None):
  165         context.can(operation_log_policy.GET_ALL_POLICY)
  166 
  167         if filters is None:
  168             filters = {}
  169 
  170         all_tenants = utils.get_bool_param('all_tenants', filters)
  171 
  172         if filters:
  173             LOG.debug("Searching by: %s.", six.text_type(filters))
  174 
  175         if context.is_admin and all_tenants:
  176             # Need to remove all_tenants to pass the filtering below.
  177             del filters['all_tenants']
  178             operation_logs = objects.OperationLogList.get_all(
  179                 context, marker, limit,
  180                 sort_keys=sort_keys,
  181                 sort_dirs=sort_dirs,
  182                 filters=filters,
  183                 offset=offset)
  184         else:
  185             operation_logs = objects.OperationLogList.get_all_by_project(
  186                 context, context.project_id, marker, limit,
  187                 sort_keys=sort_keys, sort_dirs=sort_dirs, filters=filters,
  188                 offset=offset)
  189 
  190         LOG.info("Get all operation_logs completed successfully.")
  191         return operation_logs
  192 
  193     def _get_operation_log_filter_options(self):
  194         """Return operation_log search options allowed by non-admin."""
  195         return CONF.query_operation_log_filters
  196 
  197     def _operation_log_get(self, context, operation_log_id):
  198         if not uuidutils.is_uuid_like(operation_log_id):
  199             msg = _("Invalid operation_log id provided.")
  200             raise exc.HTTPBadRequest(explanation=msg)
  201 
  202         operation_log = objects.OperationLog.get_by_id(
  203             context, operation_log_id)
  204         try:
  205             context.can(operation_log_policy.GET_POLICY, operation_log)
  206         except exception.PolicyNotAuthorized:
  207             raise exception.OperationLogFound(
  208                 operation_log_id=operation_log_id)
  209         LOG.info("Operation log info retrieved successfully.")
  210         return operation_log
  211 
  212 
  213 def create_resource():
  214     return wsgi.Resource(OperationLogsController())