524 lines
21 KiB
Python
524 lines
21 KiB
Python
#
|
|
# 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 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
|
|
import six
|
|
import six.moves.urllib.parse as urlparse
|
|
from stevedore import extension
|
|
import ujson
|
|
|
|
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."""
|
|
if six.PY2:
|
|
key = key.encode('utf-8')
|
|
return uuid.uuid5(CACHE_NAMESPACE, key).hex
|
|
|
|
|
|
EVENT_CREATE, EVENT_UPDATE, EVENT_DELETE = ("create", "update", "delete")
|
|
|
|
|
|
class ResourcesDefinition(object):
|
|
|
|
MANDATORY_FIELDS = {'resource_type': six.string_types,
|
|
'metrics': list}
|
|
|
|
MANDATORY_EVENT_FIELDS = {'id': six.string_types}
|
|
|
|
def __init__(self, definition_cfg, default_archive_policy, plugin_manager):
|
|
self._default_archive_policy = default_archive_policy
|
|
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 = {}
|
|
for t in self.cfg['metrics']:
|
|
archive_policy = self.cfg.get('archive_policy',
|
|
self._default_archive_policy)
|
|
if archive_policy is None:
|
|
self.metrics[t] = {}
|
|
else:
|
|
self.metrics[t] = dict(archive_policy_name=archive_policy)
|
|
|
|
@staticmethod
|
|
def _check_required_and_types(expected, definition):
|
|
for field, field_type in expected.items():
|
|
if field not in definition:
|
|
raise declarative.ResourceDefinitionException(
|
|
_("Required field %s not specified") % field, definition)
|
|
if not isinstance(definition[field], field_type):
|
|
raise declarative.ResourceDefinitionException(
|
|
_("Required field %(field)s should be a %(type)s") %
|
|
{'field': field, 'type': field_type}, definition)
|
|
|
|
@staticmethod
|
|
def _ensure_list(value):
|
|
if isinstance(value, list):
|
|
return value
|
|
return [value]
|
|
|
|
def metric_match(self, metric_name):
|
|
for t in self.cfg['metrics']:
|
|
if fnmatch.fnmatch(metric_name, t):
|
|
return True
|
|
return False
|
|
|
|
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 = {}
|
|
for name, definition in self._attributes.items():
|
|
value = definition.parse(sample.as_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',
|
|
[conf.dispatcher_gnocchi.filter_project])[-1]
|
|
|
|
resources_definition_file = options.get(
|
|
'resources_definition_file',
|
|
[conf.dispatcher_gnocchi.resources_definition_file])[-1]
|
|
archive_policy = options.get(
|
|
'archive_policy',
|
|
[conf.dispatcher_gnocchi.archive_policy])[-1]
|
|
self.resources_definition = self._load_resources_definitions(
|
|
conf, archive_policy, resources_definition_file)
|
|
|
|
timeout = options.get('timeout',
|
|
[conf.dispatcher_gnocchi.request_timeout])[-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()
|
|
|
|
@staticmethod
|
|
def _load_resources_definitions(conf, archive_policy,
|
|
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"))
|
|
resource_defs = []
|
|
for resource in data.get('resources', []):
|
|
try:
|
|
resource_defs.append(ResourcesDefinition(
|
|
resource,
|
|
archive_policy, plugin_manager))
|
|
except Exception as exc:
|
|
LOG.error("Failed to load resource due to error %s" %
|
|
exc)
|
|
return resource_defs
|
|
|
|
@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)
|
|
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):
|
|
return bool([rd for rd in self.resources_definition
|
|
if rd.cfg['resource_type'] == 'swift_account'
|
|
and rd.metric_match(sample.name)])
|
|
|
|
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_metric(self, metric_name):
|
|
for rd in self.resources_definition:
|
|
if rd.metric_match(metric_name):
|
|
return rd
|
|
|
|
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):
|
|
# NOTE(sileht): skip sample generated by gnocchi itself
|
|
data = [s for s in data if not self._is_gnocchi_activity(s)]
|
|
|
|
data.sort(key=lambda s: (s.resource_id, s.name))
|
|
resource_grouped_samples = itertools.groupby(
|
|
data, key=operator.attrgetter('resource_id'))
|
|
|
|
gnocchi_data = {}
|
|
measures = {}
|
|
stats = dict(measures=0, resources=0, metrics=0)
|
|
for resource_id, samples_of_resource in resource_grouped_samples:
|
|
# NOTE(sileht): / is forbidden by Gnocchi
|
|
resource_id = resource_id.replace('/', '_')
|
|
|
|
stats['resources'] += 1
|
|
metric_grouped_samples = itertools.groupby(
|
|
list(samples_of_resource),
|
|
key=operator.attrgetter('name'))
|
|
|
|
res_info = {}
|
|
for metric_name, samples in metric_grouped_samples:
|
|
stats['metrics'] += 1
|
|
|
|
samples = list(samples)
|
|
rd = self._get_resource_definition_from_metric(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
|
|
if rd.cfg.get("ignore"):
|
|
continue
|
|
|
|
res_info['resource_type'] = rd.cfg['resource_type']
|
|
res_info.setdefault("resource", {}).update({
|
|
"id": resource_id,
|
|
"user_id": samples[0].user_id,
|
|
"project_id": samples[0].project_id,
|
|
"metrics": rd.metrics,
|
|
})
|
|
|
|
for sample in samples:
|
|
res_info.setdefault("resource_extra", {}).update(
|
|
rd.sample_attributes(sample))
|
|
m = measures.setdefault(resource_id, {}).setdefault(
|
|
metric_name, [])
|
|
m.append({'timestamp': sample.timestamp,
|
|
'value': sample.volume})
|
|
unit = sample.unit
|
|
metric = sample.name
|
|
res_info['resource']['metrics'][metric]['unit'] = unit
|
|
|
|
stats['measures'] += len(measures[resource_id][metric_name])
|
|
res_info["resource"].update(res_info["resource_extra"])
|
|
if res_info:
|
|
gnocchi_data[resource_id] = res_info
|
|
|
|
try:
|
|
self.batch_measures(measures, gnocchi_data, stats)
|
|
except gnocchi_exc.ClientException as e:
|
|
LOG.error(six.text_type(e))
|
|
except Exception as e:
|
|
LOG.error(six.text_type(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("update", resource_type, resource,
|
|
self._update_resource, resource_extra)
|
|
except gnocchi_exc.ClientException as e:
|
|
LOG.error(six.text_type(e))
|
|
except Exception as e:
|
|
LOG.error(six.text_type(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'])
|
|
for rid in resource_ids]
|
|
|
|
def batch_measures(self, measures, resource_infos, stats):
|
|
# 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 in resources:
|
|
try:
|
|
self._if_not_cached("create", resource_type, resource,
|
|
self._create_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': six.text_type(e)})
|
|
# We cannot post measures for this resource
|
|
# and we can't patch it later
|
|
del measures[resource['id']]
|
|
del resource_infos[resource['id']]
|
|
|
|
# NOTE(sileht): we have created missing resources/metrics,
|
|
# now retry to post measures
|
|
self._gnocchi.metric.batch_resources_metrics_measures(
|
|
measures, create_metrics=True)
|
|
|
|
# FIXME(sileht): take care of measures removed in stats
|
|
LOG.debug("%(measures)d measures posted against %(metrics)d "
|
|
"metrics through %(resources)d resources", stats)
|
|
|
|
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, resource, resource_extra):
|
|
self._gnocchi.resource.update(resource_type,
|
|
resource["id"],
|
|
resource_extra)
|
|
LOG.debug('Resource %s updated', resource["id"])
|
|
|
|
def _if_not_cached(self, operation, resource_type, resource, method,
|
|
*args, **kwargs):
|
|
if self.cache:
|
|
cache_key = resource['id']
|
|
attribute_hash = self._check_resource_cache(cache_key, resource)
|
|
hit = False
|
|
if attribute_hash:
|
|
with self._gnocchi_resource_lock[cache_key]:
|
|
# 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.
|
|
attribute_hash = self._check_resource_cache(cache_key,
|
|
resource)
|
|
if attribute_hash:
|
|
method(resource_type, resource, *args, **kwargs)
|
|
self.cache.set(cache_key, attribute_hash)
|
|
else:
|
|
hit = True
|
|
LOG.debug('resource cache recheck hit for '
|
|
'%s %s', operation, cache_key)
|
|
self._gnocchi_resource_lock.pop(cache_key, None)
|
|
else:
|
|
hit = True
|
|
LOG.debug('Resource cache hit for %s %s', operation, cache_key)
|
|
if hit and operation == "create":
|
|
raise gnocchi_exc.ResourceAlreadyExists()
|
|
else:
|
|
method(resource_type, resource, *args, **kwargs)
|
|
|
|
def _check_resource_cache(self, key, resource_data):
|
|
cached_hash = self.cache.get(key)
|
|
attribute_hash = hash(frozenset(filter(lambda x: x[0] != "metrics",
|
|
resource_data.items())))
|
|
if not cached_hash or cached_hash != attribute_hash:
|
|
return attribute_hash
|
|
else:
|
|
return None
|
|
|
|
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)
|
|
|
|
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 _search_resource(self, resource_type, query):
|
|
try:
|
|
return self._gnocchi.resource.search(
|
|
resource_type, ujson.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_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))
|