Notification Delivery Policy

This patch introduces the delivery retry policy into Zaqar.
It will work when the notification sent from Zaqar to the
subscriber failed.
User can define the retry policy in the options of subscription or
metadata of queue.

Change-Id: I1a74c2d5b69fb82826c303468099db34b3e41b5b
Implements: bp notification-delivery-policy

doc/source/user/index.rst

@@ -9,3 +9,4 @@ User Guide
    send_request_api
    authentication_tokens
    headers_queue_api_working
+   notification_delivery_policy

doc/source/user/notification_delivery_policy.rst

@@ -0,0 +1,68 @@
+..
+      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.
+
+======================================
+The Notification Delivery Policy Guide
+======================================
+
+Support notification delivery policy in webhook type. It will work when
+the notification is sent from Zaqar to the subscriber failed.
+This guide shows how to use this feature:
+
+Webhook
+-------
+
+.. note::
+
+   You should make sure that the message notification is enabled. By default,
+   the ``message_pipeline`` config option in [storage] section should be set
+   like: message_pipeline = zaqar.notification.notifier
+
+1. Create the queue with _retry_policy metadata like this:
+
+.. code:: json
+
+    {
+        '_retry_policy': {
+            'retries_with_no_delay': <Integer value, optional>,
+            'minimum_delay_retries': <Integer value, optional>,
+            'minimum_delay': <Interger value, optional>,
+            'maximum_delay': <Interger value, optional>,
+            'maximum_delay_retries': <Integer value, optional>,
+            'retry_backoff_function': <String value, optional>,
+            'ignore_subscription_override': <Bool value, optional>}
+    }
+
+-  'minimum_delay' and 'maximum_delay' mean delay time in seconds.
+-  'retry_backoff_function' mean name of retry backoff function.
+   There will be a enum in Zaqar that contain all valid values.
+   At first step, Zaqar only supports one function: 'linear'.
+-  'minimum_delay_retries' and 'maximum_delay_retries' mean the number of
+   retries with 'minimum_delay' or 'maximum_delay' delay time.
+
+If value of retry_policy is empty dict, that Zaqar will use default
+value to those keys:
+
+-  retries_with_no_delay=3
+-  minimum_delay_retries=3
+-  minimum_delay=5
+-  maximum_delay=30
+-  maximum_delay_retries=3
+-  retry_backoff_function=linear
+-  ignore_subscription_override=False
+
+2. Create a subscription with options like queue's metadata below. If user
+   don't set the options, Zaqar will use the retry policy in queue's metadata.
+   If user do it, Zaqar will use the retry policy in options by default, if
+   user still want to use retry policy in queue's metadata, then can set the
+   ignore_subscription_override = True.

releasenotes/notes/support-notification-delivery-policy-fbc94083b4e6b8d0.yaml

@@ -0,0 +1,6 @@
+---
+features:
+  - Support notificaiton delivery policy in webhook type. It will work when
+    the notification is sent from Zaqar to the subscriber failed.
+    User can define the retry policy in the options of subscription or
+    metadata of queue.

zaqar/common/consts.py

@@ -115,3 +115,19 @@ FLAVOR_OPS = (
     'flavor_update',
     'flavor_delete',
 )
+
+RETRY_OPS = (
+    RETRIES_WITH_NO_DELAY,
+    MINIMUM_DELAY_RETRIES,
+    MINIMUM_DELAY,
+    MAXIMUM_DELAY,
+    MAXIMUM_DELA_RETRIES,
+    LINEAR_INTERVAL,
+) = (
+    3,
+    3,
+    5,
+    30,
+    3,
+    5,
+)

zaqar/notification/notifier.py

@@ -45,6 +45,7 @@ class NotifierDriver(object):
         max_workers = kwargs.get('max_notifier_workers', 10)
         self.executor = futurist.ThreadPoolExecutor(max_workers=max_workers)
         self.require_confirmation = kwargs.get('require_confirmation', False)
+        self.queue_controller = kwargs.get('queue_controller')
 
     def post(self, queue_name, messages, client_uuid, project=None):
         """Send messages to the subscribers."""
@@ -52,6 +53,9 @@ class NotifierDriver(object):
             if not isinstance(self.subscription_controller,
                               pooling.SubscriptionController):
                 marker = None
