1 # Copyright 2017 Intel Research and Development Ireland Limited
2 # *************************************************************
4 # This file is part of OSM Monitoring module
5 # All Rights Reserved to Intel Corporation
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
11 # http://www.apache.org/licenses/LICENSE-2.0
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
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
22 """Carry out OpenStack metric requests via Gnocchi API."""
27 log
= logging
.getLogger(__name__
)
31 from core
.message_bus
.producer
import KafkaProducer
33 from kafka
import KafkaConsumer
35 from plugins
.OpenStack
.common
import Common
36 from plugins
.OpenStack
.response
import OpenStack_Response
38 __author__
= "Helena McGough"
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",
61 class Metrics(object):
62 """OpenStack metric requests performed via the Gnocchi API."""
65 """Initialize the metric actions."""
66 self
._common
= Common()
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'],
73 bootstrap_servers
=server
['server'])
75 # Use the Response class to generate valid json response messages
76 self
._response
= OpenStack_Response()
78 # Initializer a producer to send responses back to SO
79 self
._producer
= KafkaProducer("metric_response")
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()
89 if vim_type
== "openstack":
90 # Generate auth_token and endpoint
91 auth_token
, endpoint
= self
.authenticate()
93 if message
.key
== "create_metric_request":
95 metric_details
= values
['metric_create']
96 metric_id
, resource_id
, status
= self
.configure_metric(
97 endpoint
, auth_token
, metric_details
)
99 # Generate and send a create metric response
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
,
109 except Exception as exc
:
110 log
.warn("Failed to create response: %s", exc
)
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
)
117 # Generate and send a response message
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
,
130 except Exception as exc
:
131 log
.warn("Failed to send read metric response:%s", exc
)
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
)
139 # Generate and send a response message
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
,
150 except Exception as exc
:
151 log
.warn("Failed to send delete response:%s", exc
)
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\
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
)
164 # Generate and send a response message
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
,
174 except Exception as exc
:
175 log
.warn("Failed to send an update response:%s", exc
)
177 elif message
.key
== "list_metric_request":
178 list_details
= values
['metrics_list_request']
180 metric_list
= self
.list_metrics(
181 endpoint
, auth_token
, list_details
)
183 # Generate and send a response message
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
,
192 except Exception as exc
:
193 log
.warn("Failed to send a list response:%s", exc
)
196 log
.warn("Unknown key, no action will be performed.")
198 log
.debug("Message is not for this OpenStack.")
202 def configure_metric(self
, endpoint
, auth_token
, values
):
203 """Create the new metric in Gnocchi."""
205 resource_id
= values
['resource_uuid']
207 log
.warn("Resource is not defined correctly.")
208 return None, None, False
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
216 # Check for an existing metric for this resource
217 metric_id
= self
.get_metric_id(
218 endpoint
, auth_token
, metric_name
, resource_id
)
220 if metric_id
is None:
221 # Try appending metric to existing resource
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.")
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",
242 url
= "{}/v1/resource/generic".format(endpoint
)
243 metric
= {'name': metric_name
,
244 'archive_policy_name': 'high',
245 'unit': values
['metric_unit'], }
247 resource_payload
= json
.dumps({'id': resource_id
,
249 metric_name
: metric
}})
251 resource
= self
._common
._perform
_request
(
252 url
, auth_token
, req_type
="post",
253 payload
=resource_payload
)
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",
260 metric_id
= self
.get_metric_id(
261 endpoint
, auth_token
, metric_name
, new_resource_id
)
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
269 log
.info("This metric already exists for this resource.")
271 return metric_id
, resource_id
, False
273 def delete_metric(self
, endpoint
, auth_token
, metric_id
):
275 url
= "{}/v1/metric/%s".format(endpoint
) % (metric_id
)
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.")
285 except Exception as exc
:
286 log
.warn("Failed to carry out delete metric request:%s", exc
)
289 def list_metrics(self
, endpoint
, auth_token
, values
):
290 """List all metrics."""
291 url
= "{}/v1/metric/".format(endpoint
)
293 # Check for a specified list
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.")
300 except KeyError as exc
:
301 log
.info("Metric name is not specified: %s", exc
)
305 resource
= values
['resource_uuid']
306 except KeyError as exc
:
307 log
.info("Resource is not specified:%s", exc
)
311 result
= self
._common
._perform
_request
(
312 url
, auth_token
, req_type
="get")
313 metrics
= json
.loads(result
.text
)
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
)
331 metric_list
= self
.response_list(metrics
)
332 log
.info("Returning a complete list of metrics")
336 log
.info("There are no metrics available")
338 except Exception as exc
:
339 log
.warn("Failed to generate any metric list. %s", exc
)
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
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
]
352 log
.info("Metric doesn't exist. No metric_id available")
355 def get_metric_name(self
, values
):
356 """Check metric name configuration and normalize."""
358 # Normalize metric name
359 metric_name
= values
['metric_name'].lower()
360 return metric_name
, METRIC_MAPPINGS
[metric_name
]
362 log
.info("Metric name %s is invalid.", metric_name
)
363 return metric_name
, None
365 def read_metric_data(self
, endpoint
, auth_token
, values
):
366 """Collectd metric measures over a specified time period."""
370 # Try and collect measures
371 metric_id
= values
['metric_uuid']
372 collection_unit
= values
['collection_unit'].upper()
373 collection_period
= values
['collection_period']
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
]
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
}
389 # Perform metric data request
390 metric_data
= self
._common
._perform
_request
(
391 url
, auth_token
, req_type
="get")
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
)
399 return timestamps
, data
400 except Exception as exc
:
401 log
.warn("Failed to gather specified measures: %s", exc
)
402 return timestamps
, data
404 def authenticate(self
):
405 """Generate an authentication token and endpoint for metric request."""
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
)
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
= [], [], []
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
))
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:
451 elif resource
is not None:
452 return list(set(res_list
).intersection(resp_list
))