diff --git a/ceilometer/api/app.py b/ceilometer/api/app.py
index 102bcc6fcc..a125bfb5f0 100644
--- a/ceilometer/api/app.py
+++ b/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.NotifierHook(),
+                 hooks.PipelineHook(),
                  hooks.TranslationHook()]
     if extra_hooks:
         app_hooks.extend(extra_hooks)
diff --git a/ceilometer/api/controllers/v2/meters.py b/ceilometer/api/controllers/v2/meters.py
index 73076a66e6..0865d555c7 100644
--- a/ceilometer/api/controllers/v2/meters.py
+++ b/ceilometer/api/controllers/v2/meters.py
@@ -21,10 +21,8 @@
 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
@@ -37,7 +35,6 @@ 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
@@ -294,20 +291,14 @@ class MeterController(rest.RestController):
                 for e in pecan.request.storage_conn.get_samples(f, limit=limit)
                 ]
 
-    @wsme_pecan.wsexpose([OldSample], str, body=[OldSample], status_code=201)
-    def post(self, direct='', samples=None):
+    @wsme_pecan.wsexpose([OldSample], body=[OldSample], status_code=201)
+    def post(self, samples):
         """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)
+        rbac.enforce('create_samples', pecan.request)
 
         now = timeutils.utcnow()
         auth_project = rbac.get_limited_to_project(pecan.request.headers)
@@ -317,6 +308,14 @@ 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)
@@ -353,22 +352,13 @@ 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
 
-            sample_dict = publisher_utils.meter_message_from_counter(
-                published_sample, cfg.CONF.publisher.telemetry_secret)
-            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, 'telemetry.api', published_samples)
+        with pecan.request.pipeline_manager.publisher(
+                context.get_admin_context()) as publisher:
+            publisher(published_samples)
 
         return samples
 
diff --git a/ceilometer/api/hooks.py b/ceilometer/api/hooks.py
index c392e36ba4..87570e17ab 100644
--- a/ceilometer/api/hooks.py
+++ b/ceilometer/api/hooks.py
@@ -21,7 +21,7 @@ from oslo_log import log
 from pecan import hooks
 
 from ceilometer.i18n import _LE
-from ceilometer import messaging
+from ceilometer import pipeline
 from ceilometer import storage
 
 LOG = log.getLogger(__name__)
@@ -67,20 +67,19 @@ class DBHook(hooks.PecanHook):
                               "retry later: %(err)s") % params)
 
 
-class NotifierHook(hooks.PecanHook):
-    """Create and attach a notifier to the request.
+class PipelineHook(hooks.PecanHook):
+    """Create and attach a pipeline to the request.
 
-    Usually, samples will be push to notification bus by notifier when they
-    are posted via /v2/meters/ API.
+    That allows new samples to be posted via the /v2/meters/ API.
     """
 
     def __init__(self):
-        transport = messaging.get_transport()
-        self.notifier = messaging.get_notifier(transport,
-                                               publisher_id="ceilometer.api")
+        # this is done here as the cfg options are not available
+        # when the file is imported.
+        self.pipeline_manager = pipeline.setup_pipeline()
 
     def before(self, state):
-        state.request.notifier = self.notifier
+        state.request.pipeline_manager = self.pipeline_manager
 
 
 class TranslationHook(hooks.PecanHook):
diff --git a/ceilometer/tests/api/test_app.py b/ceilometer/tests/api/test_app.py
index b11320417a..987df307e9 100644
--- a/ceilometer/tests/api/test_app.py
+++ b/ceilometer/tests/api/test_app.py
@@ -39,6 +39,7 @@ 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):
diff --git a/ceilometer/tests/api/v2/test_post_samples_scenarios.py b/ceilometer/tests/api/v2/test_post_samples_scenarios.py
index 473d5b4e64..8ae9fe6beb 100644
--- a/ceilometer/tests/api/v2/test_post_samples_scenarios.py
+++ b/ceilometer/tests/api/v2/test_post_samples_scenarios.py
@@ -22,6 +22,7 @@ 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
 
@@ -36,11 +37,30 @@ class TestPostSamples(v2.FunctionalTest,
     def setUp(self):
         self.published = []
         notifier = mock.Mock()
-        notifier.info.side_effect = self.fake_notifier_sample
-        self.useFixture(mockpatch.Patch('ceilometer.messaging.get_notifier',
+        notifier.sample.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',
diff --git a/ceilometer/tests/gabbi/gabbits/clean-samples.yaml b/ceilometer/tests/gabbi/gabbits/clean-samples.yaml
index 0d5927a894..e9547fc004 100644
--- a/ceilometer/tests/gabbi/gabbits/clean-samples.yaml
+++ b/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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       method: POST
       request_headers:
           content-type: application/json
diff --git a/ceilometer/tests/gabbi/gabbits/meters.yaml b/ceilometer/tests/gabbi/gabbits/meters.yaml
index d2c5a76de0..ca86471501 100644
--- a/ceilometer/tests/gabbi/gabbits/meters.yaml
+++ b/ceilometer/tests/gabbi/gabbits/meters.yaml
@@ -45,7 +45,7 @@ tests:
           - "[]"
 
     - name: meter bad accept
-      url: /v2/meters/noexist?direct=True
+      url: /v2/meters/noexist
       request_headers:
           accept: text/plain
       status: 406
@@ -56,7 +56,7 @@ tests:
       status: "404 || 405"
 
     - name: post meter no data
-      url: /v2/meters/apples?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       method: POST
       request_headers:
           content-type: application/json
@@ -167,7 +167,7 @@ tests:
 # Manipulate samples
 
     - name: post counter to meter
-      url: /v2/meters/apples?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       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?direct=True
+      url: /v2/meters/apples
       method: POST
       status: 201
       request_headers: