Skip to content

Commit 9ddbab2

Browse files
AltamashShaikhAltamashShaikh
authored
Merge pull request #262 from haristku/feature/add_redis_cluster_capability
add redis cluster capability
2 parents 0102f83 + d992c8c commit 9ddbab2

File tree

14 files changed

+462
-32
lines changed

14 files changed

+462
-32
lines changed

‎Commands/Test.php‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ protected function doExecute(): int
7575
$output->writeln('Timeout: ' . $settings->redisTimeout->getValue());
7676
$output->writeln('Password: ' . $settings->redisPassword->getValue());
7777
$output->writeln('Database: ' . $settings->redisDatabase->getValue());
78-
$output->writeln('UseSentinelBackend: ' . (int) $settings->useSentinelBackend->getValue());
78+
$output->writeln('RedisBackendType: ' . $settings->getRedisType());
7979
$output->writeln('SentinelMasterName: ' . $settings->sentinelMasterName->getValue());
8080

8181
$output->writeln('');
@@ -222,9 +222,14 @@ protected function doExecute(): int
222222
*/
223223
private function getRedisConfig($redis, $configName)
224224
{
225-
$config = $redis->config('GET', $configName);
226-
$value = strtolower(array_shift($config));
225+
if ($redis instanceof \RedisCluster) {
226+
$config = $redis->config('CONFIG', 'GET', $configName);
227+
unset($config[0]);
228+
} else {
229+
$config = $redis->config('GET', $configName);
230+
}
227231

232+
$value = strtolower(array_shift($config));
228233
return $value;
229234
}
230235

‎Queue/Backend/RedisCluster.php‎

