feature #40155 [Messenger] Support Redis Cluster (nesk)

nicolas-grekas · nicolas-grekas · commit 2928f169cefa · 2021-04-13T19:00:15.000+02:00
This PR was squashed before being merged into the 5.3-dev branch. Discussion ---------- [Messenger] Support Redis Cluster | Q | A | ------------- | --- | Branch? | 5.x | Bug fix? | no | New feature? | yes | Deprecations? | no | Tickets | Fix #38264 | License | MIT | Doc PR | symfony/symfony-docs#14956 This PR brings support for Redis Cluster in the Messenger component: - The first commit _Support RedisCluster instance_ allows to pass a `RedisCluster` object when instanciating the `Connection` class, which brings support for Redis Cluster without any friction. - The second commit _Support multiple hosts DSN for Redis Cluster_ is more opiniated and brings a DSN format to configure a Redis Cluster from `config/packages/messenger.yaml`. Instanciating `Connection` with a `RedisCluster` object: ```php $redis = new \RedisCluster(null, ['host-01:6379', 'host-02:6379', 'host-03:6379', 'host-04:6379']); $connection = new Connection([], [], [], $redis); ``` Configuring a Redis Cluster from YAML: ```yaml // config/packages/messenger.yaml framework: messenger: metadata: default: 'redis://host-01:6379,redis://host-02:6379,redis://host-03:6379' lazy: 'redis://host-01:6379?lazy=1,redis://host-02:6379,redis://host-03:6379' # Configuration will be `lazy = true` and `auto_setup = true` multipleConfig: 'redis://host-01:6379?lazy=1&auto_setup=false,redis://host-02:6379,redis://host-03:6379?auto_setup=true' ``` This format allows to define multiple hosts for a Redis Cluster and still contains valid URLs. Custom configuration is still supported, it can be specified on only one of the URLs in the DSN (see `lazy` above). If the user provides multiple configurations on different URLs, they are simply merged with the following code and if an option is defined multiple times then the latest takes precedence (see `multipleConfig` above). I understand the way the DSN is handled could not suit you. Please, if you close this PR only for the DSN part, just tell me and I will make a new PR with only the first commit. Commits ------- 04530fb2d7 [Messenger] Support Redis Cluster
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -6,6 +6,8 @@ CHANGELOG
 
  * Add `rediss://` DSN scheme support for TLS protocol
  * Deprecate TLS option, use `rediss://127.0.0.1` instead of `redis://127.0.0.1?tls=1`
+ * Add support for `\RedisCluster` instance in `Connection` constructor
+ * Add support for Redis Cluster in DSN
 
 5.2.0
 -----
diff --git a/Tests/Transport/ConnectionTest.php b/Tests/Transport/ConnectionTest.php
@@ -40,6 +40,15 @@ public static function setUpBeforeClass(): void
         }
     }
 
+    private function skipIfRedisClusterUnavailable()
+    {
+        try {
+            new \RedisCluster(null, explode(' ', getenv('REDIS_CLUSTER_HOSTS')));
+        } catch (\Exception $e) {
+            self::markTestSkipped($e->getMessage());
+        }
+    }
+
     public function testFromInvalidDsn()
     {
         $this->expectException(\InvalidArgumentException::class);
@@ -59,6 +68,20 @@ public function testFromDsn()
         );
     }
 
+    public function testFromDsnWithMultipleHosts()
+    {
+        $this->skipIfRedisClusterUnavailable();
+
+        $hosts = explode(' ', getenv('REDIS_CLUSTER_HOSTS'));
+
+        $dsn = array_map(function ($host) {
+            return 'redis://'.$host;
+        }, $hosts);
+        $dsn = implode(',', $dsn);
+
+        $this->assertInstanceOf(Connection::class, Connection::fromDsn($dsn));
+    }
+
     public function testFromDsnOnUnixSocket()
     {
         $this->assertEquals(
@@ -160,6 +183,14 @@ public function testDeprecationIfInvalidOptionIsPassedWithDsn()
         Connection::fromDsn('redis://localhost/queue?foo=bar');
     }
 
+    public function testRedisClusterInstanceIsSupported()
+    {
+        $this->skipIfRedisClusterUnavailable();
+
+        $redis = new \RedisCluster(null, explode(' ', getenv('REDIS_CLUSTER_HOSTS')));
+        $this->assertInstanceOf(Connection::class, new Connection([], [], [], $redis));
+    }
+
     public function testKeepGettingPendingMessages()
     {
         $redis = $this->createMock(\Redis::class);
@@ -429,4 +460,20 @@ public function testLazy()
         $connection->reject($message['id']);
         $redis->del('messenger-lazy');
     }
+
+    public function testLazyCluster()
+    {
+        $this->skipIfRedisClusterUnavailable();
+
+        $connection = new Connection(
+            ['lazy' => true],
+            ['host' => explode(' ', getenv('REDIS_CLUSTER_HOSTS'))]
+        );
+
+        $connection->add('1', []);
+        $this->assertNotEmpty($message = $connection->get());
+        $this->assertSame('1', $message['body']);
+        $connection->reject($message['id']);
+        $connection->cleanup();
+    }
 }
