[Proposal] Add the ability for Jobs to be watched after being dispatched

[kharysharpe]

Ability for jobs to be watched and their state determined while being able to receive basic messages from the running job, independent of the underlying queue system.

Theory of Operation QueueWatcher, QueueWatchable

• Uses a model to perform actions on the datastore, set in queue config. (could be a database, nosql, etc). job_watch: id:string, termination_requested:bool(F), time_started:int, time_completed:int, message:text, status:string
• Create simple database based model and migration script
• Jobs uses a QueueWatchable Trait
• Jobs can pingWatcher()
• Jobs can notifyWatcher($message)
• Use Queue::before to QueueWatchable::markJobStarted()
• Use Queue::after to QueueWatchable::markJobCompleted()
• Use InteractsWithQueue to markJob{Status}( ) or implement within QueueWatchable and override the behaviour we want

Trait QueueWatchable:

• setQueueWatchId($id = null): Returns a unique identifier for job
• getQueueWatchId(): Returns a unique identifier for job
• markJobStarted(): sets the time job commenced
• markJobCompleted(): sets the time job completed
• markJobReleased(): sets the time job commenced
• pingWatcher(): Updates the time the job was last seen
• notifyWatcher($message): Sets a message that can be retrieved
• terminateJob(): Returns if the job should be terminated....not so sure about this.

class QueueWatcher

• getLastSeen($id): Returns the last timestamp seen
• getDuration($id): Returns how long the job has been running.
• getNotification($id): Returns message set by job
• getStatus($id, $clearFinished = true): Returns WAITING, RUNNING, FINISHED states. Removes Finished Jobs for Watch List
• requestTermination($id): Send a kill request ...should we do this?

class FooJob extends Job
{
  use QueueWatchable;

  function __construct($data)
  {
      $this->setQueueWatchId();

      $this->data = $data;

  }

  function handle()
  {

    while($running) {

      //Do whatever

      $this->notifyWatcher('Doing bar in foo job');

      $this->pingWatcher();

      if ($this->terminateJob())
      {
        //Cleanup and exit

      }

    }

  }

}

  $job = new FooJob($data);

  $id = $job->getQueueWatchId();

  dispatch($job);

  $seconds = QueueWatch::getDuration($id)
  $timestamp = QueueWatch::getLastSeen($id);
  $status = QueueWatch::getStatus($id);

class AppServiceProvider extends ServiceProvider
{
    public function boot()
    {
        Queue::after(function (JobProcessed $event) {
            $event->job->markJobCompleted();
        });
    }
}

AFAIK there is no queue independent way to do this.

Would this be welcomed to be included in the core?

Please review and give me some feedback.

[tomschlick]

This is something that I have wanted to build for a long time. Coupled with the new notifications / broadcasting ability it could be really powerful.

I see being able to monitor it via any/all of these channels:

• By Redis Key
• By Broadcast Channel (think a front-end app waiting for an image to resize, and requesting the payload once they receive a "completed" event).
• By Database Table

Based on the methods you provided above, I think this would be the resulting data structure:

{
  "id": "ae227d8c-cc9b-4576-9843-b329bac92ad5",
  "type": "App\\Jobs\\ContactImporter",
  "status": "processing",
  "progress": "56%",
  "tries": 1,
  "last_message": "Starting to process records.",
  "last_ping_at": "2016-08-20 11:38:00",
  "created_at": "2016-08-20 11:37:43",
  "updated_at": "2016-08-20 11:38:00",
  "started_at": "2016-08-20 11:38:00",
  "completed_at": null
}

[tomschlick]

As Graham said in the laravel/framework thread, I think this is best made as a package. It's not needed by 90% of the laravel community so I don't think Taylor would want to take it on & have to maintain it.

I have created a new repo for this if you're interested in collaborating on it. https://github.com/tomschlick/laravel-micro-manager

I have a ton of experience with laravel's job queues. At @runmybusiness we process upwards of a few hundred thousand jobs per day.

[kharysharpe]

@tomschlick Wow, thanks for the feedback. I believe your suggestion is far more feature rich than what I am thinking.

The primitive use case is when was the job started, how long and is it still running, when it was completed, and optionally publish meta data to indicate current running state.

Actually, it might be more natural to enhance InteractsWithQueue with this limited additionally functionality and perhaps have a Queue::peek($jobId)->getLastSeen(); that we can use to inquire about the job.

This feels like it is a common issue that is going to arise once you start working with queues and by that measure it is a good fit for the core. Laravel Queue already provides an abstraction layer for different queue platforms, we should also have a queue platform agnostic approach to checking the state of jobs.

I want to start coding but I would appreciate as much feedback as I can get.

PS. Doing this will also aid the creation of advanced packages like yours to better manage and oversee jobs.

[kharysharpe]

I have done up more or less the concept in the following repos:

https://github.com/laravel/laravel/compare/master...kharysharpe:queue-watcher https://github.com/laravel/framework/compare/5.3...kharysharpe:queue-watcher

The queue config file will have a watcher that defaults to a null watcher that as the name suggests does nothing. A database implementation of the watcher is implemented and can be configured along with the table to be used.

The watcher implements a JobWatcher contract and so anyone can implement it as they desire for sending notifications, use memcached as a backend etc.

class FooJob extends Job
{
  use InteractsWithQueue;

  function __construct($data)
  {
      $this->jobWatcher->start(); 

      $this->data = $data;

  }

  function handle()
  {

    while($running) {

      //Do whatever

     //Update Watcher
      $this->jobWatcher->setMetaData([
           'msg' => 'Doing bar in foo job',
           'progress' => 80
      ]); 

      $this->jobWatcher->ping();

    }
  }
}

  $job = new FooJob($data);

  $watcherId = $job->getWatcherId();

  dispatch($job);

// cache $watcherId for later use

Later:

$watcher = new Queue\Job\Watchers\DatabaseWatcher(); // Or custom Job Watcher

$watcher->setWatcherId($watcherId);

$timestamp = $watcher->getTimeLastUpdated();

$status = $watcher->getStatus();

Feedback welcomed.

Update: Added migration

PS. Repos aren't tested or working, just a brain dump to get this moving.

[judgej]

Is this batch handling on L8 anything like what was being proposed here?

https://github.com/laravel/framework/pull/32830

[kharysharpe]

Thanks for pointing this out.

It's pretty similar.

At glance what I am not seeing is:

• Job can notify you of it's own state like Running Step 3/10
• Job sends a heartbeat to say it's still running

Use case for this was for the frontend to dispatch a long running job and be able to communicate to the user what's going via polling or on demand.

@taylorotwell Could we get something specific for Jobs or somehow incorporate those in the batching?

[themsaid]

You can install Laravel Horizon which will show you the job status in realtime. As for notifications, this is better implemented in userland.

[kharysharpe]

This is meant to provide end user feedback on their background process.

They started a multi step process and want visual feedback such as the percentage complete or which step of the process is being executed, like optimizing a batch of images, generating a report, compressing files, send a batch of emails.

Horizon wouldn't be a good fit.

The code is more or less already in Laravel within the Batch Jobs feature.

Refactoring, extracting and/or applying it to a Job would be cleaner and better DX.