sqsjkr

SQS Job Kicker is the simple worker to invoke command from SQS message

Install

go get github.com/kayac/sqsjkr

Usage

package main
import (
    "context"
    "log"

    "github.com/kayac/sqsjkr"
)

func main() {
    // config
    conf := sqsjkr.NewConfig()
    conf.SetAWSAccount("aws_account_id", "aws_profile", "ap-northeast-1")
    conf.SetSQSQueue("your sqs queue name")
    conf.SetConcurrentNum(5)
    conf.SetTriggerCommand("echo trigger")

    // run sqsjkr
    ctx := context.Background()
    sjkr := sqsjkr.New(conf)
    if err := sqsjkr.Run(ctx, sjkr, "debug"); err != nil {
        log.Println("[error] ", err)
    }
}

Config

• [account] section

params | type | description ------- | ------ | ----------------------- id | string | AWS account id profile | string | AWS profile name region | string | AWS region

• [sqs] section

params | type | description ----------- | ------ | ------------------------------------------ queue_name | string | AWS SQS queue name

• [kicker] section

params | type | description -------------------- | -------- | ------------------------------------------ max_concurrent_num | integer | number of jobs concurrency life_time_trigger | string | trigger command when to pass the lifetime stats_port | integer | port number of sqsjkr stats

You can load config by toml format file:

[account]
id = "12345678"
profile = "default"
region = "ap-northeast-1"

[sqs]
queue_name = "sqsjkr_queue"

[kicker]
max_concurrent_num = 5
life_time_trigger = "echo 'stdin!' | ./test/trigger_test.sh"
stats_port = 8061

conf := sqsjkr.LoadConfig("/path/to/toml")

Job definition

params | type | description ----------------- | ----------------- | ------------------------------------------------------------------------------------------------------------------- command | string | job command env | map | environment variables event_id | string | job event uniq name (for example, AWS CloudWatch Event Scheduler ID(Name)). life_time | integer or string | integer is fixed by second unit. string format requires unit name such as 'm', 's', and so on (e.g. 1s, 1m, 1h). lock_id | string | locks another job abort_if_locked | bool | if job is locked by lock_id, new job give up without retry. disable_life_time_trigger | bool | disable lifetime trigger even though a job is over the lifetime (default false).

• example:

{
    "command": "echo 'hello sqsjkr!'",
    "env": {
        "PATH": "/usr/local/bin/:/usr/bin/:/sbin/:/bin",
        "MAILTO": "example@example.com"
    },
    "event_id": "cloudwatch_event_schedule_id_name",
    "lock_id": "lock_id",
    "life_time": "1m",
    "abort_if_locked": false
}

LifeTime

The job waits for life_time if the other job which is same lock_id is executing. So, if a job requires too many time to process and don't want to execute frequently in short term, job should be set the proper life_time.

Locker

SQS Job Kicker provides Locker interface which is like a feature of 'setlock' to avoid to execute same lock_id. sqsjkr package's sample uses DynamoDB as Locker backend. Show the following Locker interface:

type Locker interface {
	Lock(string, string) error
	Unlock(string) error
}

You can set your custom Locker by SetLocker(locker Locker):

mylocker := NewMyLocker() // Your Locker
sjkr.SetLocker(mylocker)

Throttler

sqsjkr provides Throttler interface which avoid to execute duplicated job. This repository includes the Throttler sample using DynamoDB. Show following Throttler interface:

type Locker interface {
	Set(string, string) error
	UnSet(string) error
}

You can set your custom Throttler by SetThrottler(th Throttler):

myThr := NewMyThrottler() // Your Throttler
sjkr.SetLocker(myThr)

Stats HTTP endpoint

sqsjkr runs a HTTP server on port 8061 to get stats of myself.

$ curl -s localhost:8061/stats/metrics/v2
{
  "workers": {
    "busy": 4,
    "idle": 6
  },
  "invocations": {
    "succeeded": 10,
    "failed": 2,
    "errored": 3
  }
}

LICENSE

MIT