controller

Documentation

Index

• Variables
• func WriteComponent(cn *ComponentNode, debugInfo bool) *builder.Block
• type ComponentGlobals
• type ComponentID
  • func BlockComponentID(b *ast.BlockStmt) ComponentID
  • func (id ComponentID) Equals(other ComponentID) bool
  • func (id ComponentID) String() string
• type ComponentNode
  • func NewComponentNode(globals ComponentGlobals, b *ast.BlockStmt) *ComponentNode
  • func (cn *ComponentNode) Arguments() component.Arguments
  • func (cn *ComponentNode) CurrentHealth() component.Health
  • func (cn *ComponentNode) DebugInfo() interface{}
  • func (cn *ComponentNode) Evaluate(scope *vm.Scope) error
  • func (cn *ComponentNode) Exports() component.Exports
  • func (cn *ComponentNode) ID() ComponentID
  • func (cn *ComponentNode) NodeID() string
  • func (cn *ComponentNode) Run(ctx context.Context) error
  • func (cn *ComponentNode) UpdateBlock(b *ast.BlockStmt)
• type Loader
  • func NewLoader(globals ComponentGlobals) *Loader
  • func (l *Loader) Apply(parentScope *vm.Scope, blocks []*ast.BlockStmt) diag.Diagnostics
  • func (l *Loader) Components() []*ComponentNode
  • func (l *Loader) EvaluateDependencies(parentScope *vm.Scope, c *ComponentNode)
  • func (l *Loader) Graph() *dag.Graph
  • func (l *Loader) Variables() map[string]interface{}
  • func (l *Loader) WriteBlocks(debugInfo bool) []*builder.Block
• type Queue
  • func NewQueue() *Queue
  • func (q *Queue) Chan() <-chan struct{}
  • func (q *Queue) Enqueue(c *ComponentNode)
  • func (q *Queue) TryDequeue() *ComponentNode
• type Reference
  • func ComponentReferences(cn *ComponentNode, g *dag.Graph) ([]Reference, diag.Diagnostics)
• type RunnableNode
• type Scheduler
  • func NewScheduler() *Scheduler
  • func (s *Scheduler) Close() error
  • func (s *Scheduler) Synchronize(rr []RunnableNode) error
• type Traversal

Variables

var ErrUnevaluated = errors.New("managed component not built")

ErrUnevaluated is returned if ComponentNode.Run is called before a managed component is built.

Functions

func WriteComponent

func WriteComponent(cn *ComponentNode, debugInfo bool) *builder.Block

WriteComponent generates a token/builder Block from a component. Health and debug info will be included if debugInfo is true.

Types

type ComponentGlobals

type ComponentGlobals struct {
	Logger          log.Logger              // Logger shared between all managed components.
	DataPath        string                  // Shared directory where component data may be stored
	OnExportsChange func(cn *ComponentNode) // Invoked when the managed component updated its exports
	Registerer      prometheus.Registerer   // Registerer for serving agent and component metrics
}

ComponentGlobals are used by ComponentNodes to build managed components. All ComponentNodes should use the same ComponentGlobals.

type ComponentID

type ComponentID []string

ComponentID is a fully-qualified name of a component. Each element in ComponentID corresponds to a fragment of the period-delimited string; "remote.http.example" is ComponentID{"remote", "http", "example"}.

func BlockComponentID

func BlockComponentID(b *ast.BlockStmt) ComponentID

BlockComponentID returns the ComponentID specified by an River block.

func (ComponentID) Equals

func (id ComponentID) Equals(other ComponentID) bool

Equals returns true if id == other.

func (ComponentID) String

func (id ComponentID) String() string

String returns the string representation of a component ID.

type ComponentNode

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

ComponentNode is a controller node which manages a user-defined component.

ComponentNode manages the underlying component and caches its current arguments and exports. ComponentNode manages the arguments for the component from a River block.

func NewComponentNode

func NewComponentNode(globals ComponentGlobals, b *ast.BlockStmt) *ComponentNode

NewComponentNode creates a new ComponentNode from an initial ast.BlockStmt. The underlying managed component isn't created until Evaluate is called.

func (*ComponentNode) Arguments

func (cn *ComponentNode) Arguments() component.Arguments

Arguments returns the current arguments of the managed component.

func (*ComponentNode) CurrentHealth

func (cn *ComponentNode) CurrentHealth() component.Health

CurrentHealth returns the current health of the ComponentNode.

The health of a ComponentNode is tracked from three parts, in descending precedence order:

1. Exited health from a call to Run()
2. Unhealthy status from last call to Evaluate
3. Health reported by the managed component (if any)
4. Latest health from Run() or Evaluate(), if the managed component does not report health.

func (*ComponentNode) DebugInfo

func (cn *ComponentNode) DebugInfo() interface{}

DebugInfo returns debugging information from the managed component (if any).

func (*ComponentNode) Evaluate

func (cn *ComponentNode) Evaluate(scope *vm.Scope) error

Evaluate updates the arguments for the managed component by re-evaluating its River block with the provided scope. The managed component will be built the first time Evaluate is called.

Evaluate will return an error if the River block cannot be evaluated or if decoding to arguments fails.

func (*ComponentNode) Exports

func (cn *ComponentNode) Exports() component.Exports

