<commit_before>#
# 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__)


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]

    def match_type(self, meter_name):
        for t in self._event_type:
            if fnmatch.fnmatch(meter_name, t):
                return True

    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


class ProcessMeterNotifications(plugin_base.NotificationBase):

    event_types = []

    def __init__(self, manager):
        super(ProcessMeterNotifications, self).__init__(manager)
        self.definitions = self._load_definitions()

    def _load_definitions(self):
        plugin_manager = extension.ExtensionManager(
            namespace='ceilometer.event.trait_plugin')
        meters_cfg = declarative.load_definitions(
            cfg.CONF, {}, 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()

    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
        """<commit_msg>meter.notifications: stop using global conf<commit_after>#
# 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__)


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]

    def match_type(self, meter_name):
        for t in self._event_type:
            if fnmatch.fnmatch(meter_name, t):
                return True

    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


class ProcessMeterNotifications(plugin_base.NotificationBase):

    event_types = []

    def __init__(self, manager):
        super(ProcessMeterNotifications, self).__init__(manager)
        self.definitions = self._load_definitions()

    def _load_definitions(self):
        plugin_manager = extension.ExtensionManager(
            namespace='ceilometer.event.trait_plugin')
        meters_cfg = declarative.load_definitions(
            self.manager.conf, {},
            self.manager.conf.meter.meter_definitions_cfg_file,
            pkg_resources.resource_filename(__name__, "data/meters.yaml"))

        definitions = {}
        disable_non_metric_meters = (self.manager.conf.notification.
                                     disable_non_metric_meters)
        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 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()

    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
        """
{'fraction_non_alphanumeric': 0.08983917678230328, 'fraction_numerical': 0.0014788341857169265, 'mean_word_length': 2.035914702581369, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3497088', 'n_tokens_mistral': 4290, 'n_tokens_neox': 4100, 'n_words': 1189}