flywheel

README

flywheel

Flywheel is a task observability library for Spinup Go APIs. Many of our APIs must handle asynchronous tasks and this is done in a variety of ways that are less than ideal.

Sometimes...

• custom logic is encoded into the calling application (usually the UI) to check back on the resource and decide if it's been created

or...

• it's fire and forget, with the assumption that the service will get created/delete/updated eventually

or...

• we hold the connection open until the service is actually created/deleted/updated and risk a timeout

or...

• we do something I haven't thought of

Instead, using flywheel we can observe the state of a task. Tasks in flywheel are usually high level orchestration processes... usually assembling a few components into a service. Note that flywheel is currently not a job queuing or management platform. The scope is simply to track the state of an ephemeral task throughout its lifecycle in a lightweight way accessible from multiple instances of an API.

Usage

Install the library

go get github.com/YaleSpinup/flywheel
go get github.com/go-redis/redis/v8

The manager allows passing option functions to override the redis connection information and other configuration items. Some examples of creating a manager are below, see the godoc for all of the available options.

Create a default flywheel manager that can be shared by multiple tasks

    // create a new manager with the namespace using the default values
    manager, _ := flywheel.NewManager("testtesttest")

Create a flywheel manager with a custom redis client

    rdb := redis.NewClient(&redis.Options{
        Addr:     "redis-server.intha.cloud:6379",
        Username: "johnnyjohnny",
        Password: "yesspapa!",
        DB:       123,
        PoolSize: 3,
    })

    // create a new manager with the namespace using the default values
    manager, _ := flywheel.NewManager("testtesttest", WithRedis(rdb))

Create a new task and start it

    ctx := context.Background()
    task := flywheel.NewTask()
    if err := manager.Start(ctx, task); err != nil {
        log.Errorf("error starting task %s: %s", task.ID, err)
    }

Checkin periodically with the job id to extend the TTL and notify the job is active

    if err := manager.CheckIn(ctx, id); err != nil {
        log.Errorf("failed to checkin to task %s", err)
    }

Write some log messages to the event log

    if err := manager.Log(ctx, id, "some log message"); err != nil {
        log.Errorf("failed to log: %s", err)
    }

Mark the task as completed

    if err := manager.Complete(ctx, id); err != nil {
        log.Errorf("failed to mark task %s completed: %s", id, err)
    }

Mark the task as failed

    if err := manager.Fail(ctx, id, "something went horribly wrong"); err != nil {
        log.Errorf("failed to mark task %s failed: %s", id, err)
    }

Get the task details

The manager getting task details doesn't need to use the same instance of the manager as long as the namespace is the same.

    ctx := context.Background()

    out, err := manager.GetTask(ctx, id)
    if err != nil {
        log.Errorf("failed to get task %s: %s", id, err)
    }

    j, err := json.MarshalIndent(out, "", "  ")
    if err != nil {
        log.Errorf("failed to marshall task %s as json %s", id, err)
    }

    log.Infof("%s", string(j))

Use the built-in http handler

    api.Handle("/flywheel", s.flywheel.Handler())

Todo

• A task reaper running on the manager to check for tasks that have been running too long and mark them as failed so upstream systems can get see the failure before the task disappears.

License

GNU Affero General Public License v3.0 (GNU AGPLv3)
Copyright © 2020 Yale University

Documentation

Index

• Constants
• func NewID() string
• type Manager
  • func NewManager(namespace string, opts ...ManagerOption) (*Manager, error)
  • func (m *Manager) CheckIn(ctx context.Context, id string) error
  • func (m *Manager) Complete(ctx context.Context, id string) error
  • func (m *Manager) Fail(ctx context.Context, id, message string) error
  • func (m *Manager) GetTask(ctx context.Context, id string) (*Task, error)
  • func (m *Manager) Handler() http.Handler
  • func (m *Manager) Log(ctx context.Context, id, message string) error
  • func (m *Manager) Start(ctx context.Context, t *Task) error
• type ManagerOption
  • func WithRedis(client *redis.Client) ManagerOption
  • func WithRedisAddress(address string) ManagerOption
  • func WithRedisDatabase(db int) ManagerOption
  • func WithRedisPassword(password string) ManagerOption
  • func WithRedisUsername(username string) ManagerOption
  • func WithTTL(ttl time.Duration) ManagerOption
  • func WithoutRedisPing() ManagerOption
• type Task
  • func NewTask() *Task

Constants

const (
	STATUS_COMPLETED = "completed"
	STATUS_CREATED   = "created"
	STATUS_FAILED    = "failed"
	STATUS_RUNNING   = "running"
)

Functions

func NewID

func NewID() string

NewID returns a new random ID. Currently this is just a UUID string

Types

type Manager

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

Manager controls the communication with redis

func NewManager

func NewManager(namespace string, opts ...ManagerOption) (*Manager, error)

NewManager creates a new manager instance used to communicate with the redis storage engine

func (*Manager) CheckIn

func (m *Manager) CheckIn(ctx context.Context, id string) error

func (*Manager) Complete

func (m *Manager) Complete(ctx context.Context, id string) error

func (*Manager) Fail

func (m *Manager) Fail(ctx context.Context, id, message string) error

func (*Manager) GetTask

func (m *Manager) GetTask(ctx context.Context, id string) (*Task, error)

GetTask pulls the details of a task out of redis, the related events and

func (*Manager) Handler

func (m *Manager) Handler() http.Handler

func (*Manager) Log

func (m *Manager) Log(ctx context.Context, id, message string) error

func (*Manager) Start

func (m *Manager) Start(ctx context.Context, t *Task) error

type ManagerOption

type ManagerOption func(*Manager)

func WithRedis

func WithRedis(client *redis.Client) ManagerOption

WithRedis sets the redis client to support advanced options

func WithRedisAddress

func WithRedisAddress(address string) ManagerOption

WithRedisAddress sets the address for redis in the form <ipaddress|hostname>:<port>. This will be ignored if WithRedis() is passed.

func WithRedisDatabase

func WithRedisDatabase(db int) ManagerOption

WithRedisDatabase sets the redis database to be used. This will be ignored if WithRedis() is passed.

func WithRedisPassword

func WithRedisPassword(password string) ManagerOption

WithRedisPassword sets the redis password to be used. This will be ignored if WithRedis() is passed.

func WithRedisUsername

func WithRedisUsername(username string) ManagerOption

WithRedisUsername sets the redis username to be used. This will be ignored if WithRedis() is passed.

func WithTTL

func WithTTL(ttl time.Duration) ManagerOption

WithTTL sets the amount of time a job will linger in the persistence layer (redis) after the last check-in, log message, completion or failure. This sets a TTL on the redis keys used by the library.

func WithoutRedisPing

func WithoutRedisPing() ManagerOption

WithoutRedisPing disables the initial redis ping check

type Task

type Task struct {
	CheckinAt   string   `json:"checkin_at,omitempty"`
	CompletedAt string   `json:"completed_at,omitempty"`
	CreatedAt   string   `json:"created_at,omitempty"`
	ID          string   `json:"id"`
	Status      string   `json:"status"`
	Events      []string `json:"events,omitempty"`
	FailedAt    string   `json:"failed_at,omitempty"`
	Failure     string   `json:"failure,omitempty"`
}

Task is the detail about a task

func NewTask

func NewTask() *Task