Rethink the pulse:check command (#314)

Author: timacdonald

src/Commands/CheckCommand.php

@@ -3,11 +3,12 @@
 namespace Laravel\Pulse\Commands;
 
 use Carbon\CarbonImmutable;
-use Carbon\CarbonInterval;
 use Illuminate\Console\Command;
 use Illuminate\Contracts\Cache\LockProvider;
 use Illuminate\Events\Dispatcher;
+use Illuminate\Support\Env;
 use Illuminate\Support\Sleep;
+use Illuminate\Support\Str;
 use Laravel\Pulse\Events\IsolatedBeat;
 use Laravel\Pulse\Events\SharedBeat;
 use Laravel\Pulse\Pulse;
@@ -25,7 +26,7 @@ class CheckCommand extends Command
      *
      * @var string
      */
-    public $signature = 'pulse:check';
+    public $signature = 'pulse:check {--once : Take a single snapshot}';
 
     /**
      * The command's description.
@@ -42,38 +43,36 @@ public function handle(
         CacheStoreResolver $cache,
         Dispatcher $event,
     ): int {
-        $lastRestart = $cache->store()->get('laravel:pulse:restart');
+        $isVapor = (bool) Env::get('VAPOR_SSM_PATH');
 
-        $interval = CarbonInterval::seconds(5);
+        $instance = $isVapor ? 'vapor' : Str::random();
 
-        $lastSnapshotAt = CarbonImmutable::now()->floorSeconds((int) $interval->totalSeconds);
+        $lastRestart = $cache->store()->get('laravel:pulse:restart');
 
         $lock = ($store = $cache->store()->getStore()) instanceof LockProvider
-            ? $store->lock('laravel:pulse:check', (int) $interval->totalSeconds)
+            ? $store->lock('laravel:pulse:check', 1)
             : null;
 
         while (true) {
-            $now = CarbonImmutable::now();
-
-            if ($now->subSeconds((int) $interval->totalSeconds)->lessThan($lastSnapshotAt)) {
-                Sleep::for(500)->milliseconds();
-
-                continue;
-            }
-
             if ($lastRestart !== $cache->store()->get('laravel:pulse:restart')) {
                 return self::SUCCESS;
             }
 
-            $lastSnapshotAt = $now->floorSeconds((int) $interval->totalSeconds);
+            $now = CarbonImmutable::now();
 
             if ($lock?->get()) {
-                $event->dispatch(new IsolatedBeat($lastSnapshotAt, $interval));
+                $event->dispatch(new IsolatedBeat($now));
             }
 
-            $event->dispatch(new SharedBeat($lastSnapshotAt, $interval));
+            $event->dispatch(new SharedBeat($now, $instance));
 
             $pulse->ingest();
+
+            if ($isVapor || $this->option('once')) {
+                return self::SUCCESS;
+            }
+
+            Sleep::until($now->addSecond());
         }
     }
 }

src/Events/IsolatedBeat.php

@@ -3,7 +3,6 @@
 namespace Laravel\Pulse\Events;
 
 use Carbon\CarbonImmutable;
-use Carbon\CarbonInterval;
 
 class IsolatedBeat
 {
@@ -12,7 +11,6 @@ class IsolatedBeat
      */
     public function __construct(
         public CarbonImmutable $time,
-        public CarbonInterval $interval,
     ) {
         //
     }

src/Events/SharedBeat.php

@@ -3,7 +3,6 @@
 namespace Laravel\Pulse\Events;
 
 use Carbon\CarbonImmutable;
-use Carbon\CarbonInterval;
 
 class SharedBeat
 {
@@ -12,7 +11,7 @@ class SharedBeat
      */
     public function __construct(
         public CarbonImmutable $time,
-        public CarbonInterval $interval,
+        public string $instance,
     ) {
         //
     }

src/Recorders/Concerns/Throttling.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace Laravel\Pulse\Recorders\Concerns;
+
+use Carbon\CarbonImmutable;
+use DateInterval;
+use Illuminate\Support\Facades\App;
+use Illuminate\Support\InteractsWithTime;
+use Laravel\Pulse\Events\IsolatedBeat;
+use Laravel\Pulse\Events\SharedBeat;
+use Laravel\Pulse\Support\CacheStoreResolver;
+
+trait Throttling
+{
+    use InteractsWithTime;
+
+    /**
+     * Determine if the recorder is ready to record another snapshot.
+     */
+    protected function throttle(DateInterval|int $interval, SharedBeat|IsolatedBeat $event, callable $callback, ?string $key = null): void
+    {
+        $key ??= static::class;
+
+        if ($event instanceof SharedBeat) {
+            $key = $event->instance.":{$key}";
+        }
+
+        $cache = App::make(CacheStoreResolver::class);
+
+        $key = 'laravel:pulse:throttle:'.$key;
+
+        $lastRunAt = $cache->store()->get($key);
+
+        if ($lastRunAt !== null && CarbonImmutable::createFromTimestamp($lastRunAt)->addSeconds($this->secondsUntil($interval))->isFuture()) {
+            return;
+        }
+
+        $callback($event);
+
+        $cache->store()->put($key, $event->time->getTimestamp(), $interval);
+    }
+}

src/Recorders/Servers.php

@@ -13,6 +13,8 @@
  */
 class Servers
 {
+    use Concerns\Throttling;
+
     /**
      * The events to listen for.
      *
@@ -35,51 +37,49 @@ public function __construct(
      */
     public function record(SharedBeat $event): void
     {
-        if ($event->time->second % 15 !== 0) {
-            return;
-        }
-
-        $server = $this->config->get('pulse.recorders.'.self::class.'.server_name');
-        $slug = Str::slug($server);
+        $this->throttle(15, $event, function ($event) {
+            $server = $this->config->get('pulse.recorders.'.self::class.'.server_name');
+            $slug = Str::slug($server);
 
-        $memoryTotal = match (PHP_OS_FAMILY) {
-            'Darwin' => intval(`sysctl hw.memsize | grep -Eo '[0-9]+'` / 1024 / 1024),
-            'Linux' => intval(`cat /proc/meminfo | grep MemTotal | grep -E -o '[0-9]+'` / 1024),
-            'Windows' => intval(((int) trim(`wmic ComputerSystem get TotalPhysicalMemory | more +1`)) / 1024 / 1024),
-            'BSD' => intval(`sysctl hw.physmem | grep -Eo '[0-9]+'` / 1024 / 1024),
-            default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
-        };
+            $memoryTotal = match (PHP_OS_FAMILY) {
+                'Darwin' => intval(`sysctl hw.memsize | grep -Eo '[0-9]+'` / 1024 / 1024),
+                'Linux' => intval(`cat /proc/meminfo | grep MemTotal | grep -E -o '[0-9]+'` / 1024),
+                'Windows' => intval(((int) trim(`wmic ComputerSystem get TotalPhysicalMemory | more +1`)) / 1024 / 1024),
+                'BSD' => intval(`sysctl hw.physmem | grep -Eo '[0-9]+'` / 1024 / 1024),
+                default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
+            };
 
-        $memoryUsed = match (PHP_OS_FAMILY) {
-            'Darwin' => $memoryTotal - intval(intval(`vm_stat | grep 'Pages free' | grep -Eo '[0-9]+'`) * intval(`pagesize`) / 1024 / 1024), // MB
-            'Linux' => $memoryTotal - intval(`cat /proc/meminfo | grep MemAvailable | grep -E -o '[0-9]+'` / 1024), // MB
-            'Windows' => $memoryTotal - intval(((int) trim(`wmic OS get FreePhysicalMemory | more +1`)) / 1024), // MB
-            'BSD' => intval(intval(`( sysctl vm.stats.vm.v_cache_count | grep -Eo '[0-9]+' ; sysctl vm.stats.vm.v_inactive_count | grep -Eo '[0-9]+' ; sysctl vm.stats.vm.v_active_count | grep -Eo '[0-9]+' ) | awk '{s+=$1} END {print s}'`) * intval(`pagesize`) / 1024 / 1024), // MB
-            default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
-        };
+            $memoryUsed = match (PHP_OS_FAMILY) {
+                'Darwin' => $memoryTotal - intval(intval(`vm_stat | grep 'Pages free' | grep -Eo '[0-9]+'`) * intval(`pagesize`) / 1024 / 1024), // MB
+                'Linux' => $memoryTotal - intval(`cat /proc/meminfo | grep MemAvailable | grep -E -o '[0-9]+'` / 1024), // MB
+                'Windows' => $memoryTotal - intval(((int) trim(`wmic OS get FreePhysicalMemory | more +1`)) / 1024), // MB
+                'BSD' => intval(intval(`( sysctl vm.stats.vm.v_cache_count | grep -Eo '[0-9]+' ; sysctl vm.stats.vm.v_inactive_count | grep -Eo '[0-9]+' ; sysctl vm.stats.vm.v_active_count | grep -Eo '[0-9]+' ) | awk '{s+=$1} END {print s}'`) * intval(`pagesize`) / 1024 / 1024), // MB
+                default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
+            };
 
-        $cpu = match (PHP_OS_FAMILY) {
-            'Darwin' => (int) `top -l 1 | grep -E "^CPU" | tail -1 | awk '{ print $3 + $5 }'`,
-            'Linux' => (int) `top -bn1 | grep -E '^(%Cpu|CPU)' | awk '{ print $2 + $4 }'`,
-            'Windows' => (int) trim(`wmic cpu get loadpercentage | more +1`),
-            'BSD' => (int) `top -b -d 2| grep 'CPU: ' | tail -1 | awk '{print$10}' | grep -Eo '[0-9]+\.[0-9]+' | awk '{ print 100 - $1 }'`,
-            default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
-        };
+            $cpu = match (PHP_OS_FAMILY) {
+                'Darwin' => (int) `top -l 1 | grep -E "^CPU" | tail -1 | awk '{ print $3 + $5 }'`,
+                'Linux' => (int) `top -bn1 | grep -E '^(%Cpu|CPU)' | awk '{ print $2 + $4 }'`,
+                'Windows' => (int) trim(`wmic cpu get loadpercentage | more +1`),
+                'BSD' => (int) `top -b -d 2| grep 'CPU: ' | tail -1 | awk '{print$10}' | grep -Eo '[0-9]+\.[0-9]+' | awk '{ print 100 - $1 }'`,
+                default => throw new RuntimeException('The pulse:check command does not currently support '.PHP_OS_FAMILY),
+            };
 
-        $this->pulse->record('cpu', $slug, $cpu, $event->time)->avg()->onlyBuckets();
-        $this->pulse->record('memory', $slug, $memoryUsed, $event->time)->avg()->onlyBuckets();
-        $this->pulse->set('system', $slug, json_encode([
-            'name' => $server,
-            'cpu' => $cpu,
-            'memory_used' => $memoryUsed,
-            'memory_total' => $memoryTotal,
-            'storage' => collect($this->config->get('pulse.recorders.'.self::class.'.directories')) // @phpstan-ignore argument.templateType argument.templateType
-                ->map(fn (string $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
-                ])
-                ->all(),
-        ], flags: JSON_THROW_ON_ERROR), $event->time);
+            $this->pulse->record('cpu', $slug, $cpu, $event->time)->avg()->onlyBuckets();
+            $this->pulse->record('memory', $slug, $memoryUsed, $event->time)->avg()->onlyBuckets();
+            $this->pulse->set('system', $slug, json_encode([
+                'name' => $server,
+                'cpu' => $cpu,
+                'memory_used' => $memoryUsed,
+                'memory_total' => $memoryTotal,
+                'storage' => collect($this->config->get('pulse.recorders.'.self::class.'.directories')) // @phpstan-ignore argument.templateType argument.templateType
+                    ->map(fn (string $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
+                    ])
+                    ->all(),
+            ], flags: JSON_THROW_ON_ERROR), $event->time);
+        });
     }
 }