RedisMetricsRepository.php

<?php

namespace Laravel\Horizon\Repositories;

use Carbon\CarbonImmutable;
use Illuminate\Contracts\Redis\Factory as RedisFactory;
use Illuminate\Redis\Connections\PhpRedisConnection;
use Illuminate\Support\Str;
use Laravel\Horizon\Contracts\MetricsRepository;
use Laravel\Horizon\Lock;
use Laravel\Horizon\LuaScripts;
use Laravel\Horizon\WaitTimeCalculator;
use Throwable;

class RedisMetricsRepository implements MetricsRepository
{
    /**
     * The Redis connection instance.
     *
     * @var \Illuminate\Contracts\Redis\Factory
     */
    public $redis;

    /**
     * Create a new repository instance.
     *
     * @param  \Illuminate\Contracts\Redis\Factory  $redis
     * @return void
     */
    public function __construct(RedisFactory $redis)
    {
        $this->redis = $redis;
    }

    /**
     * Get all of the class names that have metrics measurements.
     *
     * @return array
     */
    public function measuredJobs()
    {
        $classes = (array) $this->connection()->smembers('measured_jobs');

        return collect($classes)
            ->map(fn ($class) => preg_match('/job:(.*)$/', $class, $matches) ? $matches[1] : $class)
            ->sort()
            ->values()
            ->all();
    }

    /**
     * Get all of the queues that have metrics measurements.
     *
     * @return array
     */
    public function measuredQueues()
    {
        $queues = (array) $this->connection()->smembers('measured_queues');

        return collect($queues)
            ->map(fn ($class) => preg_match('/queue:(.*)$/', $class, $matches) ? $matches[1] : $class)
            ->sort()
            ->values()
            ->all();
    }

    /**
     * Get the jobs processed per minute since the last snapshot.
     *
     * @return float
     */
    public function jobsProcessedPerMinute()
    {
        return round($this->throughput() / $this->minutesSinceLastSnapshot());
    }

    /**
     * Get the application's total throughput since the last snapshot.
     *
     * @return int
     */
    public function throughput()
    {
        return collect($this->measuredQueues())
            ->reduce(fn ($carry, $queue) => $carry + $this->connection()->hget('queue:'.$queue, 'throughput'), 0);
    }

    /**
     * Get the throughput for a given job.
     *
     * @param  string  $job
     * @return int
     */
    public function throughputForJob($job)
    {
        return $this->throughputFor('job:'.$job);
    }

    /**
     * Get the throughput for a given queue.
     *
     * @param  string  $queue
     * @return int
     */
    public function throughputForQueue($queue)
    {
        return $this->throughputFor('queue:'.$queue);
    }

    /**
     * Get the throughput for a given key.
     *
     * @param  string  $key
     * @return int
     */
    protected function throughputFor($key)
    {
        return (int) $this->connection()->hget($key, 'throughput');
    }

    /**
     * Get the average runtime for a given job in milliseconds.
     *
     * @param  string  $job
     * @return float
     */
    public function runtimeForJob($job)
    {
        return $this->runtimeFor('job:'.$job);
    }

    /**
     * Get the average runtime for a given queue in milliseconds.
     *
     * @param  string  $queue
     * @return float
     */
    public function runtimeForQueue($queue)
    {
        return $this->runtimeFor('queue:'.$queue);
    }

    /**
     * Get the average runtime for a given key in milliseconds.
     *
     * @param  string  $key
     * @return float
     */
    protected function runtimeFor($key)
    {
        return (float) $this->connection()->hget($key, 'runtime');
    }

    /**
     * Get the queue that has the longest runtime.
     *
     * @return int
     */
    public function queueWithMaximumRuntime()
    {
        return collect($this->measuredQueues())
            ->sortBy(function ($queue) {
                if ($snapshots = $this->connection()->zrange('snapshot:queue:'.$queue, -1, -1)) {
                    return json_decode($snapshots[0])->runtime;
                }
            })
            ->last();
    }

