Create dynamic pollster feature

The dynamic pollster feature allows system administrators to
create/update pollsters on the fly (without changing code). The system
reads YAML configures that are found in ``pollsters_definitions_dirs``,
which has the default at ``/etc/ceilometer/pollsters.d``. Each YAML file
in the dynamic pollster feature can use the following attributes to
define a dynamic pollster:

* ``name`` -- mandatory field. It specifies the name/key of the dynamic
pollster. For instance, a pollster for magnum can use the name
``dynamic.magnum.cluster``;
* ``sample_type``: mandatory field; it defines the sample type. It must
be one of the values: ``gauge``, ``delta``, ``cumulative``;
* ``unit``: mandatory field; defines the unit of the metric that is
being collected. For magnum, for instance, one can use ``cluster`` as
the unit or some other meaningful String value;
* ``value_attribute``: mandatory attribute; defines the attribute in the
JSON response from the URL of the component being polled. In our magnum
example, we can use ``status`` as the value attribute;
* ``endpoint_type``: mandatory field; defines the endpoint type that is
used to discover the base URL of the component to be monitored; for
magnum, one can use ``container-infra``. Other values are accepted such
as ``volume`` for cinder endpoints, ``object-store`` for swift, and so
on;
* ``url_path``: mandatory attribute. It defines the path of the request
that we execute on the endpoint to gather data. For example, to gather
data from magnum, one can use ``v1/clusters/detail``;
* ``metadata_fields``: optional field. It is a list of all fields that
the response of the request executed with ``url_path`` that we want to
retrieve. As an example, for magnum, one can use the following values:
```
  metadata_fields:
    - "labels"
    - "updated_at"
    - "keypair"
    - "master_flavor_id"
    - "api_address"
    - "master_addresses"
    - "node_count"
    - "docker_volume_size"
    - "master_count"
    - "node_addresses"
    - "status_reason"
    - "coe_version"
    - "cluster_template_id"
    - "name"
    - "stack_id"
    - "created_at"
    - "discovery_url"
    - "container_version"
```
* ``skip_sample_values``: optional field. It defines the values that
might come in the ``value_attribute`` that we want to ignore. For
magnun, one could for instance, ignore some of the status it has for
clusters. Therefore, data is not gathered for clusters in the defined
status.
```
  skip_sample_values:
    - "CREATE_FAILED"
    - "DELETE_FAILED"
```
* ``value_mapping``: optional attribute. It defines a mapping for the
values that the dynamic pollster is handling. This is the actual value
that is sent to Gnocchi or other backends. If there is no mapping
specified, we will use the raw value that is obtained with the use of
``value_attribute``. An example for magnum, one can use:
```
  value_mapping:
    CREATE_IN_PROGRESS: "0"
    CREATE_FAILED: "1"
    CREATE_COMPLETE: "2"
    UPDATE_IN_PROGRESS: "3"
    UPDATE_FAILED: "4"
    UPDATE_COMPLETE: "5"
    DELETE_IN_PROGRESS: "6"
    DELETE_FAILED: "7"
    DELETE_COMPLETE: "8"
    RESUME_COMPLETE: "9"
    RESUME_FAILED: "10"
    RESTORE_COMPLETE: "11"
    ROLLBACK_IN_PROGRESS: "12"
    ROLLBACK_FAILED: "13"
    ROLLBACK_COMPLETE: "14"
    SNAPSHOT_COMPLETE: "15"
    CHECK_COMPLETE: "16"
    ADOPT_COMPLETE: "17"
```
* ``default_value_mapping``: optional parameter. The default value for
the value mapping in case the variable value receives data that is not
mapped to something in the ``value_mapping`` configuration. This
attribute is only used when ``value_mapping`` is defined. Moreover, it
has a default of ``-1``.

Change-Id: I5f0614518a9e304b86b74aa5bb0f9667d2a3a787
Signed-off-by: Rafael Weingärtner <rafael@apache.org>

2 files changed, 312 insertions, 6 deletions

