Merge "Add support for posting samples to notification-agent via API"

ceilometer/api/app.py

@@ -62,7 +62,7 @@ def setup_app(pecan_config=None, extra_hooks=None):
     # FIXME: Replace DBHook with a hooks.TransactionHook
     app_hooks = [hooks.ConfigHook(),
                  hooks.DBHook(),
-                 hooks.PipelineHook(),
+                 hooks.NotifierHook(),
                  hooks.TranslationHook()]
     if extra_hooks:
         app_hooks.extend(extra_hooks)

ceilometer/api/controllers/v2/meters.py

@@ -21,8 +21,10 @@
 import base64
 import datetime
 
+from oslo_config import cfg
 from oslo_context import context
 from oslo_log import log
+from oslo_utils import strutils
 from oslo_utils import timeutils
 import pecan
 from pecan import rest
@@ -35,6 +37,7 @@ from ceilometer.api.controllers.v2 import base
 from ceilometer.api.controllers.v2 import utils as v2_utils
 from ceilometer.api import rbac
 from ceilometer.i18n import _
+from ceilometer.publisher import utils as publisher_utils
 from ceilometer import sample
 from ceilometer import storage
 from ceilometer import utils
@@ -291,15 +294,21 @@ class MeterController(rest.RestController):
                 for e in pecan.request.storage_conn.get_samples(f, limit=limit)
                 ]
 
-    @wsme_pecan.wsexpose([OldSample], body=[OldSample], status_code=201)
+    @wsme_pecan.wsexpose([OldSample], str, body=[OldSample], status_code=201)
-    def post(self, samples):
+    def post(self, direct='', samples=None):
         """Post a list of new Samples to Telemetry.
 
+        :param direct: a flag indicates whether the samples will be posted
+                       directly to storage or not.
         :param samples: a list of samples within the request body.
         """
-
         rbac.enforce('create_samples', pecan.request)
 
+        direct = strutils.bool_from_string(direct)
+        if not samples:
+            msg = _('Samples should be included in request body')
+            raise base.ClientSideError(msg)
+
         now = timeutils.utcnow()
         auth_project = rbac.get_limited_to_project(pecan.request.headers)
         def_source = pecan.request.cfg.sample_source
@@ -308,14 +317,6 @@ class MeterController(rest.RestController):
 
         published_samples = []
         for s in samples:
-            for p in pecan.request.pipeline_manager.pipelines:
-                if p.support_meter(s.counter_name):
-                    break
-            else:
-                message = _("The metric %s is not supported by metering "
-                            "pipeline configuration.") % s.counter_name
-                raise base.ClientSideError(message, status_code=409)
-
             if self.meter_name != s.counter_name:
                 raise wsme.exc.InvalidInput('counter_name', s.counter_name,
                                             'should be %s' % self.meter_name)
@@ -352,13 +353,22 @@ class MeterController(rest.RestController):
                 resource_metadata=utils.restore_nesting(s.resource_metadata,
                                                         separator='.'),
                 source=s.source)
-            published_samples.append(published_sample)
-
             s.message_id = published_sample.id
 
-        with pecan.request.pipeline_manager.publisher(
+            sample_dict = publisher_utils.meter_message_from_counter(
-                context.get_admin_context()) as publisher:
+                published_sample, cfg.CONF.publisher.telemetry_secret)
-            publisher(published_samples)
+            if direct:
+                ts = timeutils.parse_isotime(sample_dict['timestamp'])
+                sample_dict['timestamp'] = timeutils.normalize_time(ts)
+                pecan.request.storage_conn.record_metering_data(sample_dict)
+            else:
+                published_samples.append(sample_dict)
+        if not direct:
+            ctxt = context.RequestContext(user=def_user_id,
+                                          tenant=def_project_id,
+                                          is_admin=True)
+            notifier = pecan.request.notifier
+            notifier.info(ctxt.to_dict(), 'telemetry.api', published_samples)
 
         return samples
 

