Add handler of sample creation notification

This change add a notification handler of sample.create event, which is
emitted when posting samples via /v2/meters API. The sample's info is
include in the payload of notification.

Partially implements: blueprint api-no-pipeline

Change-Id: I3941b851249c0f70f864bae32da154819b3e5dd7

ceilometer/sample.py

@@ -54,10 +54,12 @@ cfg.CONF.register_opts(OPTS)
 # Resource ID: the resource ID
 # Timestamp: when the sample has been read
 # Resource metadata: various metadata
+# id: an uuid of a sample, can be taken from API  when post sample via API
 class Sample(object):
 
     def __init__(self, name, type, unit, volume, user_id, project_id,
-                 resource_id, timestamp, resource_metadata, source=None):
+                 resource_id, timestamp, resource_metadata, source=None,
+                 id=None):
         self.name = name
         self.type = type
         self.unit = unit
@@ -68,7 +70,7 @@ class Sample(object):
         self.timestamp = timestamp
         self.resource_metadata = resource_metadata
         self.source = source or cfg.CONF.sample_source
-        self.id = str(uuid.uuid1())
+        self.id = id or str(uuid.uuid1())
 
     def as_dict(self):
         return copy.copy(self.__dict__)

ceilometer/telemetry/notifications.py

@@ -0,0 +1,64 @@
+#
+# 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 oslo_config import cfg
+import oslo_messaging
+
+from ceilometer.agent import plugin_base
+from ceilometer import sample
+
+OPTS = [
+    cfg.StrOpt('ceilometer_control_exchange',
+               default='ceilometer',
+               help="Exchange name for ceilometer notifications."),
+]
+
+
+cfg.CONF.register_opts(OPTS)
+
+
+class TelemetryBase(plugin_base.NotificationBase):
+    """Convert telemetry notification into Samples."""
+
+    @staticmethod
+    def get_targets(conf):
+        """Return a sequence of oslo_messaging.Target
+
+        Sequence defining the exchange and topics to be connected for this
+        plugin.
+        """
+        return [oslo_messaging.Target(
+                topic=topic, exchange=conf.ceilometer_control_exchange)
+                for topic in conf.notification_topics]
+
+
+class TelemetryApiPost(TelemetryBase):
+    """Handle sample from notification bus, which is posted via API."""
+
+    event_types = ['telemetry.api']
+
+    def process_notification(self, message):
+        samples = message['payload']
+        for sample_dict in samples:
+            yield sample.Sample(
+                name=sample_dict['counter_name'],
+                type=sample_dict['counter_type'],
+                unit=sample_dict['counter_unit'],
+                volume=sample_dict['counter_volume'],
+                user_id=sample_dict['user_id'],
+                project_id=sample_dict['project_id'],
+                resource_id=sample_dict['resource_id'],
+                timestamp=sample_dict['timestamp'],
+                resource_metadata=sample_dict['resource_metadata'],
+                source=sample_dict['source'],
+                id=sample_dict['message_id'])

ceilometer/tests/telemetry/test_notifications.py

@@ -0,0 +1,84 @@
+# 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 oslotest import base
+
+from ceilometer.telemetry import notifications
+
+NOTIFICATION = {
+    u'_context_domain': None,
+    u'_context_request_id': u'req-da91b4bf-d2b5-43ae-8b66-c7752e72726d',
+    'event_type': u'sample.create',
+    'timestamp': u'2015-06-1909: 19: 35.786893',
+    u'_context_auth_token': None,
+    u'_context_read_only': False,
+    'payload': [{
+                u'counter_name': u'instance100',
+                u'user_id': u'e1d870e51c7340cb9d555b15cbfcaec2',
+                u'resource_id': u'instance',
+                u'timestamp': u'2015-06-19T09: 19: 35.785330',
+                u'message_signature': u'fake_signature1',
+                u'resource_metadata': {u'foo': u'bar'},
+                u'source': u'30be1fc9a03c4e94ab05c403a8a377f2: openstack',
+                u'counter_unit': u'instance',
+                u'counter_volume': 1.0,
+                u'project_id': u'30be1fc9a03c4e94ab05c403a8a377f2',
+                u'message_id': u'4d865c6e-1664-11e5-9d41-0819a6cff905',
+                u'counter_type': u'gauge'
+                },
+                {
+                u'counter_name': u'instance100',
+                u'user_id': u'e1d870e51c7340cb9d555b15cbfcaec2',
+                u'resource_id': u'instance',
+                u'timestamp': u'2015-06-19T09: 19: 35.785330',
+                u'message_signature': u'fake_signature12',
+                u'resource_metadata': {u'foo': u'bar'},
+                u'source': u'30be1fc9a03c4e94ab05c403a8a377f2: openstack',
+                u'counter_unit': u'instance',
+                u'counter_volume': 1.0,
+                u'project_id': u'30be1fc9a03c4e94ab05c403a8a377f2',
+                u'message_id': u'4d866da8-1664-11e5-9d41-0819a6cff905',
+                u'counter_type': u'gauge'
+                }],
+    u'_context_resource_uuid': None,
+    u'_context_user_identity': u'fake_user_identity---',
+    u'_context_show_deleted': False,
+    u'_context_tenant': u'30be1fc9a03c4e94ab05c403a8a377f2',
+    'priority': 'info',
+    u'_context_is_admin': True,
+    u'_context_project_domain': None,
+    u'_context_user': u'e1d870e51c7340cb9d555b15cbfcaec2',
+    u'_context_user_domain': None,
+    'publisher_id': u'ceilometer.api',
+    'message_id': u'939823de-c242-45a2-a399-083f4d6a8c3e'
+}
+
+
+class TelemetryApiPostTestCase(base.BaseTestCase):
+
+    def test_process_notification(self):
+        sample_creation = notifications.TelemetryApiPost(None)
+        samples = list(sample_creation.process_notification(NOTIFICATION))
+        self.assertEqual(2, len(samples))
+        payload = NOTIFICATION["payload"]
+        for index, sample in enumerate(samples):
+            self.assertEqual(payload[index]["user_id"], sample.user_id)
+            self.assertEqual(payload[index]["counter_name"], sample.name)
+            self.assertEqual(payload[index]["resource_id"], sample.resource_id)
+            self.assertEqual(payload[index]["timestamp"], sample.timestamp)
+            self.assertEqual(payload[index]["resource_metadata"],
+                             sample.resource_metadata)
+            self.assertEqual(payload[index]["counter_volume"], sample.volume)
+            self.assertEqual(payload[index]["source"], sample.source)
+            self.assertEqual(payload[index]["counter_type"], sample.type)
+            self.assertEqual(payload[index]["message_id"], sample.id)
+            self.assertEqual(payload[index]["counter_unit"], sample.unit)

setup.cfg

@@ -94,6 +94,7 @@ ceilometer.notification =
     network.services.vpn.connections = ceilometer.network.notifications:IPSecSiteConnection
     objectstore.request = ceilometer.objectstore.notifications:SwiftWsgiMiddleware
     objectstore.request.meters = ceilometer.objectstore.notifications:SwiftWsgiMiddlewareMeters
+    _sample = ceilometer.telemetry.notifications:TelemetryApiPost
 
 ceilometer.discover =
     local_instances = ceilometer.compute.discovery:InstanceDiscovery