Allow using pymemcache's socket keepalive and retry mechanisms

Socket keepalive capabilities have been introduced [1][2][3] with pymemcache
3.5.0 [4][5]. These changes allow to pass keepalive configuration to the
pymemcache client.

Retry mechanisms have been implemented [6][7] with pymemcache 3.5.0 [5][8].
These changes allow to instantiate a ``RetryingClient`` by using
dogpile.cache.

[1] https://github.com/pinterest/pymemcache/commit/b289c87bb89b3ab477bd5d92c8951ab42c923923
[2] https://github.com/pinterest/pymemcache/commit/c782de1cac7cfaf4f6868d17682197022dad2d6b
[3] https://github.com/pinterest/pymemcache/commit/4d46f5ad8ddbd860e5219965df0714bdc15062f6
[4] https://github.com/pinterest/pymemcache/commit/07b5ecc21ce5d388d4312c943d79f813311e349f
[5] https://pypi.org/project/pymemcache/3.5.0/
[6] https://github.com/pinterest/pymemcache/commit/75fe5c81c35d2bcfc8e6a697aef948efbfebe8ba
[7] https://pymemcache.readthedocs.io/en/latest/getting_started.html#using-the-built-in-retrying-mechanism
[8] https://github.com/pinterest/pymemcache/commit/07b5ecc21ce5d388d4312c943d79f813311e349f

Closes: #205
Pull-request: https://github.com/sqlalchemy/dogpile.cache/pull/205
Pull-request-sha: 8a7ea5dfb0653b675f2ce9eb4873d8bf08841028

Change-Id: Ia2e76475943bc8ec86b5974217d1c92110be5b43

6 files changed, 161 insertions, 50 deletions

diff --git a/docs/build/conf.py b/docs/build/conf.py
index 61d3d5d..ce463ed 100644
--- a/docs/build/conf.py
+++ b/docs/build/conf.py
@@ -39,7 +39,7 @@ extensions = [
     "sphinx_paramlinks",
 ]
 
-changelog_sections = ["feature", "bug"]
+changelog_sections = ["feature", "usecase", "bug"]
 
 changelog_render_ticket = (
     "https://github.com/sqlalchemy/dogpile.cache/issues/%s"
diff --git a/docs/build/unreleased/205.rst b/docs/build/unreleased/205.rst
new file mode 100644
index 0000000..ebf2719
--- /dev/null
+++ b/docs/build/unreleased/205.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: usecase, memcached
+
+    Added support for pymemcache socket keepalive and retrying client.
+
+    .. seealso::
+
+        :paramref:`.PyMemcacheBackend.socket_keepalive`
+
+        :paramref:`.PyMemcacheBackend.enable_retry_client`
diff --git a/dogpile/cache/backends/memcached.py b/dogpile/cache/backends/memcached.py
index 9ae6741..66e9d58 100644
--- a/dogpile/cache/backends/memcached.py
+++ b/dogpile/cache/backends/memcached.py
@@ -12,11 +12,13 @@ import time
 import typing
 from typing import Any
 from typing import Mapping
+import warnings
 
 from ..api import CacheBackend
 from ..api import NO_VALUE
 from ... import util
 
+
 if typing.TYPE_CHECKING:
     import bmemcached
     import memcache
@@ -323,16 +325,6 @@ class BMemcachedBackend(GenericMemcachedBackend):
     SASL is a standard for adding authentication mechanisms
     to protocols in a way that is protocol independent.
 
-    SSL/TLS is a security layer on end-to-end communication.
-    It provides following benefits:
-
-    * Encryption: Data is encrypted on the wire between
-      Memcached client and server.
-    * Authentication: Optionally, both server and client
-      authenticate each other.
-    * Integrity: Data is not tampered or altered when
-      transmitted between client and server
-
     A typical configuration using username/password::
 
         from dogpile.cache import make_region
@@ -441,27 +433,17 @@ class PyMemcacheBackend(GenericMemcachedBackend):
       cache misses.
 
     dogpile.cache uses the ``HashClient`` from pymemcache in order to reduce
-    API differences when compared to other memcached client drivers. In short,
-    this allows the user to provide a single server or a list of memcached
+    API differences when compared to other memcached client drivers.
+    This allows the user to provide a single server or a list of memcached
     servers.
 
-    The ``serde`` param defaults to ``pymemcache.serde.pickle_serde`` as the
-    legacy ``serde`` would always convert the stored data to binary.
-
-    The ``default_noreply`` param defaults to False, otherwise the add command
-    would always return True causing the mutex not to work.
-
-    SSL/TLS is a security layer on end-to-end communication.
-    It provides following benefits:
+    Arguments which can be passed to the ``arguments``
+    dictionary include:
 
-    * Encryption: Data is encrypted on the wire between
-      Memcached client and server.
-    * Authentication: Optionally, both server and client
-      authenticate each other.
-    * Integrity: Data is not tampered or altered when
-      transmitted between client and server
+    :param tls_context: optional TLS context, will be used for
+     TLS connections.
 
-    A typical configuration using tls_context::
+     A typical configuration using tls_context::
 
         import ssl
         from dogpile.cache import make_region
@@ -477,19 +459,85 @@ class PyMemcacheBackend(GenericMemcachedBackend):
             }
         )
 