Exports returns the current set of exports from the managed component. Exports returns nil if the managed component does not have exports.

func (*ComponentNode) ID

func (cn *ComponentNode) ID() ComponentID

ID returns the component ID of the managed component from its River block.

func (*ComponentNode) NodeID

func (cn *ComponentNode) NodeID() string

NodeID implements dag.Node and returns the unique ID for this node. The NodeID is the string representation of the component's ID from its River block.

func (*ComponentNode) Run

func (cn *ComponentNode) Run(ctx context.Context) error

Run runs the managed component in the calling goroutine until ctx is canceled. Evaluate must have been called at least once without retuning an error before calling Run.

Run will immediately return ErrUnevaluated if Evaluate has never been called successfully. Otherwise, Run will return nil.

func (*ComponentNode) UpdateBlock

func (cn *ComponentNode) UpdateBlock(b *ast.BlockStmt)

UpdateBlock updates the River block used to construct arguments for the managed component. The new block isn't used until the next time Evaluate is invoked.

UpdateBlock will panic if the block does not match the component ID of the ComponentNode.

type Loader

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

The Loader builds and evaluates ComponentNodes from River blocks.

func NewLoader

func NewLoader(globals ComponentGlobals) *Loader

NewLoader creates a new Loader. Components built by the Loader will be built with co for their options.

func (*Loader) Apply

func (l *Loader) Apply(parentScope *vm.Scope, blocks []*ast.BlockStmt) diag.Diagnostics

Apply loads a new set of components into the Loader. Apply will drop any previously loaded component which is not described in the set of River blocks.

Apply will reuse existing components if there is an existing component which matches the component ID specified by any of the provided River blocks. Reused components will be updated to point at the new River block.

Apply will perform an evaluation of all loaded components before returning. The provided parentContext can be used to provide global variables and functions to components. A child context will be constructed from the parent to expose values of other components.

func (*Loader) Components

func (l *Loader) Components() []*ComponentNode

Components returns the current set of loaded components.

func (*Loader) EvaluateDependencies

func (l *Loader) EvaluateDependencies(parentScope *vm.Scope, c *ComponentNode)

EvaluateDependencies re-evaluates components which depend directly or indirectly on c. EvaluateDependencies should be called whenever a component updates its exports.

The provided parentContext can be used to provide global variables and functions to components. A child context will be constructed from the parent to expose values of other components.

func (*Loader) Graph

func (l *Loader) Graph() *dag.Graph

Graph returns a copy of the DAG managed by the Loader.

func (*Loader) Variables

func (l *Loader) Variables() map[string]interface{}

Variables returns the Variables the Loader exposes for other Flow components to reference.

func (*Loader) WriteBlocks

func (l *Loader) WriteBlocks(debugInfo bool) []*builder.Block

WriteBlocks returns a set of evaluated token/builder blocks for each loaded component. Components are returned in the order they were supplied to Apply (i.e., the original order from the config file) and not topological order.

Blocks will include health and debug information if debugInfo is true.

type Queue

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

Queue is an unordered queue of components.

Queue is intended for tracking components that have updated their Exports for later reevaluation.

func NewQueue

func NewQueue() *Queue

NewQueue returns a new unordered component queue.

func (*Queue) Chan

func (q *Queue) Chan() <-chan struct{}

Chan returns a channel which is written to when the queue is non-empty.

func (*Queue) Enqueue

func (q *Queue) Enqueue(c *ComponentNode)

Enqueue inserts a new component into the Queue. Enqueue is a no-op if the component is already in the Queue.

func (*Queue) TryDequeue

func (q *Queue) TryDequeue() *ComponentNode

TryDequeue dequeues a randomly queued component. TryDequeue will return nil if the queue is empty.

type Reference

type Reference struct {
	Target *ComponentNode // Component being referenced

	// Traversal describes which nested field relative to Target is being
	// accessed.
	Traversal Traversal
}

Reference describes an River expression reference to a ComponentNode.

func ComponentReferences

func ComponentReferences(cn *ComponentNode, g *dag.Graph) ([]Reference, diag.Diagnostics)

ComponentReferences returns the list of references a component is making to other components.

type RunnableNode

type RunnableNode interface {
	NodeID() string
	Run(ctx context.Context) error
}

RunnableNode is any dag.Node which can also be ran.

type Scheduler

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

Scheduler runs components.

func NewScheduler

func NewScheduler() *Scheduler

NewScheduler creates a new Scheduler. Call Synchronize to manage the set of components which are running.

Call Close to stop the Scheduler and all running components.

func (*Scheduler) Close

func (s *Scheduler) Close() error

Close stops the Scheduler and returns after all running goroutines have exited.

func (*Scheduler) Synchronize

func (s *Scheduler) Synchronize(rr []RunnableNode) error

Synchronize synchronizes the running components to those defined by rr.

New RunnableNodes will be launched as new goroutines. RunnableNodes already managed by Scheduler will be kept running, while running RunnableNodes that are not in rr will be shut down and removed.

Existing components will be restarted if they stopped since the previous call to Synchronize.

type Traversal

type Traversal []*ast.Ident

Traversal describes accessing a sequence of fields relative to a component. Traversal only include uninterrupted sequences of field accessors; for an expression "component.field_a.field_b.field_c[0].inner_field", the Traversal will be (field_a, field_b, field_c).