diff --git a/Transport/Connection.php b/Transport/Connection.php
@@ -56,7 +56,10 @@ class Connection
     private $deleteAfterReject;
     private $couldHavePendingMessages = true;
 
-    public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], \Redis $redis = null)
+    /**
+     * @param \Redis|\RedisCluster|null $redis
+     */
+    public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], $redis = null)
     {
         if (version_compare(phpversion('redis'), '4.3.0', '<')) {
             throw new LogicException('The redis transport requires php-redis 4.3.0 or higher.');
@@ -71,29 +74,19 @@ public function __construct(array $configuration, array $connectionCredentials =
             $auth = null;
         }
 
-        $initializer = static function ($redis) use ($host, $port, $auth, $serializer, $dbIndex) {
-            $redis->connect($host, $port);
-            $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);
-
-            if (null !== $auth && !$redis->auth($auth)) {
-                throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
-            }
-
-            if ($dbIndex && !$redis->select($dbIndex)) {
-                throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
-            }
-
-            return true;
-        };
-
-        if (null === $redis) {
-            $redis = new \Redis();
-        }
-
-        if ($configuration['lazy'] ?? self::DEFAULT_OPTIONS['lazy']) {
-            $redis = new RedisProxy($redis, $initializer);
+        $lazy = $configuration['lazy'] ?? self::DEFAULT_OPTIONS['lazy'];
+        if (\is_array($host) || $redis instanceof \RedisCluster) {
+            $hosts = \is_string($host) ? [$host.':'.$port] : $host; // Always ensure we have an array
+            $initializer = static function ($redis) use ($hosts, $auth, $serializer) {
+                return self::initializeRedisCluster($redis, $hosts, $auth, $serializer);
+            };
+            $redis = $lazy ? new RedisClusterProxy($redis, $initializer) : $initializer($redis);
         } else {
-            $initializer($redis);
+            $redis = $redis ?? new \Redis();
+            $initializer = static function ($redis) use ($host, $port, $auth, $serializer, $dbIndex) {
+                return self::initializeRedis($redis, $host, $port, $auth, $serializer, $dbIndex);
+            };
+            $redis = $lazy ? new RedisProxy($redis, $initializer) : $initializer($redis);
         }
 
         $this->connection = $redis;
@@ -116,21 +109,57 @@ public function __construct(array $configuration, array $connectionCredentials =
         $this->claimInterval = $configuration['claim_interval'] ?? self::DEFAULT_OPTIONS['claim_interval'];
     }
 
-    public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $redis = null): self
+    private static function initializeRedis(\Redis $redis, string $host, int $port, ?string $auth, int $serializer, int $dbIndex): \Redis
     {
-        $url = $dsn;
-        $scheme = 0 === strpos($dsn, 'rediss:') ? 'rediss' : 'redis';
+        $redis->connect($host, $port);
+        $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);
 
-        if (preg_match('#^'.$scheme.':///([^:@])+$#', $dsn)) {
-            $url = str_replace($scheme.':', 'file:', $dsn);
+        if (null !== $auth && !$redis->auth($auth)) {
+            throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
         }
 
-        if (false === $parsedUrl = parse_url($url)) {
-            throw new InvalidArgumentException(sprintf('The given Redis DSN "%s" is invalid.', $dsn));
+        if ($dbIndex && !$redis->select($dbIndex)) {
+            throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
         }
-        if (isset($parsedUrl['query'])) {
-            parse_str($parsedUrl['query'], $dsnOptions);
-            $redisOptions = array_merge($redisOptions, $dsnOptions);
+
+        return $redis;
+    }
+
+    private static function initializeRedisCluster(?\RedisCluster $redis, array $hosts, ?string $auth, int $serializer): \RedisCluster
+    {
+        if (null === $redis) {
+            $redis = new \RedisCluster(null, $hosts, 0.0, 0.0, false, $auth);
+        }
+
+        $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);
+
+        return $redis;
+    }
+
+    /**
+     * @param \Redis|\RedisCluster|null $redis
+     */
+    public static function fromDsn(string $dsn, array $redisOptions = [], $redis = null): self
+    {
+        if (false === strpos($dsn, ',')) {
+            $parsedUrl = self::parseDsn($dsn, $redisOptions);
+        } else {
+            $dsns = explode(',', $dsn);
+            $parsedUrls = array_map(function ($dsn) use (&$redisOptions) {
+                return self::parseDsn($dsn, $redisOptions);
+            }, $dsns);
+
+            // Merge all the URLs, the last one overrides the previous ones
+            $parsedUrl = array_merge(...$parsedUrls);
+
+            // Regroup all the hosts in an array interpretable by RedisCluster
+            $parsedUrl['host'] = array_map(function ($parsedUrl, $dsn) {
+                if (!isset($parsedUrl['host'])) {
+                    throw new InvalidArgumentException(sprintf('Missing host in DSN part "%s", it must be defined when using Redis Cluster.', $dsn));
+                }
+
+                return $parsedUrl['host'].':'.($parsedUrl['port'] ?? 6379);
+            }, $parsedUrls, $dsns);
         }
 
         self::validateOptions($redisOptions);
@@ -165,7 +194,7 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
             unset($redisOptions['dbindex']);
         }
 
-        $tls = 'rediss' === $scheme;
+        $tls = 'rediss' === $parsedUrl['scheme'];
         if (\array_key_exists('tls', $redisOptions)) {
             trigger_deprecation('symfony/redis-messenger', '5.3', 'Providing "tls" parameter is deprecated, use "rediss://" DSN scheme instead');
             $tls = filter_var($redisOptions['tls'], \FILTER_VALIDATE_BOOLEAN);
@@ -223,6 +252,26 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
         return new self($configuration, $connectionCredentials, $redisOptions, $redis);
     }
 
+    private static function parseDsn(string $dsn, array &$redisOptions): array
+    {
+        $url = $dsn;
+        $scheme = 0 === strpos($dsn, 'rediss:') ? 'rediss' : 'redis';
+
+        if (preg_match('#^'.$scheme.':///([^:@])+$#', $dsn)) {
+            $url = str_replace($scheme.':', 'file:', $dsn);
+        }
+
+        if (false === $parsedUrl = parse_url($url)) {
+            throw new InvalidArgumentException(sprintf('The given Redis DSN "%s" is invalid.', $dsn));
+        }
+        if (isset($parsedUrl['query'])) {
+            parse_str($parsedUrl['query'], $dsnOptions);
+            $redisOptions = array_merge($redisOptions, $dsnOptions);
+        }
+
+        return $parsedUrl;
+    }
+
     private static function validateOptions(array $options): void
     {
         $availableOptions = array_keys(self::DEFAULT_OPTIONS);
diff --git a/Transport/RedisClusterProxy.php b/Transport/RedisClusterProxy.php
@@ -0,0 +1,42 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Redis\Transport;
+
+/**
+ * Allow to delay connection to Redis Cluster.
+ *
+ * @author Johann Pardanaud <johann@pardanaud.com>
+ *
+ * @internal
+ */
+class RedisClusterProxy
+{
+    private $redis;
+    private $initializer;
+    private $ready = false;
+
+    public function __construct(?\RedisCluster $redis, \Closure $initializer)
+    {
+        $this->redis = $redis;
+        $this->initializer = $initializer;
+    }
+
+    public function __call(string $method, array $args)
+    {
+        if (!$this->ready) {
+            $this->redis = $this->initializer->__invoke($this->redis);
+            $this->ready = true;
+        }
+
+        return $this->redis->{$method}(...$args);
+    }
+}
diff --git a/Transport/RedisProxy.php b/Transport/RedisProxy.php
@@ -33,7 +33,10 @@ public function __construct(\Redis $redis, \Closure $initializer)
 
     public function __call(string $method, array $args)
     {
-        $this->ready ?: $this->ready = $this->initializer->__invoke($this->redis);
+        if (!$this->ready) {
+            $this->redis = $this->initializer->__invoke($this->redis);
+            $this->ready = true;
+        }
 
         return $this->redis->{$method}(...$args);
     }

Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,10 @@ public function __construct(\Redis $redis, \Closure $initializer)`
`33`	`33`
`34`	`34`	`public function __call(string $method, array $args)`
`35`	`35`	`{`
`36`		`- $this->ready ?: $this->ready = $this->initializer->__invoke($this->redis);`
	`36`	`+ if (!$this->ready) {`
	`37`	`+ $this->redis = $this->initializer->__invoke($this->redis);`
	`38`	`+ $this->ready = true;`
	`39`	`+ }`
`37`	`40`
`38`	`41`	`return $this->redis->{$method}(...$args);`
`39`	`42`	`}`