Re-work configuration

phpstan.neon.dist → _phpstan.neon.dist

@@ -1,6 +1,5 @@
 parameters:
   paths:
-    - config
     - src
 
-  level: 0
+  level: 8

config/pulse.php

@@ -1,5 +1,6 @@
 <?php
 
+use Carbon\CarbonInterval as Interval;
 use Laravel\Pulse\Http\Middleware\Authorize;
 
 return [
@@ -17,12 +18,31 @@
     // This must be unique for each reporting server.
     'server_name' => env('PULSE_SERVER_NAME', gethostname()),
 
-    // 'ingest' => Laravel\Pulse\Ingests\Database::class,
-    'ingest' => Laravel\Pulse\Ingests\Redis::class,
+    'storage' => [
+        'driver' => env('PULSE_STORAGE_DRIVER', 'database'),
+
+        'retain' => Interval::days(7),
+
+        'database' => [
+            'connection' => env('PULSE_DB_CONNECTION') ?? env('DB_CONNECTION') ?? 'mysql',
+        ],
+    ],
+
+    'ingest' => [
+        'driver' => env('PULSE_INGEST_DRIVER', 'storage'),
+
+        // TODO how does this play with "storage" and the conflicting key above.
+        'retain' => Interval::days(7),
+
+        // TODO this might conflict with sampling lottery / whatevers
+        'lottery' => [1, 100],
+
+        'redis' => [
+            'connection' => env('PULSE_REDIS_CONNECTION') ?? 'default',
+        ],
+    ],
 
     // TODO: filter configuration?
-    // TODO: trim lottery configuration
-    // TODO: configure days of data to store? default: 7
 
     // in milliseconds
     'slow_endpoint_threshold' => 1000,

src/Checks/QueueSize.php

@@ -0,0 +1,30 @@
+<?php
+
+namespace Laravel\Pulse\Checks;
+
+use Carbon\CarbonImmutable;
+use Illuminate\Support\Collection;
+use Illuminate\Support\Facades\Config;
+use Illuminate\Support\Facades\Queue;
+use Laravel\Pulse\Entries\Entry;
+use Laravel\Pulse\Entries\Table;
+
+class QueueSize
+{
+    /**
+     * Resolve the queue size.
+     *
+     * @return \Illuminate\Support\Collection<int, \Laravel\Pulse\Entries\Entry>
+     */
+    public function __invoke(CarbonImmutable $now): Collection
+    {
+        return collect(Config::get('pulse.queues'))->map(fn ($queue) => new Entry(Table::QueueSize, [
+            'date' => $now->toDateTimeString(),
+            'queue' => $queue,
+            'size' => Queue::size($queue),
+            'failed' => collect(app('queue.failer')->all())
+                ->filter(fn ($job) => $job->queue === $queue)
+                ->count(),
+        ]));
+    }
+}

src/Checks/SystemStats.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace Laravel\Pulse\Checks;
+
+use Carbon\CarbonImmutable;
+use Illuminate\Support\Facades\Config;
+use Laravel\Pulse\Entries\Entry;
+use Laravel\Pulse\Entries\Table;
+use RuntimeException;
+
+class SystemStats
+{
+    /**
+     * Resolve the systems stats.
+     */
+    public function __invoke(CarbonImmutable $now): Entry
+    {
+        return new Entry(Table::Server, [
+            'date' => $now->toDateTimeString(),
+            'server' => Config::get('pulse.server_name'),
+            ...match (PHP_OS_FAMILY) {
+                'Darwin' => [
+                    'cpu_percent' => (int) `top -l 1 | grep -E "^CPU" | tail -1 | awk '{ print $3 + $5 }'`,
+                    'memory_total' => $memoryTotal = intval(`sysctl hw.memsize | grep -Eo '[0-9]+'` / 1024 / 1024), // MB
+                    'memory_used' => $memoryTotal - intval(intval(`vm_stat | grep 'Pages free' | grep -Eo '[0-9]+'`) * intval(`pagesize`) / 1024 / 1024), // MB
+                ],
+                'Linux' => [
+                    'cpu_percent' => (int) `top -bn1 | grep '%Cpu(s)' | tail -1 | grep -Eo '[0-9]+\.[0-9]+' | head -n 4 | tail -1 | awk '{ print 100 - $1 }'`,
+                    'memory_total' => $memoryTotal = intval(`cat /proc/meminfo | grep MemTotal | grep -E -o '[0-9]+'` / 1024), // MB
+                    'memory_used' => $memoryTotal - intval(`cat /proc/meminfo | grep MemAvailable | grep -E -o '[0-9]+'` / 1024), // MB
+                ],
+                default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
+            },
+            'storage' => collect(Config::get('pulse.directories'))->map(fn ($directory) => [
+                'directory' => $directory,
+                'total' => $total = intval(round(disk_total_space($directory) / 1024 / 1024)), // MB
+                'used' => intval(round($total - (disk_free_space($directory) / 1024 / 1024))), // MB
+            ])->toJson(),
+        ]);
+    }
+}

src/Commands/CheckCommand.php

@@ -5,10 +5,13 @@
 use Carbon\CarbonImmutable;
 use Illuminate\Console\Command;
 use Illuminate\Support\Facades\Cache;
-use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\Queue;
 use Illuminate\Support\Sleep;
-use RuntimeException;
+use Laravel\Pulse\Checks\QueueSize;
+use Laravel\Pulse\Checks\SystemStats;
+use Laravel\Pulse\Entries\Entry;
+use Laravel\Pulse\Entries\Table;
+use Laravel\Pulse\Facades\Pulse;
 use Symfony\Component\Console\Attribute\AsCommand;
 
 #[AsCommand(name: 'pulse:check')]
@@ -36,24 +39,22 @@ class CheckCommand extends Command
     /**
      * Handle the command.
      */
-    public function handle(): int
-    {
+    public function handle(
+        SystemStats $systemStats,
+        QueueSize $queueSize,
+    ): int {
         $lastRestart = Cache::get('illuminate:pulse:restart');
 
         $lastSnapshotAt = (new CarbonImmutable)->floorSeconds($this->interval);
 
         while (true) {
-            if (Cache::get('illuminate:pulse:restart') !== $lastRestart) {
-                $this->comment('Pulse restart requested. Exiting at '.$now->toDateTimeString());
+            $now = new CarbonImmutable();
 
+            if (Cache::get('illuminate:pulse:restart') !== $lastRestart) {
                 return self::SUCCESS;
             }
 
-            $now = new CarbonImmutable();
-
             if ($now->subSeconds($this->interval)->lessThan($lastSnapshotAt)) {
-                $this->comment('Sleeping for a second at '.$now->toDateTimeString());
-
                 Sleep::for(1)->second();
 
                 continue;
@@ -62,82 +63,24 @@ public function handle(): int
             $lastSnapshotAt = $now->floorSeconds($this->interval);
 
             /*
-             * Check system stats.
+             * Collect server stats.
              */
 
-            $stats = [
-                'date' => $lastSnapshotAt->toDateTimeString(),
-                'server' => config('pulse.server_name'),
-                ...$this->getStats(),
-                'storage' => collect(config('pulse.directories'))->map(fn ($directory) => [
-                    'directory' => $directory,
-                    'total' => $total = intval(round(disk_total_space($directory) / 1024 / 1024)), // MB
-                    'used' => intval(round($total - (disk_free_space($directory) / 1024 / 1024))), // MB
-                ])->toJson(),
-            ];
-
-            // TODO: do this via the injest
-            DB::table('pulse_servers')->insert($stats);
+            Pulse::record($entry = $systemStats($lastSnapshotAt));
 
-            $this->line('<fg=gray>[system stats]</> '.json_encode($stats));
+            $this->line('<fg=gray>[system stats]</> '.json_encode($entry->attributes));
 
             /*
-             * Check queue sizes.
+             * Collect queue sizes.
              */
 
             if (Cache::lock("illuminate:pulse:check-queue-sizes:{$lastSnapshotAt->timestamp}", $this->interval)->get()) {
-                $sizes = collect(config('pulse.queues'))
-                    ->map(fn ($queue) => [
-                        'date' => $lastSnapshotAt->toDateTimeString(),
-                        'queue' => $queue,
-                        'size' => Queue::size($queue),
-                        'failed' => collect(app('queue.failer')->all())
-                            ->filter(fn ($job) => $job->queue === $queue)
-                            ->count(),
-                    ]);
-
-                DB::table('pulse_queue_sizes')->insert($sizes->all());
-
-                $this->line('<fg=gray>[queue sizes]</> '.$sizes->toJson());
+                $entries = $queueSize($lastSnapshotAt)->each(fn ($entry) => Pulse::record($entry));
+
+                $this->line('<fg=gray>[queue sizes]</> '.$entries->pluck('attributes')->toJson());
             }
 
-            $this->comment('Stats and queue sizes checked at: '.$now->toDateTimeString());
+            Pulse::store();
         }
     }
-
-    /**
-     * Collect stats.
-     */
-    protected function getStats(): array
-    {
-        return match (PHP_OS_FAMILY) {
-            'Darwin' => $this->getDarwinStats(),
-            'Linux' => $this->getLinuxStats(),
-            default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
-        };
-    }
-
-    /**
-     * Collect stats for "Darwin" based systems.
-     */
-    protected function getDarwinStats(): array
-    {
-        return [
-            'cpu_percent' => (int) `top -l 1 | grep -E "^CPU" | tail -1 | awk '{ print $3 + $5 }'`,
-            'memory_total' => $memoryTotal = intval(`sysctl hw.memsize | grep -Eo '[0-9]+'` / 1024 / 1024), // MB
-            'memory_used' => $memoryTotal - intval(intval(`vm_stat | grep 'Pages free' | grep -Eo '[0-9]+'`) * intval(`pagesize`) / 1024 / 1024), // MB
-        ];
-    }
-
-    /**
-     * Collect stats for "Linux" based systems.
-     */
-    protected function getLinuxStats(): array
-    {
-        return [
-            'cpu_percent' => (int) `top -bn1 | grep '%Cpu(s)' | tail -1 | grep -Eo '[0-9]+\.[0-9]+' | head -n 4 | tail -1 | awk '{ print 100 - $1 }'`,
-            'memory_total' => $memoryTotal = intval(`cat /proc/meminfo | grep MemTotal | grep -E -o '[0-9]+'` / 1024), // MB
-            'memory_used' => $memoryTotal - intval(`cat /proc/meminfo | grep MemAvailable | grep -E -o '[0-9]+'` / 1024), // MB
-        ];
-    }
 }

src/Commands/RestartCommand.php

@@ -5,8 +5,6 @@
 use Illuminate\Console\Command;
 use Illuminate\Support\Facades\Cache;
 use Illuminate\Support\InteractsWithTime;
-use Laravel\Pulse\Ingests\Database;
-use Laravel\Pulse\Ingests\Redis;
 use Symfony\Component\Console\Attribute\AsCommand;
 
 #[AsCommand(name: 'pulse:restart')]

src/Commands/WorkCommand.php

@@ -6,8 +6,8 @@
 use Illuminate\Console\Command;
 use Illuminate\Support\Facades\Cache;
 use Illuminate\Support\Sleep;
-use Laravel\Pulse\Ingests\Database;
-use Laravel\Pulse\Ingests\Redis;
+use Laravel\Pulse\Contracts\Ingest;
+use Laravel\Pulse\Contracts\Storage;
 use Symfony\Component\Console\Attribute\AsCommand;
 
 #[AsCommand(name: 'pulse:work')]
@@ -30,34 +30,30 @@ class WorkCommand extends Command
     /**
      * Handle the command.
      */
-    public function handle(Redis $redisIngest, Database $databaseIngest): int
+    public function handle(Ingest $ingest, Storage $storage): int
     {
         $lastRestart = Cache::get('illuminate:pulse:restart');
 
-        $lastTrimmedDatabaseAt = (new CarbonImmutable)->startOfMinute();
+        $lastTrimmedStorageAt = (new CarbonImmutable)->startOfMinute();
 
         while (true) {
             $now = new CarbonImmutable;
 
             if (Cache::get('illuminate:pulse:restart') !== $lastRestart) {
-                $this->comment('Pulse restart requested. Exiting at '.$now->toDateTimeString());
-
                 return self::SUCCESS;
             }
 
-            if ($now->subMinute()->greaterThan($lastTrimmedDatabaseAt)) {
-                $this->comment('Trimming the database at '.$now->toDateTimeString());
+            if ($now->subMinute()->greaterThan($lastTrimmedStorageAt)) {
+                $storage->trim();
 
-                $databaseIngest->trim($now->subWeek());
+                $lastTrimmedStorageAt = $now;
 
-                $lastTrimmedDatabaseAt = $now;
+                $this->comment('Storage trimmed');
             }
 
-            $processed = $redisIngest->processEntries(1000);
+            $processed = $ingest->store($storage, 1000);
 
             if ($processed === 0) {
-                $this->comment('Queue finished processing. Sleeping at '.$now->toDateTimeString());
-
                 Sleep::for(1)->second();
             } else {
                 $this->comment('Processed ['.$processed.'] entries at '.$now->toDateTimeString());

src/Contracts/Ingest.php

@@ -2,17 +2,25 @@
 
 namespace Laravel\Pulse\Contracts;
 
-use Carbon\CarbonImmutable;
+use Illuminate\Support\Collection;
 
 interface Ingest
 {
     /**
-     * Ingest the entries and updates without throwing exceptions.
+     * Ingest the entries and updates.
+     *
+     * @param  \Illuminate\Support\Collection<int, \Laravel\Pulse\Entries\Entry>  $entries
+     * @param  \Illuminate\Support\Collection<int, \Laravel\Pulse\Entries\Update>  $updates
      */
-    public function ingestSilently(array $entries, array $updates): void;
+    public function ingest(Collection $entries, Collection $updates): void;
 
     /**
-     * Trim the ingest without throwing exceptions.
+     * Trim the ingested entries.
      */
-    public function trimSilently(CarbonImmutable $oldest): void;
+    public function trim(): void;
+
+    /**
+     * Store the ingested entries.
+     */
+    public function store(Storage $storage, int $count): int;
 }

src/Contracts/Storage.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace Laravel\Pulse\Contracts;
+
+use Illuminate\Support\Collection;
+
+interface Storage
+{
+    /**
+     * Store the entries and updates.
+     *
+     * @param  \Illuminate\Support\Collection<int, \Laravel\Pulse\Entries\Entry>  $entries
+     * @param  \Illuminate\Support\Collection<int, \Laravel\Pulse\Entries\Update>  $updates
+     */
+    public function store(Collection $entries, Collection $updates): void;
+
+    /**
+     * Trim the stored entries.
+     */
+    public function trim(): void;
+}