-    For advanced ways to configure TLS creating a more complex
-    tls_context visit https://docs.python.org/3/library/ssl.html
+     .. seealso::
 
-    Arguments which can be passed to the ``arguments``
-    dictionary include:
+        `<https://docs.python.org/3/library/ssl.html>`_ - additional TLS
+        documentation.
 
-    :param tls_context: optional TLS context, will be used for
-     TLS connections.
     :param serde: optional "serde". Defaults to
-     ``pymemcache.serde.pickle_serde``
-    :param default_noreply: Defaults to False
+     ``pymemcache.serde.pickle_serde``.
 
-    """
+    :param default_noreply:  defaults to False.  When set to True this flag
+     enables the pymemcache "noreply" feature.  See the pymemcache
+     documentation for further details.
+
+    :param socket_keepalive: optional socket keepalive, will be used for
+     TCP keepalive configuration.  Use of this parameter requires pymemcache
+     3.5.0 or greater.  This parameter
+     accepts a
+     `pymemcache.client.base.KeepAliveOpts
+     <https://pymemcache.readthedocs.io/en/latest/apidoc/pymemcache.client.base.html#pymemcache.client.base.KeepaliveOpts>`_
+     object.
+
+     A typical configuration using ``socket_keepalive``::
+
+        from pymemcache import KeepaliveOpts
+        from dogpile.cache import make_region
+
+        # Using the default keepalive configuration
+        socket_keepalive = KeepaliveOpts()
+
+        region = make_region().configure(
+            'dogpile.cache.pymemcache',
+            expiration_time = 3600,
+            arguments = {
+                'url':["127.0.0.1"],
+                'socket_keepalive': socket_keepalive
+            }
+        )
+
+     .. versionadded:: 1.1.4 - added support for ``socket_keepalive``.
+
+    :param enable_retry_client: optional flag to enable retry client
+     mechanisms to handle failure.  Defaults to False.  When set to ``True``,
+     the :paramref:`.PyMemcacheBackend.retry_attempts` parameter must also
+     be set, along with optional parameters
+     :paramref:`.PyMemcacheBackend.retry_delay`.
+     :paramref:`.PyMemcacheBackend.retry_for`,
+     :paramref:`.PyMemcacheBackend.do_not_retry_for`.
+
+     .. seealso::
+
+        `<https://pymemcache.readthedocs.io/en/latest/getting_started.html#using-the-built-in-retrying-mechanism>`_ -
+        in the pymemcache documentation
+
+     .. versionadded:: 1.1.4
+
+    :param retry_attempts: how many times to attempt an action before
+     failing. Must be 1 or above. Defaults to None.
+
+     .. versionadded:: 1.1.4
+
+    :param retry_delay: optional int|float, how many seconds to sleep between
+     each attempt. Defaults to None.
+
+     .. versionadded:: 1.1.4
+
+    :param retry_for: optional None|tuple|set|list, what exceptions to
+     allow retries for. Will allow retries for all exceptions if None.
+     Example: ``(MemcacheClientError, MemcacheUnexpectedCloseError)``
+     Accepts any class that is a subclass of Exception.  Defaults to None.
+
+     .. versionadded:: 1.1.4
+
+    :param do_not_retry_for: optional None|tuple|set|list, what
+     exceptions should be retried. Will not block retries for any Exception if
+     None. Example: ``(IOError, MemcacheIllegalInputError)``
+     Accepts any class that is a subclass of Exception. Defaults to None.
+
+     .. versionadded:: 1.1.4
+
+    """  # noqa E501
 
     def __init__(self, arguments):
         super().__init__(arguments)
@@ -497,15 +545,45 @@ class PyMemcacheBackend(GenericMemcachedBackend):
         self.serde = arguments.get("serde", pymemcache.serde.pickle_serde)
         self.default_noreply = arguments.get("default_noreply", False)
         self.tls_context = arguments.get("tls_context", None)