+                queue_metadata = self.queue_controller.get(queue_name,
+                                                           project)
+                retry_policy = queue_metadata.get('_retry_policy', {})
                 while True:
                     subscribers = self.subscription_controller.list(
                         queue_name, project, marker=marker)
@@ -70,7 +74,8 @@ class NotifierDriver(object):
                             continue
                         for msg in messages:
                             msg['Message_Type'] = MessageType.Notification.name
-                        self._execute(s_type, sub, messages)
+                        self._execute(s_type, sub, messages,
+                                      retry_policy=retry_policy)
                     marker = next(subscribers)
                     if not marker:
                         break
@@ -137,7 +142,8 @@ class NotifierDriver(object):
 
         self._execute(s_type, subscription, [messages], conf)
 
-    def _execute(self, s_type, subscription, messages, conf=None):
+    def _execute(self, s_type, subscription, messages, conf=None,
+                 retry_policy=None):
         if self.subscription_controller:
             data_driver = self.subscription_controller.driver
             conf = data_driver.conf
@@ -147,4 +153,4 @@ class NotifierDriver(object):
                                    s_type,
                                    invoke_on_load=True)
         self.executor.submit(mgr.driver.execute, subscription, messages,
-                             conf=conf)
+                             conf=conf, queue_retry_policy=retry_policy)

zaqar/notification/tasks/webhook.py

@@ -13,15 +13,84 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+
 import json
 from oslo_log import log as logging
 import requests
 
+from zaqar.common import consts
+
 LOG = logging.getLogger(__name__)
 
 
+def _Linear_function(minimum_delay, maximum_delay, times):
+    return range(minimum_delay, maximum_delay, times)
+
+RETRY_BACKOFF_FUNCTION_MAP = {'linear': _Linear_function}
+
+
 class WebhookTask(object):
 
+    def _post_request_success(self, subscriber, data, headers):
+        try:
+            response = requests.post(subscriber, data=data, headers=headers)
+            if response and (response.status_code in range(200, 500)):
+                return True
+        except Exception as e:
+            LOG.exception('post request got exception in retry: %s.', str(e))
+        return False
+
+    def _retry_post(self, sub_retry_policy, queue_retry_policy, subscriber,
+                    data, headers):
+        retry_policy = None
+        if sub_retry_policy.get('ignore_subscription_override') or \
+           queue_retry_policy.get('ignore_subscription_override'):
+            retry_policy = queue_retry_policy or {}
+        else:
+            retry_policy = sub_retry_policy or queue_retry_policy or {}
+        # Immediate Retry Phase
+        for retry_with_no_delay in range(
+                0, retry_policy.get('retries_with_no_delay',
+                                    consts.RETRIES_WITH_NO_DELAY)):
+            LOG.debug('Retry with no delay, count: %s', retry_with_no_delay)
+            if self._post_request_success(subscriber, data, headers):
+                return
+        # Pre-Backoff Phase
+        for minimum_delay_retry in range(
+                0, retry_policy.get('minimum_delay_retries',
+                                    consts.MINIMUM_DELAY_RETRIES)):
+            LOG.debug('Retry with minimum delay, count: %s',
+                      minimum_delay_retry)
+            time.sleep(retry_policy.get('minimum_delay', consts.MINIMUM_DELAY))
+            if self._post_request_success(subscriber, data, headers):
+                return
+        # Backoff Phase: Linear retry
+        # TODO(wanghao): Now we only support the linear function, we should
+        # support more in Queens.
+        retry_function = retry_policy.get('retry_backoff_function', 'linear')
+        backoff_function = RETRY_BACKOFF_FUNCTION_MAP[retry_function]
+        for i in backoff_function(retry_policy.get('minimum_delay',
+                                                   consts.MINIMUM_DELAY),
+                                  retry_policy.get('maximum_delay',
+                                                   consts.MAXIMUM_DELAY),
+                                  consts.LINEAR_INTERVAL):
+            LOG.debug('Retry with retry_backoff_function, sleep: %s seconds',
+                      i)
+            time.sleep(i)
+            if self._post_request_success(subscriber, data, headers):
+                return
+        # Post-Backoff Phase
+        for maximum_delay_retries in range(
+                0, retry_policy.get('maximum_delay_retries',
+                                    consts.MAXIMUM_DELA_RETRIES)):
+            LOG.debug('Retry with maximum delay, count: %s',
+                      maximum_delay_retries)
+            time.sleep(retry_policy.get('maximum_delay', consts.MAXIMUM_DELAY))
+            if self._post_request_success(subscriber, data, headers):
+                return
+        LOG.debug('Send request retries are all failed.')
+
     def execute(self, subscription, messages, headers=None, **kwargs):
         if headers is None:
             headers = {'Content-Type': 'application/json'}
