Writing a Jobs Driver

JOBS drivers are mini-plugins that are connected to the main JOBS plugin and initialized by it.

Architecture

While initializing, JOBS plugin searches for the registered drivers by the Constructor interface. Constructor and Driver (will be described below) interfaces are declared in the RR API repository.

Constructor interface:

constructor.go

// Constructor constructs Consumer interface. Endure abstraction.
type Constructor interface {
 // Name returns the name of the driver
 Name() string
 // DriverFromConfig constructs a driver (e.g. kafka, amqp) from the configuration using the provided configKey
 DriverFromConfig(configKey string, queue Queue, pipeline Pipeline) (Driver, error)
 // DriverFromPipeline constructs a driver (e.g. kafka, amqp) from the pipeline. All configuration is provided by the pipeline
 DriverFromPipeline(pipe Pipeline, queue Queue) (Driver, error)
}

Driver interface:

driver.go

// Driver represents the interface for a single jobs driver
type Driver interface {
 // Push pushes the job to the underlying driver
 Push(ctx context.Context, msg Message) error
 // Run starts consuming the pipeline
 Run(ctx context.Context, pipeline Pipeline) error
 // Stop stops the consumer and closes the underlying connection
 Stop(ctx context.Context) error
 // Pause pauses the jobs consuming (while still allowing job pushing)
 Pause(ctx context.Context, pipeline string) error
 // Resume resumes the consumer
 Resume(ctx context.Context, pipeline string) error
 // State returns information about the driver state
 State(ctx context.Context) (*State, error)
}

So every driver should implement the Constructor interface to be found by the JOBS plugin. Let's have at the methods included in the Constructor interface:

1. Name() string: This method should return a user-friendly name for the driver. It'll be used later in the pipelines <pipeline_name>.driver option. It is an important option. The name here and name in the pipeline options should match.

2. DriverFromConfig(configKey string, queue Queue, pipeline Pipeline) (Driver, error): Returns Driver implementation declared via configuration. RoadRunner in turn provide a configuration key (like: jobs.pipelines.pipeline-name.driver-name.config), queue implementation where to push the messages and pipeline with the all information about the pipeline. Later we will have a look on how to use it.

3. DriverFromPipeline(pipe Pipeline, queue Queue) (Driver, error): Returns Driver implementation declared via RPC jobs.Declare call. It doesn't have a configuration, but all info and configuration options are stored in the pipeline method argument.

Initialization

On the Initialization step, JOBS plugin searches for the JOBS drivers and saves them into a hashmap with by its name provided by the Name() string method. It is not possible to have two drivers with the same name. Drives here are things which are declared in the jobs.pipelines configuration. For the pipelines declared via configuration, RoadRunner saves them as well with their name. Pipelines can also be declared with jobs.Declare RPC method. If it is required, you may use Configurer plugin to unmarshal global driver configuration, such a connection string for example.

alt text

How to create a driver for JOBS

All code from the tutorial is here: link

To create a driver for jobs, you need to create a plugin instance:

driver.go

package jobs_driver //nolint:revive,stylecheck

import (
	"github.com/roadrunner-server/api/v4/plugins/v4/jobs"
	"github.com/roadrunner-server/errors"
	"github.com/roadrunner-server/samples/plugins/jobs_driver/driver"
	"go.uber.org/zap"
)

const pluginName string = "my_awesome_driver"

type Configurer interface {
	// UnmarshalKey takes a single key and unmarshal it into a Struct.
	UnmarshalKey(name string, out any) error
	// Has checks if a config section exists.
	Has(name string) bool
}

type Logger interface {
	NamedLogger(name string) *zap.Logger
}

type Plugin struct {
	log *zap.Logger
	cfg Configurer
}

func (p *Plugin) Init(log Logger, cfg Configurer) error {
	if !cfg.Has(pluginName) {
		return errors.E(errors.Disabled)
	}

	p.log = log.NamedLogger(pluginName)
	p.cfg = cfg
	return nil
}

func (p *Plugin) Name() string {
	return pluginName
}

func (p *Plugin) DriverFromConfig(configKey string, pq jobs.Queue, pipeline jobs.Pipeline) (jobs.Driver, error) {
	return driver.FromConfig(configKey, p.log, p.cfg, pipeline, pq)
}

func (p *Plugin) DriverFromPipeline(pipe jobs.Pipeline, pq jobs.Queue) (jobs.Driver, error) {
	return driver.FromPipeline(pipe, p.log, p.cfg, pq)
} 

This is a simple representation of the RR plugin. It is called driver because it is attached to another controlling plugin as a pluggable extender, aka: a driver. Keep in mind the plugin's name.

