Add Kwapi pollster (energy monitoring).

Given that my blueprint is not publicly available, I summarize it here.

Kwapi (kilowatt API) contains the following modules:
- Drivers: receive values from wattmeters and forward them on a bus (ZeroMQ).
Wattmeter drivers are specific to each wattmeters (Wattsup, OmegaWatt, etc).
- Plugins: listen the bus and process received data.
Currently, there is two plugins: the ceilometer plugin (REST API) and a visualization plugin (build graphs with RRDtool).

Kwapi is part of the XLcloud project (HPC cloud).
http://www.xlcloud.org

Repository:
https://github.com/stackforge/kwapi

Change-Id: Ieaaa1db9c8c569b6ee9f0815e03879f8b3f3e282

ceilometer/energy/kwapi.py

@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+#
+# Author: François Rossigneux <francois.rossigneux@inria.fr>
+#
+# 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 datetime
+
+from keystoneclient.v2_0 import client as ksclient
+import requests
+
+from ceilometer import counter
+from ceilometer.central import plugin
+from ceilometer.openstack.common import cfg
+
+
+class KwapiClient(object):
+    """Kwapi API client."""
+
+    def __init__(self, url, token=None):
+        """Initializes client."""
+        self.url = url
+        self.token = token
+
+    def iter_probes(self):
+        """Returns a list of dicts describing all probes."""
+        probes_url = self.url + '/probes/'
+        headers = {}
+        if self.token is not None:
+            headers = {'X-Auth-Token': self.token}
+        request = requests.get(probes_url, headers=headers)
+        message = request.json
+        probes = message['probes']
+        for key, value in probes.iteritems():
+            probe_dict = value
+            probe_dict['id'] = key
+            yield probe_dict
+
+
+class _Base(plugin.CentralPollster):
+    """Base class for the Kwapi pollster, derived from CentralPollster."""
+
+    @staticmethod
+    def get_kwapi_client():
+        """Returns a KwapiClient configured with the proper url and token."""
+        keystone = ksclient.Client(username=cfg.CONF.os_username,
+                                   password=cfg.CONF.os_password,
+                                   tenant_id=cfg.CONF.os_tenant_id,
+                                   tenant_name=cfg.CONF.os_tenant_name,
+                                   auth_url=cfg.CONF.os_auth_url)
+        endpoint = keystone.service_catalog.url_for(service_type='energy',
+                                                    endpoint_type='internalURL'
+                                                    )
+        return KwapiClient(endpoint, keystone.auth_token)
+
+    def iter_probes(self):
+        """Iterate over all probes."""
+        client = self.get_kwapi_client()
+        return client.iter_probes()
+
+
+class KwapiPollster(_Base):
+    """Kwapi pollster derived from the base class."""
+
+    def get_counters(self, manager, context):
+        """Returns all counters."""
+        for probe in self.iter_probes():
+            yield counter.Counter(
+                name='energy',
+                type=counter.TYPE_CUMULATIVE,
+                unit='kWh',
+                volume=probe['kwh'],
+                user_id=None,
+                project_id=None,
+                resource_id=probe['id'],
+                timestamp=datetime.datetime.fromtimestamp(
+                    probe['timestamp']).isoformat(),
+                resource_metadata={}
+            )
+            yield counter.Counter(
+                name='power',
+                type=counter.TYPE_GAUGE,
+                unit='W',
+                volume=probe['w'],
+                user_id=None,
+                project_id=None,
+                resource_id=probe['id'],
+                timestamp=datetime.datetime.fromtimestamp(
+                    probe['timestamp']).isoformat(),
+                resource_metadata={}
+            )

doc/source/measurements.rst

@@ -129,6 +129,16 @@ storage.objects.incoming.bytes  Delta                B  store ID  Number of inco
 storage.objects.outgoing.bytes  Delta                B  store ID  Number of outgoing bytes
 ==============================  ==========  ==========  ========  ==============================================
 
+Energy (Kwapi)
+======================
+
+==========================      ==========  ==========  ========  ==============================================
+Name                            Type        Volume      Resource  Note
+==========================      ==========  ==========  ========  ==============================================
+energy                          Cumulative         kWh  probe ID  Amount of energy
+power                           Gauge                W  probe ID  Power consumption
+==============================  ==========  ==========  ========  ==============================================
+
 Dynamically retrieving the Meters via ceilometer client
 =======================================================
     ceilometer meter-list -s openstack

setup.py

@@ -121,6 +121,7 @@ setuptools.setup(
     image = ceilometer.image.glance:ImagePollster
     image_size = ceilometer.image.glance:ImageSizePollster
     objectstore = ceilometer.objectstore.swift:SwiftPollster
+    kwapi = ceilometer.energy.kwapi:KwapiPollster
 
     [ceilometer.storage]
     log = ceilometer.storage.impl_log:LogStorage

tests/energy/test_kwapi.py

@@ -0,0 +1,81 @@
+# -*- coding: utf-8 -*-
+#
+# Author: François Rossigneux <francois.rossigneux@inria.fr>
+#
+# 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 datetime
+
+from ceilometer.tests import base
+from ceilometer.energy import kwapi
+from ceilometer.central import manager
+from ceilometer.openstack.common import context
+
+
+PROBE_DICT = {
+    "probes": {
+        "A": {
+            "timestamp": 1357730232.68754,
+            "w": 107.3,
+            "kwh": 0.001058255421506034
+        },
+        "B": {
+            "timestamp": 1357730232.048158,
+            "w": 15.0,
+            "kwh": 0.029019045026169896
+        },
+        "C": {
+            "timestamp": 1357730232.223375,
+            "w": 95.0,
+            "kwh": 0.17361822634312918
+        }
+    }
+}
+
+
+class TestKwapiPollster(base.TestCase):
+
+    @staticmethod
+    def fake_kwapi_iter_probes(foobar):
+        probes = PROBE_DICT['probes']
+        for key, value in probes.iteritems():
+            probe_dict = value
+            probe_dict['id'] = key
+            yield probe_dict
+
+    def setUp(self):
+        super(TestKwapiPollster, self).setUp()
+        self.context = context.get_admin_context()
+        self.manager = manager.AgentManager()
+        self.stubs.Set(kwapi._Base, 'iter_probes', self.fake_kwapi_iter_probes)
+
+    def test_kwapi_counter(self):
+        counters = list(kwapi.KwapiPollster().get_counters(self.manager,
+                                                           self.context))
+        self.assertEqual(len(counters), 6)
+        energy_counters = [counter for counter in counters
+                           if counter.name == "energy"]
+        power_counters = [counter for counter in counters
+                          if counter.name == "power"]
+        for probe in PROBE_DICT['probes'].values():
+            self.assert_(
+                any(map(lambda counter: counter.timestamp ==
+                    datetime.datetime.fromtimestamp(
+                        probe['timestamp']).isoformat(),
+                        counters)))
+            self.assert_(
+                any(map(lambda counter: counter.volume == probe['kwh'],
+                        energy_counters)))
+            self.assert_(
+                any(map(lambda counter: counter.volume == probe['w'],
+                        power_counters)))

tools/pip-requires

@@ -16,3 +16,4 @@ python-novaclient>=2.6.10
 python-keystoneclient>=0.2,<0.3
 python-swiftclient
 lxml
+requests<1.0