    /**
     * Get the queue that has the most throughput.
     *
     * @return int
     */
    public function queueWithMaximumThroughput()
    {
        return collect($this->measuredQueues())
            ->sortBy(function ($queue) {
                if ($snapshots = $this->connection()->zrange('snapshot:queue:'.$queue, -1, -1)) {
                    return json_decode($snapshots[0])->throughput;
                }
            })
            ->last();
    }

    /**
     * Increment the metrics information for a job.
     *
     * @param  string  $job
     * @param  float|null  $runtime
     * @return void
     */
    public function incrementJob($job, $runtime)
    {
        $this->connection()->eval(LuaScripts::updateMetrics(), 2,
            'job:'.$job, 'measured_jobs', str_replace(',', '.', (string) $runtime)
        );
    }

    /**
     * Increment the metrics information for a queue.
     *
     * @param  string  $queue
     * @param  float|null  $runtime
     * @return void
     */
    public function incrementQueue($queue, $runtime)
    {
        $this->connection()->eval(LuaScripts::updateMetrics(), 2,
            'queue:'.$queue, 'measured_queues', str_replace(',', '.', (string) $runtime)
        );
    }

    /**
     * Get all of the snapshots for the given job.
     *
     * @param  string  $job
     * @return array
     */
    public function snapshotsForJob($job)
    {
        return $this->snapshotsFor('job:'.$job);
    }

    /**
     * Get all of the snapshots for the given queue.
     *
     * @param  string  $queue
     * @return array
     */
    public function snapshotsForQueue($queue)
    {
        return $this->snapshotsFor('queue:'.$queue);
    }

    /**
     * Get all of the snapshots for the given key.
     *
     * @param  string  $key
     * @return array
     */
    protected function snapshotsFor($key)
    {
        return collect($this->connection()->zrange('snapshot:'.$key, 0, -1))
            ->map(fn ($snapshot) => (object) json_decode($snapshot, true))
            ->values()
            ->all();
    }

    /**
     * Store a snapshot of the metrics information.
     *
     * @return void
     */
    public function snapshot()
    {
        collect($this->measuredJobs())->each(function ($job) {
            $this->storeSnapshotForJob($job);
        });

        collect($this->measuredQueues())->each(function ($queue) {
            $this->storeSnapshotForQueue($queue);
        });

        $this->storeSnapshotTimestamp();
    }

    /**
     * Store a snapshot for the given job.
     *
     * @param  string  $job
     * @return void
     */
    protected function storeSnapshotForJob($job)
    {
        $data = $this->baseSnapshotData($key = 'job:'.$job);

        $this->connection()->zadd(
            'snapshot:'.$key, $time = CarbonImmutable::now()->getTimestamp(), json_encode([
                'throughput' => $data['throughput'],
                'runtime' => $data['runtime'],
                'time' => $time,
            ])
        );

        $this->connection()->zremrangebyrank(
            'snapshot:'.$key, 0, -abs(1 + config('horizon.metrics.trim_snapshots.job', 24))
        );
    }

    /**
     * Store a snapshot for the given queue.
     *
     * @param  string  $queue
     * @return void
     */
    protected function storeSnapshotForQueue($queue)
    {
        $data = $this->baseSnapshotData($key = 'queue:'.$queue);

        $this->connection()->zadd(
            'snapshot:'.$key, $time = CarbonImmutable::now()->getTimestamp(), json_encode([
                'throughput' => $data['throughput'],
                'runtime' => $data['runtime'],
                'wait' => app(WaitTimeCalculator::class)->calculateFor($queue),
                'time' => $time,
            ])
        );

        $this->connection()->zremrangebyrank(
            'snapshot:'.$key, 0, -abs(1 + config('horizon.metrics.trim_snapshots.queue', 24))
        );
    }

