Merge "Adding monasca push driver to receive webhook notifications"
This commit is contained in:
commit
814eac046e
@ -1,4 +1,4 @@
|
|||||||
# Copyright (c) 2015 Cisco.
|
# Copyright (c) 2015 Cisco, 2018 NEC, Inc.
|
||||||
#
|
#
|
||||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||||
# not use this file except in compliance with the License. You may obtain
|
# not use this file except in compliance with the License. You may obtain
|
||||||
@ -13,9 +13,13 @@
|
|||||||
# under the License.
|
# under the License.
|
||||||
#
|
#
|
||||||
|
|
||||||
import datetime
|
from datetime import datetime
|
||||||
|
from datetime import timedelta
|
||||||
|
|
||||||
|
import eventlet
|
||||||
|
from futurist import periodics
|
||||||
from monascaclient import client as monasca_client
|
from monascaclient import client as monasca_client
|
||||||
|
from oslo_concurrency import lockutils
|
||||||
from oslo_log import log as logging
|
from oslo_log import log as logging
|
||||||
|
|
||||||
from congress.datasources import constants
|
from congress.datasources import constants
|
||||||
@ -24,6 +28,13 @@ from congress.datasources import datasource_utils as ds_utils
|
|||||||
|
|
||||||
LOG = logging.getLogger(__name__)
|
LOG = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
DATA = "statistics.data"
|
||||||
|
DIMENSIONS = "dimensions"
|
||||||
|
METRICS = "metrics"
|
||||||
|
NOTIFICATIONS = "alarm_notification"
|
||||||
|
STATISTICS = "statistics"
|
||||||
|
value_trans = {'translation-type': 'VALUE'}
|
||||||
|
|
||||||
|
|
||||||
# TODO(thinrichs): figure out how to move even more of this boilerplate
|
# TODO(thinrichs): figure out how to move even more of this boilerplate
|
||||||
# into DataSourceDriver. E.g. change all the classes to Driver instead of
|
# into DataSourceDriver. E.g. change all the classes to Driver instead of
|
||||||
@ -32,16 +43,10 @@ LOG = logging.getLogger(__name__)
|
|||||||
class MonascaDriver(datasource_driver.PollingDataSourceDriver,
|
class MonascaDriver(datasource_driver.PollingDataSourceDriver,
|
||||||
datasource_driver.ExecutionDriver):
|
datasource_driver.ExecutionDriver):
|
||||||
|
|
||||||
METRICS = "metrics"
|
|
||||||
DIMENSIONS = "dimensions"
|
|
||||||
STATISTICS = "statistics"
|
|
||||||
DATA = "statistics.data"
|
|
||||||
# TODO(fabiog): add events and logs when fully supported in Monasca
|
# TODO(fabiog): add events and logs when fully supported in Monasca
|
||||||
# EVENTS = "events"
|
# EVENTS = "events"
|
||||||
# LOGS = "logs"
|
# LOGS = "logs"
|
||||||
|
|
||||||
value_trans = {'translation-type': 'VALUE'}
|
|
||||||
|
|
||||||
metric_translator = {
|
metric_translator = {
|
||||||
'translation-type': 'HDICT',
|
'translation-type': 'HDICT',
|
||||||
'table-name': METRICS,
|
'table-name': METRICS,
|
||||||
@ -115,9 +120,9 @@ class MonascaDriver(datasource_driver.PollingDataSourceDriver,
|
|||||||
self.add_update_method(statistics_method, self.statistics_translator)
|
self.add_update_method(statistics_method, self.statistics_translator)
|
||||||
|
|
||||||
def update_statistics(self):
|
def update_statistics(self):
|
||||||
today = datetime.datetime.now()
|
today = datetime.utcnow()
|
||||||
yesterday = datetime.timedelta(hours=24)
|
yesterday = timedelta(hours=24)
|
||||||
start_from = datetime.datetime.isoformat(today-yesterday)
|
start_from = datetime.isoformat(today-yesterday)
|
||||||
|
|
||||||
for metric in self.monasca.metrics.list_names():
|
for metric in self.monasca.metrics.list_names():
|
||||||
LOG.debug("Monasca statistics for metric %s", metric['name'])
|
LOG.debug("Monasca statistics for metric %s", metric['name'])
|
||||||
@ -158,3 +163,115 @@ class MonascaDriver(datasource_driver.PollingDataSourceDriver,
|
|||||||
func(action_args)
|
func(action_args)
|
||||||
else:
|
else:
|
||||||
self._execute_api(self.monasca, action, action_args)
|
self._execute_api(self.monasca, action, action_args)
|
||||||
|
|
||||||
|
|
||||||
|
class MonascaWebhookDriver(datasource_driver.PushedDataSourceDriver):
|
||||||
|
|
||||||
|
def flatten_alarm_webhook(alarm_obj):
|
||||||
|
flattened = []
|
||||||
|
key_to_sub_dict = 'metrics'
|
||||||
|
for alarm in alarm_obj:
|
||||||
|
sub_dict = alarm.pop(key_to_sub_dict)[0]
|
||||||
|
for k, v in sub_dict.items():
|
||||||
|
if isinstance(v, dict):
|
||||||
|
for key, value in v.items():
|
||||||
|
alarm[key_to_sub_dict + '_' + k + '_' + key] = value
|
||||||
|
else:
|
||||||
|
alarm[key_to_sub_dict + '_' + k] = v
|
||||||
|
flattened.append(alarm)
|
||||||
|
return flattened
|
||||||
|
|
||||||
|
alarm_notification_translator = {
|
||||||
|
'translation-type': 'HDICT',
|
||||||
|
'table-name': NOTIFICATIONS,
|
||||||
|
'selector-type': 'DICT_SELECTOR',
|
||||||
|
'objects-extract-fn': flatten_alarm_webhook,
|
||||||
|
'field-translators':
|
||||||
|
({'fieldname': 'alarm_id', 'translator': value_trans},
|
||||||
|
{'fieldname': 'alarm_definition_id', 'translator': value_trans},
|
||||||
|
{'fieldname': 'alarm_name', 'translator': value_trans},
|
||||||
|
{'fieldname': 'alarm_description', 'translator': value_trans},
|
||||||
|
{'fieldname': 'alarm_timestamp', 'translator': value_trans},
|
||||||
|
{'fieldname': 'state', 'translator': value_trans},
|
||||||
|
{'fieldname': 'old_state', 'translator': value_trans},
|
||||||
|
{'fieldname': 'message', 'translator': value_trans},
|
||||||
|
{'fieldname': 'tenant_id', 'translator': value_trans},
|
||||||
|
{'fieldname': 'metrics_id', 'translator': value_trans},
|
||||||
|
{'fieldname': 'metrics_name', 'translator': value_trans},
|
||||||
|
{'fieldname': 'metrics_dimensions_hostname',
|
||||||
|
'translator': value_trans},
|
||||||
|
{'fieldname': 'metrics_dimensions_service',
|
||||||
|
'translator': value_trans},)
|
||||||
|
}
|
||||||
|
TRANSLATORS = [alarm_notification_translator]
|
||||||
|
|
||||||
|
def __init__(self, name='', args=None):
|
||||||
|
super(MonascaWebhookDriver, self).__init__(name, args=args)
|
||||||
|
if args is None:
|
||||||
|
args = {}
|
||||||
|
# set default time to 10 days before deleting an active alarm
|
||||||
|
self.hours_to_keep_alarm = int(args.get('hours_to_keep_alarm', 240))
|
||||||
|
self.set_up_periodic_tasks()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def get_datasource_info():
|
||||||
|
result = {}
|
||||||
|
result['id'] = 'monasca_webhook_driver'
|
||||||
|
result['description'] = ('Datasource driver that accepts Monasca '
|
||||||
|
'webhook alarm notifications.')
|
||||||
|
result['config'] = {'persist_data': constants.OPTIONAL,
|
||||||
|
'hours_to_keep_alarm': constants.OPTIONAL}
|
||||||
|
return result
|
||||||
|
|
||||||
|
def _webhook_handler(self, alarm):
|
||||||
|
tablename = NOTIFICATIONS
|
||||||
|
# remove already existing same alarm row
|
||||||
|
alarm_id = alarm['alarm_id']
|
||||||
|
column_index_number_of_alarm_id = 0
|
||||||
|
to_remove = [row for row in self.state[tablename]
|
||||||
|
if row[column_index_number_of_alarm_id] == alarm_id]
|
||||||
|
for row in to_remove:
|
||||||
|
self.state[tablename].discard(row)
|
||||||
|
|
||||||
|
translator = self.alarm_notification_translator
|
||||||
|
row_data = MonascaWebhookDriver.convert_objs([alarm], translator)
|
||||||
|
|
||||||
|
# add alarm to table
|
||||||
|
for table, row in row_data:
|
||||||
|
if table == tablename:
|
||||||
|
self.state[tablename].add(row)
|
||||||
|
LOG.debug('publish a new state %s in %s',
|
||||||
|
self.state[tablename], tablename)
|
||||||
|
self.publish(tablename, self.state[tablename])
|
||||||
|
return [tablename]
|
||||||
|
|
||||||
|
def set_up_periodic_tasks(self):
|
||||||
|
@lockutils.synchronized('congress_monasca_webhook_ds_data')
|
||||||
|
@periodics.periodic(spacing=max(self.hours_to_keep_alarm * 3600/10, 1))
|
||||||
|
def delete_old_alarms():
|
||||||
|
tablename = NOTIFICATIONS
|
||||||
|
col_index_of_timestamp = 4
|
||||||
|
# find for removal all alarms at least self.hours_to_keep_alarm old
|
||||||
|
to_remove = [
|
||||||
|
row for row in self.state[tablename]
|
||||||
|
if (datetime.utcnow() -
|
||||||
|
datetime.utcfromtimestamp(row[col_index_of_timestamp])
|
||||||
|
>= timedelta(hours=self.hours_to_keep_alarm))]
|
||||||
|
for row in to_remove:
|
||||||
|
self.state[tablename].discard(row)
|
||||||
|
|
||||||
|
periodic_task_callables = [
|
||||||
|
(delete_old_alarms, None, {}),
|
||||||
|
(delete_old_alarms, None, {})]
|
||||||
|
self.periodic_tasks = periodics.PeriodicWorker(periodic_task_callables)
|
||||||
|
self.periodic_tasks_thread = eventlet.spawn_n(
|
||||||
|
self.periodic_tasks.start)
|
||||||
|
|
||||||
|
def __del__(self):
|
||||||
|
if self.periodic_tasks:
|
||||||
|
self.periodic_tasks.stop()
|
||||||
|
self.periodic_tasks.wait()
|
||||||
|
self.periodic_tasks = None
|
||||||
|
if self.periodic_tasks_thread:
|
||||||
|
eventlet.greenthread.kill(self.periodic_tasks_thread)
|
||||||
|
self.periodic_tasks_thread = None
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
# Copyright (c) 2015 Cisco, Inc. All rights reserved.
|
# Copyright (c) 2018 NEC, Inc. All rights reserved.
|
||||||
#
|
#
|
||||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||||
# not use this file except in compliance with the License. You may obtain
|
# not use this file except in compliance with the License. You may obtain
|
||||||
@ -14,6 +14,7 @@
|
|||||||
#
|
#
|
||||||
import mock
|
import mock
|
||||||
import sys
|
import sys
|
||||||
|
import time
|
||||||
|
|
||||||
sys.modules['monascaclient.client'] = mock.Mock()
|
sys.modules['monascaclient.client'] = mock.Mock()
|
||||||
sys.modules['monascaclient'] = mock.Mock()
|
sys.modules['monascaclient'] = mock.Mock()
|
||||||
@ -68,8 +69,8 @@ class TestMonascaDriver(base.TestCase):
|
|||||||
|
|
||||||
def test_statistics_update_from_datasource(self):
|
def test_statistics_update_from_datasource(self):
|
||||||
self.driver._translate_statistics(self.mock_statistics['elements'])
|
self.driver._translate_statistics(self.mock_statistics['elements'])
|
||||||
stats_list = list(self.driver.state[self.driver.STATISTICS])
|
stats_list = list(self.driver.state[monasca_driver.STATISTICS])
|
||||||
stats_data_list = list(self.driver.state[self.driver.DATA])
|
stats_data_list = list(self.driver.state[monasca_driver.DATA])
|
||||||
self.assertIsNotNone(stats_list)
|
self.assertIsNotNone(stats_list)
|
||||||
self.assertIsNotNone(stats_data_list)
|
self.assertIsNotNone(stats_data_list)
|
||||||
|
|
||||||
@ -129,3 +130,73 @@ class TestMonascaDriver(base.TestCase):
|
|||||||
self.driver.execute('getStatistics', api_args)
|
self.driver.execute('getStatistics', api_args)
|
||||||
|
|
||||||
self.assertEqual(expected_ans, monasca_client.testkey)
|
self.assertEqual(expected_ans, monasca_client.testkey)
|
||||||
|
|
||||||
|
|
||||||
|
class TestMonascaWebhookDriver(base.TestCase):
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
super(TestMonascaWebhookDriver, self).setUp()
|
||||||
|
self.monasca = monasca_driver.MonascaWebhookDriver('test-monasca')
|
||||||
|
|
||||||
|
@mock.patch.object(monasca_driver.MonascaWebhookDriver, 'publish')
|
||||||
|
def test_monasca_webhook_alarm(self, mocked_publish):
|
||||||
|
test_alarm = {
|
||||||
|
'metrics': [
|
||||||
|
{u'dimensions': {u'hostname': u'openstack-13.local.lan',
|
||||||
|
u'service': u'monitoring'},
|
||||||
|
u'id': None,
|
||||||
|
u'name': u'load.avg_1_min'}],
|
||||||
|
'alarm_id': u'3beb4934-053d-4f8f-9704-273bffc2441b',
|
||||||
|
'state': u'ALARM',
|
||||||
|
'alarm_timestamp': 1531821822,
|
||||||
|
'tenant_id': u'3661888238874df098988deab07c599d',
|
||||||
|
'old_state': u'UNDETERMINED',
|
||||||
|
'alarm_description': u'',
|
||||||
|
'message': u'Thresholds were exceeded for the sub-alarms',
|
||||||
|
'alarm_definition_id': u'8e5d033f-28cc-459f-91d4-813307e4ca8a',
|
||||||
|
'alarm_name': u'alarmPerHost23'}
|
||||||
|
self.monasca._webhook_handler(test_alarm)
|
||||||
|
|
||||||
|
expected_rows = set([(u'3beb4934-053d-4f8f-9704-273bffc2441b',
|
||||||
|
u'8e5d033f-28cc-459f-91d4-813307e4ca8a',
|
||||||
|
u'alarmPerHost23',
|
||||||
|
u'',
|
||||||
|
1531821822,
|
||||||
|
u'ALARM',
|
||||||
|
u'UNDETERMINED',
|
||||||
|
u'Thresholds were exceeded for the sub-alarms',
|
||||||
|
u'3661888238874df098988deab07c599d',
|
||||||
|
None,
|
||||||
|
u'load.avg_1_min',
|
||||||
|
u'openstack-13.local.lan',
|
||||||
|
u'monitoring')])
|
||||||
|
self.assertEqual(self.monasca.state['alarm_notification'],
|
||||||
|
expected_rows)
|
||||||
|
|
||||||
|
@mock.patch.object(monasca_driver.MonascaWebhookDriver, 'publish')
|
||||||
|
def test_webhook_alarm_cleanup(self, mocked_publish):
|
||||||
|
self.monasca = monasca_driver.MonascaWebhookDriver(
|
||||||
|
'test-monasca',
|
||||||
|
args={'hours_to_keep_alarm': 1 / 3600}) # set to 1 sec for test
|
||||||
|
|
||||||
|
test_alarm = {
|
||||||
|
'metrics': [
|
||||||
|
{u'dimensions': {u'hostname': u'openstack-13.local.lan',
|
||||||
|
u'service': u'monitoring'},
|
||||||
|
u'id': None,
|
||||||
|
u'name': u'load.avg_1_min'}],
|
||||||
|
'alarm_id': u'3beb4934-053d-4f8f-9704-273bffc2441b',
|
||||||
|
'state': u'ALARM',
|
||||||
|
'alarm_timestamp': 1531821822,
|
||||||
|
'tenant_id': u'3661888238874df098988deab07c599d',
|
||||||
|
'old_state': u'UNDETERMINED',
|
||||||
|
'alarm_description': u'',
|
||||||
|
'message': u'Thresholds were exceeded for the sub-alarms',
|
||||||
|
'alarm_definition_id': u'8e5d033f-28cc-459f-91d4-813307e4ca8a',
|
||||||
|
'alarm_name': u'alarmPerHost23'}
|
||||||
|
|
||||||
|
self.monasca._webhook_handler(test_alarm)
|
||||||
|
|
||||||
|
self.assertEqual(1, len(self.monasca.state['alarm_notification']))
|
||||||
|
time.sleep(3)
|
||||||
|
self.assertEqual(0, len(self.monasca.state['alarm_notification']))
|
||||||
|
@ -515,5 +515,8 @@ def supported_drivers():
|
|||||||
"description": "Datasource driver that interfaces with Mistral."},
|
"description": "Datasource driver that interfaces with Mistral."},
|
||||||
{"id": "vitrage",
|
{"id": "vitrage",
|
||||||
"description": "Datasource driver that accepts Vitrage "
|
"description": "Datasource driver that accepts Vitrage "
|
||||||
"webhook alarm notifications."}]
|
"webhook alarm notifications."},
|
||||||
|
{"id": "monasca_webhook_driver",
|
||||||
|
"description": "Datasource driver that accepts Monasca webhook "
|
||||||
|
"alarm notifications."}]
|
||||||
return results
|
return results
|
||||||
|
@ -63,6 +63,7 @@ congress.datasource.drivers =
|
|||||||
keystonev3 = congress.datasources.keystonev3_driver:KeystoneV3Driver
|
keystonev3 = congress.datasources.keystonev3_driver:KeystoneV3Driver
|
||||||
mistral = congress.datasources.mistral_driver:MistralDriver
|
mistral = congress.datasources.mistral_driver:MistralDriver
|
||||||
monasca = congress.datasources.monasca_driver:MonascaDriver
|
monasca = congress.datasources.monasca_driver:MonascaDriver
|
||||||
|
monasca_webhook_driver = congress.datasources.monasca_driver:MonascaWebhookDriver
|
||||||
murano = congress.datasources.murano_driver:MuranoDriver
|
murano = congress.datasources.murano_driver:MuranoDriver
|
||||||
neutronv2 = congress.datasources.neutronv2_driver:NeutronV2Driver
|
neutronv2 = congress.datasources.neutronv2_driver:NeutronV2Driver
|
||||||
neutronv2_qos = congress.datasources.neutronv2_qos_driver:NeutronV2QosDriver
|
neutronv2_qos = congress.datasources.neutronv2_qos_driver:NeutronV2QosDriver
|
||||||
|
Loading…
Reference in New Issue
Block a user