PhpRedisCluster patch with newrelic traces enabled. Based off redis 1.6 and the patch provided at https://www.drupal.org/project/redis/issues/2900947#comment-14779350

drupal-redis-1.6-redis-cluster-newrelic.patch

diff --git a/README.PhpRedisCluster.txt b/README.PhpRedisCluster.txt
new file mode 100644
index 0000000..0a335b9
--- /dev/null
+++ b/README.PhpRedisCluster.txt
@@ -0,0 +1,49 @@
+See README.md file.
+
+See README.PhpRedis.txt file because PhpRedisCluster requires the same PHP Redis extension.
+However the extension version should be >=3.0.0.
+Extension info can be found here: https://github.com/phpredis/phpredis
+
+See RedisCluster() class documentation:
+https://github.com/phpredis/phpredis/blob/develop/cluster.markdown#readme
+
+Example settings.php configuration for using the PhpRedisCluster:
+
+    ....
+    $settings['redis.connection']['interface']          = 'PhpRedisCluster';
+    $settings['redis.connection']['seeds']              = ['192.168.0.1:6379', '192.168.100.100:6379'];
+    $settings['redis.connection']['read_timeout']       = 1.5;
+    $settings['redis.connection']['timeout']            = 2;
+
+    // You can also use some additional parameters for PhpRedisCluster as:
+    // cluster_name - use if set in php.ini e.g.
+    //   $settings['redis.connection']['cluster_name'] = 'redis_cluster';
+    // persistent - persistent connections to each node e.g.
+    //   $settings['redis.connection']['persistent'] = FALSE;
+
+    // Set the Drupal's default cache backend.
+    $settings['cache']['default'] = 'cache.backend.redis';
+
+    // Always set the fast backend for bootstrap, discover and config, otherwise
+    // this gets lost when redis is enabled.
+    $settings['cache']['bins']['bootstrap'] = 'cache.backend.chainedfast';
+    $settings['cache']['bins']['discovery'] = 'cache.backend.chainedfast';
+    $settings['cache']['bins']['config'] = 'cache.backend.chainedfast';
+
+Also, in your project services.yml file you should change the service for
+"cache_tags.invalidator.checksum" to use Drupal\redis\Cache\PhpRedisClusterCacheTagsChecksum class.
+
+    .....
+    services:
+      # Cache tag checksum backend. Used by redis and most other cache backend
+      # to deal with cache tag invalidations.
+      cache_tags.invalidator.checksum:
+       class: Drupal\redis\Cache\PhpRedisClusterCacheTagsChecksum
+       arguments: ['@redis.factory']
+       tags:
+         - { name: cache_tags_invalidator }
+    .....
+
+You can copy/paste the example.services.yml in your settings folder, override the value for
+cache_tags.invalidator.checksum service and include the yml file in your settings.php as shown in the
+examples in README.md
diff --git a/README.PredisCluster.txt b/README.PredisCluster.txt
new file mode 100644
index 0000000..5765d02
--- /dev/null
+++ b/README.PredisCluster.txt
@@ -0,0 +1,32 @@
+See README.md file.
+
+Sample configuration.
+
+settings.php
+
+  $settings['redis.connection']['interface'] = 'PredisCluster';
+  $settings['redis.connection']['hosts'] = ['tcp://0.0.0.1:6379', 'tcp://0.0.0.2:6379', 'tcp://0.0.0.3:6379'];
+  $settings['cache']['bins']['bootstrap'] = 'cache.backend.chainedfast';
+  $settings['cache']['bins']['discovery'] = 'cache.backend.chainedfast';
+  $settings['cache']['bins']['config'] = 'cache.backend.chainedfast';
+  $settings['cache']['default'] = 'cache.backend.redis';
+  $settings['container_yamls'][] = 'redis.services.yml';
+
+redis.services.yml
+
+  services:
+    cache_tags.invalidator.checksum:
+    class: Drupal\redis\Cache\PredisClusterCacheTagsChecksum
+    arguments: ['@redis.factory']
+    tags:
+      - { name: cache_tags_invalidator }
+    lock:
+      class: Drupal\Core\Lock\LockBackendInterface
+      factory: ['@redis.lock.factory', get]
+    lock.persistent:
+      class: Drupal\Core\Lock\LockBackendInterface
+      factory: ['@redis.lock.factory', get]
+      arguments: [true]
+    flood:
+      class: Drupal\Core\Flood\FloodInterface
+      factory: ['@redis.flood.factory', get]
diff --git a/src/Cache/PhpRedisCluster.php b/src/Cache/PhpRedisCluster.php
new file mode 100644
index 0000000..ec5793d
--- /dev/null
+++ b/src/Cache/PhpRedisCluster.php
@@ -0,0 +1,290 @@
+<?php
+
+namespace Drupal\redis\Cache;
+
+use Drupal\Component\Serialization\SerializationInterface;
+use Drupal\Core\Cache\Cache;
+use Drupal\Core\Cache\CacheTagsChecksumInterface;
+use Drupal\Component\Assertion\Inspector;
+
+/**
+ * PhpRedisCluster cache backend.
+ */
+class PhpRedisCluster extends CacheBase {
+  
+  /**
+   * The client.
+   *
+   * @var \Redis
+   */
+  protected $client;
+  
+  /**
+   * The cache tags checksum provider.
+   *
+   * @var \Drupal\Core\Cache\CacheTagsChecksumInterface|\Drupal\Core\Cache\CacheTagsInvalidatorInterface
+   */
+  protected $checksumProvider;
+  
+  /**
+   * The last delete timestamp.
+   *
+   * @var float
+   */
+  protected $lastDeleteAll = NULL;
+  
+  /**
+   * The sample rate for New Relic metrics.
+   *
+   * @var float
+   */
+  private $sampleRate = 0.1;
+  
+  /**
+   * Reference for the redis hostname used in new relic traces.
+   *
+   * @var string
+   */
+  protected $redisHost;
+  
+  /**
+   * Creates a PhpRedisCluster cache backend.
+   *
+   * @param string $bin
+   *   The cache bin for which the object is created.
+   * @param \RedisCluster $client
+   *   The cluster client.
+   * @param \Drupal\Core\Cache\CacheTagsChecksumInterface $checksum_provider
+   *   The cluster checksum.
+   * @param \Drupal\Component\Serialization\SerializationInterface $serializer
+   *   The serialization class to use.
+   */
+  public function __construct($bin, \RedisCluster $client, CacheTagsChecksumInterface $checksum_provider, SerializationInterface $serializer) {
+    parent::__construct($bin, $serializer);
+    $this->client = $client;
+    $this->checksumProvider = $checksum_provider;
+    $this->sampleRate = getenv('NEW_RELIC_SAMPLE_RATE') ?: 0.1;
+    $this->redisHost = getenv('REDIS_HOST') ?: 'redis';
+  }
+  
+  /**
+   * Logs Redis event metrics to New Relic.
+   *
+   * @param array $before
+   *   The Redis info array before the operation.
+   * @param array $after
+   *   The Redis info array after the operation.
+   */
+  public function logEventMetrics($before, $after) {
+    if (!function_exists('newrelic_custom_metric')) {
+      // New Relic is not enabled.
+      return;
+    }
+    
+    if (!isset($before['event_wait']) || !isset($after['event_wait']) || !isset($before['event_no_wait']) || !isset($after['event_no_wait'])) {
+      return;
+    }
+    
+    $delta_wait = $after['event_wait'] - $before['event_wait'];
+    $delta_no_wait = $after['event_no_wait'] - $before['event_no_wait'];
+    $delta_wait_count = $after['event_wait_count'] - $before['event_wait_count'];
+    $delta_no_wait_count = $after['event_no_wait_count'] - $before['event_no_wait_count'];
+    
+    newrelic_custom_metric('Custom/Redis/Get/DeltaEventWait', $delta_wait);
+    newrelic_custom_metric('Custom/Redis/Get/DeltaEventNoWait', $delta_no_wait);
+    newrelic_custom_metric('Custom/Redis/Get/DeltaEventWaitCount', $delta_wait_count);
+    newrelic_custom_metric('Custom/Redis/Get/DeltaEventNoWaitCount', $delta_no_wait_count);
+    
+    $numerator = ($delta_wait + $delta_wait_count);
+    $denominator = ($delta_wait + $delta_no_wait + $delta_wait_count + $delta_no_wait_count);
+    $wait_ratio = ($denominator > 0) ? $numerator / $denominator : 0;
+    newrelic_custom_metric('Custom/Redis/Get/WaitRatio', $wait_ratio);
+  }
+  
+  /**
+   * {@inheritdoc}
+   */
+  public function getMultiple(&$cids, $allow_invalid = FALSE) {
+    $shouldLog = mt_rand() / mt_getrandmax() < $this->sampleRate;
+    
+    // Avoid an error when there are no cache ids.
+    if (empty($cids)) {
+      return [];
+    }
+    
+    $return = [];
+    $before = [];
+    $after = [];
+    
+    // Build the list of keys to fetch.
+    $keys = array_map([$this, 'getKey'], $cids);
+    
+    if ($shouldLog) {
+      try {
+        $before = $this->client->info('cluster');
+      }
+      catch (\RedisClusterException $e) {
+        if (\Drupal::hasService('logger.factory')) {
+          \Drupal::logger('redis')->critical('Redis cluster exception for key %key', [
+            '%key' => $key,
+          ]);
+        }
+      }
+    }
+    
+    // Optimize for the common case when only a single cache entry needs to
+    // be fetched, no pipeline is needed then.
+    if (count($keys) > 1) {
+      foreach ($keys as $key) {
+        if (function_exists('newrelic_record_datastore_segment')) {
+          $result[] = newrelic_record_datastore_segment(function () use ($key) {
+            return $this->client->hGetAll($key);
+          }, [
+            'product' => 'RedisCluster',
+            'collection' => $key,
+            'operation' => 'hGetAll',
+            'host' => $this->redisHost,
+          ]);
+        } else {
+          $result[] = $this->client->hGetAll($key);
+        }
+      }
+    }
+    else {
+      $key = reset($keys);
+      try {
+        if (function_exists('newrelic_record_datastore_segment')) {
+          $result = newrelic_record_datastore_segment(function () use ($key) {
+            return [$this->client->hGetAll($key)];
+          }, [
+            'product' => 'RedisCluster',
+            'collection' => $key,
+            'operation' => 'hGetAll',
+            'host' => $this->redisHost,
+          ]);
+        } else {
+          $result = [$this->client->hGetAll($key)];
+        }
+      }
+      catch (\RedisClusterException $e) {
+        if (\Drupal::hasService('logger.factory')) {
+          \Drupal::logger('redis')->critical('Redis cluster exception for key %key', [
+            '%key' => $key,
+          ]);
+        }
+      }
+    }
+    
+    // Loop over the cid values to ensure numeric indexes.
+    foreach (array_values($cids) as $index => $key) {
+      // Check if a valid result was returned from Redis.
+      if (isset($result[$index]) && is_array($result[$index])) {
+        // Check expiration and invalidation and convert into an object.
+        $item = $this->expandEntry($result[$index], $allow_invalid);
+        if ($item) {
+          $return[$item->cid] = $item;
+        }
+      }
+    }
+    
+    // Remove fetched cids from the list.
+    $cids = array_diff($cids, array_keys($return));
+    
+    if ($shouldLog) {
+      try {
+        $after = $this->client->info('cluster');
+      }
+      catch (\RedisClusterException $e) {
+        if (\Drupal::hasService('logger.factory')) {
+          \Drupal::logger('redis')->critical('Redis cluster exception for key %key', [
+            '%key' => $key,
+          ]);
+        }
+        throw $e;
+      }
+      $this->logEventMetrics($before, $after);
+    }
+    return $return;
+  }
+  
+  /**
+   * {@inheritdoc}
+   */
+  public function set($cid, $data, $expire = Cache::PERMANENT, array $tags = []) {
+    $shouldLog = mt_rand() / mt_getrandmax() < $this->sampleRate;
+    $ttl = $this->getExpiration($expire);
+    
+    $key = $this->getKey($cid);
+    
+    // If the item is already expired, delete it.
+    if ($ttl <= 0) {
+      $this->delete($key);
+    }
+    
+    if ($shouldLog) {
+      try {
+        $before = $this->client->info('cluster');
+      }
+      catch (\RedisClusterException $e) {
+        if (\Drupal::hasService('logger.factory')) {
+          \Drupal::logger('redis')->critical('Redis cluster exception for key %key', [
+            '%key' => $key,
+          ]);
+        }
+      }
+    }
+    
+    // Build the cache item and save it as a hash array.
+    $entry = $this->createEntryHash($cid, $data, $expire, $tags);
+    if (function_exists('newrelic_record_datastore_segment')) {
+      newrelic_record_datastore_segment(function () use ($key, $entry, $ttl) {
+        $this->client->hMset($key, $entry);
+        $this->client->expire($key, $ttl);
+      }, [
+        'product' => 'RedisCluster',
+        'collection' => $key,
+        'operation' => 'hMset',
+        'host' => $this->redisHost,
+      ]);
+    } else {
+      $this->client->hMset($key, $entry);
+      $this->client->expire($key, $ttl);
+    }
+    
+    if ($shouldLog) {
+      try {
+        $after = $this->client->info('cluster');
+        $this->logEventMetrics($before, $after);
+      }
+      catch (\RedisClusterException $e) {
+        if (\Drupal::hasService('logger.factory')) {
+          \Drupal::logger('redis')->critical('Redis cluster exception for key %key', [
+            '%key' => $key,
+          ]);
+        }
+      }
+    }
+  }
+  
+  /**
+   * {@inheritdoc}
+   */
+  public function doDeleteMultiple(array $cids) {
+    $keys = array_map([$this, 'getKey'], $cids);
+    $keys_size = count($keys);
+    $collection = ($keys_size == 1) ? $keys[0] : sprintf('Array(len:%d)', $keys_size);
+    if (function_exists('newrelic_record_datastore_segment')) {
+      newrelic_record_datastore_segment(function () use ($keys, $collection) {
+        $this->client->del($keys);
+      }, [
+        'product' => 'RedisCluster',
+        'collection' => $collection,
+        'operation' => 'del',
+        'host' => $this->redisHost,
+      ]);
+    } else {
+      $this->client->del($keys);
+    }
+  }
+  
+}
diff --git a/src/Cache/PhpRedisClusterCacheTagsChecksum.php b/src/Cache/PhpRedisClusterCacheTagsChecksum.php
new file mode 100644
index 0000000..89f1765
--- /dev/null
+++ b/src/Cache/PhpRedisClusterCacheTagsChecksum.php
@@ -0,0 +1,79 @@
+<?php
+
+namespace Drupal\redis\Cache;
+
+use Drupal\Core\Cache\CacheTagsChecksumInterface;
+use Drupal\Core\Cache\CacheTagsChecksumTrait;
+use Drupal\Core\Cache\CacheTagsInvalidatorInterface;
+use Drupal\redis\ClientFactory;
+use Drupal\redis\RedisPrefixTrait;
+
+/**
+ * Cache tags invalidations checksum implementation that uses redis.
+ */
+class PhpRedisClusterCacheTagsChecksum implements CacheTagsChecksumInterface, CacheTagsInvalidatorInterface {
+
+  use RedisPrefixTrait;
+  use CacheTagsChecksumTrait;
+
+  /**
+   * The client.
+   *
+   * @var \RedisCluster
+   */
+  protected $client;
+
+  /**
+   * Creates a PhpRedisCluster cache backend.
+   *
+   * @param \Drupal\redis\ClientFactory $factory
+   *   The ClientFactory object to initialize the client.
+   */
+  public function __construct(ClientFactory $factory) {
+    $this->client = $factory->getClient();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function doInvalidateTags(array $tags) {
+    foreach (array_map([$this, 'getTagKey'], $tags) as $key) {
+      $this->client->incr($key);
+    }
+  }
+
+  /**
+   * Return the key for the given cache tag.
+   *
+   * @param string $tag
+   *   The cache tag.
+   *
+   * @return string
+   *   The prefixed cache tag.
+   */
+  protected function getTagKey($tag) {
+    return $this->getPrefix() . ':cachetags:' . $tag;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function getTagInvalidationCounts(array $tags) {
+    $keys = array_map([$this, 'getTagKey'], $tags);
+    // The mget command returns the values as an array with numeric keys,
+    // combine it with the tags array to get the expected return value and run
+    // it through intval() to convert to integers and FALSE to 0.
+    $values = $this->client->mget($keys);
+    return $values ? array_map('intval', array_combine($tags, $values)) : [];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function getDatabaseConnection() {
+    // This is not injected to avoid a dependency on the database in the
+    // critical path. It is only needed during cache tag invalidations.
+    return \Drupal::database();
+  }
+
+}
diff --git a/src/Cache/PredisCluster.php b/src/Cache/PredisCluster.php
new file mode 100644
index 0000000..f67a608
--- /dev/null
+++ b/src/Cache/PredisCluster.php
@@ -0,0 +1,30 @@
+<?php
+
+namespace Drupal\redis\Cache;
+
+use Drupal\Component\Serialization\SerializationInterface;
+use Drupal\Core\Cache\CacheTagsChecksumInterface;
+use Predis\Client;
+
+/**
+ * Predis cache backend.
+ */
+class PredisCluster extends Predis {
+
+  /**
+   * PredisCluster constructor.
+   *
+   * @param string $bin
+   *   The bin.
+   * @param \Predis\Client $client
+   *   The client.
+   * @param \Drupal\Core\Cache\CacheTagsChecksumInterface $checksum_provider
+   *   The checksum provider.
+   * @param \Drupal\Component\Serialization\SerializationInterface $serializer
+   *   The serializer.
+   */
+  public function __construct($bin, Client $client, CacheTagsChecksumInterface $checksum_provider, SerializationInterface $serializer) {
+    parent::__construct($bin, $client, $checksum_provider, $serializer);
+  }
+
+}
\ No newline at end of file
diff --git a/src/Cache/PredisClusterCacheTagsChecksum.php b/src/Cache/PredisClusterCacheTagsChecksum.php
new file mode 100644
index 0000000..3c8aa60
--- /dev/null
+++ b/src/Cache/PredisClusterCacheTagsChecksum.php
@@ -0,0 +1,33 @@
+<?php
+
+namespace Drupal\redis\Cache;
+
+/**
+ * Cache tags invalidations checksum implementation that uses redis.
+ */
+class PredisClusterCacheTagsChecksum extends RedisCacheTagsChecksum {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function invalidateTags(array $tags) {
+    $keys_to_increment = [];
+    foreach ($tags as $tag) {
+      // Only invalidate tags once per request unless they are written again.
+      if (isset($this->invalidatedTags[$tag])) {
+        continue;
+      }
+      $this->invalidatedTags[$tag] = TRUE;
+      unset($this->tagCache[$tag]);
+      $keys_to_increment[] = $this->getTagKey($tag);
+    }
+    if ($keys_to_increment) {
+      $pipe = $this->client->pipeline();
+      foreach ($keys_to_increment as $key) {
+        $pipe->incr($key);
+      }
+      $pipe->execute();
+    }
+  }
+
+}
diff --git a/src/Client/PhpRedisCluster.php b/src/Client/PhpRedisCluster.php
new file mode 100644
index 0000000..cfcabd5
--- /dev/null
+++ b/src/Client/PhpRedisCluster.php
@@ -0,0 +1,134 @@
+<?php
+
+namespace Drupal\redis\Client;
+
+use Drupal\Core\Site\Settings;
+use Drupal\redis\ClientInterface;
+
+/**
+ * PhpRedis client specific implementation.
+ */
+class PhpRedisCluster implements ClientInterface {
+
+  const DEFAULT_READ_TIMEOUT = 1.5;
+  const DEFAULT_TIMEOUT = 2;
+
+  /**
+   * The settings.
+   *
+   * @var array
+   */
+  private $settings;
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getName() {
+    return 'PhpRedisCluster';
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getClient($host = NULL, $port = NULL, $base = NULL, $password = NULL, $replicationHosts = [], $persistent = FALSE) {
+    // Get the redis connection settings because we need some
+    // client specific one.
+    $this->initSettings();
+
+    $client = new \RedisCluster($this->getClusterName(), $this->getSeeds(), $this->getTimeout(), $this->getReadTimeout(), $this->getPersistent(),  $this->getPassword());
+
+    return $client;
+  }
+
+  /**
+   * Initialize the settings.
+   */
+  private function initSettings() {
+    $this->settings = Settings::get('redis.connection', []);
+  }
+
+  /**
+   * Get the cluster name if configured.
+   *
+   * @return string|null
+   *   Cluster name or NULL if not configured.
+   */
+  private function getClusterName() {
+    if (isset($this->settings['cluster_name'])) {
+      return $this->settings['cluster_name'];
+    }
+
+    return NULL;
+  }
+
+  /**
+   * Get the seeds for the cluster connection.
+   *
+   * @return array
+   *   An array of hosts.
+   */
+  private function getSeeds() {
+    if (isset($this->settings['seeds'])) {
+      return $this->settings['seeds'];
+    }
+
+    return [implode(':', [$this->settings['host'], $this->settings['port']])];
+  }
+
+  /**
+   * Get the configured timeout.
+   *
+   * @return float
+   *   Configured timeout or self::DEFAULT_TIMEOUT
+   */
+  private function getTimeout() {
+    if (isset($this->settings['timeout'])) {
+      return $this->settings['timeout'];
+    }
+
+    return self::DEFAULT_TIMEOUT;
+  }
+
+  /**
+   * Get the configured read timeout.
+   *
+   * @return float
+   *   Configured timeout or self::DEFAULT_READ_TIMEOUT
+   */
+  private function getReadTimeout() {
+    if (isset($this->settings['read_timeout'])) {
+      return $this->settings['read_timeout'];
+    }
+
+    return self::DEFAULT_READ_TIMEOUT;
+  }
+
+  /**
+   * Get the persistent flag for the RedisCluster option.
+   *
+   * @return bool
+   *   Return the persistent
+   */
+  private function getPersistent() {
+    if (isset($this->settings['persistent'])) {
+      return $this->settings['persistent'];
+    }
+
+    return FALSE;
+  }
+
+  /**
+   * Get the cluster password if configured.
+   *
+   * @return string|null
+   *   Cluster password or NULL if not configured.
+   */
+  private function getPassword() {
+    if (isset($this->settings['password'])) {
+      return $this->settings['password'];
+    }
+
+    return NULL;
+  }
+
+}
diff --git a/src/Client/PredisCluster.php b/src/Client/PredisCluster.php
new file mode 100644
index 0000000..2c4f053
--- /dev/null
+++ b/src/Client/PredisCluster.php
@@ -0,0 +1,35 @@
+<?php
+
+namespace Drupal\redis\Client;
+
+use Drupal\Core\Site\Settings;
+use Drupal\redis\ClientInterface;
+use Predis\Client;
+
+/**
+ * PredisCluster client specific implementation.
+ */
+class PredisCluster implements ClientInterface {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getClient($host = NULL, $port = NULL, $base = NULL, $password = NULL, $replicationHosts = [], $persistent = FALSE) {
+
+    $settings = Settings::get('redis.connection', []);
+    $parameters = $settings['hosts'];
+    $options = ['cluster' => 'redis'];
+
+    $client = new Client($parameters, $options);
+    return $client;
+
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getName() {
+    return 'PredisCluster';
+  }
+
+}
diff --git a/src/Controller/ReportController.php b/src/Controller/ReportController.php
index b021220..570f9db 100755
--- a/src/Controller/ReportController.php
+++ b/src/Controller/ReportController.php
@@ -2,12 +2,13 @@
 
 namespace Drupal\redis\Controller;
 
-use Predis\Client;
+use Drupal\Component\Utility\Unicode;
 use Drupal\Core\Controller\ControllerBase;
 use Drupal\Core\Datetime\DateFormatterInterface;
 use Drupal\Core\Url;
 use Drupal\redis\ClientFactory;
 use Drupal\redis\RedisPrefixTrait;
+use Predis\Client;
 use Predis\Collection\Iterator\Keyspace;
 use Symfony\Component\DependencyInjection\ContainerInterface;
 
@@ -23,7 +24,7 @@ class ReportController extends ControllerBase {
   /**
    * The redis client.
    *
-   * @var \Redis|\Predis\Client|false
+   * @var \Redis|\RedisCluster|Predis\Client|false
    */
   protected $redis;
 
@@ -88,7 +89,7 @@ class ReportController extends ControllerBase {
 
     $start = microtime(TRUE);
 
-    $info = $this->redis->info();
+    $info = $this->info();
 
     $prefix_length = strlen($this->getPrefix()) + 1;
 
@@ -168,20 +169,19 @@ class ReportController extends ControllerBase {
     }
 
     $end = microtime(TRUE);
-    $memory_config = $this->redis->config('get', 'maxmemory*');
 
-    if ($memory_config['maxmemory']) {
+    if ($info['maxmemory']) {
       $memory_value = $this->t('@used_memory / @max_memory (@used_percentage%), maxmemory policy: @policy', [
         '@used_memory' => $info['used_memory_human'] ?? $info['Memory']['used_memory_human'],
-        '@max_memory' => format_size($memory_config['maxmemory']),
-        '@used_percentage' => (int) ($info['used_memory'] ?? $info['Memory']['used_memory'] / $memory_config['maxmemory'] * 100),
-        '@policy' => $memory_config['maxmemory-policy'],
+        '@max_memory' => format_size($info['maxmemory']),
+        '@used_percentage' => (int) ($info['used_memory'] / $info['maxmemory'] * 100),
+        '@policy' => $info['maxmemory_policy'],
       ]);
     }
     else {
       $memory_value = $this->t('@used_memory / unlimited, maxmemory policy: @policy', [
-        '@used_memory' => $info['used_memory_human'] ?? $info['Memory']['used_memory_human'],
-        '@policy' => $memory_config['maxmemory-policy'],
+        '@used_memory' => $info['used_memory_human'],
+        '@policy' => $info['maxmemory_policy'],
       ]);
     }
 
@@ -192,15 +192,19 @@ class ReportController extends ControllerBase {
       ],
       'version' => [
         'title' => $this->t('Version'),
-        'value' => $info['redis_version'] ?? $info['Server']['redis_version'],
+        'value' => $info['redis_version'],
+      ],
+      'mode' => [
+        'title' => $this->t('Mode'),
+        'value' => Unicode::ucfirst($info['redis_mode']),
       ],
       'clients' => [
         'title' => $this->t('Connected clients'),
-        'value' => $info['connected_clients'] ?? $info['Clients']['connected_clients'],
+        'value' => $info['connected_clients'],
       ],
       'dbsize' => [
         'title' => $this->t('Keys'),
-        'value' => $this->redis->dbSize(),
+        'value' => $info['db_size'],
       ],
       'memory' => [
         'title' => $this->t('Memory'),
@@ -208,17 +212,17 @@ class ReportController extends ControllerBase {
       ],
       'uptime' => [
         'title' => $this->t('Uptime'),
-        'value' => $this->dateFormatter->formatInterval($info['uptime_in_seconds'] ?? $info['Server']['uptime_in_seconds']),
+        'value' => $this->dateFormatter->formatInterval($info['uptime_in_seconds']),
       ],
       'read_write' => [
         'title' => $this->t('Read/Write'),
         'value' => $this->t('@read read (@percent_read%), @write written (@percent_write%), @commands commands in @connections connections.', [
-          '@read' => format_size($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes']),
-          '@percent_read' => round(100 / (($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes']) + ($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes'])) * ($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes'])),
-          '@write' => format_size($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes']),
-          '@percent_write' => round(100 / (($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes']) + ($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes'])) * ($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes'])),
-          '@commands' => $info['total_commands_processed'] ?? $info['Stats']['total_commands_processed'],
-          '@connections' => $info['total_connections_received'] ?? $info['Stats']['total_connections_received'],
+          '@read' => format_size($info['total_net_output_bytes']),
+          '@percent_read' => round(100 / ($info['total_net_output_bytes'] + $info['total_net_input_bytes']) * ($info['total_net_output_bytes'])),
+          '@write' => format_size($info['total_net_input_bytes']),
+          '@percent_write' => round(100 / ($info['total_net_output_bytes'] + $info['total_net_input_bytes']) * ($info['total_net_input_bytes'])),
+          '@commands' => $info['total_commands_processed'],
+          '@connections' => $info['total_connections_received'],
         ]),
       ],
       'per_bin' => [
@@ -244,12 +248,17 @@ class ReportController extends ControllerBase {
       ],
       'time_spent' => [
         'title' => $this->t('Time spent'),
-        'value' => ['#markup' => $this->t('@count keys in @time seconds.', ['@count' => $i, '@time' => round(($end - $start), 4)])],
+        'value' => [
+          '#markup' => $this->t('@count keys in @time seconds.', [
+            '@count' => $i,
+            '@time' => round(($end - $start), 4),
+          ]),
+        ],
       ],
     ];
 
     // Warnings/hints.
-    if ($memory_config['maxmemory-policy'] == 'noeviction') {
+    if ($info['maxmemory_policy'] == 'noeviction') {
       $redis_url = Url::fromUri('https://redis.io/topics/lru-cache', [
         'fragment' => 'eviction-policies',
         'attributes' => [
@@ -297,9 +306,63 @@ class ReportController extends ControllerBase {
         yield from $keys;
       }
     }
+    elseif ($this->redis instanceof \RedisCluster) {
+      $master = current($this->redis->_masters());
+      while ($keys = $this->redis->scan($it, $master, $this->getPrefix() . '*', $count)) {
+        yield from $keys;
+      }
+    }
     elseif ($this->redis instanceof Client) {
       yield from new Keyspace($this->redis, $match, $count);
     }
   }
 
+  /**
+   * Wrapper to get various statistical information from Redis.
+   *
+   * @return array
+   *   Redis info.
+   */
+  protected function info() {
+    $normalized_info = [];
+    if ($this->redis instanceof \RedisCluster) {
+      $master = current($this->redis->_masters());
+      $info = $this->redis->info($master);
+    }
+    else {
+      $info = $this->redis->info();
+    }
+
+    $normalized_info['redis_version'] = $info['redis_version'] ?? $info['Server']['redis_version'];
+    $normalized_info['redis_mode'] = $info['redis_mode'] ?? $info['Server']['redis_mode'];
+    $normalized_info['connected_clients'] = $info['connected_clients'] ?? $info['Clients']['connected_clients'];
+    if ($this->redis instanceof \RedisCluster) {
+      $master = current($this->redis->_masters());
+      $normalized_info['db_size'] = $this->redis->dbSize($master);
+    }
+    else {
+      $normalized_info['db_size'] = $this->redis->dbSize();
+    }
+    $normalized_info['used_memory'] = $info['used_memory'] ?? $info['Memory']['used_memory'];
+    $normalized_info['used_memory_human'] = $info['used_memory_human'] ?? $info['Memory']['used_memory_human'];
+
+    if (empty($info['maxmemory_policy'])) {
+      $memory_config = $this->redis->config('get', 'maxmemory*');
+      $normalized_info['maxmemory_policy'] = $memory_config['maxmemory-policy'];
+      $normalized_info['maxmemory'] = $memory_config['maxmemory'];
+    }
+    else {
+      $normalized_info['maxmemory_policy'] = $info['maxmemory_policy'];
+      $normalized_info['maxmemory'] = $info['maxmemory'];
+    }
+
+    $normalized_info['uptime_in_seconds'] = $info['uptime_in_seconds'] ?? $info['Server']['uptime_in_seconds'];
+    $normalized_info['total_net_output_bytes'] = $info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes'];
+    $normalized_info['total_net_input_bytes'] = $info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes'];
+    $normalized_info['total_commands_processed'] = $info['total_commands_processed'] ?? $info['Stats']['total_commands_processed'];
+    $normalized_info['total_connections_received'] = $info['total_connections_received'] ?? $info['Stats']['total_connections_received'];
+
+    return $normalized_info;
+  }
+
 }
diff --git a/src/Controller/ReportController.php.orig b/src/Controller/ReportController.php.orig
new file mode 100755
index 0000000..b021220
--- /dev/null
+++ b/src/Controller/ReportController.php.orig
@@ -0,0 +1,305 @@
+<?php
+
+namespace Drupal\redis\Controller;
+
+use Predis\Client;
+use Drupal\Core\Controller\ControllerBase;
+use Drupal\Core\Datetime\DateFormatterInterface;
+use Drupal\Core\Url;
+use Drupal\redis\ClientFactory;
+use Drupal\redis\RedisPrefixTrait;
+use Predis\Collection\Iterator\Keyspace;
+use Symfony\Component\DependencyInjection\ContainerInterface;
+
+/**
+ * Redis Report page.
+ *
+ * Display status and statistics about the Redis connection.
+ */
+class ReportController extends ControllerBase {
+
+  use RedisPrefixTrait;
+
+  /**
+   * The redis client.
+   *
+   * @var \Redis|\Predis\Client|false
+   */
+  protected $redis;
+
+  /**
+   * The date formatter.
+   *
+   * @var \Drupal\Core\Datetime\DateFormatterInterface
+   */
+  protected $dateFormatter;
+
+  /**
+   * ReportController constructor.
+   *
+   * @param \Drupal\redis\ClientFactory $client_factory
+   *   The client factory.
+   * @param \Drupal\Core\Datetime\DateFormatterInterface $date_formatter
+   *   The date formatter.
+   */
+  public function __construct(ClientFactory $client_factory, DateFormatterInterface $date_formatter) {
+    if (ClientFactory::hasClient()) {
+      $this->redis = $client_factory->getClient();
+    }
+    else {
+      $this->redis = FALSE;
+    }
+
+    $this->dateFormatter = $date_formatter;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function create(ContainerInterface $container) {
+    return new static($container->get('redis.factory'), $container->get('date.formatter'));
+  }
+
+  /**
+   * Redis report overview.
+   */
+  public function overview() {
+
+    include_once DRUPAL_ROOT . '/core/includes/install.inc';
+
+    $build['report'] = [
+      '#type' => 'status_report',
+      '#requirements' => [],
+    ];
+
+    if ($this->redis === FALSE) {
+
+      $build['report']['#requirements'] = [
+        'client' => [
+          'title' => 'Redis',
+          'value' => t('Not connected.'),
+          'severity_status' => 'error',
+          'description' => t('No Redis client connected. Verify cache settings.'),
+        ],
+      ];
+
+      return $build;
+    }
+
+    $start = microtime(TRUE);
+
+    $info = $this->redis->info();
+
+    $prefix_length = strlen($this->getPrefix()) + 1;
+
+    $entries_per_bin = array_fill_keys(\Drupal::getContainer()->getParameter('cache_bins'), 0);
+
+    $required_cached_contexts = \Drupal::getContainer()->getParameter('renderer.config')['required_cache_contexts'];
+
+    $render_cache_totals = [];
+    $render_cache_contexts = [];
+    $cache_tags = [];
+    $i = 0;
+    $cache_tags_max = FALSE;
+    foreach ($this->scan($this->getPrefix() . '*') as $key) {
+      $i++;
+      $second_colon_pos = mb_strpos($key, ':', $prefix_length);
+      if ($second_colon_pos !== FALSE) {
+        $bin = mb_substr($key, $prefix_length, $second_colon_pos - $prefix_length);
+        if (isset($entries_per_bin[$bin])) {
+          $entries_per_bin[$bin]++;
+        }
+
+        if ($bin == 'render') {
+          $cache_key = mb_substr($key, $second_colon_pos + 1);
+
+          $first_context = mb_strpos($cache_key, '[');
+          if ($first_context) {
+            $cache_key_only = mb_substr($cache_key, 0, $first_context - 1);
+            if (!isset($render_cache_totals[$cache_key_only])) {
+              $render_cache_totals[$cache_key_only] = 1;
+            }
+            else {
+              $render_cache_totals[$cache_key_only]++;
+            }
+
+            if (preg_match_all('/\[([a-z0-9:_.]+)\]=([^:]*)/', $cache_key, $matches)) {
+              foreach ($matches[1] as $index => $context) {
+                $render_cache_contexts[$cache_key_only][$context][$matches[2][$index]] = $matches[2][$index];
+              }
+            }
+          }
+        }
+        elseif ($bin == 'cachetags') {
+          $cache_tag = mb_substr($key, $second_colon_pos + 1);
+          // @todo: Make the max configurable or allow ot override it through
+          // a query parameter.
+          if (count($cache_tags) < 50000) {
+            $cache_tags[$cache_tag] = $this->redis->get($key);
+          }
+          else {
+            $cache_tags_max = TRUE;
+          }
+        }
+      }
+
+      // Do not process more than 100k cache keys.
+      // @todo Adjust this after more testing or move to a separate page.
+    }
+
+    arsort($entries_per_bin);
+    arsort($render_cache_totals);
+    arsort($cache_tags);
+
+    $per_bin_string = '';
+    foreach ($entries_per_bin as $bin => $entries) {
+      $per_bin_string .= "$bin: $entries<br />";
+    }
+
+    $render_cache_string = '';
+    foreach (array_slice($render_cache_totals, 0, 50) as $cache_key => $total) {
+      $contexts = implode(', ', array_diff(array_keys($render_cache_contexts[$cache_key]), $required_cached_contexts));
+      $render_cache_string .= $contexts ? "$cache_key: $total ($contexts)<br />" : "$cache_key: $total<br />";
+    }
+
+    $cache_tags_string = '';
+    foreach (array_slice($cache_tags, 0, 50) as $cache_tag => $invalidations) {
+      $cache_tags_string .= "$cache_tag: $invalidations<br />";
+    }
+
+    $end = microtime(TRUE);
+    $memory_config = $this->redis->config('get', 'maxmemory*');
+
+    if ($memory_config['maxmemory']) {
+      $memory_value = $this->t('@used_memory / @max_memory (@used_percentage%), maxmemory policy: @policy', [
+        '@used_memory' => $info['used_memory_human'] ?? $info['Memory']['used_memory_human'],
+        '@max_memory' => format_size($memory_config['maxmemory']),
+        '@used_percentage' => (int) ($info['used_memory'] ?? $info['Memory']['used_memory'] / $memory_config['maxmemory'] * 100),
+        '@policy' => $memory_config['maxmemory-policy'],
+      ]);
+    }
+    else {
+      $memory_value = $this->t('@used_memory / unlimited, maxmemory policy: @policy', [
+        '@used_memory' => $info['used_memory_human'] ?? $info['Memory']['used_memory_human'],
+        '@policy' => $memory_config['maxmemory-policy'],
+      ]);
+    }
+
+    $requirements = [
+      'client' => [
+        'title' => $this->t('Client'),
+        'value' => t("Connected, using the <em>@name</em> client.", ['@name' => ClientFactory::getClientName()]),
+      ],
+      'version' => [
+        'title' => $this->t('Version'),
+        'value' => $info['redis_version'] ?? $info['Server']['redis_version'],
+      ],
+      'clients' => [
+        'title' => $this->t('Connected clients'),
+        'value' => $info['connected_clients'] ?? $info['Clients']['connected_clients'],
+      ],
+      'dbsize' => [
+        'title' => $this->t('Keys'),
+        'value' => $this->redis->dbSize(),
+      ],
+      'memory' => [
+        'title' => $this->t('Memory'),
+        'value' => $memory_value,
+      ],
+      'uptime' => [
+        'title' => $this->t('Uptime'),
+        'value' => $this->dateFormatter->formatInterval($info['uptime_in_seconds'] ?? $info['Server']['uptime_in_seconds']),
+      ],
+      'read_write' => [
+        'title' => $this->t('Read/Write'),
+        'value' => $this->t('@read read (@percent_read%), @write written (@percent_write%), @commands commands in @connections connections.', [
+          '@read' => format_size($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes']),
+          '@percent_read' => round(100 / (($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes']) + ($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes'])) * ($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes'])),
+          '@write' => format_size($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes']),
+          '@percent_write' => round(100 / (($info['total_net_output_bytes'] ?? $info['Stats']['total_net_output_bytes']) + ($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes'])) * ($info['total_net_input_bytes'] ?? $info['Stats']['total_net_input_bytes'])),
+          '@commands' => $info['total_commands_processed'] ?? $info['Stats']['total_commands_processed'],
+          '@connections' => $info['total_connections_received'] ?? $info['Stats']['total_connections_received'],
+        ]),
+      ],
+      'per_bin' => [
+        'title' => $this->t('Keys per cache bin'),
+        'value' => ['#markup' => $per_bin_string],
+      ],
+      'render_cache' => [
+        'title' => $this->t('Render cache entries with most variations'),
+        'value' => ['#markup' => $render_cache_string],
+      ],
+      'cache_tags' => [
+        'title' => $this->t('Most invalidated cache tags'),
+        'value' => ['#markup' => $cache_tags_string],
+      ],
+      'cache_tag_totals' => [
+        'title' => $this->t('Total cache tag invalidations'),
+        'value' => [
+          '#markup' => $this->t('@count tags with @invalidations invalidations.', [
+            '@count' => count($cache_tags),
+            '@invalidations' => array_sum($cache_tags),
+          ]),
+        ],
+      ],
+      'time_spent' => [
+        'title' => $this->t('Time spent'),
+        'value' => ['#markup' => $this->t('@count keys in @time seconds.', ['@count' => $i, '@time' => round(($end - $start), 4)])],
+      ],
+    ];
+
+    // Warnings/hints.
+    if ($memory_config['maxmemory-policy'] == 'noeviction') {
+      $redis_url = Url::fromUri('https://redis.io/topics/lru-cache', [
+        'fragment' => 'eviction-policies',
+        'attributes' => [
+          'target' => '_blank',
+        ],
+      ]);
+      $requirements['memory']['severity_status'] = 'warning';
+      $requirements['memory']['description'] = $this->t('It is recommended to configure the maxmemory policy to e.g. volatile-lru, see <a href=":documentation_url">Redis documentation</a>.', [
+        ':documentation_url' => $redis_url->toString(),
+      ]);
+    }
+    if (count($cache_tags) == 0) {
+      $requirements['cache_tag_totals']['severity_status'] = 'warning';
+      $requirements['cache_tag_totals']['description'] = $this->t('No cache tags found, make sure that the redis cache tag checksum service is used. See example.services.yml on root of this module.');
+      unset($requirements['cache_tags']);
+    }
+
+    if ($cache_tags_max) {
+      $requirements['max_cache_tags'] = [
+        'severity_status' => 'warning',
+        'title' => $this->t('Cache tags limit reached'),
+        'value' => ['#markup' => $this->t('Cache tag count incomplete, only counted @count cache tags.', ['@count' => count($cache_tags)])],
+      ];
+    }
+
+    $build['report']['#requirements'] = $requirements;
+
+    return $build;
+  }
+
+  /**
+   * Wrapper to SCAN through matching redis keys.
+   *
+   * @param string $match
+   *   The MATCH pattern.
+   * @param int $count
+   *   Count of keys per iteration (only a suggestion to Redis).
+   *
+   * @return \Generator
+   */
+  protected function scan($match, $count = 10000) {
+    $it = NULL;
+    if ($this->redis instanceof \Redis) {
+      while ($keys = $this->redis->scan($it, $this->getPrefix() . '*', $count)) {
+        yield from $keys;
+      }
+    }
+    elseif ($this->redis instanceof Client) {
+      yield from new Keyspace($this->redis, $match, $count);
+    }
+  }
+
+}
diff --git a/src/Flood/PhpRedisCluster.php b/src/Flood/PhpRedisCluster.php
new file mode 100644
index 0000000..2552924
--- /dev/null
+++ b/src/Flood/PhpRedisCluster.php
@@ -0,0 +1,13 @@
+<?php
+
+namespace Drupal\redis\Flood;
+
+use Drupal\Core\Flood\FloodInterface;
+
+/**
+ * Defines the database flood backend. This is the default Drupal backend.
+ */
+class PhpRedisCluster extends PhpRedis implements FloodInterface {
+  // Just fall back to PhpRedis implementation for now because at the moment
+  // it is 100% overlapping at the moment.
+}
diff --git a/src/Flood/PredisCluster.php b/src/Flood/PredisCluster.php
new file mode 100644
index 0000000..dae00cb
--- /dev/null
+++ b/src/Flood/PredisCluster.php
@@ -0,0 +1,11 @@
+<?php
+
+namespace Drupal\redis\Flood;
+
+/**
+ * Defines the database flood backend. This is the default Drupal backend.
+ */
+class PredisCluster extends Predis {
+  // Just fall back to PhpRedis implementation for now because at the moment
+  // it is 100% overlapping at the moment.
+}
diff --git a/src/Lock/PhpRedisCluster.php b/src/Lock/PhpRedisCluster.php
new file mode 100644
index 0000000..b1adc9f
--- /dev/null
+++ b/src/Lock/PhpRedisCluster.php
@@ -0,0 +1,145 @@
+<?php
+
+namespace Drupal\redis\Lock;
+
+use Drupal\Core\Lock\LockBackendAbstract;
+use Drupal\redis\ClientFactory;
+use Drupal\redis\RedisPrefixTrait;
+
+/**
+ * RedisCluster lock backend implementation.
+ */
+class PhpRedisCluster extends LockBackendAbstract {
+
+  use RedisPrefixTrait;
+
+  /**
+   * The client.
+   *
+   * @var \RedisCluster
+   */
+  protected $client;
+
+  /**
+   * Creates a PhpRedisCluster cache backend.
+   *
+   * @param \Drupal\redis\ClientFactory $factory
+   *   The ClientFactory object to initialize the client.
+   */
+  public function __construct(ClientFactory $factory) {
+    $this->client = $factory->getClient();
+    // __destruct() is causing problems with garbage collections, register a
+    // shutdown function instead.
+    drupal_register_shutdown_function([$this, 'releaseAll']);
+  }
+
+  /**
+   * Generate a redis key name for the current lock name.
+   *
+   * @param string $name
+   *   Lock name.
+   *
+   * @return string
+   *   The redis key for the given lock.
+   */
+  protected function getKey($name) {
+    return $this->getPrefix() . ':lock:' . $name;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function acquire($name, $timeout = 30.0) {
+    $key = $this->getKey($name);
+    $id  = $this->getLockId();
+
+    // Insure that the timeout is at least 1 ms.
+    $timeout = max($timeout, 0.001);
+
+    // If we already have the lock, check for its owner and attempt a new EXPIRE
+    // command on it.
+    if (isset($this->locks[$name])) {
+
+      // Create a new transaction, for atomicity.
+      $this->client->watch($key);
+
+      // Global tells us we are the owner, but in real life it could have
+      // expired and another process could have taken it, check that.
+      if ($this->client->get($key) != $id) {
+        // Explicit UNWATCH we are not going to run the MULTI/EXEC block.
+        $this->client->unwatch();
+        unset($this->locks[$name]);
+        return FALSE;
+      }
+
+      $result = $this->client->psetex($key, (int) ($timeout * 1000), $id);
+
+      // If the set failed, someone else wrote the key, we failed to acquire
+      // the lock.
+      if (FALSE === $result) {
+        unset($this->locks[$name]);
+        // Explicit transaction release which also frees the WATCH'ed key.
+        $this->client->discard();
+        return FALSE;
+      }
+
+      return ($this->locks[$name] = TRUE);
+    }
+    else {
+      // Use a SET with microsecond expiration and the NX flag, which will only
+      // succeed if the key does not exist yet.
+      $result = $this->client->set($key, $id, ['nx', 'px' => (int) ($timeout * 1000)]);
+
+      // If the result is FALSE, we failed to acquire the lock.
+      if (FALSE === $result) {
+        return FALSE;
+      }
+
+      // Register the lock.
+      return ($this->locks[$name] = TRUE);
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function lockMayBeAvailable($name) {
+    $key   = $this->getKey($name);
+    $value = $this->client->get($key);
+    return FALSE === $value;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function release($name) {
+    $key = $this->getKey($name);
+    $id  = $this->getLockId();
+
+    unset($this->locks[$name]);
+
+    // Ensure the lock deletion is an atomic transaction. If another thread
+    // manages to removes all lock, we can not alter it anymore else we will
+    // release the lock for the other thread and cause race conditions.
+    $this->client->watch($key);
+
+    if ($this->client->get($key) == $id) {
+      $this->client->del($key);
+    }
+    else {
+      $this->client->unwatch();
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function releaseAll($lock_id = NULL) {
+    // We can afford to deal with a slow algorithm here, this should not happen
+    // on normal run because we should have removed manually all our locks.
+    foreach ($this->locks as $name => $foo) {
+      $this->release($name);
+    }
+  }
+
+}
diff --git a/src/Lock/PredisCluster.php b/src/Lock/PredisCluster.php
new file mode 100644
index 0000000..4d8ab17
--- /dev/null
+++ b/src/Lock/PredisCluster.php
@@ -0,0 +1,10 @@
+<?php
+
+namespace Drupal\redis\Lock;
+
+/**
+ * Predis lock backend implementation.
+ */
+class PredisCluster extends Predis {
+
+}
diff --git a/src/PersistentLock/PhpRedisCluster.php b/src/PersistentLock/PhpRedisCluster.php
new file mode 100644
index 0000000..cbec965
--- /dev/null
+++ b/src/PersistentLock/PhpRedisCluster.php
@@ -0,0 +1,30 @@
+<?php
+
+namespace Drupal\redis\PersistentLock;
+
+use Drupal\redis\Lock\PhpRedisCluster as ClusterLock;
+use Drupal\redis\ClientFactory;
+
+/**
+ * PhpRedisCluster persistent lock backend.
+ */
+class PhpRedisCluster extends ClusterLock {
+
+  /**
+   * Creates a PhpRedisCluster persistent lock backend.
+   *
+   * @param \Drupal\redis\ClientFactory $factory
+   *   The ClientFactory object to initialize the client.
+   */
+  public function __construct(ClientFactory $factory) {
+    // Do not call the parent constructor to avoid registering a shutdown
+    // function that releases all the locks at the end of a request.
+    $this->client = $factory->getClient();
+    // Set the lockId to a fixed string to make the lock ID the same across
+    // multiple requests. The lock ID is used as a page token to relate all the
+    // locks set during a request to each other.
+    // @see \Drupal\Core\Lock\LockBackendInterface::getLockId()
+    $this->lockId = 'persistent';
+  }
+
+}
diff --git a/src/PersistentLock/PredisCluster.php b/src/PersistentLock/PredisCluster.php
new file mode 100644
index 0000000..4f38b00
--- /dev/null
+++ b/src/PersistentLock/PredisCluster.php
@@ -0,0 +1,10 @@
+<?php
+
+namespace Drupal\redis\PersistentLock;
+
+/**
+ * Predis persistent lock backend.
+ */
+class PredisCluster extends Predis {
+
+}