JOBS plugin will send the following data to the Constructor interface methods:

1. If declared via configuration (.rr.yaml) - configKey, you may use that key to unmarshal the configuration section related solely to this driver. If the driver was declared via jobs.Declare RPC method, all configuration options would be stored in the jobs.Pipeline interface.

2. jobs.Queue: Priority-Queue, used to push the messages and later process by the PHP workers.

3. All other things like logger, Configurer plugin which will be used to get the values from the .rr.yaml configuration you may pass if you need them from the driver's root (e.g.: p.log).

Now, let's see the simplified Driver implementation:

driver.go

package driver

import (
	"context"

	"github.com/roadrunner-server/api/v4/plugins/v4/jobs"
	"go.uber.org/zap"
)

var _ jobs.Driver = (*Driver)(nil)

type Configurer interface {
	// UnmarshalKey takes a single key and unmarshal it into a Struct.
	UnmarshalKey(name string, out any) error
	// Has checks if a config section exists.
	Has(name string) bool
}

type Driver struct {
}

func FromConfig(configKey string, log *zap.Logger, cfg Configurer, pipeline jobs.Pipeline, pq jobs.Queue) (*Driver, error) {
	return &Driver{}, nil
}

// FromPipeline initializes consumer from pipeline
func FromPipeline(pipeline jobs.Pipeline, log *zap.Logger, cfg Configurer, pq jobs.Queue) (*Driver, error) {
	return &Driver{}, nil
}

func (d *Driver) Push(ctx context.Context, job jobs.Message) error {
	return nil
}

func (d *Driver) Run(ctx context.Context, p jobs.Pipeline) error {
	return nil
}

func (d *Driver) State(ctx context.Context) (*jobs.State, error) {
	return nil, nil
}

func (d *Driver) Pause(ctx context.Context, p string) error {
	return nil
}

func (d *Driver) Resume(ctx context.Context, p string) error {
	return nil
}

func (d *Driver) Stop(ctx context.Context) error {
	return nil
}

Here you need to remember the following things:

1. FromConfig and FromPipeline methods are used to initialize the driver, but not to start the message consumption.

2. JOBS plugin will automatically call the Run method is your pipelines would be in the jobs.consume array.

3. For the pipelines, declared via jobs.Declare RPC call, method jobs.Resume should be called instead.

Pushing Jobs into the priority queue

To push the job into priority queue, you need to slightly transform it to add Ack, Nack, etc. methods to it. All interfaces are here RR API repository, but let's have a look at the Job interface.

job.go

// Job represents a binary heap item
type Job interface {
	pq.Item
	// Ack acknowledges the item after processing
	Ack() error
	// Nack discards the item
	Nack() error
	// NackWithOptions discards the item with an optional requeue flag
	NackWithOptions(requeue bool, delay int) error
	// Requeue puts the message back to the queue with an optional delay
	Requeue(headers map[string][]string, delay int) error
	// Body returns the payload associated with the item
	Body() []byte
	// Context returns any meta-information associated with the item
	Context() ([]byte, error)
	// Headers return the metadata for the item
	Headers() map[string][]string
}

Job interface also includes the pq.Item interface to satisfy a minimal priority queue requirement. You may add this (pq.Item) interface to any interface and benefit from the RoadRunner's priority queue. So, our driver's Push method would be updated as follows:

driver.go

func (d *Driver) Push(_ context.Context, job jobs.Message) error {
	item := fromJob(job)
	d.queue.Insert(item)
	return nil
}

func fromJob(job jobs.Message) *Item {
	return &Item{
		Job:     job.Name(),
		Ident:   job.ID(),
		Payload: job.Payload(),
		headers: job.Headers(),
		Options: &Options{
			Priority: job.Priority(),
			Pipeline: job.GroupID(),
			Delay:    int(job.Delay()),
			AutoAck:  job.AutoAck(),
		},
	}
}

fromJob method needed to simply transform jobs.Message into the Job. See link. The rest is to implement a driver specific Ack, Nack, etc. methods.

Configuration for your Jobs driver:

.rr.yaml

version: '3'

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php your_php_worker.php"
  relay: "pipes"

your_global_section:
  addr: "some_connection_string"

logs:
  level: error
  encoding: console
  mode: development

jobs:
  pool:
    num_workers: 10
    allocate_timeout: 60s
    destroy_timeout: 1s

  pipelines:
    test-1:
      driver: my_awesome_driver
      config:
        priority: 1
        prefetch: 100
        # rest of the options

  consume: [ "test-1" ]

Examples of existing drivers: AMQP, Kafka, In-Memory, Nats, SQS, BoltDB, Google Pub-Sub