    /**
     * Get the base snapshot data for a given key.
     *
     * @param  string  $key
     * @return array
     */
    protected function baseSnapshotData($key)
    {
        $responses = $this->connection()->transaction(function ($trans) use ($key) {
            $trans->hmget($key, ['throughput', 'runtime']);

            $trans->del($key);
        });

        if (! is_array($responses[0])) {
            return ['throughput' => null, 'runtime' => null];
        }

        $snapshot = array_values($responses[0]);

        return [
            'throughput' => $snapshot[0],
            'runtime' => $snapshot[1],
        ];
    }

    /**
     * Get the number of minutes passed since the last snapshot.
     *
     * @return float
     */
    protected function minutesSinceLastSnapshot()
    {
        $lastSnapshotAt = (int) ($this->connection()->get('last_snapshot_at')
            ?: $this->storeSnapshotTimestamp());

        return max(
            (CarbonImmutable::now()->getTimestamp() - $lastSnapshotAt) / 60, 1
        );
    }

    /**
     * Store the current timestamp as the "last snapshot timestamp".
     *
     * @return int
     */
    protected function storeSnapshotTimestamp()
    {
        return tap(CarbonImmutable::now()->getTimestamp(), function ($timestamp) {
            $this->connection()->set('last_snapshot_at', $timestamp);
        });
    }

    /**
     * Attempt to acquire a lock to monitor the queue wait times.
     *
     * @return bool
     */
    public function acquireWaitTimeMonitorLock()
    {
        return app(Lock::class)->get('monitor:time-to-clear');
    }

    /**
     * Clear the metrics for a key.
     *
     * @param  string  $key
     * @return void
     */
    public function forget($key)
    {
        $this->connection()->del($key);
    }

    /**
     * Delete all stored metrics information.
     *
     * @return void
     */
    public function clear()
    {
        $this->forget('last_snapshot_at');
        $this->forget('measured_jobs');
        $this->forget('measured_queues');
        $this->forget('metrics:snapshot');

        $connection = $this->connection();

        // phpredis 6.1+ requires the SCAN cursor to start as null, while predis and older phpredis expect "0"...
        $defaultCursorValue = match (true) {
            $connection instanceof PhpRedisConnection && version_compare(phpversion('redis'), '6.1.0', '>=') => null,
            default => '0',
        };

        foreach (['queue:*', 'job:*', 'snapshot:*'] as $pattern) {
            $cursor = $defaultCursorValue;

            do {
                $scanResult = $connection->scan(
                    $cursor, ['match' => $this->snapshotPatternToMatch($pattern)]
                );

                if (! is_array($scanResult)) {
                    break;
                }

                [$cursor, $keys] = $scanResult;

                foreach ($keys ?? [] as $key) {
                    $this->forget(Str::after($key, config('horizon.prefix')));
                }
            } while (((string) $cursor) !== $defaultCursorValue);
        }
    }

    /**
     * Get the Redis SCAN match pattern for the given metric pattern.
     *
     * @param  string  $pattern
     * @return string
     */
    protected function snapshotPatternToMatch($pattern)
    {
        return $this->usesPhpRedisScanPrefix()
            ? $pattern
            : config('horizon.prefix').$pattern;
    }

    /**
     * Determine if PhpRedis prefixes SCAN patterns itself.
     *
     * @return bool
     */
    protected function usesPhpRedisScanPrefix()
    {
        if (! defined('Redis::OPT_SCAN') || ! defined('Redis::SCAN_PREFIX')) {
            return false;
        }

        $connection = $this->connection();

        if (! method_exists($connection, 'client')) {
            return false;
        }

        $client = $connection->client();

        if (! is_object($client) || ! method_exists($client, 'getOption')) {
            return false;
        }

        try {
            return (int) $client->getOption(\Redis::OPT_SCAN) === \Redis::SCAN_PREFIX;
        } catch (Throwable) {
            return false;
        }
    }

    /**
     * Get the Redis connection instance.
     *
     * @return \Illuminate\Redis\Connections\Connection
     */
    public function connection()
    {
        return $this->redis->connection('horizon');
    }
}