support custom metadata
this patch allows us to define metadata rather than blindly capturing entire payload. Change-Id: Id445f7f6d1231f8fdfddb43389783142190db662
This commit is contained in:
gordon chung
parent
e8929f3e18
commit
d836f8f54a
@ -76,6 +76,13 @@ class MeterDefinition(object):
|
||||
continue
|
||||
elif isinstance(field, six.integer_types):
|
||||
self._field_getter[name] = field
|
||||
elif isinstance(field, dict) and name == 'metadata':
|
||||
meta = {}
|
||||
for key, val in field.items():
|
||||
parts = self.parse_jsonpath(val)
|
||||
meta[key] = functools.partial(self._parse_jsonpath_field,
|
||||
parts)
|
||||
self._field_getter['metadata'] = meta
|
||||
else:
|
||||
parts = self.parse_jsonpath(field)
|
||||
self._field_getter[name] = functools.partial(
|
||||
@ -100,6 +107,11 @@ class MeterDefinition(object):
|
||||
getter = self._field_getter.get(field)
|
||||
if not getter:
|
||||
return
|
||||
elif isinstance(getter, dict):
|
||||
dict_val = {}
|
||||
for key, val in getter.items():
|
||||
dict_val[key] = val(message, all_values)
|
||||
return dict_val
|
||||
elif callable(getter):
|
||||
return getter(message, all_values)
|
||||
else:
|
||||
@ -232,6 +244,7 @@ class ProcessMeterNotifications(plugin_base.NotificationBase):
|
||||
projectid = self.get_project_id(d, notification_body)
|
||||
resourceid = d.parse_fields('resource_id', notification_body)
|
||||
ts = d.parse_fields('timestamp', notification_body)
|
||||
metadata = d.parse_fields('metadata', notification_body)
|
||||
if d.cfg.get('lookup'):
|
||||
meters = d.parse_fields('name', notification_body, True)
|
||||
if not meters: # skip if no meters in payload
|
||||
@ -265,7 +278,8 @@ class ProcessMeterNotifications(plugin_base.NotificationBase):
|
||||
yield sample.Sample.from_notification(
|
||||
name=m, type=t, unit=unit, volume=v,
|
||||
resource_id=r, user_id=user, project_id=p,
|
||||
message=notification_body, timestamp=ts)
|
||||
message=notification_body, timestamp=ts,
|
||||
metadata=metadata)
|
||||
else:
|
||||
yield sample.Sample.from_notification(
|
||||
name=d.cfg['name'],
|
||||
@ -276,7 +290,7 @@ class ProcessMeterNotifications(plugin_base.NotificationBase):
|
||||
user_id=userid,
|
||||
project_id=projectid,
|
||||
message=notification_body,
|
||||
timestamp=ts)
|
||||
timestamp=ts, metadata=metadata)
|
||||
|
||||
@staticmethod
|
||||
def get_user_id(d, notification_body):
|
||||
|
||||
@ -82,11 +82,12 @@ class Sample(object):
|
||||
@classmethod
|
||||
def from_notification(cls, name, type, volume, unit,
|
||||
user_id, project_id, resource_id,
|
||||
message, timestamp=None, source=None):
|
||||
metadata = (copy.copy(message['payload'])
|
||||
if isinstance(message['payload'], dict) else {})
|
||||
metadata['event_type'] = message['event_type']
|
||||
metadata['host'] = message['publisher_id']
|
||||
message, timestamp=None, metadata=None, source=None):
|
||||
if not metadata:
|
||||
metadata = (copy.copy(message['payload'])
|
||||
if isinstance(message['payload'], dict) else {})
|
||||
metadata['event_type'] = message['event_type']
|
||||
metadata['host'] = message['publisher_id']
|
||||
ts = timestamp if timestamp else message['timestamp']
|
||||
return cls(name=name,
|
||||
type=type,
|
||||
|
||||
@ -349,13 +349,13 @@ class TestMeterProcessing(test.BaseTestCase):
|
||||
event = copy.deepcopy(MIDDLEWARE_EVENT)
|
||||
del event['payload']['measurements'][1]
|
||||
cfg = yaml.dump(
|
||||
{'metric': [dict(name="payload.measurements.[*].metric.[*].name",
|
||||
{'metric': [dict(name="$.payload.measurements.[*].metric.[*].name",
|
||||
event_type="objectstore.http.request",
|
||||
type="delta",
|
||||
unit="payload.measurements.[*].metric.[*].unit",
|
||||
volume="payload.measurements.[*].result",
|
||||
resource_id="payload.target_id",
|
||||
project_id="payload.initiator.project_id",
|
||||
unit="$.payload.measurements.[*].metric.[*].unit",
|
||||
volume="$.payload.measurements.[*].result",
|
||||
resource_id="$.payload.target_id",
|
||||
project_id="$.payload.initiator.project_id",
|
||||
multi="name")]})
|
||||
self.handler.definitions = notifications.load_definitions(
|
||||
self.__setup_meter_def_file(cfg))
|
||||
@ -368,15 +368,15 @@ class TestMeterProcessing(test.BaseTestCase):
|
||||
event = copy.deepcopy(MIDDLEWARE_EVENT)
|
||||
del event['payload']['measurements'][1]
|
||||
cfg = yaml.dump(
|
||||
{'metric': [dict(name="payload.measurements.[*].metric.[*].name",
|
||||
{'metric': [dict(name="$.payload.measurements.[*].metric.[*].name",
|
||||
event_type="objectstore.http.request",
|
||||
type="delta",
|
||||
unit="payload.measurements.[*].metric.[*].unit",
|
||||
volume="payload.measurements.[*].result",
|
||||
resource_id="payload.target_id",
|
||||
project_id="payload.initiator.project_id",
|
||||
unit="$.payload.measurements.[*].metric.[*].unit",
|
||||
volume="$.payload.measurements.[*].result",
|
||||
resource_id="$.payload.target_id",
|
||||
project_id="$.payload.initiator.project_id",
|
||||
multi="name",
|
||||
timestamp='payload.eventTime')]})
|
||||
timestamp='$.payload.eventTime')]})
|
||||
self.handler.definitions = notifications.load_definitions(
|
||||
self.__setup_meter_def_file(cfg))
|
||||
c = list(self.handler.process_notification(event))
|
||||
@ -385,6 +385,45 @@ class TestMeterProcessing(test.BaseTestCase):
|
||||
self.assertEqual(MIDDLEWARE_EVENT['payload']['eventTime'],
|
||||
s1['timestamp'])
|
||||
|
||||
def test_default_metadata(self):
|
||||
cfg = yaml.dump(
|
||||
{'metric': [dict(name="test1",
|
||||
event_type="test.*",
|
||||
type="delta",
|
||||
unit="B",
|
||||
volume="$.payload.volume",
|
||||
resource_id="$.payload.resource_id",
|
||||
project_id="$.payload.project_id")]})
|
||||
self.handler.definitions = notifications.load_definitions(
|
||||
self.__setup_meter_def_file(cfg))
|
||||
c = list(self.handler.process_notification(NOTIFICATION))
|
||||
self.assertEqual(1, len(c))
|
||||
s1 = c[0].as_dict()
|
||||
meta = NOTIFICATION['payload'].copy()
|
||||
meta['host'] = NOTIFICATION['publisher_id']
|
||||
meta['event_type'] = NOTIFICATION['event_type']
|
||||
self.assertEqual(meta, s1['resource_metadata'])
|
||||
|
||||
def test_custom_metadata(self):
|
||||
cfg = yaml.dump(
|
||||
{'metric': [dict(name="test1",
|
||||
event_type="test.*",
|
||||
type="delta",
|
||||
unit="B",
|
||||
volume="$.payload.volume",
|
||||
resource_id="$.payload.resource_id",
|
||||
project_id="$.payload.project_id",
|
||||
metadata={'proj': '$.payload.project_id',
|
||||
'dict': '$.payload.resource_metadata'})]})
|
||||
self.handler.definitions = notifications.load_definitions(
|
||||
self.__setup_meter_def_file(cfg))
|
||||
c = list(self.handler.process_notification(NOTIFICATION))
|
||||
self.assertEqual(1, len(c))
|
||||
s1 = c[0].as_dict()
|
||||
meta = {'proj': s1['project_id'],
|
||||
'dict': NOTIFICATION['payload']['resource_metadata']}
|
||||
self.assertEqual(meta, s1['resource_metadata'])
|
||||
|
||||
def test_multi_match_event_meter(self):
|
||||
cfg = yaml.dump(
|
||||
{'metric': [dict(name="test1",
|
||||
|
||||
Loading…
Reference in New Issue
Block a user