# # Copyright 2014-2015 eNovance # # 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 collections import defaultdict import hashlib import itertools import json import operator import pkg_resources import threading import uuid from gnocchiclient import exceptions as gnocchi_exc from keystoneauth1 import exceptions as ka_exceptions from oslo_log import log from oslo_utils import fnmatch from oslo_utils import timeutils from stevedore import extension from urllib import parse as urlparse from ceilometer import declarative from ceilometer import gnocchi_client from ceilometer.i18n import _ from ceilometer import keystone_client from ceilometer import publisher NAME_ENCODED = __name__.encode('utf-8') CACHE_NAMESPACE = uuid.UUID(bytes=hashlib.md5(NAME_ENCODED).digest()) LOG = log.getLogger(__name__) def cache_key_mangler(key): """Construct an opaque cache key.""" return uuid.uuid5(CACHE_NAMESPACE, key).hex EVENT_CREATE, EVENT_UPDATE, EVENT_DELETE = ("create", "update", "delete") class ResourcesDefinition(object): MANDATORY_FIELDS = {'resource_type': str, 'metrics': (dict, list)} MANDATORY_EVENT_FIELDS = {'id': str} def __init__(self, definition_cfg, archive_policy_default, archive_policy_override, plugin_manager): self.cfg = definition_cfg self._check_required_and_types(self.MANDATORY_FIELDS, self.cfg) if self.support_events(): self._check_required_and_types(self.MANDATORY_EVENT_FIELDS, self.cfg['event_attributes']) self._attributes = {} for name, attr_cfg in self.cfg.get('attributes', {}).items(): self._attributes[name] = declarative.Definition(name, attr_cfg, plugin_manager) self._event_attributes = {} for name, attr_cfg in self.cfg.get('event_attributes', {}).items(): self._event_attributes[name] = declarative.Definition( name, attr_cfg, plugin_manager) self.metrics = {} # NOTE(sileht): Convert old list to new dict format if isinstance(self.cfg['metrics'], list): values = [None] * len(self.cfg['metrics']) self.cfg['metrics'] = dict(zip(self.cfg['metrics'], values)) for m, extra in self.cfg['metrics'].items(): if not extra: extra = {} if not extra.get("archive_policy_name"): extra["archive_policy_name"] = archive_policy_default if archive_policy_override: extra["archive_policy_name"] = archive_policy_override # NOTE(sileht): For backward compat, this is after the override to # preserve the wierd previous behavior. We don't really care as we # deprecate it. if 'archive_policy' in self.cfg: LOG.warning("archive_policy '%s' for a resource-type (%s) is " "deprecated, set it for each metric instead.", self.cfg["archive_policy"], self.cfg["resource_type"]) extra["archive_policy_name"] = self.cfg['archive_policy'] self.metrics[m] = extra @staticmethod def _check_required_and_types(expected, definition): for field, field_types in expected.items(): if field not in definition: raise declarative.ResourceDefinitionException( _("Required field %s not specified") % field, definition) if not isinstance(definition[field], field_types): raise declarative.ResourceDefinitionException( _("Required field %(field)s should be a %(type)s") % {'field': field, 'type': field_types}, definition) @staticmethod def _ensure_list(value): if isinstance(value, list): return value return [value] def support_events(self): for e in ["event_create", "event_delete", "event_update"]: if e in self.cfg: return True return False def event_match(self, event_type): for e in self._ensure_list(self.cfg.get('event_create', [])): if fnmatch.fnmatch(event_type, e): return EVENT_CREATE for e in self._ensure_list(self.cfg.get('event_delete', [])): if fnmatch.fnmatch(event_type, e): return EVENT_DELETE for e in self._ensure_list(self.cfg.get('event_update', [])): if fnmatch.fnmatch(event_type, e): return EVENT_UPDATE def sample_attributes(self, sample): attrs = {} sample_dict = sample.as_dict() for name, definition in self._attributes.items(): value = definition.parse(sample_dict) if value is not None: attrs[name] = value return attrs def event_attributes(self, event): attrs = {'type': self.cfg['resource_type']} traits = dict([(trait.name, trait.value) for trait in event.traits]) for attr, field in self.cfg.get('event_attributes', {}).items(): value = traits.get(field) if value is not None: attrs[attr] = value return attrs class LockedDefaultDict(defaultdict): """defaultdict with lock to handle threading Dictionary only deletes if nothing is accessing dict and nothing is holding lock to be deleted. If both cases are not true, it will skip delete. """ def __init__(self, *args, **kwargs): self.lock = threading.Lock() super(LockedDefaultDict, self).__init__(*args, **kwargs) def __getitem__(self, key): with self.lock: return super(LockedDefaultDict, self).__getitem__(key) def pop(self, key, *args): with self.lock: key_lock = super(LockedDefaultDict, self).__getitem__(key) if key_lock.acquire(False): try: super(LockedDefaultDict, self).pop(key, *args) finally: key_lock.release() class GnocchiPublisher(publisher.ConfigPublisherBase): """Publisher class for recording metering data into the Gnocchi service. The publisher class records each meter into the gnocchi service configured in Ceilometer pipeline file. An example target may look like the following: gnocchi://?archive_policy=low&filter_project=gnocchi """ def __init__(self, conf, parsed_url): super(GnocchiPublisher, self).__init__(conf, parsed_url) # TODO(jd) allow to override Gnocchi endpoint via the host in the URL options = urlparse.parse_qs(parsed_url.query) self.filter_project = options.get('filter_project', ['service'])[-1] self.filter_domain = options.get('filter_domain', ['Default'])[-1] resources_definition_file = options.get( 'resources_definition_file', ['gnocchi_resources.yaml'])[-1] archive_policy_override = options.get('archive_policy', [None])[-1] self.resources_definition, self.archive_policies_definition = ( self._load_definitions(conf, archive_policy_override, resources_definition_file)) self.metric_map = dict((metric, rd) for rd in self.resources_definition for metric in rd.metrics) timeout = options.get('timeout', [6.05])[-1] self._ks_client = keystone_client.get_client(conf) self.cache = None try: import oslo_cache oslo_cache.configure(conf) # NOTE(cdent): The default cache backend is a real but # noop backend. We don't want to use that here because # we want to avoid the cache pathways entirely if the # cache has not been configured explicitly. if conf.cache.enabled: cache_region = oslo_cache.create_region() self.cache = oslo_cache.configure_cache_region( conf, cache_region) self.cache.key_mangler = cache_key_mangler except ImportError: pass except oslo_cache.exception.ConfigurationError as exc: LOG.warning('unable to configure oslo_cache: %s', exc) self._gnocchi_project_id = None self._gnocchi_project_id_lock = threading.Lock() self._gnocchi_resource_lock = LockedDefaultDict(threading.Lock) self._gnocchi = gnocchi_client.get_gnocchiclient( conf, request_timeout=timeout) self._already_logged_event_types = set() self._already_logged_metric_names = set() self._already_configured_archive_policies = False @staticmethod def _load_definitions(conf, archive_policy_override, resources_definition_file): plugin_manager = extension.ExtensionManager( namespace='ceilometer.event.trait_plugin') data = declarative.load_definitions( conf, {}, resources_definition_file, pkg_resources.resource_filename(__name__, "data/gnocchi_resources.yaml")) archive_policy_default = data.get("archive_policy_default", "ceilometer-low") resource_defs = [] for resource in data.get('resources', []): try: resource_defs.append(ResourcesDefinition( resource, archive_policy_default, archive_policy_override, plugin_manager)) except Exception: LOG.error("Failed to load resource due to error", exc_info=True) return resource_defs, data.get("archive_policies", []) def ensures_archives_policies(self): if not self._already_configured_archive_policies: for ap in self.archive_policies_definition: try: self._gnocchi.archive_policy.create(ap) except gnocchi_exc.ArchivePolicyAlreadyExists: # created in the meantime by another worker pass self._already_configured_archive_policies = True @property def gnocchi_project_id(self): if self._gnocchi_project_id is not None: return self._gnocchi_project_id with self._gnocchi_project_id_lock: if self._gnocchi_project_id is None: try: project = self._ks_client.projects.find( name=self.filter_project, domain=self.filter_domain) except ka_exceptions.NotFound: LOG.warning('filtered project not found in keystone,' ' ignoring the filter_project ' 'option') self.filter_project = None return None except Exception: LOG.exception('fail to retrieve filtered project ') raise self._gnocchi_project_id = project.id LOG.debug("filtered project found: %s", self._gnocchi_project_id) return self._gnocchi_project_id def _is_swift_account_sample(self, sample): try: return (self.metric_map[sample.name].cfg['resource_type'] == 'swift_account') except KeyError: return False def _is_gnocchi_activity(self, sample): return (self.filter_project and self.gnocchi_project_id and ( # avoid anything from the user used by gnocchi sample.project_id == self.gnocchi_project_id or # avoid anything in the swift account used by gnocchi (sample.resource_id == self.gnocchi_project_id and self._is_swift_account_sample(sample)) )) def _get_resource_definition_from_event(self, event_type): for rd in self.resources_definition: operation = rd.event_match(event_type) if operation: return rd, operation def publish_samples(self, data): self.ensures_archives_policies() # NOTE(sileht): skip sample generated by gnocchi itself data = [s for s in data if not self._is_gnocchi_activity(s)] data.sort(key=operator.attrgetter('resource_id')) resource_grouped_samples = itertools.groupby( data, key=operator.attrgetter('resource_id')) gnocchi_data = {} measures = {} for resource_id, samples_of_resource in resource_grouped_samples: for sample in samples_of_resource: metric_name = sample.name LOG.debug("Processing sample [%s] for resource ID [%s].", sample, resource_id) rd = self.metric_map.get(metric_name) if rd is None: if metric_name not in self._already_logged_metric_names: LOG.warning("metric %s is not handled by Gnocchi" % metric_name) self._already_logged_metric_names.add(metric_name) continue # NOTE(sileht): / is forbidden by Gnocchi resource_id = resource_id.replace('/', '_') if resource_id not in gnocchi_data: gnocchi_data[resource_id] = { 'resource_type': rd.cfg['resource_type'], 'resource': {"id": resource_id, "user_id": sample.user_id, "project_id": sample.project_id}} gnocchi_data[resource_id].setdefault( "resource_extra", {}).update(rd.sample_attributes(sample)) measures.setdefault(resource_id, {}).setdefault( metric_name, {"measures": [], "archive_policy_name": rd.metrics[metric_name]["archive_policy_name"], "unit": sample.unit} )["measures"].append( {'timestamp': sample.timestamp, 'value': sample.volume} ) try: self.batch_measures(measures, gnocchi_data) except gnocchi_exc.ClientException as e: LOG.error(str(e)) except Exception as e: LOG.error(str(e), exc_info=True) for info in gnocchi_data.values(): resource = info["resource"] resource_type = info["resource_type"] resource_extra = info["resource_extra"] if not resource_extra: continue try: self._if_not_cached(resource_type, resource['id'], resource_extra) except gnocchi_exc.ClientException as e: LOG.error(str(e)) except Exception as e: LOG.error(str(e), exc_info=True) @staticmethod def _extract_resources_from_error(e, resource_infos): resource_ids = set([r['original_resource_id'] for r in e.message['detail']]) return [(resource_infos[rid]['resource_type'], resource_infos[rid]['resource'], resource_infos[rid]['resource_extra']) for rid in resource_ids] def batch_measures(self, measures, resource_infos): # NOTE(sileht): We don't care about error here, we want # resources metadata always been updated try: self._gnocchi.metric.batch_resources_metrics_measures( measures, create_metrics=True) except gnocchi_exc.BadRequest as e: if not isinstance(e.message, dict): raise if e.message.get('cause') != 'Unknown resources': raise resources = self._extract_resources_from_error(e, resource_infos) for resource_type, resource, resource_extra in resources: try: resource.update(resource_extra) self._create_resource(resource_type, resource) except gnocchi_exc.ResourceAlreadyExists: # NOTE(sileht): resource created in the meantime pass except gnocchi_exc.ClientException as e: LOG.error('Error creating resource %(id)s: %(err)s', {'id': resource['id'], 'err': str(e)}) # We cannot post measures for this resource # and we can't patch it later del measures[resource['id']] del resource_infos[resource['id']] else: if self.cache and resource_extra: self.cache.set(resource['id'], self._hash_resource(resource_extra)) # NOTE(sileht): we have created missing resources/metrics, # now retry to post measures self._gnocchi.metric.batch_resources_metrics_measures( measures, create_metrics=True) LOG.debug( "%d measures posted against %d metrics through %d resources", sum(len(m["measures"]) for rid in measures for m in measures[rid].values()), sum(len(m) for m in measures.values()), len(resource_infos)) def _create_resource(self, resource_type, resource): self._gnocchi.resource.create(resource_type, resource) LOG.debug('Resource %s created', resource["id"]) def _update_resource(self, resource_type, res_id, resource_extra): self._gnocchi.resource.update(resource_type, res_id, resource_extra) LOG.debug('Resource %s updated', res_id) def _if_not_cached(self, resource_type, res_id, resource_extra): if self.cache: attribute_hash = self._hash_resource(resource_extra) if self._resource_cache_diff(res_id, attribute_hash): with self._gnocchi_resource_lock[res_id]: # NOTE(luogangyi): there is a possibility that the # resource was already built in cache by another # ceilometer-notification-agent when we get the lock here. if self._resource_cache_diff(res_id, attribute_hash): self._update_resource(resource_type, res_id, resource_extra) self.cache.set(res_id, attribute_hash) else: LOG.debug('Resource cache hit for %s', res_id) self._gnocchi_resource_lock.pop(res_id, None) else: LOG.debug('Resource cache hit for %s', res_id) else: self._update_resource(resource_type, res_id, resource_extra) @staticmethod def _hash_resource(resource): return hash(tuple(i for i in resource.items() if i[0] != 'metrics')) def _resource_cache_diff(self, key, attribute_hash): cached_hash = self.cache.get(key) return not cached_hash or cached_hash != attribute_hash def publish_events(self, events): for event in events: rd = self._get_resource_definition_from_event(event.event_type) if not rd: if event.event_type not in self._already_logged_event_types: LOG.debug("No gnocchi definition for event type: %s", event.event_type) self._already_logged_event_types.add(event.event_type) continue rd, operation = rd if operation == EVENT_DELETE: self._delete_event(rd, event) if operation == EVENT_CREATE: self._create_event(rd, event) if operation == EVENT_UPDATE: self._update_event(rd, event) def _update_event(self, rd, event): resource = rd.event_attributes(event) associated_resources = rd.cfg.get('event_associated_resources', {}) if associated_resources: to_update = itertools.chain([resource], *[ self._search_resource(resource_type, query % resource['id']) for resource_type, query in associated_resources.items() ]) else: to_update = [resource] for resource in to_update: self._set_update_attributes(resource) def _delete_event(self, rd, event): ended_at = timeutils.utcnow().isoformat() resource = rd.event_attributes(event) associated_resources = rd.cfg.get('event_associated_resources', {}) if associated_resources: to_end = itertools.chain([resource], *[ self._search_resource(resource_type, query % resource['id']) for resource_type, query in associated_resources.items() ]) else: to_end = [resource] for resource in to_end: self._set_ended_at(resource, ended_at) def _create_event(self, rd, event): resource = rd.event_attributes(event) resource_type = resource.pop('type') try: self._create_resource(resource_type, resource) except gnocchi_exc.ResourceAlreadyExists: LOG.debug("Create event received on existing resource (%s), " "ignore it.", resource['id']) except Exception: LOG.error("Failed to create resource %s", resource, exc_info=True) def _search_resource(self, resource_type, query): try: return self._gnocchi.resource.search( resource_type, json.loads(query)) except Exception: LOG.error("Fail to search resource type %(resource_type)s " "with '%(query)s'", {'resource_type': resource_type, 'query': query}, exc_info=True) return [] def _set_update_attributes(self, resource): try: resource_id = resource.pop('id') resource_type = resource.pop('type') self._if_not_cached(resource_type, resource_id, resource) except gnocchi_exc.ResourceNotFound: LOG.debug("Update event received on unexisting resource (%s), " "ignore it.", resource['id']) except Exception: LOG.error("Fail to update the resource %s", resource, exc_info=True) def _set_ended_at(self, resource, ended_at): try: self._gnocchi.resource.update(resource['type'], resource['id'], {'ended_at': ended_at}) except gnocchi_exc.ResourceNotFound: LOG.debug("Delete event received on unexisting resource (%s), " "ignore it.", resource['id']) except Exception: LOG.error("Fail to update the resource %s", resource, exc_info=True) LOG.debug('Resource %s ended at %s' % (resource["id"], ended_at))