Lines changed: 348 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,348 @@
1+
<?php
2+
/**
3+
* Matomo - free/libre analytics platform
4+
*
5+
* @link https://matomo.org
6+
* @license http://www.gnu.org/licenses/gpl-3.0.html GPL v3 or later
7+
*
8+
*/
9+
namespace Piwik\Plugins\QueuedTracking\Queue\Backend;
10+
11+
use Piwik\Log;
12+
use Piwik\Plugins\QueuedTracking\Queue\Backend;
13+
14+
class RedisCluster extends Redis
15+
{
16+
/**
17+
* @var \RedisCluster
18+
*/
19+
protected $redis;
20+
protected $host;
21+
protected $port;
22+
protected $timeout;
23+
protected $password;
24+
25+
/**
26+
* @var int
27+
*/
28+
protected $database;
29+
30+
public function testConnection()
31+
{
32+
try {
33+
$this->connectIfNeeded();
34+
return 'TEST' === $this->redis->echo('TEST_ECHO', 'TEST');
35+
36+
} catch (\Exception $e) {
37+
Log::debug($e->getMessage());
38+
}
39+
40+
return false;
41+
}
42+
43+
public function getServerVersion()
44+
{
45+
$this->connectIfNeeded();
46+
47+
$server = $this->redis->info('server');
48+
49+
if (empty($server)) {
50+
return '';
51+
}
52+
53+
$version = $server['redis_version'];
54+
55+
return $version;
56+
}
57+
58+
public function getLastError()
59+
{
60+
$this->connectIfNeeded();
61+
62+
return $this->redis->getLastError();
63+
}
64+
65+
/**
66+
* Returns converted bytes to B,K,M,G,T.
67+
* @param int|float|double $bytes byte number.
68+
* @param int $precision decimal round.
69+
* * @return string
70+
*/
71+
private function formatBytes($bytes, $precision = 2) {
72+
$units = array('B', 'K', 'M', 'G', 'T');
73+
74+
$bytes = max($bytes, 0);
75+
$pow = floor(($bytes ? log($bytes) : 0) / log(1024));
76+
$pow = min($pow, count($units) - 1);
77+
$bytes /= (1 << (10 * $pow));
78+
79+
return round($bytes, $precision) . $units[$pow];
80+
}
81+
82+
public function getMemoryStats()
83+
{
84+
$this->connectIfNeeded();
85+
86+
$hosts = explode(',', $this->host);
87+
$ports = explode(',', $this->port);
88+
89+
$memory = array (
90+
'used_memory_human' => 0,
91+
'used_memory_peak_human' => 0
92+
);
93+
94+
foreach ($hosts as $idx=>$host) {
95+
$info = $this->redis->info(array($host, (int)$ports[$idx]), 'memory');
96+
$memory['used_memory_human'] += $info['used_memory'] ?? 0;
97+
$memory['used_memory_peak_human'] += $info['used_memory_peak'] ?? 0;
98+
}
99+
100+
$memory['used_memory_human'] = $this->formatBytes($memory['used_memory_human']);
101+
$memory['used_memory_peak_human'] = $this->formatBytes($memory['used_memory_peak_human']);
102+
103+
return $memory;
104+
}
105+
106+
/**
107+
* Returns the time to live of a key that can expire in ms.
108+
* @param $key
109+
* @return int
110+
*/
111+
public function getTimeToLive($key)
112+
{
113+
$this->connectIfNeeded();
114+
115+
$ttl = $this->redis->pttl($key);
116+
117+
if ($ttl == -1) {
118+
// key exists but has no associated expire
119+
return 99999999;
120+
}
121+
122+
if ($ttl == -2) {
123+
// key does not exist
124+
return 0;
125+
}
126+
127+
return $ttl;
128+
}
129+
130+
public function appendValuesToList($key, $values)
131+
{
132+
$this->connectIfNeeded();
133+
134+
foreach ($values as $value) {
135+
$this->redis->rPush($key, gzcompress($value));
136+
}
137+
138+
// usually we would simply do call_user_func_array(array($redis, 'rPush'), $values); as rpush supports multiple values
139+
// at once but it seems to be not implemented yet see https://github.com/nicolasff/phpredis/issues/366
140+
// doing it in one command should be much faster as it requires less tcp communication. Anyway, we currently do
141+
// not write multiple values at once ... so it is ok!
142+
}
143+
144+
public function getFirstXValuesFromList($key, $numValues)
145+
{
146+
if ($numValues <= 0) {
147+
return array();
148+
}
149+
150+
$this->connectIfNeeded();
151+
$values = $this->redis->lRange($key, 0, $numValues - 1);
152+
foreach($values as $key => $value) {
153+
$tmpValue = @gzuncompress($value); // Avoid warning if not compressed
154+
155+
// if empty, string is not compressed. Use original value
156+
if(empty($tmpValue)) {
157+
$values[$key] = $value;
158+
} else {
159+
$values[$key] = $tmpValue;
160+
}
161+
}
162+
163+
return $values;
164+
}
165+
166+
public function removeFirstXValuesFromList($key, $numValues)
167+
{
168+
if ($numValues <= 0) {
169+
return;
170+
}
171+
172+
$this->connectIfNeeded();
173+
$this->redis->ltrim($key, $numValues, -1);
174+
}
175+
176+
public function hasAtLeastXRequestsQueued($key, $numValuesRequired)
177+
{
178+
if ($numValuesRequired <= 0) {
179+
return true;
180+
}
181+
182+
$numActual = $this->getNumValuesInList($key);
183+
184+
return $numActual >= $numValuesRequired;
185+
}
186+
187+
public function getNumValuesInList($key)
188+
{
189+
$this->connectIfNeeded();
190+
191+
return $this->redis->lLen($key);
192+
}
193+
194+
public function setIfNotExists($key, $value, $ttlInSeconds)
195+
{
196+
$this->connectIfNeeded();
197+
$wasSet = $this->redis->set($key, $value, array('nx', 'ex' => $ttlInSeconds));
198+
199+
return $wasSet;
200+
}
201+
202+
/**
203+
* @internal for tests only
204+
* @return \RedisCluster
205+
*/
206+
public function getConnection()
207+
{
208+
return $this->redis;
209+
}
210+
211+
/**
212+
* @internal for tests only
213+
*/
214+
public function delete($key)
215+
{
216+
$this->connectIfNeeded();
217+
218+
return $this->redis->del($key) > 0;
219+
}
220+
221+
public function deleteIfKeyHasValue($key, $value)
222+
{
223+
if (empty($value)) {
224+
return false;
225+
}
226+
227+
$this->connectIfNeeded();
228+
229+
// see http://redis.io/topics/distlock
230+
$script = 'if redis.call("GET",KEYS[1]) == ARGV[1] then
231+
return redis.call("DEL",KEYS[1])
232+
else
233+
return 0
234+
end';
235+
236+
// ideally we would use evalSha to reduce bandwidth!
237+
return (bool) $this->evalScript($script, array($key), array($value));
238+
}
239+
240+
protected function evalScript($script, $keys, $args)
241+
{
242+
return $this->redis->eval($script, array_merge($keys, $args), count($keys));
243+
}
244+
245+
public function getKeysMatchingPattern($pattern)
246+
{
247+
$this->connectIfNeeded();
248+
249+
return $this->redis->keys($pattern);
250+
}
251+
252+
public function expireIfKeyHasValue($key, $value, $ttlInSeconds)
253+
{
254+
if (empty($value)) {
255+
return false;
256+
}
257+
258+
$this->connectIfNeeded();
259+
260+
$script = 'if redis.call("GET",KEYS[1]) == ARGV[1] then
261+
return redis.call("EXPIRE",KEYS[1], ARGV[2])
262+
else
263+
return 0
264+
end';
265+
// ideally we would use evalSha to reduce bandwidth!
266+
return (bool) $this->evalScript($script, array($key), array($value, (int) $ttlInSeconds));
267+
}
268+
269+
public function get($key)
270+
{
271+
$this->connectIfNeeded();
272+
273+
return $this->redis->get($key);
274+
}
275+
276+
/**
277+
* @internal
278+
*/
279+
public function flushAll()
280+
{
281+
$this->connectIfNeeded();
282+
283+
$hosts = explode(',', $this->host);
284+
$ports = explode(',', $this->port);
285+
286+
foreach ($hosts as $idx=>$host) {
287+
$this->redis->flushDB(array($host, (int)$ports[$idx]));
288+
}
289+
}
290+
291+
private function connectIfNeeded()
292+
{
293+
if (!$this->isConnected()) {
294+
$this->connect();
295+
}
296+
}
297+
298+
protected function connect()
299+
{
300+
$hosts = explode(',', $this->host);
301+
$ports = explode(',', $this->port);
302+
303+
if (count($hosts) !== count($ports)) {
304+
throw new Exception(Piwik::translate('QueuedTracking_NumHostsNotMatchNumPorts'));
305+
}
306+
307+
$hostsPorts = array_map(fn($host, $port): string => "$host:$port", $hosts, $ports);
308+
309+
try {
310+
$this->redis = new \RedisCluster(NULL, $hostsPorts, $this->timeout, $this->timeout, true, $this->password);
311+
return true;
312+
} catch (Exception $e) {
313+
throw new Exception('Could not connect to redis cluster: ' . $e->getMessage());
314+
}
315+
}
316+
317+
public function setConfig($host, $port, $timeout, $password)
318+
{
319+
$this->disconnect();
320+
321+
$this->host = $host;
322+
$this->port = $port;
323+
$this->timeout = $timeout;
324+
325+
if (!empty($password)) {
326+
$this->password = $password;
327+
}
328+
}
329+
330+
private function disconnect()
331+
{
332+
if ($this->isConnected()) {
333+
$this->redis->close();
334+
}
335+
336+
$this->redis = null;
337+
}
338+
339+
private function isConnected()
340+
{
341+
return isset($this->redis);
342+
}
343+
344+
public function setDatabase($database)
345+
{
346+
$this->database = $database;
347+
}
348+
}

‎Queue/Factory.php‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -70,14 +70,17 @@ public static function makeBackendFromSettings(SystemSettings $settings)
7070
} else {
7171
$redis = new Queue\Backend\Sentinel();
7272
$redis->setSentinelMasterName($masterName);
73+
$redis->setDatabase($database);
7374
}
75+
}
76+
elseif($settings->isUsingClusterBackend()) {
77+
$redis = new Queue\Backend\RedisCluster();
7478
} else {
7579
$redis = new Queue\Backend\Redis();
80+
$redis->setDatabase($database);
7681
}
7782

7883
$redis->setConfig($host, $port, $timeout, $password);
79-
$redis->setDatabase($database);
80-
8184
return $redis;
8285
}
8386

0 commit comments

Comments
 (0)