PHP code example of cboxdk / laravel-queue-autoscale

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

cboxdk / laravel-queue-autoscale example snippets

use Cbox\LaravelQueueAutoscale\Configuration\Profiles\BalancedProfile;
use Cbox\LaravelQueueAutoscale\Configuration\Profiles\ConnectionLimitedProfile;
use Cbox\LaravelQueueAutoscale\Configuration\Profiles\CriticalProfile;

return [
    'enabled' => true,

    // Shipped profiles: BalancedProfile, CriticalProfile, HighVolumeProfile,
    // BurstyProfile, BackgroundProfile, ExclusiveProfile, ConnectionLimitedProfile.
    'sla_defaults' => BalancedProfile::class,

    'queues' => [
        // A profile class...
        'payments' => CriticalProfile::class,

        // ...or a partial override merged over sla_defaults.
        'reports' => [
            'sla' => ['target_seconds' => 120],
            'workers' => ['min' => 0, 'max' => 4],
        ],

        // ...or a glob, for queue names generated at runtime. An exact name
        // above always wins over a pattern.
        'scrape-tenant-*' => [
            'profile' => ConnectionLimitedProfile::class,
            'workers' => ['max' => 5],
        ],
    ],
];

use Cbox\LaravelQueueAutoscale\Events\SlaBreachPredicted;
use Cbox\LaravelQueueAutoscale\Events\WorkersScaled;
use Illuminate\Support\Facades\Event;
use Illuminate\Support\Facades\Log;

Event::listen(WorkersScaled::class, function (WorkersScaled $event): void {
    Log::info("Scaled {$event->queue}: {$event->from} → {$event->to} workers");
    Log::info("Reason: {$event->reason}");
});

Event::listen(SlaBreachPredicted::class, function (SlaBreachPredicted $event): void {
    $decision = $event->decision;

    Log::warning("SLA breach predicted for {$decision->queue}", [
        'predicted_pickup' => $decision->predictedPickupTime,
        'sla_target' => $decision->slaTarget,
    ]);
});

'sla_defaults' => BalancedProfile::class,   // ProfileContract class or literal array
'queues' => [],                             // per-queue: profile class or partial override array
'excluded' => [],                           // fnmatch globs never managed, e.g. 'legacy-*'
'groups' => [],                             // multi-queue workers with strict priority

'scaling' => [
    'fallback_job_time_seconds' => 2.0,     // used when metrics have no avg duration
    'breach_threshold' => 0.5,              // fraction of SLA budget before backlog drain engages
    'cooldown_seconds' => 60,               // anti-flapping window for downward reversals
],

'limits' => [
    'max_cpu_percent' => 85,
    'max_memory_percent' => 85,
    'worker_memory_mb_estimate' => 128,     // cold-start estimate; measured data wins once available
    'worker_cpu_core_estimate' => 0.2,
    'reserve_cpu_cores' => 0.2,
],

'manager' => [
    // Fallback drain window for workers whose queue config no longer resolves.
    'shutdown_grace_seconds' => 30,
],

'strategy' => HybridStrategy::class,        // a plain class string, not an array

'policies' => [
    ConservativeScaleDownPolicy::class,
    BreachNotificationPolicy::class,
],

'queues' => [
    'video-encode' => [
        'workers' => ['min' => 0, 'max' => 4],
        'resources' => ['cpu_cores' => 1.0, 'memory_mb' => 2048],
    ],
],

use Cbox\LaravelQueueAutoscale\Configuration\QueueConfiguration;
use Cbox\LaravelQueueAutoscale\Contracts\ScalingStrategyContract;
use Cbox\LaravelQueueMetrics\DataTransferObjects\QueueMetricsData;

final class CustomStrategy implements ScalingStrategyContract
{
    private int $lastTarget = 0;

