Add log-volume-generator

.dockerignore

@@ -11,3 +11,4 @@
 /Justfile
 /storage/
 /vendor/
+/bootstrap/cache/*.php

.github/workflows/deploy.yml

@@ -35,6 +35,11 @@ jobs:
             --tag $DOCKER_REGISTRY_IMAGE:${PHP_VERSION} \
             .
 
+      - name: Run application tests
+        env:
+          PHP_VERSION: ${{ matrix.php-version }}
+        run: docker run --rm $DOCKER_REGISTRY_IMAGE:${PHP_VERSION} php artisan test
+
 #      - name: Cache vendor folder
 #        uses: actions/cache@v3
 #        with:

README.md

@@ -11,6 +11,13 @@ just start
 just test
 ```
 
+The PHPUnit suite runs in CI against PHP 8.3, 8.4, and 8.5. To run it in a
+built container:
+
+```bash
+docker compose run --rm php83 php artisan test
+```
+
 ## Deployment
 
 This repo is deployed to apps in all regions on the old platform. Further documentation:  
@@ -19,3 +26,17 @@ https://github.com/fortrabbit/knowledge-base/blob/main/Old%20platform/Overview/t
 ### Initial setup for a new App
 
 The app expects an environment variable: `APP_TYPE`. Possible values are `uni` or `pro`.
+
+### Log volume generator
+
+Visit `/log-volume` to start, monitor, and stop a log-volume run. The UI stores
+run state in MySQL and dispatches short chained jobs through TestRabbit's
+existing database queue. A normal worker must be running:
+
+```bash
+php artisan queue:work --sleep=5
+```
+
+Run `php artisan migrate --force` after deploying the feature. This also adds
+the standard `jobs` table required by TestRabbit's configured database queue if
+the platform database does not already have it.

app/Http/Controllers/LogVolumeController.php

@@ -0,0 +1,145 @@
+<?php
+
+namespace App\Http\Controllers;
+
+use App\Jobs\GenerateLogVolume;
+use App\Models\LogVolumeRun;
+use Illuminate\Http\JsonResponse;
+use Illuminate\Http\Request;
+use Illuminate\Routing\Controller;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Validation\ValidationException;
+use Throwable;
+
+class LogVolumeController extends Controller
+{
+    public function index()
+    {
+        return view('log-volume');
+    }
+
+    public function status(): JsonResponse
+    {
+        return response()->json([
+            'run' => $this->serialize(LogVolumeRun::latest('id')->first()),
+        ]);
+    }
+
+    public function start(Request $request): JsonResponse
+    {
+        $values = $request->validate([
+            'target_size' => ['required', 'string', 'max:32'],
+            'bytes_per_second' => ['required', 'integer', 'min:1', 'max:100000000'],
+            'payload_bytes' => ['required', 'integer', 'min:48', 'max:65536'],
+            'run_id' => ['nullable', 'string', 'max:100', 'regex:/^[A-Za-z0-9._-]+$/'],
+        ]);
+
+        $targetBytes = $this->parseBytes($values['target_size'], 10_000_000_000);
+        $progressBytes = min(100_000_000, $targetBytes);
+
+        $run = DB::transaction(function () use ($values, $targetBytes, $progressBytes) {
+            $active = LogVolumeRun::whereIn('status', ['queued', 'running', 'cancelling'])
+                ->lockForUpdate()
+                ->first();
+
+            if ($active) {
+                throw ValidationException::withMessages([
+                    'run' => 'A log-volume run is already active.',
+                ]);
+            }
+
+            return LogVolumeRun::create([
+                'run_id' => $values['run_id'] ?: bin2hex(random_bytes(8)),
+                'status' => 'queued',
+                'target_bytes' => $targetBytes,
+                'bytes_per_second' => $values['bytes_per_second'],
+                'payload_bytes' => $values['payload_bytes'],
+                'progress_bytes' => $progressBytes,
+                'written_bytes' => 0,
+                'lines' => 0,
+            ]);
+        });
+
+        try {
+            GenerateLogVolume::dispatch($run->id);
+        } catch (Throwable $exception) {
+            $run->forceFill([
+                'status' => 'failed',
+                'error' => 'Unable to queue the worker job: '.$exception->getMessage(),
+                'finished_at' => now(),
+            ])->save();
+
+            return response()->json([
+                'message' => 'Unable to queue the log-volume run.',
+                'run' => $this->serialize($run),
+            ], 500);
+        }
+
+        return response()->json(['run' => $this->serialize($run)], 202);
+    }
+
+    public function stop(LogVolumeRun $run): JsonResponse
+    {
+        if ($run->isActive()) {
+            $run->forceFill([
+                'status' => 'cancelled',
+                'cancel_requested_at' => $run->cancel_requested_at ?? now(),
+                'finished_at' => now(),
+            ])->save();
+        }
+
+        return response()->json(['run' => $this->serialize($run->fresh())]);
+    }
+
+    private function parseBytes(string $value, int $maximum): int
+    {
+        if (! preg_match('/^(\d+(?:\.\d+)?)\s*(b|kb|mb|gb|kib|mib|gib)?$/i', trim($value), $matches)) {
+            throw ValidationException::withMessages([
+                'size' => 'Use a size such as 200mb, 1gb, or 2gib.',
+            ]);
+        }
+
+        $multiplier = match (strtolower($matches[2] ?? 'b')) {
+            'b' => 1,
+            'kb' => 1_000,
+            'mb' => 1_000_000,
+            'gb' => 1_000_000_000,
+            'kib' => 1_024,
+            'mib' => 1_048_576,
+            'gib' => 1_073_741_824,
+        };
+        $bytes = (int) round((float) $matches[1] * $multiplier);
+
+        if ($bytes < 1 || $bytes > $maximum) {
+            throw ValidationException::withMessages([
+                'size' => 'The requested size must be between 1 byte and '.number_format($maximum).' bytes.',
+            ]);
+        }
+
+        return $bytes;
+    }
+
+    private function serialize(?LogVolumeRun $run): ?array
+    {
+        if (! $run) {
+            return null;
+        }
+
+        return [
+            'id' => $run->id,
+            'run_id' => $run->run_id,
+            'status' => $run->status,
+            'target_bytes' => $run->target_bytes,
+            'bytes_per_second' => $run->bytes_per_second,
+            'payload_bytes' => $run->payload_bytes,
+            'progress_bytes' => $run->progress_bytes,
+            'written_bytes' => $run->written_bytes,
+            'lines' => $run->lines,
+            'error' => $run->error,
+            'cancel_requested_at' => $run->cancel_requested_at?->toIso8601String(),
+            'started_at' => $run->started_at?->toIso8601String(),
+            'finished_at' => $run->finished_at?->toIso8601String(),
+            'created_at' => $run->created_at?->toIso8601String(),
+        ];
+    }
+}

app/Jobs/GenerateLogVolume.php

@@ -0,0 +1,173 @@
+<?php
+
+namespace App\Jobs;
+
+use App\Models\LogVolumeRun;
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use RuntimeException;
+use Throwable;
+
+class GenerateLogVolume implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+
+    public int $tries = 1;
+
+    private const CHUNK_SECONDS = 10;
+    private const CANCEL_CHECK_MICROSECONDS = 500_000;
+    private const MAX_LINES_PER_SECOND = 12;
+
+    public function __construct(public int $runId)
+    {
+    }
+
+    public function handle(): void
+    {
+        $run = LogVolumeRun::find($this->runId);
+        if (! $run || ! $run->isActive()) {
+            return;
+        }
+
+        if ($run->cancel_requested_at) {
+            $this->finish($run, 'cancelled');
+            return;
+        }
+
+        if ($run->status === 'queued') {
+            $run->forceFill([
+                'status' => 'running',
+                'started_at' => now(),
+            ])->save();
+        }
+
+        $chunkStartedAt = hrtime(true);
+        $lastCancelCheckAt = $chunkStartedAt;
+        $chunkWritten = 0;
+        $chunkLines = 0;
+        $nextProgressAt = (intdiv($run->written_bytes, $run->progress_bytes) + 1) * $run->progress_bytes;
+
+        while ($run->written_bytes < $run->target_bytes) {
+            $now = hrtime(true);
+            if (($now - $chunkStartedAt) >= self::CHUNK_SECONDS * 1_000_000_000) {
+                break;
+            }
+
+            if (($now - $lastCancelCheckAt) >= self::CANCEL_CHECK_MICROSECONDS * 1_000) {
+                if ($this->cancellationRequested($run)) {
+                    $this->finish($run, 'cancelled');
+                    return;
+                }
+                $lastCancelCheckAt = $now;
+            }
+
+            $line = sprintf(
+                "log-volume-test run_id=%s sequence=%d emitted_at=%s payload=%s\n",
+                $run->run_id,
+                $run->lines,
+                gmdate('Y-m-d\TH:i:s\Z'),
+                base64_encode(random_bytes($run->payload_bytes)),
+            );
+
+            $this->writeAll(STDERR, $line);
+            $length = strlen($line);
+            $run->written_bytes += $length;
+            ++$run->lines;
+            $chunkWritten += $length;
+            ++$chunkLines;
+
+            if ($run->written_bytes >= $nextProgressAt) {
+                $run->save();
+                $this->writeProgress($run);
+                $nextProgressAt = (intdiv($run->written_bytes, $run->progress_bytes) + 1) * $run->progress_bytes;
+            }
+
+            $expectedSeconds = max(
+                $chunkWritten / $run->bytes_per_second,
+                $chunkLines / self::MAX_LINES_PER_SECOND,
+            );
+            $actualSeconds = (hrtime(true) - $chunkStartedAt) / 1_000_000_000;
+            $delay = (int) (($expectedSeconds - $actualSeconds) * 1_000_000);
+            if ($delay > 0) {
+                usleep(min($delay, 1_000_000));
+            }
+        }
+
+        $run->save();
+
+        if ($run->written_bytes >= $run->target_bytes) {
+            $this->finish($run, 'complete');
+            return;
+        }
+
+        self::dispatch($run->id);
+    }
+
+    protected function cancellationRequested(LogVolumeRun $run): bool
+    {
+        return $run->newQuery()
+            ->whereKey($run->getKey())
+            ->whereNotNull('cancel_requested_at')
+            ->exists();
+    }
+
+    public function failed(?Throwable $exception): void
+    {
+        LogVolumeRun::whereKey($this->runId)
+            ->whereIn('status', ['queued', 'running', 'cancelling'])
+            ->update([
+                'status' => 'failed',
+                'error' => $exception?->getMessage() ?? 'The worker job failed.',
+                'finished_at' => now(),
+            ]);
+    }
+
+    private function finish(LogVolumeRun $run, string $status): void
+    {
+        $run->forceFill([
+            'status' => $status,
+            'finished_at' => now(),
+        ])->save();
+
+        $elapsed = max($run->started_at?->diffInMilliseconds(now()) / 1000, 0.001);
+        $this->writeAll(STDERR, sprintf(
+            "log-volume-finished run_id=%s status=%s bytes=%d lines=%d elapsed_seconds=%.3f average_bytes_per_second=%.0f\n",
+            $run->run_id,
+            $status,
+            $run->written_bytes,
+            $run->lines,
+            $elapsed,
+            $run->written_bytes / $elapsed,
+        ));
+    }
+
+    private function writeProgress(LogVolumeRun $run): void
+    {
+        $elapsed = max($run->started_at?->diffInMilliseconds(now()) / 1000, 0.001);
+        $this->writeAll(STDERR, sprintf(
+            "log-volume-progress run_id=%s bytes=%d target_bytes=%d elapsed_seconds=%.3f average_bytes_per_second=%.0f\n",
+            $run->run_id,
+            $run->written_bytes,
+            $run->target_bytes,
+            $elapsed,
+            $run->written_bytes / $elapsed,
+        ));
+    }
+
+    /** @param resource $stream */
+    private function writeAll($stream, string $contents): void
+    {
+        $length = strlen($contents);
+        $offset = 0;
+        while ($offset < $length) {
+            $written = fwrite($stream, substr($contents, $offset));
+            if ($written === false || $written === 0) {
+                throw new RuntimeException('Unable to write generated log data.');
+            }
+            $offset += $written;
+        }
+    }
+}

app/Models/LogVolumeRun.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace App\Models;
+
+use Illuminate\Database\Eloquent\Model;
+
+class LogVolumeRun extends Model
+{
+    protected $guarded = [];
+
+    protected $casts = [
+        'target_bytes' => 'integer',
+        'bytes_per_second' => 'integer',
+        'payload_bytes' => 'integer',
+        'progress_bytes' => 'integer',
+        'written_bytes' => 'integer',
+        'lines' => 'integer',
+        'cancel_requested_at' => 'datetime',
+        'started_at' => 'datetime',
+        'finished_at' => 'datetime',
+    ];
+
+    public function isActive(): bool
+    {
+        return in_array($this->status, ['queued', 'running', 'cancelling'], true);
+    }
+}