ceilometer/api/hooks.py

@@ -17,15 +17,19 @@ import threading
 
 from oslo_config import cfg
 from oslo_log import log
+import oslo_messaging
 
 from pecan import hooks
 
 from ceilometer.i18n import _LE
-from ceilometer import pipeline
+from ceilometer import messaging
 from ceilometer import storage
 
 LOG = log.getLogger(__name__)
 
+cfg.CONF.import_opt('telemetry_driver', 'ceilometer.publisher.messaging',
+                    group='publisher_notifier')
+
 
 class ConfigHook(hooks.PecanHook):
     """Attach the configuration object to the request.
@@ -67,19 +71,21 @@ class DBHook(hooks.PecanHook):
                               "retry later: %(err)s") % params)
 
 
-class PipelineHook(hooks.PecanHook):
+class NotifierHook(hooks.PecanHook):
-    """Create and attach a pipeline to the request.
+    """Create and attach a notifier to the request.
 
-    That allows new samples to be posted via the /v2/meters/ API.
+    Usually, samples will be push to notification bus by notifier when they
+    are posted via /v2/meters/ API.
     """
 
     def __init__(self):
-        # this is done here as the cfg options are not available
+        transport = messaging.get_transport()
-        # when the file is imported.
+        self.notifier = oslo_messaging.Notifier(
-        self.pipeline_manager = pipeline.setup_pipeline()
+            transport, driver=cfg.CONF.publisher_notifier.telemetry_driver,
+            publisher_id="ceilometer.api")
 
     def before(self, state):
-        state.request.pipeline_manager = self.pipeline_manager
+        state.request.notifier = self.notifier
 
 
 class TranslationHook(hooks.PecanHook):

ceilometer/tests/api/test_app.py

@@ -39,7 +39,6 @@ class TestApp(base.BaseTestCase):
 
     @mock.patch('ceilometer.storage.get_connection_from_config',
                 mock.MagicMock())
-    @mock.patch('ceilometer.api.hooks.PipelineHook', mock.MagicMock())
     @mock.patch('pecan.make_app')
     def test_pecan_debug(self, mocked):
         def _check_pecan_debug(g_debug, p_debug, expected, workers=1):

ceilometer/tests/api/test_hooks.py

@@ -0,0 +1,35 @@
+# Copyright 2015 Huawei Technologies Co., Ltd.
+# All Rights Reserved.
+#
+#    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 fixture as fixture_config
+import oslo_messaging
+
+from ceilometer.api import hooks
+from ceilometer.tests import base
+
+
+class TestTestNotifierHook(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestTestNotifierHook, self).setUp()
+        self.CONF = self.useFixture(fixture_config.Config()).conf
+
+    def test_init_notifier_with_drivers(self):
+        self.CONF.set_override('telemetry_driver', 'messagingv2',
+                               group='publisher_notifier')
+        hook = hooks.NotifierHook()
+        notifier = hook.notifier
+        self.assertIsInstance(notifier, oslo_messaging.Notifier)
+        self.assertEqual(['messagingv2'], notifier._driver_names)

ceilometer/tests/api/v2/test_post_samples_scenarios.py

@@ -22,7 +22,6 @@ import mock
 from oslo_utils import timeutils
 from oslotest import mockpatch
 
-from ceilometer import pipeline
 from ceilometer.tests.api import v2
 from ceilometer.tests import db as tests_db
 
