captain

README

capn'

What's Captain?

The question you should be asking is Who is Cap'n? He's captain of your ship, he controls all the jobs you need done. From cleaning up to polishing. Captain takes your command, and runs it.

Why can't I do this myself?

Well, technically you can. But if you use Capn', he'll tell you if the job failed, he'll report to you how long is it taking, etc. You can setup a adapter which will publish the information to all your crew members on slack.

Okay, tell me more

Imagine a situation, where you GET data from some random endpoint and store it on db, we can write like this:

package main

import (
	"net/http"
	"log"
)
func main() {
    resp, _ := http.Get("http://example.com/users")
    // save resp in db
    log.Print(resp)
}

You'd do much more error handling, and you can live with this.

You decide to use cron job to run this every couple of minutes.

Now imagine some day, they decided to throw an error, or not respond to you, or time out. Now unless you write that logic all in this main thread, we won't know there's something important worth paying attention to. And there's so many other factor which might slow something down, or maybe kill your worker. We won't be able to catch every single thing.

With captain, we wrap our worker in a handler, then we can handle log messages, time outs, or longer execution times.

Now consider you've got cron job which imports new Products from your e-commerce api endpoint every minute, and uses capn'

package main

import (
	"github.com/fossapps/captain"
	"time"
	"net/http"
	"sync"
	"log"
)

func main() {
	job := captain.CreateJob()
	job.WithRuntimeProcessingFrequency(100 * time.Millisecond)
	job.WithRuntimeProcessor(func(tick time.Time, message string, startTime time.Time) {
		if time.Since(startTime) > 2 * time.Minutes {
			// report this incident via email/slack/anything
		}
	})

	job.SetWorker(func(Channel captain.CommChan) {
		last_import := LastImportDate.Get() // maybe from db
		resp, _ := http.Get("http://example.com/products/new?since=" + last_import)
		// log resp in db
		log.Print(resp)
	})
	job.Run()
}

Now, say your worker didn't execute for 2 days for some reason, it'll take longer than your usual time, which can be reported back to you.

What else can this do?

As of now, it takes your job and your run time processor, and runs them in goroutines, for every tick (configurable), your runtime processor is called where you can note things like how long has it been running, did it send something in channel, do we need to tell our crew that this job is taking too long?

It also supports lock provider. Imagine a situation where you want a cron to run every single minute, and if one instance of this job is already running, lock provider will make sure we don't run duplicate ones. You can remove this restriction by not implementing a lock provider for a job.

I'm working on ResultProcessor right now, which is called after the job is finished, which will take in the summary of what the job did and can do stuff with it like: reporting to your crew on slack/irc/telegram/whatever

Why would I use this?

Cron jobs comes to mind. So many times, cron jobs ends up not running/misbehaving/failing, etc. And we pretty much don't know what happened. With this, I can setup in a way that if a particular job takes more than 1 minute, it reports to slack, it reports the summary of job to slack, it reports if something goes wrong. You are always in the for {} (See what I did there? you're always in the loop)

How good is it?

Honestly, it's not. As of now I don't know if I can use this on production, but I'm working hard on improving this. Since I don't get to go that much (see again?), I'll try to give my free time for this. And slowly start to move forward so that this can be stable in future. I believe together we can get this capn' sail the sea which he dreams of.

Can I help?

Of course you can. There's can always be something that I didn't think about, some edge case, some missing tests, some feature which might be handy, etc. I'd love to get any help I can. Be it a new bug report, a new feature request, a pull request to help me fix something. I'll use labels like "help-wanted" or "good-first-pr" for the things I'd want help with. You can grab anything you like too.

Documentation

Overview

Package captain is used to run your job, and monitor during runtime

Index

• type CommChan
• type Config
  • func CreateJob() Config
  • func (config *Config) Run()
  • func (config *Config) SetWorker(worker Worker)
  • func (config *Config) WithLockProvider(lockProvider LockProvider)
  • func (config *Config) WithResultProcessor(processor ResultProcessor)
  • func (config *Config) WithRuntimeProcessingFrequency(frequency time.Duration)
  • func (config *Config) WithRuntimeProcessor(processor RuntimeProcessor)
• type LockProvider
• type ResultProcessor
• type RuntimeProcessor
• type Worker

Examples

• Config.WithResultProcessor

Types

type CommChan

type CommChan struct {
	Logs   chan string
	Result chan string
}

CommChan is a basic struct containing channels used for communication between your worker, runtime and result processor

type Config

type Config struct {
	ResultProcessor            ResultProcessor
	RuntimeProcessor           RuntimeProcessor
	RuntimeProcessingFrequency time.Duration
	LockProvider               LockProvider
	Worker                     Worker
	SummaryBuffer              int
}

Config represents a configuration of a job. You can either create your own, or use CreateJob function which will initialize basic configuration.

func CreateJob

func CreateJob() Config

CreateJob creates a basic empty configuration with some defaults.

func (*Config) Run

func (config *Config) Run()

Run starts the job

func (*Config) SetWorker

func (config *Config) SetWorker(worker Worker)

SetWorker is used to set Worker.

func (*Config) WithLockProvider

func (config *Config) WithLockProvider(lockProvider LockProvider)

WithLockProvider is used to set LockProvider.

func (*Config) WithResultProcessor

func (config *Config) WithResultProcessor(processor ResultProcessor)

WithResultProcessor is used to set ResultProcessor.

Example

package main

import (
	"fmt"
	"github.com/fossapps/captain"
	"strconv"
)

func main() {
	job := captain.CreateJob()
	job.SetWorker(func(channels captain.CommChan) {
		channels.Result <- "Total Items: " + strconv.Itoa(80)
	})
	job.WithResultProcessor(func(results []string) {
		fmt.Printf("%+v\n", results[0])
	})
	job.Run()
}

Output:

Total Items: 80

func (*Config) WithRuntimeProcessingFrequency

func (config *Config) WithRuntimeProcessingFrequency(frequency time.Duration)

WithRuntimeProcessingFrequency is used to set how frequently RuntimeProcessor is called.

func (*Config) WithRuntimeProcessor

func (config *Config) WithRuntimeProcessor(processor RuntimeProcessor)

WithRuntimeProcessor is used to set the RuntimeProcessor.

type LockProvider

type LockProvider interface {
	Acquire() error
	Release() error
}

LockProvider is used if we need to make sure that two jobs aren't running at the same time.

type ResultProcessor

type ResultProcessor func(results []string)

ResultProcessor is called after execution of worker.

type RuntimeProcessor

type RuntimeProcessor func(tick time.Time, message string, startTime time.Time)

RuntimeProcessor gets called every `RuntimeProcessingFrequency` duration.

type Worker

type Worker func(channels CommChan)

Worker is called and is expected to do the real work.