celery

README

xk6-celery

A simple Celery tasks submitter/checker extension for k6. It relies on github.com/gocelery/gocelery library & use redis connection pooling at K6 object level.

Known current limitations

• This extension is only meant to sumbit Celery tasks and (eventually) check task completion.
• This extension does not validate task success, it only checks if the tasks has a result.
• Redis is the only Celery backend currently supported.
• Tasks using kwargs are not supported.

Also, we do not intend to support & bind all gocelery functions in this project.

Build

To build a k6 binary with this extension, first ensure you have the prerequisites:

• Go toolchain
• Git
• xk6

1. Build with xk6:

xk6 build --with github.com/ubbleai/xk6-celery

This will result in a k6 binary in the current directory.

1. Run with the just build `k6:

./k6 run example.js

Functions

import celery from 'k6/x/celery';

// Create a new Celery Redis Client
// No worker are spawned. We are running in client-only mode.
const client = new celery.Redis({
  url: "redis://127.0.0.1:6379",
  queue: "taskqueue",
});

// Publish a new task with a three positional arguments
// Task id is returned as a string
const taskID = client.delay("my_task", "text-value", 101, ["any", "arg", "type", "allowed"]);

// Check if task have been completed (whether it's a success or not)
// boolean returned
const processed = client.taskCompleted(taskID);
if processed === True {
  console.log("Task processed");
} else {
  console.log("Task still pending");
}

// Wait for task completion using a blocking func call
// boolean returned (returns false if we hit timeout)
const deadlineCompleted = client.waitForTaskCompleted(taskID);
console.log(`Task completed within a timeframe = ${deadlineCompleted}`);

Javascript client configuration

JSON Key	Default value	Description
url	"redis://127.0.0.1:6379"	Celery Client endpoint URL (only Redis supported ATM)
queue	"celery"	Celery queue where to publish tasks
timeout	"30s"	Timeout used in waitForTaskCompleted function
getinterval	"50ms"	Check interval used in waitForTaskCompleted function

example :

const client = new celery.Redis({
  "url": "redis://my-redis:6379/0",
  "queue": "worker-queue-1",
  "timeout": "10s",
  "getinterval": "100ms",
});

Future

• add check success functions
• support AMQP
• clean code/tests & add more opts

Documentation

Index

• type Celery
  • func (c *Celery) Delay(taskName string, args ...interface{}) (string, error)
  • func (c *Celery) TaskCompleted(taskID string) (bool, error)
  • func (c *Celery) WaitForTaskCompleted(taskID string) (bool, error)
• type CeleryInstance
  • func (mi *CeleryInstance) Exports() modules.Exports
  • func (mi *CeleryInstance) NewCeleryRedis(call goja.ConstructorCall) *goja.Object
• type CeleryModule
  • func New() *CeleryModule
  • func (*CeleryModule) NewModuleInstance(vu modules.VU) modules.Instance
• type Duration
  • func (d Duration) MarshalJSON() ([]byte, error)
  • func (d *Duration) UnmarshalJSON(b []byte) error

Types

type Celery

type Celery struct {
	// contains filtered or unexported fields
}

Celery is the exported module instance.

func (*Celery) Delay

func (c *Celery) Delay(taskName string, args ...interface{}) (string, error)

Submits a new task to celery broker It only supports args (no kwargs)

func (*Celery) TaskCompleted

func (c *Celery) TaskCompleted(taskID string) (bool, error)

Check if task result is filled or still empty It's a sync call with instant result.

func (*Celery) WaitForTaskCompleted

func (c *Celery) WaitForTaskCompleted(taskID string) (bool, error)

Wait for task to be completed until timeout is reached It's a blocking call that do a periodic check for any task result It returns true if task is processed, or false if timeout is reached.

type CeleryInstance

type CeleryInstance struct {

	// Celery is the exported module instance.
	*Celery
	// contains filtered or unexported fields
}

CeleryInstance represents an instance of the JS module.

func (*CeleryInstance) Exports

func (mi *CeleryInstance) Exports() modules.Exports

Exports implements the modules.Instance interface and returns the exports of the JS module.

func (*CeleryInstance) NewCeleryRedis

func (mi *CeleryInstance) NewCeleryRedis(call goja.ConstructorCall) *goja.Object

type CeleryModule

type CeleryModule struct{}

CeleryModule is the global module instance that will create Celery client module instances for each VU.

func New

func New() *CeleryModule

New returns a pointer to a new RootModule instance.

func (*CeleryModule) NewModuleInstance

func (*CeleryModule) NewModuleInstance(vu modules.VU) modules.Instance

NewModuleInstance implements the modules.Module interface and returns a new instance for each VU.

type Duration

type Duration struct {
	time.Duration
}

Duration is a wrapper of time.Duration enabling correct json marshalling

func (Duration) MarshalJSON

func (d Duration) MarshalJSON() ([]byte, error)

func (*Duration) UnmarshalJSON

func (d *Duration) UnmarshalJSON(b []byte) error