@@ -37,30 +36,11 @@ class TestPostSamples(v2.FunctionalTest,
     def setUp(self):
         self.published = []
         notifier = mock.Mock()
-        notifier.sample.side_effect = self.fake_notifier_sample
+        notifier.info.side_effect = self.fake_notifier_sample
         self.useFixture(mockpatch.Patch('oslo_messaging.Notifier',
                                         return_value=notifier))
         super(TestPostSamples, self).setUp()
 
-    @mock.patch.object(pipeline.SampleSource, "support_meter")
-    def test_post_not_supported_sample(self, mocked):
-        mocked.return_value = False
-        s = [{'counter_name': 'apples',
-              'counter_type': 'gauge',
-              'counter_unit': 'instance',
-              'counter_volume': 1,
-              'resource_id': 'bd9431c1-8d69-4ad3-803a-8d4a6b89fd36',
-              'project_id': '35b17138-b364-4e6a-a131-8f3099c5be68',
-              'user_id': 'efd87807-12d2-4b38-9c70-5f5c2ac427ff',
-              'resource_metadata': {'name1': 'value1',
-                                    'name2': 'value2'}}]
-        resp = self.post_json('/meters/apples/', s, expect_errors=True)
-        self.assertEqual(409, resp.status_code)
-        expected_msg = ("The metric apples is not supported by metering "
-                        "pipeline configuration.")
-        self.assertEqual(expected_msg,
-                         resp.json['error_message']['faultstring'])
-
     def test_one(self):
         s1 = [{'counter_name': 'apples',
                'counter_type': 'gauge',

ceilometer/tests/gabbi/gabbits/clean-samples.yaml

@@ -8,7 +8,7 @@ tests:
 
     - name: post sample for meter
       desc: post a single sample
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -41,7 +41,7 @@ tests:
     - name: post a sample expect location
       desc: https://bugs.launchpad.net/ceilometer/+bug/1426426
       xfail: true
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json

ceilometer/tests/gabbi/gabbits/meters.yaml

@@ -45,7 +45,7 @@ tests:
           - "[]"
 
     - name: meter bad accept
-      url: /v2/meters/noexist
+      url: /v2/meters/noexist?direct=True
       request_headers:
           accept: text/plain
       status: 406
@@ -56,7 +56,7 @@ tests:
       status: "404 || 405"
 
     - name: post meter no data
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -66,7 +66,7 @@ tests:
     - name: post meter error is JSON
       desc: https://bugs.launchpad.net/ceilometer/+bug/1426483
       xfail: true
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -81,7 +81,7 @@ tests:
     - name: post meter bad content-type
       desc: https://bugs.launchpad.net/wsme/+bug/1419110
       xfail: true
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: text/plain
@@ -91,7 +91,7 @@ tests:
     - name: post bad samples to meter
       desc: https://bugs.launchpad.net/ceilometer/+bug/1428185
       xfail: true
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -105,7 +105,7 @@ tests:
 # POST variations on a malformed sample
 
     - name: post limited counter to meter
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -118,7 +118,7 @@ tests:
           - "Invalid input for field/attribute counter_name"
 
     - name: post mismatched counter name to meter
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -134,7 +134,7 @@ tests:
           - "should be apples"
 
     - name: post counter no resource to meter
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -149,7 +149,7 @@ tests:
           - "Mandatory field missing."
 
     - name: post counter bad type to meter
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -167,7 +167,7 @@ tests:
 # Manipulate samples
 
     - name: post counter to meter
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -207,7 +207,7 @@ tests:
           - unable to convert to int
 
     - name: post counter to meter different resource
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       status: 201
       request_headers:
@@ -230,7 +230,7 @@ tests:
     - name: post counter with bad timestamp
       desc: https://bugs.launchpad.net/wsme/+bug/1428624
       xfail: true
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -243,7 +243,7 @@ tests:
             timestamp: "2013-01-bad 23:23:20"
 
     - name: post counter with good timestamp
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       status: 201
       request_headers:
@@ -259,7 +259,7 @@ tests:
     - name: post counter with wrong metadata
       desc: https://bugs.launchpad.net/ceilometer/+bug/1428628
       xfail: true
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       request_headers:
           content-type: application/json
@@ -273,7 +273,7 @@ tests:
             resource_metadata: "a string"
 
     - name: post counter with empty metadata
-      url: /v2/meters/apples
+      url: /v2/meters/apples?direct=True
       method: POST
       status: 201
       request_headers: