decalog / includes / libraries / prometheus / Storage / Redis.php

Redis.php in DecaLog 4.4.0, at includes/libraries/prometheus/Storage/Redis.php

472 lines 13.8 KB

Raw Download Zip

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Prometheus\Storage;
6
7	use InvalidArgumentException;
8	use Prometheus\Counter;
9	use Prometheus\Exception\StorageException;
10	use Prometheus\Gauge;
11	use Prometheus\Histogram;
12	use Prometheus\MetricFamilySamples;
13
14	class Redis implements Adapter
15	{
16	const PROMETHEUS_METRIC_KEYS_SUFFIX = '_METRIC_KEYS';
17
18	/**
19	* @var mixed[]
20	*/
21	private static $defaultOptions = [
22	'host' => '127.0.0.1',
23	'port' => 6379,
24	'timeout' => 0.1,
25	'read_timeout' => '10',
26	'persistent_connections' => false,
27	'password' => null,
28	];
29
30	/**
31	* @var string
32	*/
33	private static $prefix = 'PROMETHEUS_';
34
35	/**
36	* @var mixed[]
37	*/
38	private $options = [];
39
40	/**
41	* @var \Redis
42	*/
43	private $redis;
44
45	/**
46	* @var boolean
47	*/
48	private $connectionInitialized = false;
49
50	/**
51	* Redis constructor.
52	* @param mixed[] $options
53	*/
54	public function __construct(array $options = [])
55	{
56	$this->options = array_merge(self::$defaultOptions, $options);
57	$this->redis = new \Redis();
58	}
59
60	/**
61	* @param \Redis $redis
62	* @return self
63	* @throws StorageException
64	*/
65	public static function fromExistingConnection(\Redis $redis): self
66	{
67	if ($redis->isConnected() === false) {
68	throw new StorageException('Connection to Redis server not established');
69	}
70
71	$self = new self();
72	$self->connectionInitialized = true;
73	$self->redis = $redis;
74
75	return $self;
76	}
77
78	/**
79	* @param mixed[] $options
80	*/
81	public static function setDefaultOptions(array $options): void
82	{
83	self::$defaultOptions = array_merge(self::$defaultOptions, $options);
84	}
85
86	/**
87	* @param string $prefix
88	*/
89	public static function setPrefix(string $prefix): void
90	{
91	self::$prefix = $prefix;
92	}
93
94	/**
95	* @deprecated use replacement method wipeStorage from Adapter interface
96	* @throws StorageException
97	*/
98	public function flushRedis(): void
99	{
100	$this->wipeStorage();
101	}
102
103	/**
104	* @inheritDoc
105	*/
106	public function wipeStorage(): void
107	{
108	$this->ensureOpenConnection();
109
110	$searchPattern = "";
111
112	$globalPrefix = $this->redis->getOption(\Redis::OPT_PREFIX);
113	// @phpstan-ignore-next-line false positive, phpstan thinks getOptions returns int
114	if (is_string($globalPrefix)) {
115	$searchPattern .= $globalPrefix;
116	}
117
118	$searchPattern .= self::$prefix;
119	$searchPattern .= '*';
120
121	$this->redis->eval(
122	<<<LUA
123	local cursor = "0"
124	repeat
125	local results = redis.call('SCAN', cursor, 'MATCH', ARGV[1])
126	cursor = results[1]
127	for _, key in ipairs(results[2]) do
128	redis.call('DEL', key)
129	end
130	until cursor == "0"
131	LUA
132	,
133	[$searchPattern],
134	0
135	);
136	}
137
138	/**
139	* @return MetricFamilySamples[]
140	* @throws StorageException
141	*/
142	public function collect(): array
143	{
144	$this->ensureOpenConnection();
145	$metrics = $this->collectHistograms();
146	$metrics = array_merge($metrics, $this->collectGauges());
147	$metrics = array_merge($metrics, $this->collectCounters());
148	return array_map(
149	function (array $metric): MetricFamilySamples {
150	return new MetricFamilySamples($metric);
151	},
152	$metrics
153	);
154	}
155
156	/**
157	* @throws StorageException
158	*/
159	private function ensureOpenConnection(): void
160	{
161	if ($this->connectionInitialized === true) {
162	return;
163	}
164
165	$this->connectToServer();
166
167	if ($this->options['password'] !== null) {
168	$this->redis->auth($this->options['password']);
169	}
170
171	if (isset($this->options['database'])) {
172	$this->redis->select($this->options['database']);
173	}
174
175	$this->redis->setOption(\Redis::OPT_READ_TIMEOUT, $this->options['read_timeout']);
176	}
177
178	/**
179	* @throws StorageException
180	*/
181	private function connectToServer(): void
182	{
183	try {
184	$connection_successful = false;
185	if ($this->options['persistent_connections'] !== null) {
186	$connection_successful = $this->redis->pconnect(
187	$this->options['host'],
188	(int) $this->options['port'],
189	(float) $this->options['timeout']
190	);
191	} else {
192	$connection_successful = $this->redis->connect($this->options['host'], (int) $this->options['port'], (float) $this->options['timeout']);
193	}
194	if (!$connection_successful) {
195	throw new StorageException("Can't connect to Redis server", 0);
196	}
197	} catch (\RedisException $e) {
198	throw new StorageException("Can't connect to Redis server", 0, $e);
199	}
200	}
201
202	/**
203	* @param mixed[] $data
204	* @throws StorageException
205	*/
206	public function updateHistogram(array $data): void
207	{
208	$this->ensureOpenConnection();
209	$bucketToIncrease = '+Inf';
210	foreach ($data['buckets'] as $bucket) {
211	if ($data['value'] <= $bucket) {
212	$bucketToIncrease = $bucket;
213	break;
214	}
215	}
216	$metaData = $data;
217	unset($metaData['value'], $metaData['labelValues']);
218
219	$this->redis->eval(
220	<<<LUA
221	local increment = redis.call('hIncrByFloat', KEYS[1], ARGV[1], ARGV[3])
222	redis.call('hIncrBy', KEYS[1], ARGV[2], 1)
223	if increment == ARGV[3] then
224	redis.call('hSet', KEYS[1], '__meta', ARGV[4])
225	redis.call('sAdd', KEYS[2], KEYS[1])
226	end
227	LUA
228	,
229	[
230	$this->toMetricKey($data),
231	self::$prefix . Histogram::TYPE . self::PROMETHEUS_METRIC_KEYS_SUFFIX,
232	json_encode(['b' => 'sum', 'labelValues' => $data['labelValues']]),
233	json_encode(['b' => $bucketToIncrease, 'labelValues' => $data['labelValues']]),
234	$data['value'],
235	json_encode($metaData),
236	],
237	2
238	);
239	}
240
241	/**
242	* @param mixed[] $data
243	* @throws StorageException
244	*/
245	public function updateGauge(array $data): void
246	{
247	$this->ensureOpenConnection();
248	$metaData = $data;
249	unset($metaData['value'], $metaData['labelValues'], $metaData['command']);
250	$this->redis->eval(
251	<<<LUA
252	local result = redis.call(ARGV[1], KEYS[1], ARGV[2], ARGV[3])
253
254	if ARGV[1] == 'hSet' then
255	if result == 1 then
256	redis.call('hSet', KEYS[1], '__meta', ARGV[4])
257	redis.call('sAdd', KEYS[2], KEYS[1])
258	end
259	else
260	if result == ARGV[3] then
261	redis.call('hSet', KEYS[1], '__meta', ARGV[4])
262	redis.call('sAdd', KEYS[2], KEYS[1])
263	end
264	end
265	LUA
266	,
267	[
268	$this->toMetricKey($data),
269	self::$prefix . Gauge::TYPE . self::PROMETHEUS_METRIC_KEYS_SUFFIX,
270	$this->getRedisCommand($data['command']),
271	json_encode($data['labelValues']),
272	$data['value'],
273	json_encode($metaData),
274	],
275	2
276	);
277	}
278
279	/**
280	* @param mixed[] $data
281	* @throws StorageException
282	*/
283	public function updateCounter(array $data): void
284	{
285	$this->ensureOpenConnection();
286	$metaData = $data;
287	unset($metaData['value'], $metaData['labelValues'], $metaData['command']);
288	$this->redis->eval(
289	<<<LUA
290	local result = redis.call(ARGV[1], KEYS[1], ARGV[3], ARGV[2])
291	if result == tonumber(ARGV[2]) then
292	redis.call('hMSet', KEYS[1], '__meta', ARGV[4])
293	redis.call('sAdd', KEYS[2], KEYS[1])
294	end
295	return result
296	LUA
297	,
298	[
299	$this->toMetricKey($data),
300	self::$prefix . Counter::TYPE . self::PROMETHEUS_METRIC_KEYS_SUFFIX,
301	$this->getRedisCommand($data['command']),
302	$data['value'],
303	json_encode($data['labelValues']),
304	json_encode($metaData),
305	],
306	2
307	);
308	}
309
310	/**
311	* @return mixed[]
312	*/
313	private function collectHistograms(): array
314	{
315	$keys = $this->redis->sMembers(self::$prefix . Histogram::TYPE . self::PROMETHEUS_METRIC_KEYS_SUFFIX);
316	sort($keys);
317	$histograms = [];
318	foreach ($keys as $key) {
319	$raw = $this->redis->hGetAll(str_replace($this->redis->_prefix(''), '', $key));
320	$histogram = json_decode($raw['__meta'], true);
321	unset($raw['__meta']);
322	$histogram['samples'] = [];
323
324	// Add the Inf bucket so we can compute it later on
325	$histogram['buckets'][] = '+Inf';
326
327	$allLabelValues = [];
328	foreach (array_keys($raw) as $k) {
329	$d = json_decode($k, true);
330	if ($d['b'] == 'sum') {
331	continue;
332	}
333	$allLabelValues[] = $d['labelValues'];
334	}
335
336	// We need set semantics.
337	// This is the equivalent of array_unique but for arrays of arrays.
338	$allLabelValues = array_map("unserialize", array_unique(array_map("serialize", $allLabelValues)));
339	sort($allLabelValues);
340
341	foreach ($allLabelValues as $labelValues) {
342	// Fill up all buckets.
343	// If the bucket doesn't exist fill in values from
344	// the previous one.
345	$acc = 0;
346	foreach ($histogram['buckets'] as $bucket) {
347	$bucketKey = json_encode(['b' => $bucket, 'labelValues' => $labelValues]);
348	if (!isset($raw[$bucketKey])) {
349	$histogram['samples'][] = [
350	'name' => $histogram['name'] . '_bucket',
351	'labelNames' => ['le'],
352	'labelValues' => array_merge($labelValues, [$bucket]),
353	'value' => $acc,
354	];
355	} else {
356	$acc += $raw[$bucketKey];
357	$histogram['samples'][] = [
358	'name' => $histogram['name'] . '_bucket',
359	'labelNames' => ['le'],
360	'labelValues' => array_merge($labelValues, [$bucket]),
361	'value' => $acc,
362	];
363	}
364	}
365
366	// Add the count
367	$histogram['samples'][] = [
368	'name' => $histogram['name'] . '_count',
369	'labelNames' => [],
370	'labelValues' => $labelValues,
371	'value' => $acc,
372	];
373
374	// Add the sum
375	$histogram['samples'][] = [
376	'name' => $histogram['name'] . '_sum',
377	'labelNames' => [],
378	'labelValues' => $labelValues,
379	'value' => $raw[json_encode(['b' => 'sum', 'labelValues' => $labelValues])],
380	];
381	}
382	$histograms[] = $histogram;
383	}
384	return $histograms;
385	}
386
387	/**
388	* @return mixed[]
389	*/
390	private function collectGauges(): array
391	{
392	$keys = $this->redis->sMembers(self::$prefix . Gauge::TYPE . self::PROMETHEUS_METRIC_KEYS_SUFFIX);
393	sort($keys);
394	$gauges = [];
395	foreach ($keys as $key) {
396	$raw = $this->redis->hGetAll(str_replace($this->redis->_prefix(''), '', $key));
397	$gauge = json_decode($raw['__meta'], true);
398	unset($raw['__meta']);
399	$gauge['samples'] = [];
400	foreach ($raw as $k => $value) {
401	$gauge['samples'][] = [
402	'name' => $gauge['name'],
403	'labelNames' => [],
404	'labelValues' => json_decode($k, true),
405	'value' => $value,
406	];
407	}
408	usort($gauge['samples'], function ($a, $b): int {
409	return strcmp(implode("", $a['labelValues']), implode("", $b['labelValues']));
410	});
411	$gauges[] = $gauge;
412	}
413	return $gauges;
414	}
415
416	/**
417	* @return mixed[]
418	*/
419	private function collectCounters(): array
420	{
421	$keys = $this->redis->sMembers(self::$prefix . Counter::TYPE . self::PROMETHEUS_METRIC_KEYS_SUFFIX);
422	sort($keys);
423	$counters = [];
424	foreach ($keys as $key) {
425	$raw = $this->redis->hGetAll(str_replace($this->redis->_prefix(''), '', $key));
426	$counter = json_decode($raw['__meta'], true);
427	unset($raw['__meta']);
428	$counter['samples'] = [];
429	foreach ($raw as $k => $value) {
430	$counter['samples'][] = [
431	'name' => $counter['name'],
432	'labelNames' => [],
433	'labelValues' => json_decode($k, true),
434	'value' => $value,
435	];
436	}
437	usort($counter['samples'], function ($a, $b): int {
438	return strcmp(implode("", $a['labelValues']), implode("", $b['labelValues']));
439	});
440	$counters[] = $counter;
441	}
442	return $counters;
443	}
444
445	/**
446	* @param int $cmd
447	* @return string
448	*/
449	private function getRedisCommand(int $cmd): string
450	{
451	switch ($cmd) {
452	case Adapter::COMMAND_INCREMENT_INTEGER:
453	return 'hIncrBy';
454	case Adapter::COMMAND_INCREMENT_FLOAT:
455	return 'hIncrByFloat';
456	case Adapter::COMMAND_SET:
457	return 'hSet';
458	default:
459	throw new InvalidArgumentException("Unknown command");
460	}
461	}
462
463	/**
464	* @param mixed[] $data
465	* @return string
466	*/
467	private function toMetricKey(array $data): string
468	{
469	return implode(':', [self::$prefix, $data['type'], $data['name']]);
470	}
471	}
472