flow

Documentation

Overview

Package flow implements the Flow component graph system. Flow configuration files are parsed from River, which contain a listing of components to run.

Components

Each component has a set of arguments (River attributes and blocks) and optionally a set of exported fields. Components can reference the exports of other components using River expressions.

See the top-level component package for more information on components, and subpackages for defined components.

Component Health

A component will have various health states during its lifetime:

1. Unknown: The initial health state for new components.
2. Healthy: A healthy component
3. Unhealthy: An unhealthy component.
4. Exited: A component which is no longer running.

Health states are paired with a time for when the health state was generated and a message providing more detail for the health state.

Components can report their own health states. The health state reported by a component is merged with the Flow-level health of that component: an error when evaluating the configuration for a component will always be reported as unhealthy until the next successful evaluation.

Component Evaluation

The process of converting the River block associated with a component into the appropriate Go struct is called "component evaluation."

Components are only evaluated after all components they reference have been evaluated; cyclic dependencies are invalid.

If a component updates its Exports at runtime, other components which directly or indirectly reference the updated component will have their Arguments re-evaluated.

The arguments and exports for a component will be left in their last valid state if a component shuts down or is given an invalid config. This prevents a domino effect of a single failed component taking down other components which are otherwise healthy.

Index

• type Argument
• type ComponentHealth
• type ComponentInfo
• type File
  • func ReadFile(name string, bb []byte) (*File, error)
• type Flow
  • func New(o Options) *Flow
  • func (f *Flow) ComponentHandler() http.HandlerFunc
  • func (c *Flow) ComponentInfos() []*ComponentInfo
  • func (f *Flow) ComponentJSON(w io.Writer, ci *ComponentInfo) error
  • func (c *Flow) LoadFile(file *File, args map[string]any) error
  • func (c *Flow) Ready() bool
  • func (c *Flow) Run(ctx context.Context)
• type Options

Types

type Argument

type Argument struct {
	// Name of the argument.
	Name string `river:",label"`

	// Whether the Argument must be provided when evaluating the file.
	Optional bool `river:"optional,attr,optional"`

	// Description for the Argument.
	Comment string `river:"comment,attr,optional"`

	// Default value for the argument.
	Default any `river:"default,attr,optional"`
}

An Argument is an input to a Flow module.

type ComponentHealth

type ComponentHealth struct {
	State       string    `json:"state"`
	Message     string    `json:"message"`
	UpdatedTime time.Time `json:"updatedTime"`
}

ComponentHealth represents the health of a component.

type ComponentInfo

type ComponentInfo struct {
	Name         string           `json:"name,omitempty"`
	Type         string           `json:"type,omitempty"`
	ID           string           `json:"id,omitempty"`
	Label        string           `json:"label,omitempty"`
	References   []string         `json:"referencesTo"`
	ReferencedBy []string         `json:"referencedBy"`
	Health       *ComponentHealth `json:"health"`
	Original     string           `json:"original"`
	Arguments    json.RawMessage  `json:"arguments,omitempty"`
	Exports      json.RawMessage  `json:"exports,omitempty"`
	DebugInfo    json.RawMessage  `json:"debugInfo,omitempty"`
}

ComponentInfo represents a component in flow.

type File

type File struct {
	Name string    // File name given to ReadFile.
	Node *ast.File // Raw File node.

	// Components holds the list of raw River AST blocks describing components.
	// The Flow controller can interpret them.
	Components   []*ast.BlockStmt
	ConfigBlocks []*ast.BlockStmt
}

File holds the contents of a parsed Flow file.

func ReadFile

func ReadFile(name string, bb []byte) (*File, error)

ReadFile parses the River file specified by bb into a File. name should be the name of the file used for reporting errors.

type Flow

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

Flow is the Flow system.

func New

func New(o Options) *Flow

New creates and starts a new Flow controller. Call Close to stop the controller.

func (*Flow) ComponentHandler

func (f *Flow) ComponentHandler() http.HandlerFunc

ComponentHandler returns an http.HandlerFunc which will delegate all requests to a component named by the first path segment

func (*Flow) ComponentInfos

func (c *Flow) ComponentInfos() []*ComponentInfo

ComponentInfos returns the component infos.

func (*Flow) ComponentJSON

func (f *Flow) ComponentJSON(w io.Writer, ci *ComponentInfo) error

ComponentJSON returns the json representation of the flow component.

func (*Flow) LoadFile

func (c *Flow) LoadFile(file *File, args map[string]any) error

LoadFile synchronizes the state of the controller with the current config file. Components in the graph will be marked as unhealthy if there was an error encountered during Load.

The controller will only start running components after Load is called once without any configuration errors.

func (*Flow) Ready

func (c *Flow) Ready() bool

Ready returns whether the Flow controller has finished its initial load.

func (*Flow) Run

func (c *Flow) Run(ctx context.Context)

Run starts the Flow controller, blocking until the provided context is canceled. Run must only be called once.

type Options

type Options struct {
	// ControllerID is an identifier used to represent the controller.
	// ControllerID is used to generate a globally unique display name for
	// components in a binary where multiple controllers are used.
	//
	// If running multiple Flow controllers, each controller must have a
	// different value for ControllerID to be able to differentiate between
	// components in telemetry data.
	ControllerID string

	// LogSink to use for controller logs and components. A no-op logger will be
	// created if this is nil.
	LogSink *logging.Sink

	// Tracer for components to use. A no-op tracer will be created if this is
	// nil.
	Tracer *tracing.Tracer

	// Clusterer for implementing distributed behavior among components running
	// on different nodes.
	Clusterer *cluster.Clusterer

	// Directory where components can write data. Constructed components will be
	// given a subdirectory of DataPath using the local ID of the component.
	//
	// If running multiple Flow controllers, each controller must have a
	// different value for DataPath to prevent components from colliding.
	DataPath string

	// Reg is the prometheus register to use
	Reg prometheus.Registerer

	// HTTPPathPrefix is the path prefix given to managed components. May be
	// empty. When provided, it should be an absolute path.
	//
	// Components will be given a path relative to HTTPPathPrefix using their
	// local ID.
	//
	// If running multiple Flow controllers, each controller must have a
	// different value for HTTPPathPrefix to prevent components from colliding.
	HTTPPathPrefix string

	// HTTPListenAddr is the base address (host:port) where component APIs are
	// exposed to other components.
	HTTPListenAddr string

	// OnExportsChange is called when the exports of the controller change.
	// Exports are controlled by "export" configuration blocks. If
	// OnExportsChange is nil, export configuration blocks are not allowed in the
	// loaded config file.
	OnExportsChange func(exports map[string]any)

	// DialFunc is a function to use for components to properly connect to
	// HTTPListenAddr. If nil, DialFunc defaults to (&net.Dialer{}).DialContext.
	DialFunc func(ctx context.Context, network, address string) (net.Conn, error)
}

Options holds static options for a flow controller.