controller

package
v0.43.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 11, 2024 License: Apache-2.0 Imports: 35 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

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

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

Functions

func ExtractImportAndDeclare 

func ExtractImportAndDeclare(componentName string) (importNamespace, customComponentName string)

ExtractImportAndDeclare extracts an importNamespace and a customComponentName from a componentName. TODO: this function assumes that CustomComponentNames and ImportNamespaces don't contain "." In other words, there are two possible scenarios: - [customComponentName] LABEL -> instance of a local declare - [importNamespace].[customComponentName] LABEL -> instance of an imported declare To address this limitation in the future, we will need to find a different approach to identify the importNamespaces and CustomComponentNames.

Types

type ApplyOptions 

type ApplyOptions struct {
	Args map[string]any // input values of a module (nil for the root module)

	// TODO: rename ComponentBlocks because it also contains services
	ComponentBlocks []*ast.BlockStmt // pieces of config that can be used to instantiate builtin components and services
	ConfigBlocks    []*ast.BlockStmt // pieces of config that can be used to instantiate config nodes
	DeclareBlocks   []*ast.BlockStmt // pieces of config that can be used as templates to instantiate custom components

	// CustomComponentRegistry holds custom component templates.
	// The definition of a custom component instantiated inside of the loaded config
	// should be passed via this field if it's not declared or imported in the config.
	CustomComponentRegistry *CustomComponentRegistry
}

ApplyOptions are options that can be provided when loading a new River config.

type ArgumentConfigNode 

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

func NewArgumentConfigNode 

func NewArgumentConfigNode(block *ast.BlockStmt, globals ComponentGlobals) *ArgumentConfigNode

NewArgumentConfigNode creates a new ArgumentConfigNode from an initial ast.BlockStmt. The underlying config isn't applied until Evaluate is called.

func (*ArgumentConfigNode) Block 

