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."""
29 from core
.message_bus
.producer
import KafkaProducer
31 from kafka
import KafkaConsumer
33 from plugins
.OpenStack
.common
import Common
34 from plugins
.OpenStack
.response
import OpenStack_Response
36 __author__
= "Helena McGough"
39 "average_memory_utilization": "memory.percent",
40 "disk_read_ops": "disk.disk_ops",
41 "disk_write_ops": "disk.disk_ops",
42 "disk_read_bytes": "disk.disk_octets",
43 "disk_write_bytes": "disk.disk_octets",
44 "packets_dropped": "interface.if_dropped",
45 "packets_received": "interface.if_packets",
46 "packets_sent": "interface.if_packets",
47 "cpu_utilization": "cpu.percent",
59 class Metrics(object):
60 """OpenStack metric requests performed via the Gnocchi API."""
63 """Initialize the metric actions."""
64 self
._common
= Common()
66 # TODO(mcgoughh): Initialize a generic consumer object to consume
67 # message from the SO. This is hardcoded for now
68 server
= {'server': 'localhost:9092', 'topic': 'metric_request'}
69 self
._consumer
= KafkaConsumer(server
['topic'],
71 bootstrap_servers
=server
['server'])
73 # Use the Response class to generate valid json response messages
74 self
._response
= OpenStack_Response()
76 # Initializer a producer to send responses back to SO
77 self
._producer
= KafkaProducer("metric_response")
79 def metric_calls(self
):
80 """Consume info from the message bus to manage metric requests."""
81 # Consumer check for metric messages
82 for message
in self
._consumer
:
83 # Check if this plugin should carry out this request
84 values
= json
.loads(message
.value
)
85 vim_type
= values
['vim_type'].lower()
87 if vim_type
== "openstack":
88 # Generate auth_token and endpoint
89 auth_token
, endpoint
= self
.authenticate()
91 if message
.key
== "create_metric_request":
93 metric_details
= values
['metric_create']
94 metric_id
, resource_id
, status
= self
.configure_metric(
95 endpoint
, auth_token
, metric_details
)
97 # Generate and send a create metric response
99 resp_message
= self
._response
.generate_response(
100 'create_metric_response', status
=status
,
101 cor_id
=values
['correlation_id'],
102 metric_id
=metric_id
, r_id
=resource_id
)
103 self
._producer
.create_metrics_resp(
104 'create_metric_response', resp_message
,
106 except Exception as exc
:
107 log
.warn("Failed to create response: %s", exc
)
109 elif message
.key
== "read_metric_data_request":
110 # Read all metric data related to a specified metric
111 timestamps
, metric_data
= self
.read_metric_data(
112 endpoint
, auth_token
, values
)
114 # Generate and send a response message
116 resp_message
= self
._response
.generate_response(
117 'read_metric_data_response',
118 m_id
=values
['metric_uuid'],
119 m_name
=values
['metric_name'],
120 r_id
=values
['resource_uuid'],
121 cor_id
=values
['correlation_id'],
122 times
=timestamps
, metrics
=metric_data
)
123 self
._producer
.read_metric_data_response(
124 'read_metric_data_response', resp_message
,
126 except Exception as exc
:
127 log
.warn("Failed to send read metric response:%s", exc
)
129 elif message
.key
== "delete_metric_request":
130 # delete the specified metric in the request
131 metric_id
= values
['metric_uuid']
132 status
= self
.delete_metric(
133 endpoint
, auth_token
, metric_id
)
135 # Generate and send a response message
137 resp_message
= self
._response
.generate_response(
138 'delete_metric_response', m_id
=metric_id
,
139 m_name
=values
['metric_name'],
140 status
=status
, r_id
=values
['resource_uuid'],
141 cor_id
=values
['correlation_id'])
142 self
._producer
.delete_metric_response(
143 'delete_metric_response', resp_message
,
145 except Exception as exc
:
146 log
.warn("Failed to send delete response:%s", exc
)
148 elif message
.key
== "update_metric_request":
149 # Gnocchi doesn't support configuration updates
150 # Log and send a response back to this effect
151 log
.warn("Gnocchi doesn't support metric configuration\
153 req_details
= values
['metric_create']
154 metric_name
= req_details
['metric_name']
155 resource_id
= req_details
['resource_uuid']
156 metric_id
= self
.get_metric_id(
157 endpoint
, auth_token
, metric_name
, resource_id
)
159 # Generate and send a response message
161 resp_message
= self
._response
.generate_response(
162 'update_metric_response', status
=False,
163 cor_id
=values
['correlation_id'],
164 r_id
=resource_id
, m_id
=metric_id
)
165 self
._producer
.update_metric_response(
166 'update_metric_response', resp_message
,
168 except Exception as exc
:
169 log
.warn("Failed to send an update response:%s", exc
)
171 elif message
.key
== "list_metric_request":
172 list_details
= values
['metrics_list_request']
174 metric_list
= self
.list_metrics(
175 endpoint
, auth_token
, list_details
)
177 # Generate and send a response message
179 resp_message
= self
._response
.generate_response(
180 'list_metric_response', m_list
=metric_list
,
181 cor_id
=list_details
['correlation_id'])
182 self
._producer
.list_metric_response(
183 'list_metric_response', resp_message
,
185 except Exception as exc
:
186 log
.warn("Failed to send a list response:%s", exc
)
189 log
.warn("Unknown key, no action will be performed.")
191 log
.debug("Message is not for this OpenStack.")
195 def configure_metric(self
, endpoint
, auth_token
, values
):
196 """Create the new metric in Gnocchi."""
198 resource_id
= values
['resource_uuid']
200 log
.warn("Resource is not defined correctly.")
201 return None, None, False
203 # Check/Normalize metric name
204 metric_name
, norm_name
= self
.get_metric_name(values
)
205 if norm_name
is None:
206 log
.warn("This metric is not supported by this plugin.")
207 return None, resource_id
, False
209 # Check for an existing metric for this resource
210 metric_id
= self
.get_metric_id(
211 endpoint
, auth_token
, metric_name
, resource_id
)
213 if metric_id
is None:
214 # Try appending metric to existing resource
216 base_url
= "{}/v1/resource/generic/%s/metric"
217 res_url
= base_url
.format(endpoint
) % resource_id
218 payload
= {metric_name
: {'archive_policy_name': 'high',
219 'unit': values
['metric_unit']}}
220 result
= self
._common
._perform
_request
(
221 res_url
, auth_token
, req_type
="post",
222 payload
=json
.dumps(payload
))
223 # Get id of newly created metric
224 for row
in json
.loads(result
.text
):
225 if row
['name'] == metric_name
:
226 metric_id
= row
['id']
227 log
.info("Appended metric to existing resource.")
229 return metric_id
, resource_id
, True
230 except Exception as exc
:
231 # Gnocchi version of resource does not exist creating a new one
232 log
.info("Failed to append metric to existing resource:%s",
235 url
= "{}/v1/resource/generic".format(endpoint
)
236 metric
= {'name': metric_name
,
237 'archive_policy_name': 'high',
238 'unit': values
['metric_unit'], }
240 resource_payload
= json
.dumps({'id': resource_id
,
242 metric_name
: metric
}})
244 resource
= self
._common
._perform
_request
(
245 url
, auth_token
, req_type
="post",
246 payload
=resource_payload
)
248 # Return the newly created resource_id for creating alarms
249 new_resource_id
= json
.loads(resource
.text
)['id']
250 log
.info("Created new resource for metric: %s",
253 metric_id
= self
.get_metric_id(
254 endpoint
, auth_token
, metric_name
, new_resource_id
)
256 return metric_id
, new_resource_id
, True
257 except Exception as exc
:
258 log
.warn("Failed to create a new resource:%s", exc
)
259 return None, None, False
262 log
.info("This metric already exists for this resource.")
264 return metric_id
, resource_id
, False
266 def delete_metric(self
, endpoint
, auth_token
, metric_id
):
268 url
= "{}/v1/metric/%s".format(endpoint
) % (metric_id
)
271 result
= self
._common
._perform
_request
(
272 url
, auth_token
, req_type
="delete")
273 if str(result
.status_code
) == "404":
274 log
.warn("Failed to delete the metric.")
278 except Exception as exc
:
279 log
.warn("Failed to carry out delete metric request:%s", exc
)
282 def list_metrics(self
, endpoint
, auth_token
, values
):
283 """List all metrics."""
284 url
= "{}/v1/metric/".format(endpoint
)
287 # Check if the metric_name was specified for the list
288 metric_name
= values
['metric_name']
289 result
= self
._common
._perform
_request
(
290 url
, auth_token
, req_type
="get")
291 metric_list
= json
.loads(result
.text
)
293 # Format the list response
294 metrics
= self
.response_list(
295 metric_list
, metric_name
=metric_name
)
298 log
.debug("Metric name is not specified for this list.")
301 # Check if a resource_id was specified
302 resource_id
= values
['resource_uuid']
303 result
= self
._common
._perform
_request
(
304 url
, auth_token
, req_type
="get")
305 metric_list
= json
.loads(result
.text
)
306 # Format the list response
307 metrics
= self
.response_list(
308 metric_list
, resource
=resource_id
)
311 log
.debug("Resource id not specificed either, will return a\
314 result
= self
._common
._perform
_request
(
315 url
, auth_token
, req_type
="get")
316 metric_list
= json
.loads(result
.text
)
317 # Format the list response
318 metrics
= self
.response_list(metric_list
)
321 except Exception as exc
:
322 log
.warn("Failed to generate any metric list. %s", exc
)
325 def get_metric_id(self
, endpoint
, auth_token
, metric_name
, resource_id
):
326 """Check if the desired metric already exists for the resource."""
327 url
= "{}/v1/resource/generic/%s".format(endpoint
) % resource_id
330 # Try return the metric id if it exists
331 result
= self
._common
._perform
_request
(
332 url
, auth_token
, req_type
="get")
333 return json
.loads(result
.text
)['metrics'][metric_name
]
335 log
.info("Metric doesn't exist. No metric_id available")
338 def get_metric_name(self
, values
):
339 """Check metric name configuration and normalize."""
341 # Normalize metric name
342 metric_name
= values
['metric_name'].lower()
343 return metric_name
, METRIC_MAPPINGS
[metric_name
]
345 log
.info("Metric name %s is invalid.", metric_name
)
346 return metric_name
, None
348 def read_metric_data(self
, endpoint
, auth_token
, values
):
349 """Collectd metric measures over a specified time period."""
353 # Try and collect measures
354 metric_id
= values
['metric_uuid']
355 collection_unit
= values
['collection_unit'].upper()
356 collection_period
= values
['collection_period']
358 # Define the start and end time based on configurations
359 stop_time
= time
.strftime("%Y-%m-%d") + "T" + time
.strftime("%X")
360 end_time
= int(round(time
.time() * 1000))
361 if collection_unit
== 'YEAR':
362 diff
= PERIOD_MS
[collection_unit
]
364 diff
= collection_period
* PERIOD_MS
[collection_unit
]
365 s_time
= (end_time
- diff
)/1000.0
366 start_time
= datetime
.datetime
.fromtimestamp(s_time
).strftime(
367 '%Y-%m-%dT%H:%M:%S.%f')
368 base_url
= "{}/v1/metric/%(0)s/measures?start=%(1)s&stop=%(2)s"
369 url
= base_url
.format(endpoint
) % {
370 "0": metric_id
, "1": start_time
, "2": stop_time
}
372 # Perform metric data request
373 metric_data
= self
._common
._perform
_request
(
374 url
, auth_token
, req_type
="get")
376 # Generate a list of the requested timestamps and data
377 for r
in json
.loads(metric_data
.text
):
378 timestamp
= r
[0].replace("T", " ")
379 timestamps
.append(timestamp
)
382 return timestamps
, data
383 except Exception as exc
:
384 log
.warn("Failed to gather specified measures: %s", exc
)
385 return timestamps
, data
387 def authenticate(self
):
388 """Generate an authentication token and endpoint for metric request."""
390 # Check for a tenant_id
391 auth_token
= self
._common
._authenticate
()
392 endpoint
= self
._common
.get_endpoint("metric")
393 return auth_token
, endpoint
394 except Exception as exc
:
395 log
.warn("Authentication to Keystone failed: %s", exc
)
399 def response_list(self
, metric_list
, metric_name
=None, resource
=None):
400 """Create the appropriate lists for a list response."""
403 for row
in metric_list
:
404 if metric_name
is not None:
405 if row
['name'] == metric_name
:
406 metric
= {"metric_name": row
['name'],
407 "metric_uuid": row
['id'],
408 "metric_unit": row
['unit'],
409 "resource_uuid": row
['resource_id']}
410 resp_list
.append(metric
)
411 elif resource
is not None:
412 if row
['resource_id'] == resource
:
413 metric
= {"metric_name": row
['name'],
414 "metric_uuid": row
['id'],
415 "metric_unit": row
['unit'],
416 "resource_uuid": row
['resource_id']}
417 resp_list
.append(metric
)
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(metric
)