etcd

Documentation

Overview

Package etcd contains types which enable etcd as a repository backend for the adagio workflow engine.

Keyspace Design (etcd internals)

Namespaces: v0/runs/ : runs namespace v0/nodes/ : nodes namespace v0/states/ : states namespace

Objects: v0/agents/<agent-id> : Agent{} serialized agent object (leased) v0/runs/<run-id> : Run{} serialized run object v0/nodes/<run-id>/node/<name> : Node{} serialized node object v0/states/<state>/run/<run-id>/node/<name> : "" empty string to identify state

States: waiting, ready, running, completed

Index

• type Option
  • func ForList(name string) Option
  • func WithNamespace(ns string) Option
• type Options
  • func (o Options) Apply(r *Repository)
• type Repository
  • func New(kv clientv3.KV, watcher clientv3.Watcher, leaser clientv3.Lease, ...) *Repository
  • func (r *Repository) ClaimNode(ctx context.Context, runID, name string, claim *adagio.Claim) (node *adagio.Node, claimed bool, err error)
  • func (r *Repository) FinishNode(ctx context.Context, runID, name string, result *adagio.Node_Result, ...) (err error)
  • func (r *Repository) InspectRun(ctx context.Context, id string) (*adagio.Run, error)
  • func (r *Repository) ListAgents(ctx context.Context) (agents []*adagio.Agent, err error)
  • func (r *Repository) ListRuns(ctx context.Context, req controlplane.ListRequest) (runs []*adagio.Run, err error)
  • func (r *Repository) StartRun(ctx context.Context, spec *adagio.GraphSpec) (run *adagio.Run, err error)
  • func (r *Repository) Stats(ctx context.Context) (*adagio.Stats, error)
  • func (r *Repository) Subscribe(ctx context.Context, a *adagio.Agent, events chan<- *adagio.Event, ...) error
  • func (r *Repository) UnsubscribeAll(ctx context.Context, a *adagio.Agent, ch chan<- *adagio.Event) error

Types

type Option

type Option func(*Repository)

Option is a functional option for repository

func ForList

func ForList(name string) Option

ForList constructs a repository client for a specific list (other than the default one)

func WithNamespace

func WithNamespace(ns string) Option

WithNamespace configures the etcd client to use a particular top-level prefix

type Options

type Options []Option

Options is a slice of Option

func (Options) Apply

func (o Options) Apply(r *Repository)

Apply calls each option on r in turn

type Repository

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

Repository is the etcd backed implementation of an adagio Repository type (control plane and agent) It facilitates a distributed set of agents and control plane consumers using etcd as the store and etcd v3 transactions to ensure a consistent behavior It adheres to the repository test harness

func New

func New(kv clientv3.KV, watcher clientv3.Watcher, leaser clientv3.Lease, opts ...Option) *Repository

New constructs and configure a new repository service from the provided etcd client and a set of function options

func (*Repository) ClaimNode

func (r *Repository) ClaimNode(ctx context.Context, runID, name string, claim *adagio.Claim) (node *adagio.Node, claimed bool, err error)

ClaimNode attempts to claim a node identified by name for a specified run ID and providing a unique claim Given the node is found and the claim is successful the node is returned and the claimed boolean with be true

func (*Repository) FinishNode

func (r *Repository) FinishNode(ctx context.Context, runID, name string, result *adagio.Node_Result, claim *adagio.Claim) (err error)

FinishNode records a result for a node identified by name for a specified run ID and given a unique and active claim

func (*Repository) InspectRun

func (r *Repository) InspectRun(ctx context.Context, id string) (*adagio.Run, error)

InspectRun takes an ID and returns the associated Run if found within etcd

func (*Repository) ListAgents

func (r *Repository) ListAgents(ctx context.Context) (agents []*adagio.Agent, err error)

ListAgents returns at agents recorded within etcd at the time of the call

func (*Repository) ListRuns

func (r *Repository) ListRuns(ctx context.Context, req controlplane.ListRequest) (runs []*adagio.Run, err error)

ListRuns returns a list of runs given a set of predicates Given no start time is provided now is assumed Given no finish time is provided epoch is assumed Given no limit is provided all runs are returned

func (*Repository) StartRun

func (r *Repository) StartRun(ctx context.Context, spec *adagio.GraphSpec) (run *adagio.Run, err error)

StartRun takes a graph specification and instantiates it within etcd an returns the resulting Run representation

func (*Repository) Stats

func (r *Repository) Stats(ctx context.Context) (*adagio.Stats, error)

Stats returns the state of world represented as counts within the etcd database

func (*Repository) Subscribe

func (r *Repository) Subscribe(ctx context.Context, a *adagio.Agent, events chan<- *adagio.Event, typ ...adagio.Event_Type) error

Subscribe registers the agent as a subscriber and sends events regarding node readiness and orphanage on the provided events channel

func (*Repository) UnsubscribeAll

func (r *Repository) UnsubscribeAll(ctx context.Context, a *adagio.Agent, ch chan<- *adagio.Event) error

UnsubscribeAll unsubscribes the provided agent and channel as a listener