diff --git a/ceilometer/polling/dynamic_pollster.py b/ceilometer/polling/dynamic_pollster.py
new file mode 100644
index 00000000..c2066ac3
--- /dev/null
+++ b/ceilometer/polling/dynamic_pollster.py
@@ -0,0 +1,231 @@
+#
+# 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.
+
+"""Dynamic pollster component
+    This component enables operators to create new pollsters on the fly
+    via configuration. The configuration files are read from
+    '/etc/ceilometer/pollsters.d/'. The pollster are defined in YAML files
+    similar to the idea used for handling notifications.
+"""
+
+from oslo_log import log
+from oslo_utils import timeutils
+from requests import RequestException
+
+from ceilometer import declarative
+from ceilometer.polling import plugin_base
+from ceilometer import sample
+
+
+import requests
+from six.moves.urllib import parse as url_parse
+
+LOG = log.getLogger(__name__)
+
+
+class DynamicPollster(plugin_base.PollsterBase):
+
+    OPTIONAL_POLLSTER_FIELDS = ['metadata_fields', 'skip_sample_values',
+                                'value_mapping', 'default_value',
+                                'metadata_mapping',
+                                'preserve_mapped_metadata']
+
+    REQUIRED_POLLSTER_FIELDS = ['name', 'sample_type', 'unit',
+                                'value_attribute', 'endpoint_type',
+                                'url_path']
+
+    ALL_POLLSTER_FIELDS = OPTIONAL_POLLSTER_FIELDS + REQUIRED_POLLSTER_FIELDS
+
+    name = ""
+
+    def __init__(self, pollster_definitions, conf=None):
+        super(DynamicPollster, self).__init__(conf)
+        LOG.debug("Dynamic pollster created with [%s]",
+                  pollster_definitions)
+
+        self.pollster_definitions = pollster_definitions
+        self.validate_pollster_definition()
+
+        if 'metadata_fields' in self.pollster_definitions:
+            LOG.debug("Metadata fields configured to [%s].",
+                      self.pollster_definitions['metadata_fields'])
+
+        self.name = self.pollster_definitions['name']
+        self.obj = self
+
+        if 'skip_sample_values' not in self.pollster_definitions:
+            self.pollster_definitions['skip_sample_values'] = []
+
+        if 'value_mapping' not in self.pollster_definitions:
+            self.pollster_definitions['value_mapping'] = {}
+
+        if 'default_value' not in self.pollster_definitions:
+            self.pollster_definitions['default_value'] = -1
+
+        if 'preserve_mapped_metadata' not in self.pollster_definitions:
+            self.pollster_definitions['preserve_mapped_metadata'] = True
+
+        if 'metadata_mapping' not in self.pollster_definitions:
+            self.pollster_definitions['metadata_mapping'] = {}
+
+    def validate_pollster_definition(self):
+        missing_required_fields = \
+            [field for field in self.REQUIRED_POLLSTER_FIELDS
+             if field not in self.pollster_definitions]
+
+        if missing_required_fields:
+            raise declarative.DynamicPollsterDefinitionException(
+                "Required fields %s not specified."
+                % missing_required_fields, self.pollster_definitions)
+
+        sample_type = self.pollster_definitions['sample_type']
+        if sample_type not in sample.TYPES:
+            raise declarative.DynamicPollsterDefinitionException(
+                "Invalid sample type [%s]. Valid ones are [%s]."
+                % (sample_type, sample.TYPES), self.pollster_definitions)
+
+        for definition_key in self.pollster_definitions:
+            if definition_key not in self.ALL_POLLSTER_FIELDS:
+                LOG.warning(
+                    "Field [%s] defined in [%s] is unknown "
+                    "and will be ignored. Valid fields are [%s].",
+                    definition_key, self.pollster_definitions,
+                    self.ALL_POLLSTER_FIELDS)
+
+    def get_samples(self, manager, cache, resources):
+        if not resources:
+            LOG.debug("No resources received for processing.")
+            yield None
+
+        for endpoint in resources:
+            LOG.debug("Executing get sample on URL [%s].", endpoint)
+
+            samples = list([])
+            try:
+                samples = self.execute_request_get_samples(
+                    keystone_client=manager._keystone, endpoint=endpoint)
+            except RequestException as e:
+                LOG.warning("Error [%s] while loading samples for [%s] "
+                            "for dynamic pollster [%s].",
+                            e, endpoint, self.name)
+
+            for pollster_sample in samples:
+                response_value_attribute_name = self.pollster_definitions[
+                    'value_attribute']
+                value = pollster_sample[response_value_attribute_name]
+
+                skip_sample_values = \
+                    self.pollster_definitions['skip_sample_values']
+                if skip_sample_values and value in skip_sample_values:
+                    LOG.debug("Skipping sample [%s] because value [%s] "
+                              "is configured to be skipped in skip list [%s].",
+                              pollster_sample, value, skip_sample_values)
+                    continue
+
+                value = self.execute_value_mapping(value)
+
+                user_id = None
+                if 'user_id' in pollster_sample:
+                    user_id = pollster_sample["user_id"]
+
+                project_id = None
+                if 'project_id' in pollster_sample:
+                    project_id = pollster_sample["project_id"]
+
+                metadata = []
+                if 'metadata_fields' in self.pollster_definitions:
+                    metadata = dict((k, pollster_sample.get(k))
+                                    for k in self.pollster_definitions[
+                                        'metadata_fields'])
+                self.generate_new_metadata_fields(metadata=metadata)
+                yield sample.Sample(
+                    timestamp=timeutils.isotime(),
+
+                    name=self.pollster_definitions['name'],
+                    type=self.pollster_definitions['sample_type'],
+                    unit=self.pollster_definitions['unit'],
+                    volume=value,
+
+                    user_id=user_id,
+                    project_id=project_id,
+                    resource_id=pollster_sample["id"],
+
+                    resource_metadata=metadata
+                )
+
+    def execute_value_mapping(self, value):
+        value_mapping = self.pollster_definitions['value_mapping']
+        if value_mapping:
+            if value in value_mapping:
+                old_value = value
+                value = value_mapping[value]
+                LOG.debug("Value mapped from [%s] to [%s]",
+                          old_value, value)
+            else:
+                default_value = \
+                    self.pollster_definitions['default_value']
+                LOG.warning(
+                    "Value [%s] was not found in value_mapping [%s]; "
+                    "therefore, we will use the default [%s].",
+                    value, value_mapping, default_value)
+                value = default_value
+        return value
+
+    def generate_new_metadata_fields(self, metadata=None):
+        metadata_mapping = self.pollster_definitions['metadata_mapping']
+        if not metadata_mapping or not metadata:
+            return
+
+        metadata_keys = list(metadata.keys())
+        for k in metadata_keys:
+            if k not in metadata_mapping:
+                continue
+
+            new_key = metadata_mapping[k]
+            metadata[new_key] = metadata[k]
+            LOG.debug("Generating new key [%s] with content [%s] of key [%s]",
+                      new_key, metadata[k], k)
+            if self.pollster_definitions['preserve_mapped_metadata']:
+                continue
+
+            k_value = metadata.pop(k)
+            LOG.debug("Removed key [%s] with value [%s] from "
+                      "metadata set that is sent to Gnocchi.", k, k_value)
+
+    @property
+    def default_discovery(self):
+        return 'endpoint:' + self.pollster_definitions['endpoint_type']
+
+    def execute_request_get_samples(self, keystone_client, endpoint):
+        url = url_parse.urljoin(
+            endpoint, self.pollster_definitions['url_path'])
+        resp = keystone_client.session.get(url, authenticated=True)
+        if resp.status_code != requests.codes.ok:
+            resp.raise_for_status()
+
+        response_json = resp.json()
+
+        entry_size = len(response_json)
+        LOG.debug("Entries [%s] in the JSON for request [%s] "
+                  "for dynamic pollster [%s].",
+                  response_json, url, self.name)
+
+        if entry_size > 0:
+            first_entry_name = None
+            try:
+                first_entry_name = next(iter(response_json))
+            except RuntimeError as e:
+                LOG.debug("Generator threw a StopIteration "
+                          "and we need to catch it [%s].",  e)
+            return response_json[first_entry_name]
+        return []
diff --git a/ceilometer/polling/manager.py b/ceilometer/polling/manager.py
index dfeb3ba2..240e8b6a 100644
--- a/ceilometer/polling/manager.py
+++ b/ceilometer/polling/manager.py
@@ -15,8 +15,10 @@
 # under the License.
 
 import collections
