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."""
30 from osm_mon
.core
.message_bus
.producer
import KafkaProducer
32 from osm_mon
.plugins
.OpenStack
.response
import OpenStack_Response
33 from osm_mon
.plugins
.OpenStack
.settings
import Config
35 log
= logging
.getLogger(__name__
)
38 "average_memory_utilization": "memory.percent",
39 "disk_read_ops": "disk.disk_ops",
40 "disk_write_ops": "disk.disk_ops",
41 "disk_read_bytes": "disk.disk_octets",
42 "disk_write_bytes": "disk.disk_octets",
43 "packets_dropped": "interface.if_dropped",
44 "packets_received": "interface.if_packets",
45 "packets_sent": "interface.if_packets",
46 "cpu_utilization": "cpu.percent",
58 class Metrics(object):
59 """OpenStack metric requests performed via the Gnocchi API."""
62 """Initialize the metric actions."""
63 # Configure an instance of the OpenStack metric plugin
64 config
= Config
.instance()
65 config
.read_environ("gnocchi")
67 # Initialise authentication for API requests
68 self
.auth_token
= None
72 # Use the Response class to generate valid json response messages
73 self
._response
= OpenStack_Response()
75 # Initializer a producer to send responses back to SO
76 self
._producer
= KafkaProducer("metric_response")
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
)
82 log
.info("OpenStack metric action required.")
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
90 log
.info("Auth_token has not been updated.")
92 log
.info("Using environment variables to set Gnocchi auth_token.")
93 self
.auth_token
= self
._common
._authenticate
()
95 if self
.endpoint
is None:
96 log
.info("Generating a new endpoint for Gnocchi.")
97 self
.endpoint
= self
._common
.get_endpoint("metric")
99 if message
.key
== "create_metric_request":
101 metric_details
= values
['metric_create']
102 metric_id
, resource_id
, status
= self
.configure_metric(
103 self
.endpoint
, self
.auth_token
, metric_details
)
105 # Generate and send a create metric response
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
,
115 except Exception as exc
:
116 log
.warn("Failed to create response: %s", exc
)
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
)
123 # Generate and send a response message
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
,
136 except Exception as exc
:
137 log
.warn("Failed to send read metric response:%s", exc
)
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
)
145 # Generate and send a response message
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
,
156 except Exception as exc
:
157 log
.warn("Failed to send delete response:%s", exc
)
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\
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
)
170 # Generate and send a response message
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
,
180 except Exception as exc
:
181 log
.warn("Failed to send an update response:%s", exc
)
183 elif message
.key
== "list_metric_request":
184 list_details
= values
['metrics_list_request']
186 metric_list
= self
.list_metrics(
187 self
.endpoint
, self
.auth_token
, list_details
)
189 # Generate and send a response message
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
,
198 except Exception as exc
:
199 log
.warn("Failed to send a list response:%s", exc
)
202 log
.warn("Unknown key, no action will be performed.")
206 def configure_metric(self
, endpoint
, auth_token
, values
):
207 """Create the new metric in Gnocchi."""
209 resource_id
= values
['resource_uuid']
211 log
.warn("Resource is not defined correctly.")
212 return None, None, False
214 # Check/Normalize metric name
215 metric_name
, norm_name
= self
.get_metric_name(values
)
216 if norm_name
is None:
217 log
.warn("This metric is not supported by this plugin.")
218 return None, resource_id
, False
220 # Check for an existing metric for this resource
221 metric_id
= self
.get_metric_id(
222 endpoint
, auth_token
, metric_name
, resource_id
)
224 if metric_id
is None:
225 # Try appending metric to existing resource
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.")
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",
246 url
= "{}/v1/resource/generic".format(endpoint
)
247 metric
= {'name': metric_name
,
248 'archive_policy_name': 'high',
249 'unit': values
['metric_unit'], }
251 resource_payload
= json
.dumps({'id': resource_id
,
253 metric_name
: metric
}})
255 resource
= self
._common
._perform
_request
(
256 url
, auth_token
, req_type
="post",
257 payload
=resource_payload
)
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",
264 metric_id
= self
.get_metric_id(
265 endpoint
, auth_token
, metric_name
, new_resource_id
)
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
273 log
.info("This metric already exists for this resource.")
275 return metric_id
, resource_id
, False
277 def delete_metric(self
, endpoint
, auth_token
, metric_id
):
279 url
= "{}/v1/metric/%s".format(endpoint
) % (metric_id
)
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.")
289 except Exception as exc
:
290 log
.warn("Failed to carry out delete metric request:%s", exc
)
293 def list_metrics(self
, endpoint
, auth_token
, values
):
294 """List all metrics."""
295 url
= "{}/v1/metric/".format(endpoint
)
297 # Check for a specified list
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.")
304 except KeyError as exc
:
305 log
.info("Metric name is not specified: %s", exc
)
309 resource
= values
['resource_uuid']
310 except KeyError as exc
:
311 log
.info("Resource is not specified:%s", exc
)
315 result
= self
._common
._perform
_request
(
316 url
, auth_token
, req_type
="get")
317 metrics
= json
.loads(result
.text
)
319 if metrics
is not None:
320 # Format the list response
321 if metric_name
is not None and resource
is not None:
322 metric_list
= self
.response_list(
323 metrics
, metric_name
=metric_name
, resource
=resource
)
324 log
.info("Returning an %s resource list for %s metrics",
325 metric_name
, resource
)
326 elif metric_name
is not None:
327 metric_list
= self
.response_list(
328 metrics
, metric_name
=metric_name
)
329 log
.info("Returning a list of %s metrics", metric_name
)
330 elif resource
is not None:
331 metric_list
= self
.response_list(
332 metrics
, resource
=resource
)
333 log
.info("Return a list of %s resource metrics", resource
)
335 metric_list
= self
.response_list(metrics
)
336 log
.info("Returning a complete list of metrics")
340 log
.info("There are no metrics available")
342 except Exception as exc
:
343 log
.warn("Failed to generate any metric list. %s", exc
)
346 def get_metric_id(self
, endpoint
, auth_token
, metric_name
, resource_id
):
347 """Check if the desired metric already exists for the resource."""
348 url
= "{}/v1/resource/generic/%s".format(endpoint
) % resource_id
351 # Try return the metric id if it exists
352 result
= self
._common
._perform
_request
(
353 url
, auth_token
, req_type
="get")
354 return json
.loads(result
.text
)['metrics'][metric_name
]
356 log
.info("Metric doesn't exist. No metric_id available")
359 def get_metric_name(self
, values
):
360 """Check metric name configuration and normalize."""
362 # Normalize metric name
363 metric_name
= values
['metric_name'].lower()
364 return metric_name
, METRIC_MAPPINGS
[metric_name
]
366 log
.info("Metric name %s is invalid.", metric_name
)
367 return metric_name
, None
369 def read_metric_data(self
, endpoint
, auth_token
, values
):
370 """Collectd metric measures over a specified time period."""
374 # Try and collect measures
375 metric_id
= values
['metric_uuid']
376 collection_unit
= values
['collection_unit'].upper()
377 collection_period
= values
['collection_period']
379 # Define the start and end time based on configurations
380 stop_time
= time
.strftime("%Y-%m-%d") + "T" + time
.strftime("%X")
381 end_time
= int(round(time
.time() * 1000))
382 if collection_unit
== 'YEAR':
383 diff
= PERIOD_MS
[collection_unit
]
385 diff
= collection_period
* PERIOD_MS
[collection_unit
]
386 s_time
= (end_time
- diff
) / 1000.0
387 start_time
= datetime
.datetime
.fromtimestamp(s_time
).strftime(
388 '%Y-%m-%dT%H:%M:%S.%f')
389 base_url
= "{}/v1/metric/%(0)s/measures?start=%(1)s&stop=%(2)s"
390 url
= base_url
.format(endpoint
) % {
391 "0": metric_id
, "1": start_time
, "2": stop_time
}
393 # Perform metric data request
394 metric_data
= self
._common
._perform
_request
(
395 url
, auth_token
, req_type
="get")
397 # Generate a list of the requested timestamps and data
398 for r
in json
.loads(metric_data
.text
):
399 timestamp
= r
[0].replace("T", " ")
400 timestamps
.append(timestamp
)
403 return timestamps
, data
404 except Exception as exc
:
405 log
.warn("Failed to gather specified measures: %s", exc
)
406 return timestamps
, data
408 def response_list(self
, metric_list
, metric_name
=None, resource
=None):
409 """Create the appropriate lists for a list response."""
410 resp_list
, name_list
, res_list
= [], [], []
412 # Create required lists
413 for row
in metric_list
:
414 # Only list OSM metrics
415 if row
['name'] in METRIC_MAPPINGS
.keys():
416 metric
= {"metric_name": row
['name'],
417 "metric_uuid": row
['id'],
418 "metric_unit": row
['unit'],
419 "resource_uuid": row
['resource_id']}
420 resp_list
.append(str(metric
))
421 # Generate metric_name specific list
422 if metric_name
is not None:
423 if row
['name'] == metric_name
:
424 metric
= {"metric_name": row
['name'],
425 "metric_uuid": row
['id'],
426 "metric_unit": row
['unit'],
427 "resource_uuid": row
['resource_id']}
428 name_list
.append(str(metric
))
429 # Generate resource specific list
430 if resource
is not None:
431 if row
['resource_id'] == resource
:
432 metric
= {"metric_name": row
['name'],
433 "metric_uuid": row
['id'],
434 "metric_unit": row
['unit'],
435 "resource_uuid": row
['resource_id']}
436 res_list
.append(str(metric
))
438 # Join required lists
439 if metric_name
is not None and resource
is not None:
440 return list(set(res_list
).intersection(name_list
))
441 elif metric_name
is not None:
443 elif resource
is not None:
444 return list(set(res_list
).intersection(resp_list
))