redis_job_dispatcher

README

Redis

• Key/Value Store: where all the jobs are stored
• Todo Queue: Contains the ids of the jobs that are currently todo

Data Structure

Jobs

Key	Val
uuid of job	json object of jobs

{worker_instance} : todo

A hash map with the todo jobs

Key	Val
uuid of job	1

{worker_instance} : queue

A simple queue with the uids of the jobs

Queue Items
uuid of job

{worker_instance} : doing

A hash map with the current jobs

Key	Val
uuid of job	unix-timestamp

{worker_instance} : done

A hash map with the done jobs

Key	Val
uuid of job	1

Documentation

Index

• func GenerateKeyDoingMap(workerInstance string) string
• func GenerateKeyDoneMap(workerInstance string) string
• func GenerateKeyJobsMap() string
• func GenerateKeyTodoMap(workerInstance string) string
• func GenerateKeyTodoQueue(workerInstance string) string
• func GenerateKeyWorkerJobs() string
• func GetInstances() (result []string, err error)
• func GetKeyPrefix() string
• func GetKeys() (results []string, err error)
• func GetStalledJobs() (stalledJobs []uuid.UUID, err error)
• func GetStats() (results []job_dispatcher.Stats, err error)
• func ParseJob(cmd *redis.StringCmd) (job job_dispatcher.Job, err error)
• func Pop(instance string) (job job_dispatcher.Job, err error)
• func Push(job job_dispatcher.Job) (err error)
• type RedisService
  • func (m *RedisService) BulkNew(newJobs []*job_dispatcher.Job) (err error)
  • func (m *RedisService) Clean() error
  • func (m *RedisService) Complete(job *job_dispatcher.Job) (err error)
  • func (m *RedisService) DeleteWorker(workerUID, jobUID uuid.UUID) (err error)
  • func (m *RedisService) GetCurrentJobOfWorker(workerInstances []string, currentWorkerUID uuid.UUID) (job *job_dispatcher.Job, found bool, err error)
  • func (m *RedisService) GetJobByUUID(uid uuid.UUID) (result *job_dispatcher.Job, err error)
  • func (m *RedisService) GetLatestJob(workerInstances []string, workerUUID uuid.UUID) (job *job_dispatcher.Job, err error)
  • func (m *RedisService) GetStats() (results []job_dispatcher.Stats, err error)
  • func (m *RedisService) GetWorkerJob(workerUID uuid.UUID) (jobUID uuid.UUID, err error)
  • func (m *RedisService) HeartBeat(job *job_dispatcher.Job, status map[string]interface{}) error
  • func (m *RedisService) MapWorker(workerUID, jobUID uuid.UUID) (err error)
  • func (m *RedisService) New(job *job_dispatcher.Job) (err error)
  • func (m *RedisService) Release(job *job_dispatcher.Job) (err error)
  • func (m *RedisService) SaveJob(job *job_dispatcher.Job) (err error)
  • func (m *RedisService) Start(job *job_dispatcher.Job) error

Functions

func GenerateKeyDoingMap

func GenerateKeyDoingMap(workerInstance string) string

func GenerateKeyDoneMap

func GenerateKeyDoneMap(workerInstance string) string

func GenerateKeyJobsMap

func GenerateKeyJobsMap() string

func GenerateKeyTodoMap

func GenerateKeyTodoMap(workerInstance string) string

func GenerateKeyTodoQueue

func GenerateKeyTodoQueue(workerInstance string) string

func GenerateKeyWorkerJobs

func GenerateKeyWorkerJobs() string

func GetInstances

func GetInstances() (result []string, err error)

GetInstances returns the instances present in the database

func GetKeyPrefix

func GetKeyPrefix() string

GetKeyPrefix returns the prefix of the redis keys

func GetKeys

func GetKeys() (results []string, err error)

GetKeys returns all the key from the database

func GetStalledJobs

func GetStalledJobs() (stalledJobs []uuid.UUID, err error)

func GetStats

func GetStats() (results []job_dispatcher.Stats, err error)

GetStats calculates the stats by iterating over the keys in the redis store

func ParseJob

func ParseJob(cmd *redis.StringCmd) (job job_dispatcher.Job, err error)

ParseJob extracts the job from a redis result object

func Pop

func Pop(instance string) (job job_dispatcher.Job, err error)

Pop retrieves a job to the bulk job list

func Push

func Push(job job_dispatcher.Job) (err error)

Push adds a new job to the bulk job list

Types

type RedisService

type RedisService struct{}

func (*RedisService) BulkNew

func (m *RedisService) BulkNew(newJobs []*job_dispatcher.Job) (err error)

BulkNew creates multiple new keyJobs

func (*RedisService) Clean

func (m *RedisService) Clean() error

Clean cleans all stalled keyJobs get different job instances find all jobs of the instances in the active map iterate over the jobs and compare the timestamps if the timestamp is larger than the threshold add the job to the queue again

func (*RedisService) Complete

func (m *RedisService) Complete(job *job_dispatcher.Job) (err error)

Complete marks the job as completed

func (*RedisService) DeleteWorker

func (m *RedisService) DeleteWorker(workerUID, jobUID uuid.UUID) (err error)

DeleteWorker removes a worker from the job

func (*RedisService) GetCurrentJobOfWorker

func (m *RedisService) GetCurrentJobOfWorker(
	workerInstances []string,
	currentWorkerUID uuid.UUID,
) (job *job_dispatcher.Job, found bool, err error)

GetCurrentJobOfWorker retrieves the job of the worker if one exists in the database

func (*RedisService) GetJobByUUID

func (m *RedisService) GetJobByUUID(uid uuid.UUID) (result *job_dispatcher.Job, err error)

GetJobByUUID returns a job by its unique identifier

func (*RedisService) GetLatestJob

func (m *RedisService) GetLatestJob(workerInstances []string, workerUUID uuid.UUID) (job *job_dispatcher.Job, err error)

GetLatestJob returns the latest job

func (*RedisService) GetStats

func (m *RedisService) GetStats() (results []job_dispatcher.Stats, err error)

GetStats returns the stats of all the jobs

func (*RedisService) GetWorkerJob

func (m *RedisService) GetWorkerJob(workerUID uuid.UUID) (jobUID uuid.UUID, err error)

GetWorkerJob returns the uid of the job

func (*RedisService) HeartBeat

func (m *RedisService) HeartBeat(job *job_dispatcher.Job, status map[string]interface{}) error

HeartBeat marks a job as alive

func (*RedisService) MapWorker

func (m *RedisService) MapWorker(workerUID, jobUID uuid.UUID) (err error)

MapWorker adds a mapping which stores what worker works on which job Todo: enhance this by adding a timestamp

func (*RedisService) New

func (m *RedisService) New(job *job_dispatcher.Job) (err error)

New Creates a new job

func (*RedisService) Release

func (m *RedisService) Release(job *job_dispatcher.Job) (err error)

Release releases the job

func (*RedisService) SaveJob

func (m *RedisService) SaveJob(job *job_dispatcher.Job) (err error)

SaveJob saves a job in the database

func (*RedisService) Start

func (m *RedisService) Start(job *job_dispatcher.Job) error

Start marks a job as started