@@ -37,11 +106,23 @@ class WebhookTask(object):
                     data = data.replace('"$zaqar_message$"', json.dumps(msg))
                 else:
                     data = json.dumps(msg)
-                requests.post(subscription['subscriber'],
+                response = requests.post(subscription['subscriber'],
                                          data=data,
                                          headers=headers)
+                if response and (response.status_code not in range(200, 500)):
+                    LOG.info("Response is %s, begin to retry",
+                             response.status_code)
+                    self._retry_post(
+                        subscription['options'].get('_retry_policy', {}),
+                        kwargs.get('queue_retry_policy'),
+                        subscription['subscriber'],
+                        data, headers)
         except Exception as e:
             LOG.exception('webhook task got exception: %s.', str(e))
+            self._retry_post(subscription['options'].get('_retry_policy', {}),
+                             kwargs.get('queue_retry_policy'),
+                             subscription['subscriber'],
+                             data, headers)
 
     def register(self, subscriber, options, ttl, project_id, request_data):
         pass

zaqar/storage/pipeline.py

@@ -155,7 +155,9 @@ class DataDriver(base.DataDriverBase):
                   'max_notifier_workers':
                   self.conf.notification.max_notifier_workers,
                   'require_confirmation':
-                  self.conf.notification.require_confirmation}
+                  self.conf.notification.require_confirmation,
+                  'queue_controller':
+                  self._storage.queue_controller}
         stages.extend(_get_storage_pipeline('message', self.conf, **kwargs))
         stages.append(self._storage.message_controller)
         return common.Pipeline(stages)

zaqar/tests/unit/notification/test_notifier.py

@@ -70,9 +70,13 @@ class NotifierTest(testing.TestBase):
                          'options': {}}]
         ctlr = mock.MagicMock()
         ctlr.list = mock.Mock(return_value=iter([subscription, {}]))
-        driver = notifier.NotifierDriver(subscription_controller=ctlr)
+        queue_ctlr = mock.MagicMock()
+        queue_ctlr.get = mock.Mock(return_value={})
+        driver = notifier.NotifierDriver(subscription_controller=ctlr,
+                                         queue_controller=queue_ctlr)
         headers = {'Content-Type': 'application/json'}
         with mock.patch('requests.post') as mock_post:
+            mock_post.return_value = None
             driver.post('fake_queue', self.messages, self.client_id,
                         self.project)
             driver.executor.shutdown()
@@ -114,9 +118,13 @@ class NotifierTest(testing.TestBase):
                          'options': {'post_data': json.dumps(post_data)}}]
         ctlr = mock.MagicMock()
         ctlr.list = mock.Mock(return_value=iter([subscription, {}]))
-        driver = notifier.NotifierDriver(subscription_controller=ctlr)
+        queue_ctlr = mock.MagicMock()
+        queue_ctlr.get = mock.Mock(return_value={})
+        driver = notifier.NotifierDriver(subscription_controller=ctlr,
+                                         queue_controller=queue_ctlr)
         headers = {'Content-Type': 'application/json'}
         with mock.patch('requests.post') as mock_post:
+            mock_post.return_value = None
             driver.post('fake_queue', self.messages, self.client_id,
                         self.project)
             driver.executor.shutdown()
@@ -155,9 +163,13 @@ class NotifierTest(testing.TestBase):
                 return iter([subscription2, {}])
 
         ctlr.list = mock_list
-        driver = notifier.NotifierDriver(subscription_controller=ctlr)
+        queue_ctlr = mock.MagicMock()
+        queue_ctlr.get = mock.Mock(return_value={})
+        driver = notifier.NotifierDriver(subscription_controller=ctlr,
+                                         queue_controller=queue_ctlr)
         headers = {'Content-Type': 'application/json'}
         with mock.patch('requests.post') as mock_post:
+            mock_post.return_value = None
             driver.post('fake_queue', self.messages, self.client_id,
                         self.project)
             driver.executor.shutdown()
@@ -192,7 +204,10 @@ class NotifierTest(testing.TestBase):
                                      'from': 'zaqar@example.com'}}]
         ctlr = mock.MagicMock()
         ctlr.list = mock.Mock(return_value=iter([subscription, {}]))
