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