Source code for ceilometer.meter.notifications

#
# 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.

import itertools
import pkg_resources
import six

from oslo_config import cfg
from oslo_log import log
import oslo_messaging
from oslo_utils import fnmatch
from stevedore import extension

from ceilometer.agent import plugin_base
from ceilometer import declarative
from ceilometer.i18n import _LE, _LW
from ceilometer import sample

OPTS = [
    cfg.StrOpt('meter_definitions_cfg_file',
               default="meters.yaml",
               help="Configuration file for defining meter notifications."
               ),
]

cfg.CONF.register_opts(OPTS, group='meter')
cfg.CONF.import_opt('disable_non_metric_meters', 'ceilometer.notification',
                    group='notification')

LOG = log.getLogger(__name__)


[docs]class MeterDefinition(object): 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, plugin_manager): self.cfg = definition_cfg missing = [field for field in self.REQUIRED_FIELDS if not self.cfg.get(field)] if missing: raise declarative.MeterDefinitionException( _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 declarative.MeterDefinitionException( _LE("Invalid type %s specified") % self.cfg['type'], self.cfg) self._fallback_user_id = declarative.Definition( 'user_id', "_context_user_id|_context_user", plugin_manager) self._fallback_project_id = declarative.Definition( 'project_id', "_context_tenant_id|_context_tenant", plugin_manager) self._attributes = {} self._metadata_attributes = {} for name in self.SAMPLE_ATTRIBUTES: attr_cfg = self.cfg.get(name) if attr_cfg: self._attributes[name] = declarative.Definition( name, attr_cfg, plugin_manager) metadata = self.cfg.get('metadata', {}) for name in metadata: self._metadata_attributes[name] = declarative.Definition( 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]
[docs] def match_type(self, meter_name): for t in self._event_type: if fnmatch.fnmatch(meter_name, t): return True
[docs] def to_samples(self, message, all_values=False): # Sample defaults sample = { 'name': self.cfg["name"], 'type': self.cfg["type"], 'unit': self.cfg["unit"], 'volume': None, 'timestamp': None, 'user_id': self._fallback_user_id.parse(message), 'project_id': self._fallback_project_id.parse(message), 'resource_id': None, 'message': message, 'metadata': {}, } for name, parser in self._metadata_attributes.items(): value = parser.parse(message) if value: sample['metadata'][name] = value # NOTE(sileht): We expect multiple samples in the payload # so put each attribute into a list if self.lookup: for name in sample: sample[name] = [sample[name]] for name in self.SAMPLE_ATTRIBUTES: 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 are overridden even we don'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
[docs]class ProcessMeterNotifications(plugin_base.NotificationBase): event_types = [] def __init__(self, manager): super(ProcessMeterNotifications, self).__init__(manager) self.definitions = self._load_definitions() @staticmethod def _load_definitions(): plugin_manager = extension.ExtensionManager( namespace='ceilometer.event.trait_plugin') meters_cfg = declarative.load_definitions( {}, cfg.CONF.meter.meter_definitions_cfg_file, pkg_resources.resource_filename(__name__, "data/meters.yaml")) definitions = {} for meter_cfg in reversed(meters_cfg['metric']): if meter_cfg.get('name') in definitions: # skip duplicate meters LOG.warning(_LW("Skipping duplicate meter definition %s") % meter_cfg) continue if (meter_cfg.get('volume') != 1 or not cfg.CONF.notification.disable_non_metric_meters): try: md = MeterDefinition(meter_cfg, plugin_manager) except declarative.DefinitionException as e: errmsg = _LE("Error loading meter definition: %s") LOG.error(errmsg, six.text_type(e)) else: definitions[meter_cfg['name']] = md return definitions.values()
[docs] def get_targets(self, conf): """Return a sequence of oslo_messaging.Target It is defining the exchange and topics to be connected for this plugin. :param conf: Configuration. #TODO(prad): This should be defined in the notification agent """ targets = [] exchanges = [ conf.nova_control_exchange, conf.cinder_control_exchange, conf.glance_control_exchange, conf.neutron_control_exchange, conf.heat_control_exchange, conf.keystone_control_exchange, conf.sahara_control_exchange, conf.trove_control_exchange, conf.zaqar_control_exchange, conf.swift_control_exchange, conf.ceilometer_control_exchange, conf.magnum_control_exchange, conf.dns_control_exchange, ] for exchange in exchanges: targets.extend(oslo_messaging.Target(topic=topic, exchange=exchange) for topic in self.get_notification_topics(conf)) return targets
[docs] def process_notification(self, notification_body): for d in self.definitions: if d.match_type(notification_body['event_type']): for s in d.to_samples(notification_body): yield sample.Sample.from_notification(**s)

Project Source