func (cn *ArgumentConfigNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed config node.

func (*ArgumentConfigNode) Default 

func (cn *ArgumentConfigNode) Default() any

func (*ArgumentConfigNode) Evaluate 

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

Evaluate implements BlockNode and updates the arguments for the managed config block by re-evaluating its River block with the provided scope. The managed config block 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 (*ArgumentConfigNode) Label 

func (cn *ArgumentConfigNode) Label() string

func (*ArgumentConfigNode) NodeID 

func (cn *ArgumentConfigNode) NodeID() string

NodeID implements dag.Node and returns the unique ID for the config node.

func (*ArgumentConfigNode) Optional 

func (cn *ArgumentConfigNode) Optional() bool

func (*ArgumentConfigNode) UpdateBlock 

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

UpdateBlock updates the River block used to construct arguments. 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 ArgumentConfigNode.

type BlockNode 

type BlockNode interface {
	dag.Node

	// Block returns the current block managed by the node.
	Block() *ast.BlockStmt

	// Evaluate updates the arguments by re-evaluating the River block with the provided scope.
	//
	// Evaluate will return an error if the River block cannot be evaluated or if
	// decoding to arguments fails.
	Evaluate(scope *vm.Scope) error

	// UpdateBlock updates the River block used to construct arguments.
	UpdateBlock(b *ast.BlockStmt)
}

BlockNode is a node in the DAG which manages a River block and can be evaluated.

func NewConfigNode 

func NewConfigNode(block *ast.BlockStmt, globals ComponentGlobals) (BlockNode, diag.Diagnostics)

NewConfigNode creates a new ConfigNode from an initial ast.BlockStmt. The underlying config isn't applied until Evaluate is called.

type BuiltinComponentNode 

type BuiltinComponentNode struct {
	OnBlockNodeUpdate func(cn BlockNode) // Informs controller that we need to reevaluate
	// contains filtered or unexported fields
}

BuiltinComponentNode is a controller node which manages a builtin component.

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

func NewBuiltinComponentNode 

func NewBuiltinComponentNode(globals ComponentGlobals, reg component.Registration, b *ast.BlockStmt) *BuiltinComponentNode

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

func (*BuiltinComponentNode) Arguments 

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

Arguments returns the current arguments of the managed component.

func (*BuiltinComponentNode) Block 

func (cn *BuiltinComponentNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed component.

func (*BuiltinComponentNode) Component 

func (cn *BuiltinComponentNode) Component() component.Component

Component returns the instance of the managed component. Component may be nil if the BuiltinComponentNode has not been successfully evaluated yet.

func (*BuiltinComponentNode) ComponentName 

func (cn *BuiltinComponentNode) ComponentName() string

ComponentName returns the component's type, i.e. `local.file.test` returns `local.file`.

func (*BuiltinComponentNode) CurrentHealth 

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

CurrentHealth returns the current health of the BuiltinComponentNode.

The health of a BuiltinComponentNode is determined by combining:

  1. Health from the call to Run().
  2. Health from the last call to Evaluate().
  3. Health reported from the component.

func (*BuiltinComponentNode) DebugInfo 

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

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

func (*BuiltinComponentNode) Evaluate 

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

Evaluate implements BlockNode and 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 (*BuiltinComponentNode) Exports 

func (cn *BuiltinComponentNode) 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 (*BuiltinComponentNode) ID 

func (cn *BuiltinComponentNode) ID() ComponentID

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

func (*BuiltinComponentNode) Label 

func (cn *BuiltinComponentNode) Label() string

Label returns the label for the block or "" if none was specified.

func (*BuiltinComponentNode) ModuleIDs 

func (cn *BuiltinComponentNode) ModuleIDs() []string

ModuleIDs returns the current list of modules that this component is managing.

func (*BuiltinComponentNode) NodeID 

func (cn *BuiltinComponentNode) 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 (*BuiltinComponentNode) Registration 

func (cn *BuiltinComponentNode) Registration() component.Registration

Registration returns the original registration of the component.

func (*BuiltinComponentNode) Run 

func (cn *BuiltinComponentNode) 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 returning an error before calling Run.

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

func (*BuiltinComponentNode) UpdateBlock 

func (cn *BuiltinComponentNode) 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 BuiltinComponentNode.

type ComponentGlobals 

type ComponentGlobals struct {
	Logger              *logging.Logger                        // Logger shared between all managed components.
	TraceProvider       trace.TracerProvider                   // Tracer shared between all managed components.
	DataPath            string                                 // Shared directory where component data may be stored
	MinStability        featuregate.Stability                  // Minimum allowed stability level for features
	OnBlockNodeUpdate   func(cn BlockNode)                     // Informs controller that we need to reevaluate
	OnExportsChange     func(exports map[string]any)           // Invoked when the managed component updated its exports
	Registerer          prometheus.Registerer                  // Registerer for serving agent and component metrics
	ControllerID        string                                 // ID of controller.
	NewModuleController func(id string) ModuleController       // Func to generate a module controller.
	GetServiceData      func(name string) (interface{}, error) // Get data for a service.
}

ComponentGlobals are used by BuiltinComponentNodes to build managed components. All BuiltinComponentNodes 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 interface {
	RunnableNode

	// CurrentHealth returns the current health of the component.
	CurrentHealth() component.Health

	// Arguments returns the current arguments of the managed component.
	Arguments() component.Arguments

	// Exports returns the current set of exports from the managed component.
	Exports() component.Exports

	// Label returns the component label.
	Label() string

	// ComponentName returns the name of the component.
	ComponentName() string

	// ID returns the component ID of the managed component from its River block.
	ID() ComponentID

	// ModuleIDs returns the current list of modules managed by the component.
	ModuleIDs() []string
}

ComponentNode is a generic representation of a Flow component.

type ComponentNodeManager 

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

ComponentNodeManager is responsible for creating new component nodes and obtaining the necessary information to run them.

func NewComponentNodeManager 

func NewComponentNodeManager(globals ComponentGlobals, componentReg ComponentRegistry) *ComponentNodeManager

NewComponentNodeManager creates a new ComponentNodeManager without custom component registry.

type ComponentRegistry 

type ComponentRegistry interface {
	// Get looks up a component by name. It returns an error if the component does not exist or its usage is restricted,
	// for example, because of the component's stability level.
	Get(name string) (component.Registration, error)
}

ComponentRegistry is a collection of registered components.

func NewDefaultComponentRegistry 

func NewDefaultComponentRegistry(minStability featuregate.Stability) ComponentRegistry

NewDefaultComponentRegistry creates a new ComponentRegistry which gets components registered to github.com/grafana/agent/component.

func NewRegistryMap 

func NewRegistryMap(
	minStability featuregate.Stability,
	registrations map[string]component.Registration,
) ComponentRegistry

NewRegistryMap creates a new ComponentRegistry which uses a map to store components. Currently, it is only used in tests.

type ConfigNodeMap 

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

ConfigNodeMap represents the config BlockNodes in their explicit types. This is helpful when validating node conditions specific to config node types.

func NewConfigNodeMap 

func NewConfigNodeMap() *ConfigNodeMap

NewConfigNodeMap will create an initial ConfigNodeMap. Append must be called to populate NewConfigNodeMap.

func (*ConfigNodeMap) Append 

func (nodeMap *ConfigNodeMap) Append(configNode BlockNode) diag.Diagnostics

Append will add a config node to the ConfigNodeMap. This will overwrite values on the ConfigNodeMap that are matched and previously set.

func (*ConfigNodeMap) Validate 

func (nodeMap *ConfigNodeMap) Validate(isInModule bool, args map[string]any) diag.Diagnostics

Validate wraps all validators for ConfigNodeMap.

func (*ConfigNodeMap) ValidateModuleConstraints 

func (nodeMap *ConfigNodeMap) ValidateModuleConstraints(isInModule bool) diag.Diagnostics

ValidateModuleConstraints will make sure config blocks with module constraints get followed.

func (*ConfigNodeMap) ValidateUnsupportedArguments 

func (nodeMap *ConfigNodeMap) ValidateUnsupportedArguments(args map[string]any) diag.Diagnostics

ValidateUnsupportedArguments will validate each provided argument is supported in the config.

type CustomComponent 

type CustomComponent interface {
	// LoadBody loads a River AST body into the CustomComponent. LoadBody can be called
	// multiple times, and called prior to [CustomComponent.Run].
	// customComponentRegistry provides custom component definitions for the loaded config.
	LoadBody(body ast.Body, args map[string]any, customComponentRegistry *CustomComponentRegistry) error

	// Run starts the CustomComponent. No components within the CustomComponent
	// will be run until Run is called.
	//
	// Run blocks until the provided context is canceled. The ID of a CustomComponent as defined in
	// ModuleController.NewCustomComponent will not be released until Run returns.
	Run(context.Context) error
}

CustomComponent is a controller for running components within a CustomComponent.

type CustomComponentNode 

type CustomComponentNode struct {
	OnBlockNodeUpdate func(cn BlockNode) // Informs controller that we need to reevaluate
	// contains filtered or unexported fields
}

CustomComponentNode is a controller node which manages a custom component.

CustomComponentNode manages the underlying custom component and caches its current arguments and exports.

func NewCustomComponentNode 

func NewCustomComponentNode(globals ComponentGlobals, b *ast.BlockStmt, getConfig getCustomComponentConfig) *CustomComponentNode

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

func (*CustomComponentNode) Arguments 

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

Arguments returns the current arguments of the managed custom component.

func (*CustomComponentNode) Block 

func (cn *CustomComponentNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed custom component.

func (*CustomComponentNode) ComponentName 

func (cn *CustomComponentNode) ComponentName() string

ComponentName returns the name of the component.

func (*CustomComponentNode) CurrentHealth 

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

CurrentHealth returns the current health of the CustomComponentNode.

The health of a CustomComponentNode is determined by combining:

  1. Health from the call to Run().
  2. Health from the last call to Evaluate().

func (*CustomComponentNode) Evaluate 

func (cn *CustomComponentNode) Evaluate(evalScope *vm.Scope) error

Evaluate implements BlockNode and updates the arguments by re-evaluating its River block with the provided scope and the custom component by retrieving the component definition from the corresponding import or declare node. The managed custom component will be built the first time Evaluate is called.

Evaluate will return an error if the River block cannot be evaluated, if decoding to arguments fails or if the custom component definition cannot be retrieved.

func (*CustomComponentNode) Exports 

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

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

func (*CustomComponentNode) ID 

func (cn *CustomComponentNode) ID() ComponentID

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

func (*CustomComponentNode) Label 

func (cn *CustomComponentNode) Label() string

Label returns the label for the block or "" if none was specified.

func (*CustomComponentNode) ModuleIDs 

func (cn *CustomComponentNode) ModuleIDs() []string

TODO: currently used by the component provider to access the components running within the custom components. Change it when getting rid of old modules.

func (*CustomComponentNode) NodeID 

func (cn *CustomComponentNode) 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 (*CustomComponentNode) Run 

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

func (*CustomComponentNode) UpdateBlock 

func (cn *CustomComponentNode) 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 CustomComponentNode.

type CustomComponentRegistry 

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

CustomComponentRegistry holds custom component definitions that are available in the context. The definitions are either imported, declared locally, or declared in a parent registry. Imported definitions are stored inside of the corresponding import registry.

func NewCustomComponentRegistry 

func NewCustomComponentRegistry(parent *CustomComponentRegistry) *CustomComponentRegistry

NewCustomComponentRegistry creates a new CustomComponentRegistry with a parent. parent can be nil.

type DeclareNode 

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

DeclareNode represents a declare block in the DAG.

func NewDeclareNode 

func NewDeclareNode(block *ast.BlockStmt) *DeclareNode

NewDeclareNode creates a new declare node with a content which will be loaded by custom components.

func (*DeclareNode) Block 

func (cn *DeclareNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed config node.

func (*DeclareNode) Evaluate 

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

Evaluate does nothing for this node.

func (*DeclareNode) Label 

func (cn *DeclareNode) Label() string

Label returns the label of the block.

func (*DeclareNode) NodeID 

func (cn *DeclareNode) NodeID() string

NodeID implements dag.Node and returns the unique ID for the config node.

func (*DeclareNode) UpdateBlock 

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

UpdateBlock updates the managed River block.

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

type DialFunc 

type DialFunc func(ctx context.Context, network, address string) (net.Conn, error)

DialFunc is a function to establish a network connection.

type ExportConfigNode 

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

func NewExportConfigNode 

func NewExportConfigNode(block *ast.BlockStmt, globals ComponentGlobals) *ExportConfigNode

NewExportConfigNode creates a new ExportConfigNode from an initial ast.BlockStmt. The underlying config isn't applied until Evaluate is called.

func (*ExportConfigNode) Block 

func (cn *ExportConfigNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed config node.

func (*ExportConfigNode) Evaluate 

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

Evaluate implements BlockNode and updates the arguments for the managed config block by re-evaluating its River block with the provided scope. The managed config block 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 (*ExportConfigNode) Label 

func (cn *ExportConfigNode) Label() string

func (*ExportConfigNode) NodeID 

func (cn *ExportConfigNode) NodeID() string

NodeID implements dag.Node and returns the unique ID for the config node.

func (*ExportConfigNode) UpdateBlock 

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

UpdateBlock updates the River block used to construct arguments. 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 ExportConfigNode.

func (*ExportConfigNode) Value 

func (cn *ExportConfigNode) Value() any

Value returns the value of the export.

type ImportConfigNode 

type ImportConfigNode struct {
	OnBlockNodeUpdate func(cn BlockNode) // notifies the controller or the parent for reevaluation
	// contains filtered or unexported fields
}

ImportConfigNode imports declare and import blocks via a managed import source. The imported declare are stored in importedDeclares. For every imported import block, the ImportConfigNode will create ImportConfigNode children. The children are evaluated and ran by the parent. When an ImportConfigNode receives new content from its source, it updates its importedDeclares and recreates its children. Then an update call is propagated to the root ImportConfigNode to inform the controller for reevaluation.

func NewImportConfigNode 

func NewImportConfigNode(block *ast.BlockStmt, globals ComponentGlobals, sourceType importsource.SourceType) *ImportConfigNode

NewImportConfigNode creates a new ImportConfigNode from an initial ast.BlockStmt. The underlying config isn't applied until Evaluate is called.

func (*ImportConfigNode) Block 

func (cn *ImportConfigNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed config node.

func (*ImportConfigNode) CurrentHealth 

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

CurrentHealth returns the current health of the ImportConfigNode.

The health of a ImportConfigNode is determined by combining:

  1. Health from the call to Run().
  2. Health from the last call to Evaluate().
  3. Health from the last call to OnContentChange().
  4. Health reported from the source.
  5. Health reported from the nested imports.

func (*ImportConfigNode) Equals 

func (cn *ImportConfigNode) Equals(other runner.Task) bool

We don't want to reuse previous running tasks. On every updates, the previous workers should be stopped and new ones should spawn.

func (*ImportConfigNode) Evaluate 

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

Evaluate implements BlockNode and evaluates the import source.

func (*ImportConfigNode) Hash 

func (cn *ImportConfigNode) Hash() uint64

func (*ImportConfigNode) ImportConfigNodesChildren 

func (cn *ImportConfigNode) ImportConfigNodesChildren() map[string]*ImportConfigNode

ImportConfigNodesChildren returns the ImportConfigNodesChildren of this ImportConfigNode.

func (*ImportConfigNode) ImportedDeclares 

func (cn *ImportConfigNode) ImportedDeclares() map[string]ast.Body

ImportedDeclares returns all declare blocks that it imported.

func (*ImportConfigNode) Label 

func (cn *ImportConfigNode) Label() string

func (*ImportConfigNode) NodeID 

func (cn *ImportConfigNode) NodeID() string

NodeID implements dag.Node and returns the unique ID for the config node.

func (*ImportConfigNode) Run 

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

Run runs the managed source and the import children until ctx is canceled. Evaluate must have been called at least once without returning an error before calling Run.

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

func (*ImportConfigNode) UpdateBlock 

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

UpdateBlock updates the River block used to construct arguments. 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 ImportConfigNode.

type Loader 

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

The Loader builds and evaluates ComponentNodes from River blocks.

func NewLoader 

func NewLoader(opts LoaderOptions) *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(options ApplyOptions) 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) Cleanup 

func (l *Loader) Cleanup(stopWorkerPool bool)

Cleanup unregisters any existing metrics and optionally stops the worker pool.

func (*Loader) Components 

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

Components returns the current set of loaded components.

func (*Loader) EvaluateDependants 

func (l *Loader) EvaluateDependants(ctx context.Context, updatedNodes []*QueuedNode)

EvaluateDependants sends nodes which depend directly on nodes in updatedNodes for evaluation to the workerPool. It should be called whenever nodes update their exports. It is beneficial to call EvaluateDependants with a batch of nodes, as it will enqueue the entire batch before the worker pool starts to evaluate them, resulting in smaller number of total evaluations when node updates are frequent. If the worker pool's queue is full, EvaluateDependants will retry with a backoff until it succeeds or until the ctx is cancelled.

func (*Loader) Graph 

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

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

func (*Loader) Imports 

func (l *Loader) Imports() map[string]*ImportConfigNode

Imports returns the current set of import nodes.

func (*Loader) OriginalGraph 

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

OriginalGraph returns a copy of the graph before Reduce was called. This can be used if you want to show a UI of the original graph before the reduce function was called.

func (*Loader) Services 

func (l *Loader) Services() []*ServiceNode

Services returns the current set of service nodes.

func (*Loader) Variables 

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

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

type LoaderOptions 

type LoaderOptions struct {
	// ComponentGlobals contains data to use when creating components.
	ComponentGlobals ComponentGlobals

	Services          []service.Service // Services to load into the DAG.
	Host              service.Host      // Service host (when running services).
	ComponentRegistry ComponentRegistry // Registry to search for components.
	WorkerPool        worker.Pool       // Worker pool to use for async tasks.
}

LoaderOptions holds options for creating a Loader.

type LoggingConfigNode 

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

func NewDefaultLoggingConfigNode 

func NewDefaultLoggingConfigNode(globals ComponentGlobals) *LoggingConfigNode

NewDefaultLoggingConfigNode creates a new LoggingConfigNode with nil block and eval. This will force evaluate to use the default logging options for this node.

func NewLoggingConfigNode 

func NewLoggingConfigNode(block *ast.BlockStmt, globals ComponentGlobals) *LoggingConfigNode

NewLoggingConfigNode creates a new LoggingConfigNode from an initial ast.BlockStmt. The underlying config isn't applied until Evaluate is called.

func (*LoggingConfigNode) Block 

func (cn *LoggingConfigNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed config node.

func (*LoggingConfigNode) Evaluate 

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

Evaluate implements BlockNode and updates the arguments for the managed config block by re-evaluating its River block with the provided scope. The managed config block 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 (*LoggingConfigNode) NodeID 

func (cn *LoggingConfigNode) NodeID() string

NodeID implements dag.Node and returns the unique ID for the config node.

func (*LoggingConfigNode) UpdateBlock 

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

UpdateBlock updates the River block used to construct arguments. 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 LoggingConfigNode.

type ModuleController 

type ModuleController interface {
	component.ModuleController

	// ModuleIDs returns the list of managed modules in unspecified order.
	ModuleIDs() []string

	// Creates a new custom component.
	NewCustomComponent(id string, export component.ExportFunc) (CustomComponent, error)
}

ModuleController is a lower-level interface for module controllers which allows probing for the list of managed modules.

type Queue 

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

Queue is a thread-safe, insertion-ordered set of nodes.

Queue is intended for tracking nodes that have been updated for later reevaluation.

func NewQueue 

func NewQueue() *Queue

NewQueue returns a new 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) DequeueAll 

func (q *Queue) DequeueAll() []*QueuedNode

DequeueAll removes all BlockNode from the queue and returns them.

func (*Queue) Enqueue 

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

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

type QueuedNode 

type QueuedNode struct {
	Node            BlockNode
	LastUpdatedTime time.Time
}

type Reference 

type Reference struct {
	Target BlockNode // BlockNode being referenced

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

Reference describes an River expression reference to a BlockNode.

func ComponentReferences 

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

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

type RunnableNode 

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

RunnableNode is any BlockNode which can also be run.

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 ServiceMap 

type ServiceMap map[string]service.Service

ServiceMap is a map of service name to services.

func NewServiceMap 

func NewServiceMap(services []service.Service) ServiceMap

NewServiceMap creates a ServiceMap from a slice of services.

func (ServiceMap) FilterByName 

func (sm ServiceMap) FilterByName(keepNames []string) ServiceMap

FilterByName creates a new ServiceMap where services that are not defined in keepNames are removed.

func (ServiceMap) Get 

func (sm ServiceMap) Get(name string) (svc service.Service, found bool)

Get looks up a service by name.

func (ServiceMap) List 

func (sm ServiceMap) List() []service.Service

List returns a slice of all the services.

type ServiceNode 

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

ServiceNode is a Flow DAG node which represents a running service.

func NewServiceNode 

func NewServiceNode(host service.Host, svc service.Service) *ServiceNode

NewServiceNode creates a new instance of a ServiceNode from an instance of a Service. The provided host is used when running the service.

func (*ServiceNode) Block 

func (sn *ServiceNode) Block() *ast.BlockStmt

Block implements BlockNode. It returns nil, since ServiceNodes don't have associated configs.

func (*ServiceNode) Definition 

func (sn *ServiceNode) Definition() service.Definition

Definition returns the service definition associated with the node.

func (*ServiceNode) Evaluate 

func (sn *ServiceNode) Evaluate(scope *vm.Scope) error

Evaluate implements BlockNode, evaluating the configuration for a service. Evalute returns an error if the service doesn't support being configured and the ServiceNode has an associated block from a call to UpdateBlock.

func (*ServiceNode) NodeID 

func (sn *ServiceNode) NodeID() string

NodeID returns the ID of the ServiceNode, which is equal to the service's name.

func (*ServiceNode) Run 

func (sn *ServiceNode) Run(ctx context.Context) error

func (*ServiceNode) Service 

func (sn *ServiceNode) Service() service.Service

Service returns the service instance associated with the node.

func (*ServiceNode) UpdateBlock 

func (sn *ServiceNode) UpdateBlock(b *ast.BlockStmt)

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

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

Call UpdateBlock with a nil block to remove the block associated with the ServiceNode.

type TracingConfigNode 

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

func NewDefaulTracingConfigNode 

func NewDefaulTracingConfigNode(globals ComponentGlobals) *TracingConfigNode

NewDefaulTracingConfigNode creates a new TracingConfigNode with nil block and eval. This will force evaluate to use the default tracing options for this node.

func NewTracingConfigNode 

func NewTracingConfigNode(block *ast.BlockStmt, globals ComponentGlobals) *TracingConfigNode

NewTracingConfigNode creates a new TracingConfigNode from an initial ast.BlockStmt. The underlying config isn't applied until Evaluate is called.

func (*TracingConfigNode) Block 

func (cn *TracingConfigNode) Block() *ast.BlockStmt

Block implements BlockNode and returns the current block of the managed config node.

func (*TracingConfigNode) Evaluate 

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

Evaluate implements BlockNode and updates the arguments for the managed config block by re-evaluating its River block with the provided scope. The managed config block 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 (*TracingConfigNode) NodeID 

func (cn *TracingConfigNode) NodeID() string

NodeID implements dag.Node and returns the unique ID for the config node.

func (*TracingConfigNode) UpdateBlock 

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

UpdateBlock updates the River block used to construct arguments. 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 LoggingConfigNode.

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).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.