dag

Documentation

Overview

Dag provides a way to describe a directed acyclic graph of work to be done. It starts with a root node, then you add nodes to it. Outputs are automatically connected to subsequent inputs.

Index

• Variables
• func AllNodesFinished(ctx context.Context, nodes []Node[IOSpec]) (bool, error)
• func GetDagStartTime[T any](ctx context.Context, dag []Node[T]) (time.Time, error)
• func GetEndTimeIfDagComplete[T any](ctx context.Context, dag []Node[T]) (time.Time, error)
• type ExecutionInfo
• type IOSpec
  • func NewIOSpec(name, id, value, path string, root bool, context string) IOSpec
  • func NilFunc(ctx context.Context, inputs []IOSpec, statusChan chan NodeResult) []IOSpec
• type Node
  • func FilterForRootNodes(ctx context.Context, dags []Node[IOSpec]) ([]Node[IOSpec], error)
  • func NewNode(ctx context.Context, persistence db.NodePersistence, ...) (Node[IOSpec], error)
  • func NodeMapToList[T any](dags map[T]Node[IOSpec]) (nodes []Node[IOSpec])
• type NodeExecutor
  • func NewNodeExecutor[T any](ctx context.Context, analytics analytics.AnalyticsRepository) (NodeExecutor[T], error)
• type NodeMetadata
• type NodeRepresentation
• type NodeResult
• type NodeSpec
• type NodeStore
  • func NewNodeStore(ctx context.Context, p db.NodePersistence, wr WorkRepository[IOSpec]) (NodeStore[IOSpec], error)
• type Status
  • func (s Status) String() string
• type Work
• type WorkRepository
  • func NewInMemWorkRepository[T any]() WorkRepository[T]

Variables

var (
	ErrWorkNotFound      = fmt.Errorf("work not found")
	ErrWorkAlreadyExists = fmt.Errorf("work already exists")
)

var (
	MaxTime = time.Unix(1<<63-62135596801, 999999999)
)

Functions

func AllNodesFinished

func AllNodesFinished(ctx context.Context, nodes []Node[IOSpec]) (bool, error)

func GetDagStartTime

func GetDagStartTime[T any](ctx context.Context, dag []Node[T]) (time.Time, error)

func GetEndTimeIfDagComplete

func GetEndTimeIfDagComplete[T any](ctx context.Context, dag []Node[T]) (time.Time, error)

Types

type ExecutionInfo

type ExecutionInfo struct {
	ID     string // External ID of the job
	Stdout string // Stdout of the job
	Stderr string // Stderr of the job
	Status string // Status of the job
}

type IOSpec

type IOSpec interface {
	NodeName() string // The node ID from the graph spec (not the internal node ID)
	ID() string       // The node's input ID from the graph spec
	CID() string
	Context() string
	Path() string
	IsRoot() bool
}

IOSpec is a generic input/output specification for a DAG

func NewIOSpec

func NewIOSpec(name, id, value, path string, root bool, context string) IOSpec

func NilFunc

func NilFunc(ctx context.Context, inputs []IOSpec, statusChan chan NodeResult) []IOSpec

type Node

type Node[T any] interface {
	ID() int32
	Get(context.Context) (NodeRepresentation[T], error)
	AddChild(context.Context, Node[T]) error
	AddParentChildRelationship(context.Context, Node[T]) error
	AddParent(context.Context, Node[T]) error
	AddInput(context.Context, T) error
	AddOutput(context.Context, T) error
	SetMetadata(context.Context, NodeMetadata) error
	SetResults(context.Context, NodeResult) error
}

func FilterForRootNodes

func FilterForRootNodes(ctx context.Context, dags []Node[IOSpec]) ([]Node[IOSpec], error)

func NewNode

func NewNode(ctx context.Context, persistence db.NodePersistence, workRepo WorkRepository[IOSpec], n NodeSpec[IOSpec]) (Node[IOSpec], error)

func NodeMapToList

func NodeMapToList[T any](dags map[T]Node[IOSpec]) (nodes []Node[IOSpec])

type NodeExecutor

type NodeExecutor[T any] interface {
	Execute(ctx context.Context, executionId uuid.UUID, node Node[T])
}

func NewNodeExecutor

func NewNodeExecutor[T any](ctx context.Context, analytics analytics.AnalyticsRepository) (NodeExecutor[T], error)

type NodeMetadata

type NodeMetadata struct {
	CreatedAt time.Time
	StartedAt time.Time
	EndedAt   time.Time
	Status    string // Status of the execution
}

NodeMetadata contains metadata about a node

type NodeRepresentation

type NodeRepresentation[T any] struct {
	Id          int32        // Keep track of the node's ID, useful during debugging
	Name        string       // Name of the node
	QueueItemID uuid.UUID    // ID of the queue item
	Work        Work[T]      // The work to be done by the node
	Children    []Node[T]    // Children of the node
	Parents     []Node[T]    // Parents of the node
	Inputs      []T          // Input data
	Outputs     []T          // Output data (which is fed into the inputs of its children)
	Metadata    NodeMetadata // Metadata about the node
	Results     NodeResult   // Result of the node
}

type NodeResult

type NodeResult struct {
	ID      string // External ID of the execution
	StdOut  string // Stdout of the execution
	StdErr  string // Stderr of the execution
	Skipped bool   // Whether the execution was skipped
}

type NodeSpec

type NodeSpec[T any] struct {
	OwnerID uuid.UUID
	Name    string
	Work    Work[T]
}

type NodeStore

type NodeStore[T any] interface {
	NewNode(context.Context, NodeSpec[T]) (Node[T], error)
	GetNode(context.Context, int32) (Node[T], error)
}

func NewNodeStore

func NewNodeStore(ctx context.Context, p db.NodePersistence, wr WorkRepository[IOSpec]) (NodeStore[IOSpec], error)

type Status

type Status int64

const (
	Queued Status = iota
	Started
	Finished
)

func (Status) String

func (s Status) String() string

type Work

type Work[T any] func(ctx context.Context, inputs []T, statusChan chan NodeResult) []T

Work is shorthand for a function that accepts inputs and returns outputs.

type WorkRepository

type WorkRepository[T any] interface {
	Get(context.Context, int32) (Work[T], error)
	Set(context.Context, int32, Work[T]) error
}

func NewInMemWorkRepository

func NewInMemWorkRepository[T any]() WorkRepository[T]