+import glob
 import itertools
 import logging
+import os
 import random
 import uuid
 
@@ -34,8 +36,10 @@ from stevedore import extension
 from tooz import coordination
 
 from ceilometer import agent
+from ceilometer import declarative
 from ceilometer import keystone_client
 from ceilometer import messaging
+from ceilometer.polling import dynamic_pollster
 from ceilometer.polling import plugin_base
 from ceilometer.publisher import utils as publisher_utils
 from ceilometer import utils
@@ -58,6 +62,10 @@ POLLING_OPTS = [
                default=50,
                help='Batch size of samples to send to notification agent, '
                     'Set to 0 to disable'),
+    cfg.MultiStrOpt('pollsters_definitions_dirs',
+                    default=["/etc/ceilometer/pollsters.d"],
+                    help="List of directories with YAML files used "
+                         "to created pollsters.")
 ]
 
 
@@ -93,7 +101,7 @@ class Resources(object):
                     not self.agent_manager.partition_coordinator or
                     self.agent_manager.hashrings[
                         static_resources_group].belongs_to_self(
-                            six.text_type(v))] + source_discovery
+                        six.text_type(v))] + source_discovery
 
         return source_discovery
 
@@ -245,8 +253,12 @@ class AgentManager(cotyledon.Service):
         extensions_fb = (self._extensions_from_builder('poll', namespace)
                          for namespace in namespaces)
 