+        self.socket_keepalive = arguments.get("socket_keepalive", None)
+        self.enable_retry_client = arguments.get("enable_retry_client", False)
+        self.retry_attempts = arguments.get("retry_attempts", None)
+        self.retry_delay = arguments.get("retry_delay", None)
+        self.retry_for = arguments.get("retry_for", None)
+        self.do_not_retry_for = arguments.get("do_not_retry_for", None)
+        if (
+            self.retry_delay is not None
+            or self.retry_attempts is not None
+            or self.retry_for is not None
+            or self.do_not_retry_for is not None
+        ) and not self.enable_retry_client:
+            warnings.warn(
+                "enable_retry_client is not set; retry options "
+                "will be ignored"
+            )
 
     def _imports(self):
         global pymemcache
         import pymemcache
 
     def _create_client(self):
-        return pymemcache.client.hash.HashClient(
-            self.url,
-            serde=self.serde,
-            default_noreply=self.default_noreply,
-            tls_context=self.tls_context,
-        )
+        _kwargs = {
+            "serde": self.serde,
+            "default_noreply": self.default_noreply,
+            "tls_context": self.tls_context,
+        }
+        if self.socket_keepalive is not None:
+            _kwargs.update({"socket_keepalive": self.socket_keepalive})
+
+        client = pymemcache.client.hash.HashClient(self.url, **_kwargs)
+
+        if self.enable_retry_client:
+            return pymemcache.client.retrying.RetryingClient(
+                client,
+                attempts=self.retry_attempts,
+                retry_delay=self.retry_delay,
+                retry_for=self.retry_for,
+                do_not_retry_for=self.do_not_retry_for,
+            )
+
+        return client
diff --git a/dogpile/cache/backends/redis.py b/dogpile/cache/backends/redis.py
index a319d2d..5bff3ea 100644
--- a/dogpile/cache/backends/redis.py
+++ b/dogpile/cache/backends/redis.py
@@ -173,11 +173,7 @@ class RedisBackend(BytesBackend):
 
     def set_serialized(self, key, value):
         if self.redis_expiration_time:
-            self.writer_client.setex(
-                key,
-                self.redis_expiration_time,
-                value,
-            )
+            self.writer_client.setex(key, self.redis_expiration_time, value)
         else:
             self.writer_client.set(key, value)
 
@@ -283,7 +279,7 @@ class RedisSentinelBackend(RedisBackend):
                 "distributed_lock": True,
                 "thread_local_lock": False,
                 **arguments,
-            },
+            }
         )
 
     def _imports(self):
diff --git a/tests/cache/test_memcached_backend.py b/tests/cache/test_memcached_backend.py
index bac7445..407502a 100644
--- a/tests/cache/test_memcached_backend.py
+++ b/tests/cache/test_memcached_backend.py
@@ -2,11 +2,13 @@ import os
 import ssl
 from threading import Thread
 import time
+from unittest import mock
 from unittest import TestCase
 import weakref
 
 import pytest
 
+from dogpile.cache import make_region
 from dogpile.cache.backends.memcached import GenericMemcachedBackend
 from dogpile.cache.backends.memcached import MemcachedBackend
 from dogpile.cache.backends.memcached import PylibmcBackend
@@ -200,6 +202,17 @@ class PyMemcacheSerializerTest(
     backend = "dogpile.cache.pymemcache"
 
 
+class PyMemcacheRetryTest(_NonDistributedMemcachedTest):
+    backend = "dogpile.cache.pymemcache"
+    config_args = {
+        "arguments": {
+            "url": MEMCACHED_URL,
+            "enable_retry_client": True,
+            "retry_attempts": 3,
+        }
+    }
+
+
 class MemcachedTest(_NonDistributedMemcachedTest):
     backend = "dogpile.cache.memcached"
 
@@ -327,6 +340,20 @@ class MemcachedArgstest(TestCase):
         backend.set("foo", "bar")
         eq_(backend._clients.memcached.canary, [{"min_compress_len": 20}])
 
+    def test_pymemcache_enable_retry_client_not_set(self):
+        with mock.patch("warnings.warn") as warn_mock:
+            _ = make_region().configure(
+                "dogpile.cache.pymemcache",
+                arguments={"url": "foo", "retry_attempts": 2},
+            )
+            eq_(
+                warn_mock.mock_calls[0],
+                mock.call(
+                    "enable_retry_client is not set; retry options "
+                    "will be ignored"
+                ),
+            )
+
 
 class LocalThreadTest(TestCase):
     def setUp(self):
diff --git a/tox.ini b/tox.ini
index 2ccbca9..59715fb 100644
--- a/tox.ini
+++ b/tox.ini
@@ -38,7 +38,7 @@ deps=
     {memcached}: python-memcached
     {memcached}: python-binary-memcached>=0.29.0
     {memcached}: pifpaf>=2.5.0
-    {memcached}: pymemcache>=3.1.0
+    {memcached}: pymemcache>=3.5.0
     {redis}: redis
     {redis}: pifpaf
     {redis_sentinel}: redis