Merge pull request #4 from worksome/memory

feat(memory): JIRA-5517 Peak memory usage per job

README.md

@@ -64,6 +64,19 @@ The default schedule for this is `true`, to disable this event listener,
 add `MeterName::ProcessedJobs->value => false` under a `horizon` key in your `telemetry.php` config file.
 
 
+#### [`ProcessedJobsPeakMemoryUsageListener`](src/Listeners/ProcessedJobsPeakMemoryUsageListener.php)
+
+The `ProcessedJobsPeakMemoryUsageListener` listener will create a histogram over peak memory usage each time a job is processed.
+This metric will be registered under the name `horizon_processed_jobs_peak_memory_usage`.
+
+Two listeners are actually registered here, the secondary listener is [`ProcessedJobsPeakMemoryUsagePreparationListener`](src/Listeners/ProcessedJobsPeakMemoryUsagePreparationListener.php)
+which will take care of clearing the peak memory usage before a job starts, so for long-running queue workers, we get
+the correct number.
+
+The default schedule for this is `true`, to disable this event listener,
+add `MeterName::ProcessedJobsPeakMemoryUsage->value => false` under a `horizon` key in your `telemetry.php` config file.
+
+
 ## Testing
 
 ```bash

config/telemetry.php

@@ -46,5 +46,12 @@
          * The default is `true`, set to `false` to disable this.
          */
         MeterName::ProcessedJobs->value => true,
+
+        /**
+         * Whether the processed job memory usage metric listener is enabled.
+         *
+         * The default is `true`, set to `false` to disable this.
+         */
+        MeterName::ProcessedJobsPeakMemoryUsage->value => true,
     ],
 ];

src/Enums/MeterName.php

@@ -11,6 +11,7 @@ enum MeterName: string
     case CurrentJobs = 'horizon_current_jobs';
     case FailedJobs = 'horizon_failed_jobs';
     case ProcessedJobs = 'horizon_processed_jobs';
+    case ProcessedJobsPeakMemoryUsage = 'horizon_processed_jobs_peak_memory_usage';
 
     public function with(string ...$names): string
     {

src/Enums/MeterUnit.php

@@ -9,4 +9,5 @@ enum MeterUnit: string
     case MasterSupervisors = 'master supervisors';
     case Processes = 'processes';
     case Jobs = 'jobs';
+    case Bytes = 'bytes';
 }

src/HorizonTelemetryServiceProvider.php

@@ -9,10 +9,13 @@
 use Illuminate\Contracts\Events\Dispatcher;
 use Illuminate\Queue\Events\JobFailed;
 use Illuminate\Queue\Events\JobProcessed;
+use Illuminate\Queue\Events\JobProcessing;
 use Illuminate\Support\ServiceProvider;
 use Worksome\HorizonTelemetry\Enums\MeterName;
 use Worksome\HorizonTelemetry\Listeners\FailedJobsListener;
 use Worksome\HorizonTelemetry\Listeners\ProcessedJobsListener;
+use Worksome\HorizonTelemetry\Listeners\ProcessedJobsPeakMemoryUsageListener;
+use Worksome\HorizonTelemetry\Listeners\ProcessedJobsPeakMemoryUsagePreparationListener;
 use Worksome\HorizonTelemetry\Metrics\CurrentJobsMetric;
 use Worksome\HorizonTelemetry\Metrics\CurrentMasterSupervisorsMetric;
 use Worksome\HorizonTelemetry\Metrics\CurrentProcessesMetric;
@@ -29,13 +32,18 @@ public function boot(): void
             /** @var Repository $config */
             $config = $this->app->make(Repository::class);
 
-            if ($config->get(self::CONFIG_PREFIX . MeterName::FailedJobs->value)) {
+            if ($config->get(self::CONFIG_PREFIX . MeterName::FailedJobs->value, true)) {
                 $dispatcher->listen(JobFailed::class, FailedJobsListener::class);
             }
 
-            if ($config->get(self::CONFIG_PREFIX . MeterName::ProcessedJobs->value)) {
+            if ($config->get(self::CONFIG_PREFIX . MeterName::ProcessedJobs->value, true)) {
                 $dispatcher->listen(JobProcessed::class, ProcessedJobsListener::class);
             }
+
+            if ($config->get(self::CONFIG_PREFIX . MeterName::ProcessedJobsPeakMemoryUsage->value, true)) {
+                $dispatcher->listen(JobProcessed::class, ProcessedJobsPeakMemoryUsageListener::class);
+                $dispatcher->listen(JobProcessing::class, ProcessedJobsPeakMemoryUsagePreparationListener::class);
+            }
         });
 
         $this->callAfterResolving(Schedule::class, function (Schedule $schedule) {

src/Listeners/ProcessedJobsPeakMemoryUsageListener.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Worksome\HorizonTelemetry\Listeners;
+
+use Illuminate\Queue\Events\JobProcessed;
+use OpenTelemetry\API\Metrics\ObserverInterface;
+use Worksome\HorizonTelemetry\Enums\MeterName;
+use Worksome\HorizonTelemetry\Enums\MeterUnit;
+use Worksome\HorizonTelemetry\MeterProvider;
+
+readonly class ProcessedJobsPeakMemoryUsageListener
+{
+    public function __construct(
+        private MeterProvider $meterProvider,
+    ) {
+    }
+
+    public function __invoke(JobProcessed $event): void
+    {
+        $meter = $this->meterProvider->getMeter(MeterName::ProcessedJobsPeakMemoryUsage);
+
+        $histogram = $meter->createHistogram(
+            MeterName::ProcessedJobsPeakMemoryUsage->value,
+            MeterUnit::Bytes->value,
+            'The memory usage per job',
+        );
+
+
+        $histogram->record(
+            memory_get_peak_usage(),
+            [
+                'name' => $event->job->resolveName(),
+                'queue' => $event->job->getQueue(),
+            ]
+        );
+    }
+}

src/Listeners/ProcessedJobsPeakMemoryUsagePreparationListener.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Worksome\HorizonTelemetry\Listeners;
+
+use Illuminate\Queue\Events\JobProcessed;
+use OpenTelemetry\API\Metrics\ObserverInterface;
+use Worksome\HorizonTelemetry\Enums\MeterName;
+use Worksome\HorizonTelemetry\Enums\MeterUnit;
+use Worksome\HorizonTelemetry\MeterProvider;
+
+/**
+ * Needed for cleaning the peak memory usage variable inside PHP between jobs.
+ */
+readonly class ProcessedJobsPeakMemoryUsagePreparationListener
+{
+    public function __invoke(JobProcessed $event): void
+    {
+        memory_reset_peak_usage();
+    }
+}