    public function calculateTargetWorkers(QueueMetricsData $metrics, QueueConfiguration $config): int
    {
        // avgDuration is already in seconds by the time a strategy sees it.
        $jobsPerSecond = $metrics->throughputPerMinute / 60.0;
        $target = (int) ceil($jobsPerSecond * max($metrics->avgDuration, 0.1) * 2);

        return $this->lastTarget = max(
            $config->workers->min,
            min($config->workers->max, $target),
        );
    }

    public function getLastReason(): string
    {
        return "Custom strategy: doubled steady-state demand → {$this->lastTarget} workers";
    }

    public function getLastPrediction(): ?float
    {
        return null; // Optional: predicted pickup time in seconds
    }
}

'strategy' => \App\Scaling\CustomStrategy::class,

use Cbox\LaravelQueueAutoscale\Contracts\ScalingPolicy;
use Cbox\LaravelQueueAutoscale\Scaling\ScalingDecision;

final class BusinessHoursFloorPolicy implements ScalingPolicy
{
    public function beforeScaling(ScalingDecision $decision): ?ScalingDecision
    {
        if (! $decision->shouldScaleDown() || ! now()->isWeekday()) {
            return null;
        }

        if ($decision->targetWorkers >= 2) {
            return null;
        }

        return new ScalingDecision(
            connection: $decision->connection,
            queue: $decision->queue,
            currentWorkers: $decision->currentWorkers,
            targetWorkers: 2,
            reason: "business-hours floor of 2 applied (was: {$decision->reason})",
            predictedPickupTime: $decision->predictedPickupTime,
            slaTarget: $decision->slaTarget,
        );
    }

    public function afterScaling(ScalingDecision $decision): void
    {
        //
    }
}

'policies' => [
    \App\Policies\BusinessHoursFloorPolicy::class,
],

use Cbox\LaravelQueueAutoscale\Events\ScalingDecisionMade;
use Cbox\LaravelQueueAutoscale\Events\WorkersScaled;

Event::listen(ScalingDecisionMade::class, function (ScalingDecisionMade $event): void {
    $decision = $event->decision;

    Log::info('Scaling decision', [
        'queue' => $decision->queue,
        'current' => $decision->currentWorkers,
        'target' => $decision->targetWorkers,
        'action' => $decision->action(),   // 'scale_up' | 'scale_down' | 'hold'
        'reason' => $decision->reason,
    ]);
});

Event::listen(WorkersScaled::class, function (WorkersScaled $event): void {
    Metrics::gauge('queue.workers', $event->to, [
        'queue' => $event->queue,
        'action' => $event->action,        // 'up' | 'down'
    ]);
});

use Cbox\LaravelQueueMetrics\Facades\QueueMetrics;

$metrics = QueueMetrics::getQueueMetrics('redis', 'payments');

echo "Queue: {$metrics->connection}/{$metrics->queue}\n";
echo "Pending: {$metrics->pending} jobs\n";
echo "Oldest job age: {$metrics->oldestJobAge}s\n";
echo "Throughput: {$metrics->throughputPerMinute} jobs/min\n";
echo "Avg duration: {$metrics->avgDuration}\n";
echo "Active workers: {$metrics->activeWorkers}\n";

use Cbox\LaravelQueueAutoscale\Testing\InteractsWithAutoscaling;
use Cbox\LaravelQueueAutoscale\Testing\QueueMetricsFactory;

uses(InteractsWithAutoscaling::class);

test('no tenant ever gets a sixth connection', function () {
    $this->assertWorkersCappedAt(5, 'scrape-tenant-42');
});

test('a failing provider stops the queue scaling up', function () {
    $behind = QueueMetricsFactory::behind(1000, oldestJobAge: 300, queue: 'payments');

    $this->tripFuseFor('payments');

    expect($this->workersDemandedFor($behind))->toBe(0);
});

bash
php artisan queue:autoscale:install

bash
php artisan vendor:publish --tag=queue-autoscale-config

bash
php artisan vendor:publish --tag=queue-metrics-config

bash
php artisan vendor:publish --tag=queue-metrics-migrations
php artisan migrate

bash
php artisan queue:autoscale:restart