Factorize field definition of declarative code

Currently we have three differents parts of code that parse samples
and notifications. All of them do the same thing.
The event one have some additionals feature "TraitPlugin".

This change removes the code duplication and allows to use TraitPlugin
into gnocchi and meter definitions.

Change-Id: Id125de92a5893d7afa5a3d55c3f183bd2035a733

ceilometer/declarative.py

@@ -0,0 +1,118 @@
+#
+# 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
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+
+from jsonpath_rw_ext import parser
+import six
+
+from ceilometer.i18n import _
+
+
+class DefinitionException(Exception):
+    def __init__(self, message, definition_cfg):
+        super(DefinitionException, self).__init__(message)
+        self.definition_cfg = definition_cfg
+
+
+class Definition(object):
+    JSONPATH_RW_PARSER = parser.ExtentedJsonPathParser()
+
+    def __init__(self, name, cfg, plugin_manager):
+        self.cfg = cfg
+        self.name = name
+        self.plugin = None
+        if isinstance(cfg, dict):
+            if 'fields' not in cfg:
+                raise DefinitionException(
+                    _("The field 'fields' is required for %s") % name,
+                    self.cfg)
+
+            if 'plugin' in cfg:
+                plugin_cfg = cfg['plugin']
+                if isinstance(plugin_cfg, six.string_types):
+                    plugin_name = plugin_cfg
+                    plugin_params = {}
+                else:
+                    try:
+                        plugin_name = plugin_cfg['name']
+                    except KeyError:
+                        raise DefinitionException(
+                            _('Plugin specified, but no plugin name supplied '
+                              'for %s') % name, self.cfg)
+                    plugin_params = plugin_cfg.get('parameters')
+                    if plugin_params is None:
+                        plugin_params = {}
+                try:
+                    plugin_ext = plugin_manager[plugin_name]
+                except KeyError:
+                    raise DefinitionException(
+                        _('No plugin named %(plugin)s available for '
+                          '%(name)s') % dict(
+                              plugin=plugin_name,
+                              name=name), self.cfg)
+                plugin_class = plugin_ext.plugin
+                self.plugin = plugin_class(**plugin_params)
+
+            fields = cfg['fields']
+        else:
+            # Simple definition "foobar: jsonpath"
+            fields = cfg
+
+        if isinstance(fields, list):
+            # NOTE(mdragon): if not a string, we assume a list.
+            if len(fields) == 1:
+                fields = fields[0]
+            else:
+                fields = '|'.join('(%s)' % path for path in fields)
+
+        if isinstance(fields, six.integer_types):
+            self.getter = fields
+        else:
+            try:
+                self.getter = self.JSONPATH_RW_PARSER.parse(fields).find
+            except Exception as e:
+                raise DefinitionException(
+                    _("Parse error in JSONPath specification "
+                      "'%(jsonpath)s' for %(name)s: %(err)s")
+                    % dict(jsonpath=fields, name=name, err=e), self.cfg)
+
+    def _get_path(self, match):
+        if match.context is not None:
+            for path_element in self._get_path(match.context):
+                yield path_element
+            yield str(match.path)
+
+    def parse(self, obj, return_all_values=False):
+        if callable(self.getter):
+            values = self.getter(obj)
+        else:
+            return self.getter
+
+        values = [match for match in values
+                  if return_all_values or match.value is not None]
+
+        if self.plugin is not None:
+            if return_all_values and not self.plugin.support_return_all_values:
+                raise DefinitionException("Plugin %s don't allows to "
+                                          "return multiple values" %
+                                          self.cfg["plugin"]["name"])
+            values_map = [('.'.join(self._get_path(match)), match.value) for
+                          match in values]
+            values = [v for v in self.plugin.trait_values(values_map)
+                      if v is not None]
+        else:
+            values = [match.value for match in values if match is not None]
+        if return_all_values:
+            return values
+        else:
+            return values[0] if values else None

ceilometer/dispatcher/gnocchi.py

@@ -13,18 +13,18 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 import fnmatch
-import functools
 import itertools
 import operator
 import os
 import threading
 
-from jsonpath_rw_ext import parser
 from oslo_config import cfg
 from oslo_log import log
 import six
+from stevedore import extension
 import yaml
 
+from ceilometer import declarative
 from ceilometer import dispatcher
 from ceilometer.dispatcher import gnocchi_client
 from ceilometer.i18n import _, _LE
@@ -81,42 +81,23 @@ class ResourcesDefinition(object):
     MANDATORY_FIELDS = {'resource_type': six.string_types,
                         'metrics': list}
 
-    JSONPATH_RW_PARSER = parser.ExtentedJsonPathParser()
+    def __init__(self, definition_cfg, default_archive_policy, plugin_manager):
-
-    def __init__(self, definition_cfg, default_archive_policy):
         self._default_archive_policy = default_archive_policy
         self.cfg = definition_cfg
 
         for field, field_type in self.MANDATORY_FIELDS.items():
             if field not in self.cfg:
