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                         log.info("Response messages: %s", resp_message)
 106                         self._producer.create_metrics_resp(
 107                             'create_metric_response', resp_message,
 108                             'metric_response')
 109                     except Exception as exc:
 110                         log.warn("Failed to create response: %s", exc)
 111
 112                 elif message.key == "read_metric_data_request":
 113                     # Read all metric data related to a specified metric
 114                     timestamps, metric_data = self.read_metric_data(
 115                         endpoint, auth_token, values)
 116
 117                     # Generate and send a response message
 118                     try:
 119                         resp_message = self._response.generate_response(
 120                             'read_metric_data_response',
 121                             m_id=values['metric_uuid'],
 122                             m_name=values['metric_name'],
 123                             r_id=values['resource_uuid'],
 124                             cor_id=values['correlation_id'],
 125                             times=timestamps, metrics=metric_data)
 126                         log.info("Response message: %s", resp_message)
 127                         self._producer.read_metric_data_response(
 128                             'read_metric_data_response', resp_message,
 129                             'metric_response')
 130                     except Exception as exc:
 131                         log.warn("Failed to send read metric response:%s", exc)
 132
 133                 elif message.key == "delete_metric_request":
 134                     # delete the specified metric in the request
 135                     metric_id = values['metric_uuid']
 136                     status = self.delete_metric(
 137                         endpoint, auth_token, metric_id)
 138
 139                     # Generate and send a response message
 140                     try:
 141                         resp_message = self._response.generate_response(
 142                             'delete_metric_response', m_id=metric_id,
 143                             m_name=values['metric_name'],
 144                             status=status, r_id=values['resource_uuid'],
 145                             cor_id=values['correlation_id'])
 146                         log.info("Response message: %s", resp_message)
 147                         self._producer.delete_metric_response(
 148                             'delete_metric_response', resp_message,
 149                             'metric_response')
 150                     except Exception as exc:
 151                         log.warn("Failed to send delete response:%s", exc)
 152
 153                 elif message.key == "update_metric_request":
 154                     # Gnocchi doesn't support configuration updates
 155                     # Log and send a response back to this effect
 156                     log.warn("Gnocchi doesn't support metric configuration\
 157                               updates.")
 158                     req_details = values['metric_create']
 159                     metric_name = req_details['metric_name']
 160                     resource_id = req_details['resource_uuid']
 161                     metric_id = self.get_metric_id(
 162                         endpoint, auth_token, metric_name, resource_id)
 163
 164                     # Generate and send a response message
 165                     try:
 166                         resp_message = self._response.generate_response(
 167                             'update_metric_response', status=False,
 168                             cor_id=values['correlation_id'],
 169                             r_id=resource_id, m_id=metric_id)
 170                         log.info("Response message: %s", resp_message)
 171                         self._producer.update_metric_response(
 172                             'update_metric_response', resp_message,
 173                             'metric_response')
 174                     except Exception as exc:
 175                         log.warn("Failed to send an update response:%s", exc)
 176
 177                 elif message.key == "list_metric_request":
 178                     list_details = values['metrics_list_request']
 179
 180                     metric_list = self.list_metrics(
 181                         endpoint, auth_token, list_details)
 182
 183                     # Generate and send a response message
 184                     try:
 185                         resp_message = self._response.generate_response(
 186                             'list_metric_response', m_list=metric_list,
 187                             cor_id=list_details['correlation_id'])
 188                         log.info("Response message: %s", resp_message)
 189                         self._producer.list_metric_response(
 190                             'list_metric_response', resp_message,
 191                             'metric_response')
 192                     except Exception as exc:
 193                         log.warn("Failed to send a list response:%s", exc)
 194
 195                 else:
 196                     log.warn("Unknown key, no action will be performed.")
 197             else:
 198                 log.debug("Message is not for this OpenStack.")
 199
 200         return
 201
 202     def configure_metric(self, endpoint, auth_token, values):
 203         """Create the new metric in Gnocchi."""
 204         try:
 205             resource_id = values['resource_uuid']
 206         except KeyError:
 207             log.warn("Resource is not defined correctly.")
 208             return None, None, False
 209
 210         # Check/Normalize metric name
 211         metric_name, norm_name = self.get_metric_name(values)
 212         if norm_name is None:
 213             log.warn("This metric is not supported by this plugin.")
 214             return None, resource_id, False
 215
 216         # Check for an existing metric for this resource
 217         metric_id = self.get_metric_id(
 218             endpoint, auth_token, metric_name, resource_id)
 219
 220         if metric_id is None:
 221             # Try appending metric to existing resource
 222             try:
 223                 base_url = "{}/v1/resource/generic/%s/metric"
 224                 res_url = base_url.format(endpoint) % resource_id
 225                 payload = {metric_name: {'archive_policy_name': 'high',
 226                                          'unit': values['metric_unit']}}
 227                 result = self._common._perform_request(
 228                     res_url, auth_token, req_type="post",
 229                     payload=json.dumps(payload))
 230                 # Get id of newly created metric
 231                 for row in json.loads(result.text):
 232                     if row['name'] == metric_name:
 233                         metric_id = row['id']
 234                 log.info("Appended metric to existing resource.")
 235
 236                 return metric_id, resource_id, True
 237             except Exception as exc:
 238                 # Gnocchi version of resource does not exist creating a new one
 239                 log.info("Failed to append metric to existing resource:%s",
 240                          exc)
 241                 try:
 242                     url = "{}/v1/resource/generic".format(endpoint)
 243                     metric = {'name': metric_name,
 244                               'archive_policy_name': 'high',
 245                               'unit': values['metric_unit'], }
 246
 247                     resource_payload = json.dumps({'id': resource_id,
 248                                                    'metrics': {
 249                                                        metric_name: metric}})
 250
 251                     resource = self._common._perform_request(
 252                         url, auth_token, req_type="post",
 253                         payload=resource_payload)
 254
 255                     # Return the newly created resource_id for creating alarms
 256                     new_resource_id = json.loads(resource.text)['id']
 257                     log.info("Created new resource for metric: %s",
 258                              new_resource_id)
 259
 260                     metric_id = self.get_metric_id(
 261                         endpoint, auth_token, metric_name, new_resource_id)
 262
 263                     return metric_id, new_resource_id, True
 264                 except Exception as exc:
 265                     log.warn("Failed to create a new resource:%s", exc)
 266             return None, None, False
 267
 268         else:
 269             log.info("This metric already exists for this resource.")
 270
 271         return metric_id, resource_id, False
 272
 273     def delete_metric(self, endpoint, auth_token, metric_id):
 274         """Delete metric."""
 275         url = "{}/v1/metric/%s".format(endpoint) % (metric_id)
 276
 277         try:
 278             result = self._common._perform_request(
 279                 url, auth_token, req_type="delete")
 280             if str(result.status_code) == "404":
 281                 log.warn("Failed to delete the metric.")
 282                 return False
 283             else:
 284                 return True
 285         except Exception as exc:
 286             log.warn("Failed to carry out delete metric request:%s", exc)
 287         return False
 288
 289     def list_metrics(self, endpoint, auth_token, values):
 290         """List all metrics."""
 291         url = "{}/v1/metric/".format(endpoint)
 292
 293         # Check for a specified list
 294         try:
 295             # Check if the metric_name was specified for the list
 296             metric_name = values['metric_name'].lower()
 297             if metric_name not in METRIC_MAPPINGS.keys():
 298                 log.warn("This metric is not supported, won't be listed.")
 299                 metric_name = None
 300         except KeyError as exc:
 301             log.info("Metric name is not specified: %s", exc)
 302             metric_name = None
 303
 304         try:
 305             resource = values['resource_uuid']
 306         except KeyError as exc:
 307             log.info("Resource is not specified:%s", exc)
 308             resource = None
 309
 310         try:
 311             result = self._common._perform_request(
 312                 url, auth_token, req_type="get")
 313             metrics = json.loads(result.text)
 314
 315             if metrics is not None:
 316                 # Format the list response
 317                 if metric_name is not None and resource is not None:
 318                     metric_list = self.response_list(
 319                         metrics, metric_name=metric_name, resource=resource)
 320                     log.info("Returning an %s resource list for %s metrics",
 321                              metric_name, resource)
 322                 elif metric_name is not None:
 323                     metric_list = self.response_list(
 324                         metrics, metric_name=metric_name)
 325                     log.info("Returning a list of %s metrics", metric_name)
 326                 elif resource is not None:
 327                     metric_list = self.response_list(
 328                         metrics, resource=resource)
 329                     log.info("Return a list of %s resource metrics", resource)
 330                 else:
 331                     metric_list = self.response_list(metrics)
 332                     log.info("Returning a complete list of metrics")
 333
 334                 return metric_list
 335             else:
 336                 log.info("There are no metrics available")
 337                 return []
 338         except Exception as exc:
 339             log.warn("Failed to generate any metric list. %s", exc)
 340         return None
 341
 342     def get_metric_id(self, endpoint, auth_token, metric_name, resource_id):
 343         """Check if the desired metric already exists for the resource."""
 344         url = "{}/v1/resource/generic/%s".format(endpoint) % resource_id
 345
 346         try:
 347             # Try return the metric id if it exists
 348             result = self._common._perform_request(
 349                 url, auth_token, req_type="get")
 350             return json.loads(result.text)['metrics'][metric_name]
 351         except Exception:
 352             log.info("Metric doesn't exist. No metric_id available")
 353         return None
 354
 355     def get_metric_name(self, values):
 356         """Check metric name configuration and normalize."""
 357         try:
 358             # Normalize metric name
 359             metric_name = values['metric_name'].lower()
 360             return metric_name, METRIC_MAPPINGS[metric_name]
 361         except KeyError:
 362             log.info("Metric name %s is invalid.", metric_name)
 363         return metric_name, None
 364
 365     def read_metric_data(self, endpoint, auth_token, values):
 366         """Collectd metric measures over a specified time period."""
 367         timestamps = []
 368         data = []
 369         try:
 370             # Try and collect measures
 371             metric_id = values['metric_uuid']
 372             collection_unit = values['collection_unit'].upper()
 373             collection_period = values['collection_period']
 374
 375             # Define the start and end time based on configurations
 376             stop_time = time.strftime("%Y-%m-%d") + "T" + time.strftime("%X")
 377             end_time = int(round(time.time() * 1000))
 378             if collection_unit == 'YEAR':
 379                 diff = PERIOD_MS[collection_unit]
 380             else:
 381                 diff = collection_period * PERIOD_MS[collection_unit]
 382             s_time = (end_time - diff) / 1000.0
 383             start_time = datetime.datetime.fromtimestamp(s_time).strftime(
 384                 '%Y-%m-%dT%H:%M:%S.%f')
 385             base_url = "{}/v1/metric/%(0)s/measures?start=%(1)s&stop=%(2)s"
 386             url = base_url.format(endpoint) % {
 387                 "0": metric_id, "1": start_time, "2": stop_time}
 388
 389             # Perform metric data request
 390             metric_data = self._common._perform_request(
 391                 url, auth_token, req_type="get")
 392
 393             # Generate a list of the requested timestamps and data
 394             for r in json.loads(metric_data.text):
 395                 timestamp = r[0].replace("T", " ")
 396                 timestamps.append(timestamp)
 397                 data.append(r[2])
 398
 399             return timestamps, data
 400         except Exception as exc:
 401             log.warn("Failed to gather specified measures: %s", exc)
 402         return timestamps, data
 403
 404     def authenticate(self):
 405         """Generate an authentication token and endpoint for metric request."""
 406         try:
 407             # Check for a tenant_id
 408             auth_token = self._common._authenticate()
 409             endpoint = self._common.get_endpoint("metric")
 410             return auth_token, endpoint
 411         except Exception as exc:
 412             log.warn("Authentication to Keystone failed: %s", exc)
 413
 414         return None, None
 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             if row['name'] in METRIC_MAPPINGS.keys():
 424                 metric = {"metric_name": row['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:
 431                 if row['name'] == metric_name:
 432                     metric = {"metric_name": row['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:
 439                 if row['resource_id'] == resource:
 440                     metric = {"metric_name": row['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             return list(set(res_list).intersection(name_list))
 449         elif metric_name is not None:
 450             return name_list
 451         elif resource is not None:
 452             return list(set(res_list).intersection(resp_list))
 453         else:
 454             return resp_list