+        # Create dynamic pollsters
+        extensions_dynamic_pollsters = self.create_dynamic_pollsters()
+
         self.extensions = list(itertools.chain(*list(extensions))) + list(
-            itertools.chain(*list(extensions_fb)))
+            itertools.chain(*list(extensions_fb))) + list(
+            extensions_dynamic_pollsters)
 
         if not self.extensions:
             LOG.warning('No valid pollsters can be loaded from %s '
@@ -280,6 +292,70 @@ class AgentManager(cotyledon.Service):
         self._keystone = None
         self._keystone_last_exception = None
 
+    def create_dynamic_pollsters(self):
+        """Creates dynamic pollsters
+
+        This method Creates dynamic pollsters based on configurations placed on
+        'pollsters_definitions_dirs'
+
+        :return: a list with the dynamic pollsters defined by the operator.
+        """
+
+        pollsters_definitions_dirs = self.conf.pollsters_definitions_dirs
+        if not pollsters_definitions_dirs:
+            LOG.info("Variable 'pollsters_definitions_dirs' not defined.")
+            return []
+
+        LOG.info("Looking for dynamic pollsters configurations at [%s].",
+                 pollsters_definitions_dirs)
+        pollsters_definitions_files = []
+        for directory in pollsters_definitions_dirs:
+            files = glob.glob(os.path.join(directory, "*.yaml"))
+            if not files:
+                LOG.info("No dynamic pollsters found in folder [%s].",
+                         directory)
+                continue
+            for filepath in sorted(files):
+                if filepath is not None:
+                    pollsters_definitions_files.append(filepath)
+
+        if not pollsters_definitions_files:
+            LOG.info("No dynamic pollsters file found in dirs [%s].",
+                     pollsters_definitions_dirs)
+            return []
+
+        pollsters_definitions = {}
+        for pollsters_definitions_file in pollsters_definitions_files:
+            pollsters_cfg = declarative.load_definitions(
+                self.conf, {}, pollsters_definitions_file)
+
+            LOG.info("File [%s] has [%s] dynamic pollster configurations.",
+                     pollsters_definitions_file, len(pollsters_cfg))
+
+            for pollster_cfg in pollsters_cfg:
+                pollster_name = pollster_cfg['name']
+                if pollster_name not in pollsters_definitions:
+                    LOG.info("Loading dynamic pollster [%s] from file [%s].",
+                             pollster_name, pollsters_definitions_file)
+                    try:
+                        dynamic_pollster_object = dynamic_pollster.\
+                            DynamicPollster(pollster_cfg, self.conf)
+                        pollsters_definitions[pollster_name] = \
+                            dynamic_pollster_object
+                    except Exception as e:
+                        LOG.error(
+                            "Error [%s] while loading dynamic pollster [%s].",
+                            e, pollster_name)
+
+                else:
+                    LOG.info(
+                        "Dynamic pollster [%s] is already defined."
+                        "Therefore, we are skipping it.", pollster_name)
+
+        LOG.debug("Total of dynamic pollsters [%s] loaded.",
+                  len(pollsters_definitions))
+        return pollsters_definitions.values()
+
     @staticmethod
     def _get_ext_mgr(namespace, *args, **kwargs):
         def _catch_extension_load_error(mgr, ep, exc):
@@ -371,7 +447,6 @@ class AgentManager(cotyledon.Service):
             futures.ThreadPoolExecutor(max_workers=len(data)))
 
         for interval, polling_task in data.items():
-
             @periodics.periodic(spacing=interval, run_immediately=True)
             def task(running_task):
                 self.interval_task(running_task)
@@ -461,9 +536,9 @@ class AgentManager(cotyledon.Service):
                         service_type = getattr(
                             self.conf.service_types,
                             discoverer.KEYSTONE_REQUIRED_FOR_SERVICE)
-                        if not keystone_client.get_service_catalog(
-                                self.keystone).get_endpoints(
-                                    service_type=service_type):
+                        if not keystone_client.\
+                                get_service_catalog(self.keystone).\
+                                get_endpoints(service_type=service_type):
                             LOG.warning(
                                 'Skipping %(name)s, %(service_type)s service '
                                 'is not registered in keystone',