-                raise ResourcesDefinitionException(
+                raise declarative.DefinitionException(
                     _LE("Required field %s not specified") % field, self.cfg)
             if not isinstance(self.cfg[field], field_type):
-                raise ResourcesDefinitionException(
+                raise declarative.DefinitionException(
                     _LE("Required field %(field)s should be a %(type)s") %
                     {'field': field, 'type': field_type}, self.cfg)
 
-        self._field_getter = {}
+        self._attributes = {}
-        for name, fval in self.cfg.get('attributes', {}).items():
+        for name, attr_cfg in self.cfg.get('attributes', {}).items():
-            if isinstance(fval, six.integer_types):
+            self._attributes[name] = declarative.Definition(name, attr_cfg,
-                self._field_getter[name] = fval
+                                                            plugin_manager)
-            else:
-                try:
-                    parts = self.JSONPATH_RW_PARSER.parse(fval)
-                except Exception as e:
-                    raise ResourcesDefinitionException(
-                        _LE("Parse error in JSONPath specification "
-                            "'%(jsonpath)s': %(err)s")
-                        % dict(jsonpath=fval, err=e), self.cfg)
-                self._field_getter[name] = functools.partial(
-                    self._parse_jsonpath_field, parts)
-
-    @staticmethod
-    def _parse_jsonpath_field(parts, sample):
-        values = [match.value for match in parts.find(sample)
-                  if match.value is not None]
-        if values:
-            return values[0]
 
     def match(self, metric_name):
         for t in self.cfg['metrics']:
@@ -126,13 +107,10 @@ class ResourcesDefinition(object):
 
     def attributes(self, sample):
         attrs = {}
-        for attr, getter in self._field_getter.items():
+        for name, definition in self._attributes.items():
-            if callable(getter):
+            value = definition.parse(sample)
-                value = getter(sample)
-            else:
-                value = getter
             if value is not None:
-                attrs[attr] = value
+                attrs[name] = value
         return attrs
 
     def metrics(self):
@@ -169,6 +147,8 @@ class GnocchiDispatcher(dispatcher.Base):
 
     @classmethod
     def _load_resources_definitions(cls, conf):
+        plugin_manager = extension.ExtensionManager(
+            namespace='ceilometer.event.trait_plugin')
         res_def_file = cls._get_config_file(
             conf, conf.dispatcher_gnocchi.resources_definition_file)
         data = {}
@@ -179,7 +159,8 @@ class GnocchiDispatcher(dispatcher.Base):
                 except ValueError:
                     data = {}
 
-        return [ResourcesDefinition(r, conf.dispatcher_gnocchi.archive_policy)
+        return [ResourcesDefinition(r, conf.dispatcher_gnocchi.archive_policy,
+                                    plugin_manager)
                 for r in data.get('resources', [])]
 
     @property

ceilometer/event/converter.py

@@ -16,13 +16,14 @@
 import fnmatch
 import os
 
-from jsonpath_rw_ext import parser
+from debtcollector import moves
 from oslo_config import cfg
 from oslo_log import log
 from oslo_utils import timeutils
 import six
 import yaml
 
+from ceilometer import declarative
 from ceilometer.event.storage import models
 from ceilometer.i18n import _, _LI
 
@@ -46,97 +47,26 @@ cfg.CONF.register_opts(OPTS, group='event')
 
 LOG = log.getLogger(__name__)
 
-
+EventDefinitionException = moves.moved_class(declarative.DefinitionException,
-class EventDefinitionException(Exception):
+                                             'EventDefinitionException',
-    def __init__(self, message, definition_cfg):
+                                             __name__,
-        super(EventDefinitionException, self).__init__(message)
+                                             version=6.0,
-        self.definition_cfg = definition_cfg
+                                             removal_version="?")
-
-    def __str__(self):
-        return '%s %s: %s' % (self.__class__.__name__,
-                              self.definition_cfg, self.message)
 
 
-class TraitDefinition(object):
+class TraitDefinition(declarative.Definition):
-
-    JSONPATH_RW_PARSER = parser.ExtentedJsonPathParser()
-
     def __init__(self, name, trait_cfg, plugin_manager):
-        self.cfg = trait_cfg
+        super(TraitDefinition, self).__init__(name, trait_cfg, plugin_manager)
-        self.name = name
+        type_name = (trait_cfg.get('type', 'text')
-
+                     if isinstance(trait_cfg, dict) else 'text')
-        type_name = trait_cfg.get('type', 'text')
-
-        if 'plugin' in trait_cfg:
-            plugin_cfg = trait_cfg['plugin']
-            if isinstance(plugin_cfg, six.string_types):
-                plugin_name = plugin_cfg
-                plugin_params = {}
-            else:
-                try:
-                    plugin_name = plugin_cfg['name']
-                except KeyError:
-                    raise EventDefinitionException(
-                        _('Plugin specified, but no plugin name supplied for '
-                          'trait %s') % name, self.cfg)
-                plugin_params = plugin_cfg.get('parameters')
-                if plugin_params is None:
-                    plugin_params = {}
-            try:
-                plugin_ext = plugin_manager[plugin_name]
-            except KeyError:
-                raise EventDefinitionException(
-                    _('No plugin named %(plugin)s available for '
-                      'trait %(trait)s') % dict(plugin=plugin_name,
-                                                trait=name), self.cfg)
-            plugin_class = plugin_ext.plugin
-            self.plugin = plugin_class(**plugin_params)
-        else:
-            self.plugin = None
-
-        if 'fields' not in trait_cfg:
-            raise EventDefinitionException(
-                _("Required field in trait definition not specified: "
-                  "'%s'") % 'fields',
-                self.cfg)
-
-        fields = trait_cfg['fields']
-        if not isinstance(fields, six.string_types):
-            # NOTE(mdragon): if not a string, we assume a list.
-            if len(fields) == 1:
-                fields = fields[0]
-            else:
-                fields = '|'.join('(%s)' % path for path in fields)
-        try:
-            self.fields = self.JSONPATH_RW_PARSER.parse(fields)
-        except Exception as e:
-            raise EventDefinitionException(
-                _("Parse error in JSONPath specification "
-                  "'%(jsonpath)s' for %(trait)s: %(err)s")
-                % dict(jsonpath=fields, trait=name, err=e), self.cfg)
         self.trait_type = models.Trait.get_type_by_name(type_name)
         if self.trait_type is None:
-            raise EventDefinitionException(
+            raise declarative.DefinitionException(
                 _("Invalid trait type '%(type)s' for trait %(trait)s")
                 % dict(type=type_name, trait=name), self.cfg)
 
-    def _get_path(self, match):
-        if match.context is not None:
-            for path_element in self._get_path(match.context):
-                yield path_element
-            yield str(match.path)
-
     def to_trait(self, notification_body):
-        values = [match for match in self.fields.find(notification_body)
+        value = self.parse(notification_body)
-                  if match.value is not None]
-
-        if self.plugin is not None:
-            value_map = [('.'.join(self._get_path(match)), match.value) for
-                         match in values]
-            value = self.plugin.trait_value(value_map)
-        else:
-            value = values[0].value if values else None
-
         if value is None:
             return None
 
@@ -175,7 +105,7 @@ class EventDefinition(object):
             event_type = definition_cfg['event_type']
             traits = definition_cfg['traits']
         except KeyError as err:
-            raise EventDefinitionException(
+            raise declarative.DefinitionException(
                 _("Required field %s not specified") % err.args[0], self.cfg)
 
         if isinstance(event_type, six.string_types):

ceilometer/event/trait_plugins.py

@@ -15,6 +15,7 @@
 
 import abc
 
+from debtcollector import moves
 from oslo_log import log
 import six
 
@@ -30,6 +31,12 @@ class TraitPluginBase(object):
     It converts notification fields to Trait values.
     """
 
+    support_return_all_values = False
+    """If True, an exception will be raised if the user expect
+    the plugin to return one trait per match_list, but
+    the plugin doesn't allow/support that.
+    """
+
     def __init__(self, **kw):
         """Setup the trait plugin.
 
@@ -43,9 +50,12 @@ class TraitPluginBase(object):
         """
         super(TraitPluginBase, self).__init__()
 
-    @abc.abstractmethod
+    @moves.moved_method('trait_values', version=6.0, removal_version="?")
     def trait_value(self, match_list):
-        """Convert a set of fields to a Trait value.
+        pass
+
+    def trait_values(self, match_list):
+        """Convert a set of fields to one or multiple Trait values.
 
         This method is called each time a trait is attempted to be extracted
         from a notification. It will be called *even if* no matching fields
@@ -93,13 +103,18 @@ class TraitPluginBase(object):
               def trait_value(self, match_list):
                   if not match_list:
                       return None
-                  return match_list[0][1]
+                  return [ match[1] for match in match_list]
         """
 
+        # For backwards compatibility for the renamed method.
+        return [self.trait_value(match_list)]
+
 
 class SplitterTraitPlugin(TraitPluginBase):
     """Plugin that splits a piece off of a string value."""
 
+    support_return_all_values = True
+
     def __init__(self, separator=".", segment=0, max_split=None, **kw):
         """Setup how do split the field.
 
@@ -120,10 +135,12 @@ class SplitterTraitPlugin(TraitPluginBase):
         self.max_split = max_split
         super(SplitterTraitPlugin, self).__init__(**kw)
 
-    def trait_value(self, match_list):
+    def trait_values(self, match_list):
-        if not match_list:
+        return [self._trait_value(match)
-            return None
+                for match in match_list]
-        value = six.text_type(match_list[0][1])
+
+    def _trait_value(self, match):
+        value = six.text_type(match[1])
         if self.max_split is not None:
             values = value.split(self.separator, self.max_split)
         else:
@@ -158,7 +175,7 @@ class BitfieldTraitPlugin(TraitPluginBase):
         self.flags = flags
         super(BitfieldTraitPlugin, self).__init__(**kw)
 
-    def trait_value(self, match_list):
+    def trait_values(self, match_list):
         matches = dict(match_list)
         bitfield = self.initial_bitfield
         for flagdef in self.flags:
@@ -170,4 +187,4 @@ class BitfieldTraitPlugin(TraitPluginBase):
                         bitfield |= bit
                 else:
                     bitfield |= bit
-        return bitfield
+        return [bitfield]

ceilometer/meter/notifications.py

@@ -12,19 +12,20 @@
 # under the License.
 
 import fnmatch
-import functools
 import itertools
 import os
 import pkg_resources
 import six
 import yaml
 
-from jsonpath_rw_ext import parser
+from debtcollector import moves
 from oslo_config import cfg
 from oslo_log import log
 import oslo_messaging
+from stevedore import extension
 
 from ceilometer.agent import plugin_base
+from ceilometer import declarative
 from ceilometer.i18n import _LE, _LI
 from ceilometer import sample
 
@@ -42,95 +43,129 @@ cfg.CONF.import_opt('disable_non_metric_meters', 'ceilometer.notification',
 LOG = log.getLogger(__name__)
 
 
-class MeterDefinitionException(Exception):
+MeterDefinitionException = moves.moved_class(declarative.DefinitionException,
-    def __init__(self, message, definition_cfg):
+                                             'MeterDefinitionException',
-        super(MeterDefinitionException, self).__init__(message)
+                                             __name__,
-        self.message = message
+                                             version=6.0,
-        self.definition_cfg = definition_cfg
+                                             removal_version="?")
-
-    def __str__(self):
-        return '%s %s: %s' % (self.__class__.__name__,
-                              self.definition_cfg, self.message)
 
 
 class MeterDefinition(object):
 
-    JSONPATH_RW_PARSER = parser.ExtentedJsonPathParser()
+    SAMPLE_ATTRIBUTES = ["name", "type", "volume", "unit", "timestamp",
+                         "user_id", "project_id", "resource_id"]
 
     REQUIRED_FIELDS = ['name', 'type', 'event_type', 'unit', 'volume',
                        'resource_id']
 
-    def __init__(self, definition_cfg):
+    def __init__(self, definition_cfg, plugin_manager):
         self.cfg = definition_cfg
         missing = [field for field in self.REQUIRED_FIELDS
                    if not self.cfg.get(field)]
         if missing:
-            raise MeterDefinitionException(
+            raise declarative.DefinitionException(
                 _LE("Required fields %s not specified") % missing, self.cfg)
+
         self._event_type = self.cfg.get('event_type')
         if isinstance(self._event_type, six.string_types):
             self._event_type = [self._event_type]
 
         if ('type' not in self.cfg.get('lookup', []) and
                 self.cfg['type'] not in sample.TYPES):
-            raise MeterDefinitionException(
+            raise declarative.DefinitionException(
                 _LE("Invalid type %s specified") % self.cfg['type'], self.cfg)
 
-        self._field_getter = {}
+        self._fallback_user_id = declarative.Definition(
-        for name, field in self.cfg.items():
+            'user_id', "_context_user_id|_context_user", plugin_manager)
-            if name in ["event_type", "lookup"] or not field:
+        self._fallback_project_id = declarative.Definition(
-                continue
+            'project_id', "_context_tenant_id|_context_tenant", plugin_manager)
-            elif isinstance(field, six.integer_types):
+        self._attributes = {}
-                self._field_getter[name] = field
+        self._metadata_attributes = {}
-            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(
-                    self._parse_jsonpath_field, parts)
 
-    def parse_jsonpath(self, field):
+        for name in self.SAMPLE_ATTRIBUTES:
-        try:
+            attr_cfg = self.cfg.get(name)
-            parts = self.JSONPATH_RW_PARSER.parse(field)
+            if attr_cfg:
-        except Exception as e:
+                self._attributes[name] = declarative.Definition(
-            raise MeterDefinitionException(_LE(
+                    name, attr_cfg, plugin_manager)
-                "Parse error in JSONPath specification "
+        metadata = self.cfg.get('metadata', {})
-                "'%(jsonpath)s': %(err)s")
+        for name in metadata:
-                % dict(jsonpath=field, err=e), self.cfg)
+            self._metadata_attributes[name] = declarative.Definition(
-        return parts
+                name, metadata[name], plugin_manager)
+
+        # List of fields we expected when multiple meter are in the payload
+        self.lookup = self.cfg.get('lookup')
+        if isinstance(self.lookup, six.string_types):
+            self.lookup = [self.lookup]
 
     def match_type(self, meter_name):
         for t in self._event_type:
             if fnmatch.fnmatch(meter_name, t):
                 return True
 
-    def parse_fields(self, field, message, all_values=False):
+    def to_samples(self, message, all_values=False):
-        getter = self._field_getter.get(field)
+        # Sample defaults
-        if not getter:
+        sample = {
-            return
+            'name': self.cfg["name"], 'type': self.cfg["type"],
-        elif isinstance(getter, dict):
+            'unit': self.cfg["unit"], 'volume': None, 'timestamp': None,
-            dict_val = {}
+            'user_id': self._fallback_user_id.parse(message),
-            for key, val in getter.items():
+            'project_id': self._fallback_project_id.parse(message),
-                dict_val[key] = val(message, all_values)
+            'resource_id': None, 'message': message, 'metadata': {},
-            return dict_val
+        }
-        elif callable(getter):
+        for name, parser in self._metadata_attributes.items():
-            return getter(message, all_values)
+            value = parser.parse(message)
-        else:
+            if value:
-            return getter
+                sample['metadata'][name] = value
 
-    @staticmethod
+        # NOTE(sileht): We expect multiple samples in the payload
-    def _parse_jsonpath_field(parts, message, all_values):
+        # so put each attribute into a list
-        values = [match.value for match in parts.find(message)
+        if self.lookup:
-                  if match.value is not None]
+            for name in sample:
-        if values:
+                sample[name] = [sample[name]]
-            if not all_values:
+
-                return values[0]
+        for name in self.SAMPLE_ATTRIBUTES:
-            return values
+            parser = self._attributes.get(name)
+            if parser is not None:
+                value = parser.parse(message, bool(self.lookup))
+                # NOTE(sileht): If we expect multiple samples
+                # some attributes and overriden even we doesn't get any
+                # result. Also note in this case value is always a list
+                if ((not self.lookup and value is not None) or
+                        (self.lookup and ((name in self.lookup + ["name"])
+                                          or value))):
+                    sample[name] = value
+
+        if self.lookup:
+            nb_samples = len(sample['name'])
+            # skip if no meters in payload
+            if nb_samples <= 0:
+                raise StopIteration
+
+            attributes = self.SAMPLE_ATTRIBUTES + ["message", "metadata"]
+
+            samples_values = []
+            for name in attributes:
+                values = sample.get(name)
+                nb_values = len(values)
+                if nb_values == nb_samples:
+                    samples_values.append(values)
+                elif nb_values == 1 and name not in self.lookup:
+                    samples_values.append(itertools.cycle(values))
+                else:
+                    nb = (0 if nb_values == 1 and values[0] is None
+                          else nb_values)
+                    LOG.warning('Only %(nb)d fetched meters contain '
+                                '"%(name)s" field instead of %(total)d.' %
+                                dict(name=name, nb=nb,
+                                     total=nb_samples))
+                    raise StopIteration
+
+            # NOTE(sileht): Transform the sample with multiple values per
+            # attribute into multiple samples with one value per attribute.
+            for values in zip(*samples_values):
+                yield dict((attributes[idx], value)
+                           for idx, value in enumerate(values))
+        else:
+            yield sample
 
 
 def get_config_file():
@@ -182,23 +217,23 @@ def setup_meters_config():
 def load_definitions(config_def):
     if not config_def:
         return []
+
+    plugin_manager = extension.ExtensionManager(
+        namespace='ceilometer.event.trait_plugin')
+
     meter_defs = []
     for event_def in reversed(config_def['metric']):
         try:
             if (event_def['volume'] != 1 or
                     not cfg.CONF.notification.disable_non_metric_meters):
-                meter_defs.append(MeterDefinition(event_def))
+                meter_defs.append(MeterDefinition(event_def, plugin_manager))
-        except MeterDefinitionException as me:
+        except declarative.DefinitionException as me:
             errmsg = (_LE("Error loading meter definition : %(err)s")
-                      % dict(err=me.message))
+                      % dict(err=six.text_type(me)))
             LOG.error(errmsg)
     return meter_defs
 
 
-class InvalidPayload(Exception):
-    pass
-
-
 class ProcessMeterNotifications(plugin_base.NotificationBase):
 
     event_types = []
@@ -237,82 +272,8 @@ class ProcessMeterNotifications(plugin_base.NotificationBase):
                            for topic in conf.notification_topics)
         return targets
 
-    @staticmethod
-    def _normalise_as_list(value, d, body, length):
-        values = d.parse_fields(value, body, True)
-        if not values:
-            if value in d.cfg.get('lookup'):
-                LOG.warning('Could not find %s values', value)
-                raise InvalidPayload
-            values = [d.cfg[value]]
-        elif value in d.cfg.get('lookup') and length != len(values):
-            LOG.warning('Not all fetched meters contain "%s" field', value)
-            raise InvalidPayload
-        return values if isinstance(values, list) else [values]
-
     def process_notification(self, notification_body):
         for d in self.definitions:
             if d.match_type(notification_body['event_type']):
-                userid = self.get_user_id(d, notification_body)
+                for s in d.to_samples(notification_body):
-                projectid = self.get_project_id(d, notification_body)
+                    yield sample.Sample.from_notification(**s)
-                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
-                        break
-                    try:
-                        resources = self._normalise_as_list(
-                            'resource_id', d, notification_body, len(meters))
-                        volumes = self._normalise_as_list(
-                            'volume', d, notification_body, len(meters))
-                        units = self._normalise_as_list(
-                            'unit', d, notification_body, len(meters))
-                        types = self._normalise_as_list(
-                            'type', d, notification_body, len(meters))
-                        users = (self._normalise_as_list(
-                            'user_id', d, notification_body, len(meters))
-                            if 'user_id' in d.cfg['lookup'] else [userid])
-                        projs = (self._normalise_as_list(
-                            'project_id', d, notification_body, len(meters))
-                            if 'project_id' in d.cfg['lookup']
-                            else [projectid])
-                        times = (self._normalise_as_list(
-                            'timestamp', d, notification_body, len(meters))
-                            if 'timestamp' in d.cfg['lookup'] else [ts])
-                    except InvalidPayload:
-                        break
-                    for m, v, unit, t, r, p, user, ts in zip(
-                            meters, volumes, itertools.cycle(units),
-                            itertools.cycle(types), itertools.cycle(resources),
-                            itertools.cycle(projs), itertools.cycle(users),
-                            itertools.cycle(times)):
-                        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,
-                            metadata=metadata)
-                else:
-                    yield sample.Sample.from_notification(
-                        name=d.cfg['name'],
-                        type=d.cfg['type'],
-                        unit=d.cfg['unit'],
-                        volume=d.parse_fields('volume', notification_body),
-                        resource_id=resourceid,
-                        user_id=userid,
-                        project_id=projectid,
-                        message=notification_body,
-                        timestamp=ts, metadata=metadata)
-
-    @staticmethod
-    def get_user_id(d, notification_body):
-        return (d.parse_fields('user_id', notification_body) or
-                notification_body.get('_context_user_id') or
-                notification_body.get('_context_user', None))
-
-    @staticmethod
-    def get_project_id(d, notification_body):
-        return (d.parse_fields('project_id', notification_body) or
-                notification_body.get('_context_tenant_id') or
-                notification_body.get('_context_tenant', None))

ceilometer/tests/unit/dispatcher/test_gnocchi.py

@@ -28,6 +28,7 @@ import six
 import six.moves.urllib.parse as urlparse
 import testscenarios
 
+from ceilometer import declarative
 from ceilometer.dispatcher import gnocchi
 from ceilometer import service as ceilometer_service
 from ceilometer.tests import base
@@ -133,7 +134,7 @@ class DispatcherTest(base.BaseTestCase):
             self.conf.config(filter_service_activity=False,
                              resources_definition_file=temp,
                              group='dispatcher_gnocchi')
-            self.assertRaises(gnocchi.ResourcesDefinitionException,
+            self.assertRaises(declarative.DefinitionException,
                               gnocchi.GnocchiDispatcher, self.conf.conf)
 
     @mock.patch('ceilometer.dispatcher.gnocchi.GnocchiDispatcher'

ceilometer/tests/unit/event/test_converter.py

@@ -20,6 +20,7 @@ import mock
 from oslo_config import fixture as fixture_config
 import six
 
+from ceilometer import declarative
 from ceilometer.event import converter
 from ceilometer.event.storage import models
 from ceilometer.tests import base
@@ -112,13 +113,13 @@ class TestTraitDefinition(ConverterBase):
         self.ext1 = mock.MagicMock(name='mock_test_plugin')
         self.test_plugin_class = self.ext1.plugin
         self.test_plugin = self.test_plugin_class()
-        self.test_plugin.trait_value.return_value = 'foobar'
+        self.test_plugin.trait_values.return_value = ['foobar']
         self.ext1.reset_mock()
 
         self.ext2 = mock.MagicMock(name='mock_nothing_plugin')
         self.nothing_plugin_class = self.ext2.plugin
         self.nothing_plugin = self.nothing_plugin_class()
-        self.nothing_plugin.trait_value.return_value = None
+        self.nothing_plugin.trait_values.return_value = [None]
         self.ext2.reset_mock()
 
         self.fake_plugin_mgr = dict(test=self.ext1, nothing=self.ext2)
@@ -136,7 +137,7 @@ class TestTraitDefinition(ConverterBase):
         self.assertEqual(models.Trait.TEXT_TYPE, t.dtype)
         self.assertEqual('foobar', t.value)
         self.test_plugin_class.assert_called_once_with()
-        self.test_plugin.trait_value.assert_called_once_with([
+        self.test_plugin.trait_values.assert_called_once_with([
             ('payload.instance_id', 'id-for-instance-0001'),
             ('payload.instance_uuid', 'uuid-for-instance-0001')])
 
@@ -153,7 +154,7 @@ class TestTraitDefinition(ConverterBase):
         self.assertEqual(models.Trait.TEXT_TYPE, t.dtype)
         self.assertEqual('foobar', t.value)
         self.test_plugin_class.assert_called_once_with()
-        self.test_plugin.trait_value.assert_called_once_with([])
+        self.test_plugin.trait_values.assert_called_once_with([])
 
     def test_to_trait_with_plugin_null(self):
         cfg = dict(type='text',
@@ -165,7 +166,7 @@ class TestTraitDefinition(ConverterBase):
         t = tdef.to_trait(self.n1)
         self.assertIs(None, t)
         self.nothing_plugin_class.assert_called_once_with()
-        self.nothing_plugin.trait_value.assert_called_once_with([
+        self.nothing_plugin.trait_values.assert_called_once_with([
             ('payload.instance_id', 'id-for-instance-0001'),
             ('payload.instance_uuid', 'uuid-for-instance-0001')])
 
@@ -182,7 +183,7 @@ class TestTraitDefinition(ConverterBase):
         self.assertEqual(models.Trait.TEXT_TYPE, t.dtype)
         self.assertEqual('foobar', t.value)
         self.test_plugin_class.assert_called_once_with(a=1, b='foo')
-        self.test_plugin.trait_value.assert_called_once_with([
+        self.test_plugin.trait_values.assert_called_once_with([
             ('payload.instance_id', 'id-for-instance-0001'),
             ('payload.instance_uuid', 'uuid-for-instance-0001')])
 
@@ -287,7 +288,7 @@ class TestTraitDefinition(ConverterBase):
         self.assertIs(None, t)
 
     def test_missing_fields_config(self):
-        self.assertRaises(converter.EventDefinitionException,
+        self.assertRaises(declarative.DefinitionException,
                           converter.TraitDefinition,
                           'bogus_trait',
                           dict(),
@@ -296,19 +297,20 @@ class TestTraitDefinition(ConverterBase):
     def test_string_fields_config(self):
         cfg = dict(fields='payload.test')
         t = converter.TraitDefinition('test_trait', cfg, self.fake_plugin_mgr)
-        self.assertPathsEqual(t.fields, jsonpath_rw_ext.parse('payload.test'))
+        self.assertPathsEqual(t.getter.__self__,
+                              jsonpath_rw_ext.parse('payload.test'))
 
     def test_list_fields_config(self):
         cfg = dict(fields=['payload.test', 'payload.other'])
         t = converter.TraitDefinition('test_trait', cfg, self.fake_plugin_mgr)
         self.assertPathsEqual(
-            t.fields,
+            t.getter.__self__,
             jsonpath_rw_ext.parse('(payload.test)|(payload.other)'))
 
     def test_invalid_path_config(self):
         # test invalid jsonpath...
         cfg = dict(fields='payload.bogus(')
-        self.assertRaises(converter.EventDefinitionException,
+        self.assertRaises(declarative.DefinitionException,
                           converter.TraitDefinition,
                           'bogus_trait',
                           cfg,
@@ -317,7 +319,7 @@ class TestTraitDefinition(ConverterBase):
     def test_invalid_plugin_config(self):
         # test invalid jsonpath...
         cfg = dict(fields='payload.test', plugin=dict(bogus="true"))
-        self.assertRaises(converter.EventDefinitionException,
+        self.assertRaises(declarative.DefinitionException,
                           converter.TraitDefinition,
                           'test_trait',
                           cfg,
@@ -326,7 +328,7 @@ class TestTraitDefinition(ConverterBase):
     def test_unknown_plugin(self):
         # test invalid jsonpath...
         cfg = dict(fields='payload.test', plugin=dict(name='bogus'))
-        self.assertRaises(converter.EventDefinitionException,
+        self.assertRaises(declarative.DefinitionException,
                           converter.TraitDefinition,
                           'test_trait',
                           cfg,
@@ -352,7 +354,7 @@ class TestTraitDefinition(ConverterBase):
     def test_invalid_type_config(self):
         # test invalid jsonpath...
         cfg = dict(type='bogus', fields='payload.test')
-        self.assertRaises(converter.EventDefinitionException,
+        self.assertRaises(declarative.DefinitionException,
                           converter.TraitDefinition,
                           'bogus_trait',
                           cfg,
@@ -438,14 +440,14 @@ class TestEventDefinition(ConverterBase):
 
     def test_bogus_cfg_no_traits(self):
         bogus = dict(event_type='test.foo')
-        self.assertRaises(converter.EventDefinitionException,
+        self.assertRaises(declarative.DefinitionException,
                           converter.EventDefinition,
                           bogus,
                           self.fake_plugin_mgr)
 
     def test_bogus_cfg_no_type(self):
         bogus = dict(traits=self.traits_cfg)
-        self.assertRaises(converter.EventDefinitionException,
+        self.assertRaises(declarative.DefinitionException,
                           converter.EventDefinition,
                           bogus,
                           self.fake_plugin_mgr)

ceilometer/tests/unit/event/test_trait_plugins.py

@@ -27,41 +27,41 @@ class TestSplitterPlugin(base.BaseTestCase):
         param = dict(separator='-', segment=0)
         plugin = self.pclass(**param)
         match_list = [('test.thing', 'test-foobar-baz')]
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)[0]
         self.assertEqual('test', value)
 
         param = dict(separator='-', segment=1)
         plugin = self.pclass(**param)
         match_list = [('test.thing', 'test-foobar-baz')]
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)[0]
         self.assertEqual('foobar', value)
 
         param = dict(separator='-', segment=1, max_split=1)
         plugin = self.pclass(**param)
         match_list = [('test.thing', 'test-foobar-baz')]
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)[0]
         self.assertEqual('foobar-baz', value)
 
     def test_no_sep(self):
         param = dict(separator='-', segment=0)
         plugin = self.pclass(**param)
         match_list = [('test.thing', 'test.foobar.baz')]
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)[0]
         self.assertEqual('test.foobar.baz', value)
 
     def test_no_segment(self):
         param = dict(separator='-', segment=5)
         plugin = self.pclass(**param)
         match_list = [('test.thing', 'test-foobar-baz')]
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)[0]
         self.assertIs(None, value)
 
     def test_no_match(self):
         param = dict(separator='-', segment=0)
         plugin = self.pclass(**param)
         match_list = []
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)
-        self.assertIs(None, value)
+        self.assertEqual([], value)
 
 
 class TestBitfieldPlugin(base.BaseTestCase):
@@ -86,8 +86,8 @@ class TestBitfieldPlugin(base.BaseTestCase):
                       ('thingy.boink', 'testagain')]
 
         plugin = self.pclass(**self.params)
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)
-        self.assertEqual(0x412, value)
+        self.assertEqual(0x412, value[0])
 
     def test_initial(self):
         match_list = [('payload.foo', 12),
@@ -95,14 +95,14 @@ class TestBitfieldPlugin(base.BaseTestCase):
                       ('thingy.boink', 'testagain')]
         self.params['initial_bitfield'] = 0x2000
         plugin = self.pclass(**self.params)
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)
-        self.assertEqual(0x2412, value)
+        self.assertEqual(0x2412, value[0])
 
     def test_no_match(self):
         match_list = []
         plugin = self.pclass(**self.params)
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)
-        self.assertEqual(self.init, value)
+        self.assertEqual(self.init, value[0])
 
     def test_multi(self):
         match_list = [('payload.foo', 12),
@@ -111,5 +111,5 @@ class TestBitfieldPlugin(base.BaseTestCase):
                       ('thingy.boink', 'testagain')]
 
         plugin = self.pclass(**self.params)
-        value = plugin.trait_value(match_list)
+        value = plugin.trait_values(match_list)
-        self.assertEqual(0x412, value)
+        self.assertEqual(0x412, value[0])

ceilometer/tests/unit/meter/test_notifications.py

@@ -18,9 +18,11 @@ import six
 import yaml
 
 from oslo_config import fixture as fixture_config
+from oslo_utils import encodeutils
 from oslo_utils import fileutils
 from oslotest import mockpatch
 
+from ceilometer import declarative
 from ceilometer.meter import notifications
 from ceilometer import service as ceilometer_service
 from ceilometer.tests import base as test
@@ -223,31 +225,34 @@ class TestMeterDefinition(test.BaseTestCase):
                    volume="$.payload.volume",
                    resource_id="$.payload.resource_id",
                    project_id="$.payload.project_id")
-        handler = notifications.MeterDefinition(cfg)
+        handler = notifications.MeterDefinition(cfg, mock.Mock())
         self.assertTrue(handler.match_type("test.create"))
-        self.assertEqual(1.0, handler.parse_fields("volume", NOTIFICATION))
+        sample = list(handler.to_samples(NOTIFICATION))[0]
+        self.assertEqual(1.0, sample["volume"])
         self.assertEqual("bea70e51c7340cb9d555b15cbfcaec23",
-                         handler.parse_fields("resource_id", NOTIFICATION))
+                         sample["resource_id"])
         self.assertEqual("30be1fc9a03c4e94ab05c403a8a377f2",
-                         handler.parse_fields("project_id", NOTIFICATION))
+                         sample["project_id"])
 
     def test_config_required_missing_fields(self):
         cfg = dict()
         try:
-            notifications.MeterDefinition(cfg)
+            notifications.MeterDefinition(cfg, mock.Mock())
-        except notifications.MeterDefinitionException as e:
+        except declarative.DefinitionException as e:
             self.assertEqual("Required fields ['name', 'type', 'event_type',"
                              " 'unit', 'volume', 'resource_id']"
-                             " not specified", e.message)
+                             " not specified",
+                             encodeutils.exception_to_unicode(e))
 
     def test_bad_type_cfg_definition(self):
         cfg = dict(name="test", type="foo", event_type="bar.create",
                    unit="foo", volume="bar",
                    resource_id="bea70e51c7340cb9d555b15cbfcaec23")
         try:
-            notifications.MeterDefinition(cfg)
+            notifications.MeterDefinition(cfg, mock.Mock())
-        except notifications.MeterDefinitionException as e:
+        except declarative.DefinitionException as e:
-            self.assertEqual("Invalid type foo specified", e.message)
+            self.assertEqual("Invalid type foo specified",
+                             encodeutils.exception_to_unicode(e))
 
 
 class TestMeterProcessing(test.BaseTestCase):
@@ -601,7 +606,8 @@ class TestMeterProcessing(test.BaseTestCase):
             self.__setup_meter_def_file(cfg))
         c = list(self.handler.process_notification(event))
         self.assertEqual(0, len(c))
-        LOG.warning.assert_called_with('Could not find %s values', 'volume')
+        LOG.warning.assert_called_with('Only 0 fetched meters contain '
+                                       '"volume" field instead of 2.')
 
     @mock.patch('ceilometer.meter.notifications.LOG')
     def test_multi_meter_payload_invalid_short(self, LOG):
@@ -620,8 +626,8 @@ class TestMeterProcessing(test.BaseTestCase):
             self.__setup_meter_def_file(cfg))
         c = list(self.handler.process_notification(event))
         self.assertEqual(0, len(c))
-        LOG.warning.assert_called_with('Not all fetched meters contain "%s" '
+        LOG.warning.assert_called_with('Only 1 fetched meters contain '
-                                       'field', 'volume')
+                                       '"volume" field instead of 2.')
 
     def test_arithmetic_expr_meter(self):
         cfg = yaml.dump(