osm_mon/plugins/OpenStack/Gnocchi/metrics.py

   1 # Copyright 2017 Intel Research and Development Ireland Limited
   2 # *************************************************************
   3
   4 # This file is part of OSM Monitoring module
   5 # All Rights Reserved to Intel Corporation
   6
   7 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   8 # not use this file except in compliance with the License. You may obtain
   9 # a copy of the License at
  10
  11 #         http://www.apache.org/licenses/LICENSE-2.0
  12
  13 # Unless required by applicable law or agreed to in writing, software
  14 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  15 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  16 # License for the specific language governing permissions and limitations
  17 # under the License.
  18
  19 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact: helena.mcgough@intel.com or adrian.hoban@intel.com
  21 ##
  22 """Carry out OpenStack metric requests via Gnocchi API."""
  23
  24 import datetime
  25 import json
  26 import logging
  27
  28 import time
  29
  30 import six
  31
  32 from osm_mon.core.message_bus.producer import KafkaProducer
  33 from osm_mon.plugins.OpenStack.common import Common
  34
  35 from osm_mon.plugins.OpenStack.response import OpenStack_Response
  36 from osm_mon.plugins.OpenStack.settings import Config
  37
  38 log = logging.getLogger(__name__)
  39
  40 METRIC_MAPPINGS = {
  41     "average_memory_utilization": "memory.percent",
  42     "disk_read_ops": "disk.read.requests",
  43     "disk_write_ops": "disk.write.requests",
  44     "disk_read_bytes": "disk.read.bytes",
  45     "disk_write_bytes": "disk.write.bytes",
  46     "packets_dropped": "interface.if_dropped",
  47     "packets_received": "interface.if_packets",
  48     "packets_sent": "interface.if_packets",
  49     "cpu_utilization": "cpu_util",
  50 }
  51
  52 PERIOD_MS = {
  53     "HR": 3600000,
  54     "DAY": 86400000,
  55     "WEEK": 604800000,
  56     "MONTH": 2629746000,
  57     "YEAR": 31556952000
  58 }
  59
  60
  61 class Metrics(object):
  62     """OpenStack metric requests performed via the Gnocchi API."""
  63
  64     def __init__(self):
  65         """Initialize the metric actions."""
  66         # Configure an instance of the OpenStack metric plugin
  67         config = Config.instance()
  68         config.read_environ()
  69
  70         # Initialise authentication for API requests
  71         self._common = Common()
  72
  73         # Use the Response class to generate valid json response messages
  74         self._response = OpenStack_Response()
  75
  76         # Initializer a producer to send responses back to SO
  77         self._producer = KafkaProducer("metric_response")
  78
  79     def metric_calls(self, message):
  80         """Consume info from the message bus to manage metric requests."""
  81         values = json.loads(message.value)
  82         log.info("OpenStack metric action required.")
  83
  84         auth_token = Common.get_auth_token(values['vim_uuid'])
  85
  86         endpoint = Common.get_endpoint("metric", values['vim_uuid'])
  87
  88         if message.key == "create_metric_request":
  89             # Configure metric
  90             metric_details = values['metric_create']
  91             metric_id, resource_id, status = self.configure_metric(
  92                 endpoint, auth_token, metric_details)
  93
  94             # Generate and send a create metric response
  95             try:
  96                 resp_message = self._response.generate_response(
  97                     'create_metric_response', status=status,
  98                     cor_id=values['correlation_id'],
  99                     metric_id=metric_id, r_id=resource_id)
 100                 log.info("Response messages: %s", resp_message)
 101                 self._producer.create_metrics_resp(
 102                     'create_metric_response', resp_message,
 103                     'metric_response')
 104             except Exception as exc:
 105                 log.warn("Failed to create response: %s", exc)
 106
 107         elif message.key == "read_metric_data_request":
 108             # Read all metric data related to a specified metric
 109             timestamps, metric_data = self.read_metric_data(
 110                 endpoint, auth_token, values)
 111
 112             # Generate and send a response message
 113             try:
 114
 115                 metric_id = self.get_metric_id(endpoint, auth_token, METRIC_MAPPINGS[values['metric_name']],
 116                                           values['resource_uuid'])
 117                 resp_message = self._response.generate_response(
 118                     'read_metric_data_response',
 119                     m_id=metric_id,
 120                     m_name=values['metric_name'],
 121                     r_id=values['resource_uuid'],
 122                     cor_id=values['correlation_id'],
 123                     times=timestamps, metrics=metric_data)
 124                 log.info("Response message: %s", resp_message)
 125                 self._producer.read_metric_data_response(
 126                     'read_metric_data_response', resp_message,
 127                     'metric_response')
 128             except Exception as exc:
 129                 log.warn("Failed to send read metric response:%s", exc)
 130
 131         elif message.key == "delete_metric_request":
 132             # delete the specified metric in the request
 133             metric_id = self.get_metric_id(endpoint, auth_token, METRIC_MAPPINGS[values['metric_name']],
 134                                            values['resource_uuid'])
 135             status = self.delete_metric(
 136                 endpoint, auth_token, metric_id)
 137
 138             # Generate and send a response message
 139             try:
 140                 resp_message = self._response.generate_response(
 141                     'delete_metric_response', m_id=metric_id,
 142                     m_name=values['metric_name'],
 143                     status=status, r_id=values['resource_uuid'],
 144                     cor_id=values['correlation_id'])
 145                 log.info("Response message: %s", resp_message)
 146                 self._producer.delete_metric_response(
 147                     'delete_metric_response', resp_message,
 148                     'metric_response')
 149             except Exception as exc:
 150                 log.warn("Failed to send delete response:%s", exc)
 151
 152         elif message.key == "update_metric_request":
 153             # Gnocchi doesn't support configuration updates
 154             # Log and send a response back to this effect
 155             log.warn("Gnocchi doesn't support metric configuration\
 156                       updates.")
 157             req_details = values['metric_create']
 158             metric_name = req_details['metric_name']
 159             resource_id = req_details['resource_uuid']
 160             metric_id = self.get_metric_id(
 161                 endpoint, auth_token, metric_name, resource_id)
 162
 163             # Generate and send a response message
 164             try:
 165                 resp_message = self._response.generate_response(
 166                     'update_metric_response', status=False,
 167                     cor_id=values['correlation_id'],
 168                     r_id=resource_id, m_id=metric_id)
 169                 log.info("Response message: %s", resp_message)
 170                 self._producer.update_metric_response(
 171                     'update_metric_response', resp_message,
 172                     'metric_response')
 173             except Exception as exc:
 174                 log.warn("Failed to send an update response:%s", exc)
 175
 176         elif message.key == "list_metric_request":
 177             list_details = values['metrics_list_request']
 178
 179             metric_list = self.list_metrics(
 180                 endpoint, auth_token, list_details)
 181
 182             # Generate and send a response message
 183             try:
 184                 resp_message = self._response.generate_response(
 185                     'list_metric_response', m_list=metric_list,
 186                     cor_id=list_details['correlation_id'])
 187                 log.info("Response message: %s", resp_message)
 188                 self._producer.list_metric_response(
 189                     'list_metric_response', resp_message,
 190                     'metric_response')
 191             except Exception as exc:
 192                 log.warn("Failed to send a list response:%s", exc)
 193
 194         else:
 195             log.warn("Unknown key, no action will be performed.")
 196
 197         return
 198
 199     def configure_metric(self, endpoint, auth_token, values):
 200         """Create the new metric in Gnocchi."""
 201         try:
 202             resource_id = values['resource_uuid']
 203         except KeyError:
 204             log.warn("Resource is not defined correctly.")
 205             return None, None, False
 206
 207         # Check/Normalize metric name
 208         norm_name, metric_name = self.get_metric_name(values)
 209         if metric_name is None:
 210             log.warn("This metric is not supported by this plugin.")
 211             return None, resource_id, False
 212
 213         # Check for an existing metric for this resource
 214         metric_id = self.get_metric_id(
 215             endpoint, auth_token, metric_name, resource_id)
 216
 217         if metric_id is None:
 218             # Try appending metric to existing resource
 219             try:
 220                 base_url = "{}/v1/resource/generic/%s/metric"
 221                 res_url = base_url.format(endpoint) % resource_id
 222                 payload = {metric_name: {'archive_policy_name': 'high',
 223                                          'unit': values['metric_unit']}}
 224                 result = Common.perform_request(
 225                     res_url, auth_token, req_type="post",
 226                     payload=json.dumps(payload))
 227                 # Get id of newly created metric
 228                 for row in json.loads(result.text):
 229                     if row['name'] == metric_name:
 230                         metric_id = row['id']
 231                 log.info("Appended metric to existing resource.")
 232
 233                 return metric_id, resource_id, True
 234             except Exception as exc:
 235                 # Gnocchi version of resource does not exist creating a new one
 236                 log.info("Failed to append metric to existing resource:%s",
 237                          exc)
 238                 try:
 239                     url = "{}/v1/resource/generic".format(endpoint)
 240                     metric = {'name': metric_name,
 241                               'archive_policy_name': 'high',
 242                               'unit': values['metric_unit'], }
 243
 244                     resource_payload = json.dumps({'id': resource_id,
 245                                                    'metrics': {
 246                                                        metric_name: metric}})
 247
 248                     resource = Common.perform_request(
 249                         url, auth_token, req_type="post",
 250                         payload=resource_payload)
 251
 252                     # Return the newly created resource_id for creating alarms
 253                     new_resource_id = json.loads(resource.text)['id']
 254                     log.info("Created new resource for metric: %s",
 255                              new_resource_id)
 256
 257                     metric_id = self.get_metric_id(
 258                         endpoint, auth_token, metric_name, new_resource_id)
 259
 260                     return metric_id, new_resource_id, True
 261                 except Exception as exc:
 262                     log.warn("Failed to create a new resource:%s", exc)
 263             return None, None, False
 264
 265         else:
 266             log.info("This metric already exists for this resource.")
 267
 268         return metric_id, resource_id, False
 269
 270     def delete_metric(self, endpoint, auth_token, metric_id):
 271         """Delete metric."""
 272         url = "{}/v1/metric/%s".format(endpoint) % metric_id
 273
 274         try:
 275             result = Common.perform_request(
 276                 url, auth_token, req_type="delete")
 277             if str(result.status_code) == "404":
 278                 log.warn("Failed to delete the metric.")
 279                 return False
 280             else:
 281                 return True
 282         except Exception as exc:
 283             log.warn("Failed to carry out delete metric request:%s", exc)
 284         return False
 285
 286     def list_metrics(self, endpoint, auth_token, values):
 287         """List all metrics."""
 288
 289         # Check for a specified list
 290         try:
 291             # Check if the metric_name was specified for the list
 292             metric_name = values['metric_name'].lower()
 293             if metric_name not in METRIC_MAPPINGS.keys():
 294                 log.warn("This metric is not supported, won't be listed.")
 295                 metric_name = None
 296         except KeyError as exc:
 297             log.info("Metric name is not specified: %s", exc)
 298             metric_name = None
 299
 300         try:
 301             resource = values['resource_uuid']
 302         except KeyError as exc:
 303             log.info("Resource is not specified:%s", exc)
 304             resource = None
 305
 306         try:
 307             url = "{}/v1/metric?sort=name:asc".format(endpoint)
 308             result = Common.perform_request(
 309                 url, auth_token, req_type="get")
 310             metrics = []
 311             metrics_partial = json.loads(result.text)
 312             for metric in metrics_partial:
 313                 metrics.append(metric)
 314
 315             while len(json.loads(result.text)) > 0:
 316                 last_metric_id = metrics_partial[-1]['id']
 317                 url = "{}/v1/metric?sort=name:asc&marker={}".format(endpoint, last_metric_id)
 318                 result = Common.perform_request(
 319                     url, auth_token, req_type="get")
 320                 if len(json.loads(result.text)) > 0:
 321                     metrics_partial = json.loads(result.text)
 322                     for metric in metrics_partial:
 323                         metrics.append(metric)
 324
 325             if metrics is not None:
 326                 # Format the list response
 327                 if metric_name is not None and resource is not None:
 328                     metric_list = self.response_list(
 329                         metrics, metric_name=metric_name, resource=resource)
 330                     log.info("Returning an %s resource list for %s metrics",
 331                              metric_name, resource)
 332                 elif metric_name is not None:
 333                     metric_list = self.response_list(
 334                         metrics, metric_name=metric_name)
 335                     log.info("Returning a list of %s metrics", metric_name)
 336                 elif resource is not None:
 337                     metric_list = self.response_list(
 338                         metrics, resource=resource)
 339                     log.info("Return a list of %s resource metrics", resource)
 340                 else:
 341                     metric_list = self.response_list(metrics)
 342                     log.info("Returning a complete list of metrics")
 343
 344                 return metric_list
 345             else:
 346                 log.info("There are no metrics available")
 347                 return []
 348         except Exception as exc:
 349             log.warn("Failed to generate any metric list. %s", exc)
 350         return None
 351
 352     def get_metric_id(self, endpoint, auth_token, metric_name, resource_id):
 353         """Check if the desired metric already exists for the resource."""
 354         url = "{}/v1/resource/generic/%s".format(endpoint) % resource_id
 355         try:
 356             # Try return the metric id if it exists
 357             result = Common.perform_request(
 358                 url, auth_token, req_type="get")
 359             return json.loads(result.text)['metrics'][metric_name]
 360         except Exception:
 361             log.info("Metric doesn't exist. No metric_id available")
 362         return None
 363
 364     def get_metric_name(self, values):
 365         """Check metric name configuration and normalize."""
 366         metric_name = None
 367         try:
 368             # Normalize metric name
 369             metric_name = values['metric_name'].lower()
 370             return metric_name, METRIC_MAPPINGS[metric_name]
 371         except KeyError:
 372             log.info("Metric name %s is invalid.", metric_name)
 373         return metric_name, None
 374
 375     def read_metric_data(self, endpoint, auth_token, values):
 376         """Collect metric measures over a specified time period."""
 377         timestamps = []
 378         data = []
 379         try:
 380             #get metric_id
 381             metric_id = self.get_metric_id(endpoint, auth_token, METRIC_MAPPINGS[values['metric_name']], values['resource_uuid'])
 382             # Try and collect measures
 383             collection_unit = values['collection_unit'].upper()
 384             collection_period = values['collection_period']
 385
 386             # Define the start and end time based on configurations
 387             # FIXME: Local timezone may differ from timezone set in Gnocchi, causing discrepancies in measures
 388             stop_time = time.strftime("%Y-%m-%d") + "T" + time.strftime("%X")
 389             end_time = int(round(time.time() * 1000))
 390             if collection_unit == 'YEAR':
 391                 diff = PERIOD_MS[collection_unit]
 392             else:
 393                 diff = collection_period * PERIOD_MS[collection_unit]
 394             s_time = (end_time - diff) / 1000.0
 395             start_time = datetime.datetime.fromtimestamp(s_time).strftime(
 396                 '%Y-%m-%dT%H:%M:%S.%f')
 397             base_url = "{}/v1/metric/%(0)s/measures?start=%(1)s&stop=%(2)s"
 398             url = base_url.format(endpoint) % {
 399                 "0": metric_id, "1": start_time, "2": stop_time}
 400
 401             # Perform metric data request
 402             metric_data = Common.perform_request(
 403                 url, auth_token, req_type="get")
 404
 405             # Generate a list of the requested timestamps and data
 406             for r in json.loads(metric_data.text):
 407                 timestamp = r[0].replace("T", " ")
 408                 timestamps.append(timestamp)
 409                 data.append(r[2])
 410
 411             return timestamps, data
 412         except Exception as exc:
 413             log.warn("Failed to gather specified measures: %s", exc)
 414         return timestamps, data
 415
 416     def response_list(self, metric_list, metric_name=None, resource=None):
 417         """Create the appropriate lists for a list response."""
 418         resp_list, name_list, res_list = [], [], []
 419
 420         # Create required lists
 421         for row in metric_list:
 422             # Only list OSM metrics
 423             name = None
 424             if row['name'] in METRIC_MAPPINGS.values():
 425                 for k,v in six.iteritems(METRIC_MAPPINGS):
 426                     if row['name'] == v:
 427                         name = k
 428                 metric = {"metric_name": name,
 429                           "metric_uuid": row['id'],
 430                           "metric_unit": row['unit'],
 431                           "resource_uuid": row['resource_id']}
 432                 resp_list.append(metric)
 433             # Generate metric_name specific list
 434             if metric_name is not None and name is not None:
 435                 if metric_name in METRIC_MAPPINGS.keys() and row['name'] == METRIC_MAPPINGS[metric_name]:
 436                     metric = {"metric_name": metric_name,
 437                               "metric_uuid": row['id'],
 438                               "metric_unit": row['unit'],
 439                               "resource_uuid": row['resource_id']}
 440                     name_list.append(metric)
 441             # Generate resource specific list
 442             if resource is not None and name is not None:
 443                 if row['resource_id'] == resource:
 444                     metric = {"metric_name": name,
 445                               "metric_uuid": row['id'],
 446                               "metric_unit": row['unit'],
 447                               "resource_uuid": row['resource_id']}
 448                     res_list.append(metric)
 449
 450         # Join required lists
 451         if metric_name is not None and resource is not None:
 452             # Return intersection of res_list and name_list
 453             return [i for i in res_list for j in name_list if i['metric_uuid'] == j['metric_uuid']]
 454         elif metric_name is not None:
 455             return name_list
 456         elif resource is not None:
 457             return res_list
 458         else:
 459             return resp_list