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