-        driver = notifier.NotifierDriver(subscription_controller=ctlr)
+        queue_ctlr = mock.MagicMock()
+        queue_ctlr.get = mock.Mock(return_value={})
+        driver = notifier.NotifierDriver(subscription_controller=ctlr,
+                                         queue_controller=queue_ctlr)
         called = set()
         msg = ('Content-Type: text/plain; charset="us-ascii"\n'
                'MIME-Version: 1.0\nContent-Transfer-Encoding: 7bit\nto:'
@@ -242,7 +257,10 @@ class NotifierTest(testing.TestBase):
     def test_post_no_subscriber(self):
         ctlr = mock.MagicMock()
         ctlr.list = mock.Mock(return_value=iter([[], {}]))
-        driver = notifier.NotifierDriver(subscription_controller=ctlr)
+        queue_ctlr = mock.MagicMock()
+        queue_ctlr.get = mock.Mock(return_value={})
+        driver = notifier.NotifierDriver(subscription_controller=ctlr,
+                                         queue_controller=queue_ctlr)
         with mock.patch('requests.post') as mock_post:
             driver.post('fake_queue', self.messages, self.client_id,
                         self.project)
@@ -255,8 +273,12 @@ class NotifierTest(testing.TestBase):
                          'options': {}}]
         ctlr = mock.MagicMock()
         ctlr.list = mock.Mock(return_value=iter([subscription, {}]))
-        driver = notifier.NotifierDriver(subscription_controller=ctlr)
+        queue_ctlr = mock.MagicMock()
+        queue_ctlr.get = mock.Mock(return_value={})
+        driver = notifier.NotifierDriver(subscription_controller=ctlr,
+                                         queue_controller=queue_ctlr)
         with mock.patch('requests.post') as mock_post:
+            mock_post.return_value = None
             driver.post('fake_queue', self.messages, self.client_id,
                         self.project)
             driver.executor.shutdown()

zaqar/transport/validation.py

@@ -232,6 +232,39 @@ class Validator(object):
         path_list = self._decode_json_pointer(path)
         return op, path_list
 
+    def _validate_retry_policy(self, metadata):
+        retry_policy = metadata.get('_retry_policy') if metadata else None
+        if retry_policy and not isinstance(retry_policy, dict):
+            msg = _('retry_policy must be a dict.')
+            raise ValidationFailed(msg)
+
+        if retry_policy:
+            valid_keys = ['retries_with_no_delay', 'minimum_delay_retries',
+                          'minimum_delay', 'maximum_delay',
+                          'maximum_delay_retries', 'retry_backoff_function',
+                          'ignore_subscription_override']
+            for key in valid_keys:
+                retry_value = retry_policy.get(key)
+                if key == 'retry_backoff_function':
+                    if retry_value and not isinstance(retry_value, str):
+                        msg = _('retry_backoff_function must be a string.')
+                        raise ValidationFailed(msg)
+                    # TODO(wanghao): Now we only support linear function.
+                    # This will be removed after we support more functions.
+                    if retry_value and retry_value != 'linear':
+                        msg = _('retry_backoff_function only supports linear '
+                                'now.')
+                        raise ValidationFailed(msg)
+                elif key == 'ignore_subscription_override':
+                    if retry_value and not isinstance(retry_value, bool):
+                        msg = _('ignore_subscription_override must be a '
+                                'boolean.')
+                        raise ValidationFailed(msg)
+                else:
+                    if retry_value and not isinstance(retry_value, int):
+                        msg = _('Retry policy: %s must be a integer.') % key
+                        raise ValidationFailed(msg)
+
     def queue_patching(self, request, changes):
         washed_changes = []
         content_types = {
@@ -344,6 +377,8 @@ class Validator(object):
             raise ValidationFailed(msg, self._limits_conf.max_message_ttl,
                                    MIN_MESSAGE_TTL)
 
+        self._validate_retry_policy(queue_metadata)
+
     def queue_purging(self, document):
         """Restrictions the resource types to be purged for a queue.
 
@@ -559,6 +594,8 @@ class Validator(object):
             msg = _(u'Options must be a dict.')
             raise ValidationFailed(msg)
 
+        self._validate_retry_policy(options)
+
         ttl = subscription.get('ttl')
         if ttl:
             if not isinstance(ttl, int):