herd

README

🏁 herd

Herd is a Embedded Runnable DAG (H.E.R.D.). it aims to be a tiny library that allows to define arbitrary DAG, and associate job operations on them.

Why?

I've found couple of nice libraries (fx, or dag for instance), however none of them satisfied my constraints:

• Tiny
• Completely tested (TDD)
• Define jobs in a DAG, runs them in sequence, execute the ones that can be done in parallel (parallel topological sorting) in separate go routines
• Provide some sorta of similarity with systemd concepts

Usage

herd can be used as a library as such:

package main

import (
    "context"

    "github.com/spectrocloud-labs/herd"
)

func main() {

    // Generic usage
    g := herd.DAG()
    g.Add("name", ...)
    g.Run(context.TODO())

    // Example
    f := ""
    g.Add("foo", herd.WithCallback(func(ctx context.Context) error {
        f += "foo"
        // This executes after "bar" has ended successfully.
        return nil
    }), herd.WithDeps("bar"))

    g.Add("bar", herd.WithCallback(func(ctx context.Context) error {
        f += "bar"
        // This execute first
        return nil
    }))

    // Execute the DAG
    g.Run(context.Background())
    // f is "barfoo"
}

Documentation

Index

• type Graph
  • func DAG(opts ...GraphOption) *Graph
  • func (g *Graph) Add(name string, opts ...OpOption) error
  • func (g *Graph) Analyze() (graph [][]GraphEntry)
  • func (g *Graph) Run(ctx context.Context) error
  • func (g *Graph) State(name string) GraphEntry
• type GraphEntry
• type GraphOption
• type OpOption
  • func ConditionalOption(condition func() bool, op OpOption) OpOption
  • func EnableIf(conditional func() bool) OpOption
  • func IfElse(condition bool, op, noOp OpOption) OpOption
  • func WithCallback(fn ...func(context.Context) error) OpOption
  • func WithDeps(deps ...string) OpOption
  • func WithWeakDeps(deps ...string) OpOption
• type OpState

Types

type Graph

type Graph struct {
	*depgraph.Graph
	// contains filtered or unexported fields
}

Graph represents a directed graph.

func DAG

func DAG(opts ...GraphOption) *Graph

DAG creates a new instance of a runnable Graph. A DAG is a Direct Acyclic Graph. The graph is walked, and depending on the dependencies it will run the jobs as requested. The Graph can be explored with `Analyze()`, extended with new operations with Add(), and finally being run with Run(context.Context).

func (*Graph) Add

func (g *Graph) Add(name string, opts ...OpOption) error

Add adds a new operation to the graph. Requires a name (string), and accepts a list of options.

func (*Graph) Analyze

func (g *Graph) Analyze() (graph [][]GraphEntry)

Analyze returns the DAG and the Graph in the execution order. It will also return eventual updates if called after Run().

func (*Graph) Run

func (g *Graph) Run(ctx context.Context) error

Run starts the jobs defined in the DAG with a context. It returns error in case of failure.

func (*Graph) State

func (g *Graph) State(name string) GraphEntry

Stage returns the DAG item state. Note: it locks to be thread-safe.

type GraphEntry

type GraphEntry struct {
	WithCallback                       bool
	Background                         bool
	Callback                           []func(context.Context) error
	Error                              error
	Ignored, Fatal, WeakDeps, Executed bool
	Name                               string
	Dependencies                       []string
	WeakDependencies                   []string
}

GraphEntry is the external representation of the operation to execute (OpState).

type GraphOption

type GraphOption func(g *Graph)

GraphOption it's the option for the DAG graph.

var CollectOrphans GraphOption = func(g *Graph) {
	g.collectOrphans = true
}

CollectOrphans enables orphan job collection.

var EnableInit GraphOption = func(g *Graph) {
	g.init = true
}

EnableInit enables an Init jobs that takes paternity of orphan jobs without dependencies.

type OpOption

type OpOption func(string, *OpState, *Graph) error

OpOption defines the operation settings.

var Background OpOption = func(key string, os *OpState, g *Graph) error {
	os.background = true
	return nil
}

Background runs the operation in the background.

var FatalOp OpOption = func(key string, os *OpState, g *Graph) error {
	os.fatal = true
	return nil
}

FatalOp makes the operation fatal. Any error will make the DAG to stop and return the error immediately.

var NoOp OpOption = func(s string, os *OpState, g *Graph) error { return nil }

var WeakDeps OpOption = func(key string, os *OpState, g *Graph) error {
	os.weak = true
	return nil
}

WeakDeps sets all the dependencies of the job as "weak". Any failure of the jobs which depends on won't impact running the job. By default, a failure job will make also fail all the children - this is option disables this behavor and make the child start too.

func ConditionalOption

func ConditionalOption(condition func() bool, op OpOption) OpOption

ConditionalOption defines an option that is enabled only if the conditional callback returns true.

func EnableIf

func EnableIf(conditional func() bool) OpOption

EnableIf defines an operation dependency. Dependencies can be expressed as a string. Note: before running the DAG you must define all the operations.

func IfElse

func IfElse(condition bool, op, noOp OpOption) OpOption

IfElse defines options that are enabled if the condition passess or not It is just syntax sugar.

func WithCallback

func WithCallback(fn ...func(context.Context) error) OpOption

WithCallback associates a callback to the operation to be executed when the DAG is walked-by.

func WithDeps

func WithDeps(deps ...string) OpOption

WithDeps defines an operation dependency. Dependencies can be expressed as a string. Note: before running the DAG you must define all the operations.

func WithWeakDeps

func WithWeakDeps(deps ...string) OpOption

WithWeakDeps defines dependencies that doesn't prevent the op to trigger.

type OpState

type OpState struct {
	sync.Mutex
	// contains filtered or unexported fields
}