component

Documentation

Overview

Package component describes the interfaces which Flow components implement.

A Flow component is a distinct piece of business logic that accepts inputs (Arguments) for its configuration and can optionally export a set of outputs (Exports).

Arguments and Exports do not need to be static for the lifetime of a component. A component will be given a new Config if the runtime configuration changes. A component may also update its Exports throughout its lifetime, such as a component which outputs the current day of the week.

Components are built by users with River, where they can use River expressions to refer to any input or exported field from other components. This allows users to connect components together to declaratively form a pipeline.

Defining Arguments and Exports structs

Arguments and Exports implemented by new components must be able to be encoded to and from River. "river" struct field tags are used for encoding; refer to the package documentation at pkg/river for a description of how to write these tags.

The set of River element names of a given component's Arguments and Exports types must not overlap. Additionally, the following River field and block names are reserved for use by the Flow controller:

• for_each
• enabled
• health
• debug

Default values for Arguments may be provided by implementing river.Unmarshaler.

Arguments and Exports immutability

Arguments passed to a component should be treated as immutable, as memory can be shared between components as an optimization. Components should make copies for fields they need to modify. An exception to this is for fields which are expected to be mutable (e.g., interfaces which expose a goroutine-safe API).

Similarly, Exports and the fields within Exports must be considered immutable after they are written for the same reason.

Mapping River strings to custom types

Custom encoding and decoding of fields is available by implementing encoding.TextMarshaler and encoding.TextUnmarshaler. Types implementing these interfaces will be represented as strings in River.

Component registration

Components are registered globally by calling Register. These components are then made available by including them in the import path. The "all" child package imports all known component packages and should be updated when creating a new one.

Index

• func Register(r Registration)
• type Arguments
• type Component
• type DebugComponent
• type Exports
• type Health
• type HealthComponent
• type HealthType
  • func (ht HealthType) MarshalText() (text []byte, err error)
  • func (ht HealthType) String() string
  • func (ht *HealthType) UnmarshalText(text []byte) error
• type Options
• type Registration
  • func Get(name string) (Registration, bool)
  • func (r Registration) CloneArguments() Arguments

Functions

func Register

func Register(r Registration)

Register registers a component. Register will panic if the name is in use by another component, if the name is invalid, or if the component name has a suffix length mismatch with an existing component.

Types

type Arguments

type Arguments interface{}

The Arguments contains the input fields for a specific component, which is unmarshaled from River.

Refer to the package documentation for details around how to build proper Arguments implementations.

type Component

type Component interface {
	// Run starts the component, blocking until ctx is canceled or the component
	// suffers a fatal error. Run is guaranteed to be called exactly once per
	// Component.
	//
	// Implementations of Component should perform any necessary cleanup before
	// returning from Run.
	Run(ctx context.Context) error

	// Update provides a new Config to the component. The type of newConfig will
	// always match the struct type which the component registers.
	//
	// Update will be called concurrently with Run. The component must be able to
	// gracefully handle updating its config will still running.
	//
	// An error may be returned if the provided config is invalid.
	Update(args Arguments) error
}

Component is the base interface for a Flow component. Components may implement extension interfaces (named <Extension>Component) to implement extra known behavior.

type DebugComponent

type DebugComponent interface {
	Component

	// DebugInfo returns the current debug information of the component. May
	// return nil if there is no debug info to currently report. The result of
	// DebugInfo must be encodable to River like Arguments and Exports.
	//
	// Values from DebugInfo are not exposed to other components for use in
	// expressions.
	//
	// DebugInfo must be safe for calling concurrently.
	DebugInfo() interface{}
}

DebugComponent is an extension interface for components which can report debugging information upon request.

type Exports

type Exports interface{}

Exports contains the current set of outputs for a specific component, which is then marshaled to River.

Refer to the package documentation for details around how to build proper Exports implementations.

type Health

type Health struct {
	// The specific health value.
	Health HealthType `river:"state,attr"`

	// An optional message to describe the health; useful to say why a component
	// is unhealthy.
	Message string `river:"message,attr,optional"`

	// An optional time to indicate when the component last modified something
	// which updated its health.
	UpdateTime time.Time `river:"update_time,attr,optional"`
}

Health is the reported health state of a component. It can be encoded to River.

type HealthComponent

type HealthComponent interface {
	Component

	// CurrentHealth returns the current Health status for the component.
	//
	// CurrentHealth may be overridden by the Flow controller if there is a
	// higher-level issue, such as a config file being invalid or a Component
	// shutting down unexpectedly.
	CurrentHealth() Health
}

HealthComponent is an optional extension interface for Components which report health information.

Health information is exposed to the end user for informational purposes and cannot be referened in a River expression.

type HealthType

type HealthType uint8

HealthType holds the health value for a component.

const (
	// HealthTypeUnknown is the initial health of components, set when they're
	// first created.
	HealthTypeUnknown HealthType = iota

	// HealthTypeHealthy represents a component which is working as expected.
	HealthTypeHealthy

	// HealthTypeUnhealthy represents a component which is not working as
	// expected.
	HealthTypeUnhealthy

	// HealthTypeExited represents a component which has stopped running.
	HealthTypeExited
)

func (HealthType) MarshalText

func (ht HealthType) MarshalText() (text []byte, err error)

MarshalText implements encoding.TextMarshaler.

func (HealthType) String

func (ht HealthType) String() string

String returns the string representation of ht.

func (*HealthType) UnmarshalText

func (ht *HealthType) UnmarshalText(text []byte) error

UnmarshalText implments encoding.TextUnmarshaler.

type Options

type Options struct {
	// ID of the component. Guaranteed to be globally unique across all running
	// components.
	ID string

	// Logger the component may use for logging. The component ID will always be
	// set as a field.
	Logger log.Logger

	// A path to a directory with this component may use for storage. The path is
	// guaranteed to be unique across all running components.
	//
	// The directory may not exist when the component is created; components
	// should create the directory if needed.
	DataPath string

	// OnStateChange may be invoked at any time by a component whose Export value
	// changes. The Flow controller then will queue re-processing components
	// which depend on the changed component.
	//
	// OnStateChange will panic if e does not match the Exports type registered
	// by the component; a component must use the same Exports type for its
	// lifetime.
	OnStateChange func(e Exports)

	// Registerer allows components to add their own metrics. The register will come pre-wrapped with the component ID. It is not necessary for components to unregister metrics on shutdown.
	Registerer prometheus.Registerer
}

Options are provided to a component when it is being constructed. Options are static for the lifetime of a component.

type Registration

type Registration struct {
	// Name of the component. Must be a list of period-delimited valid
	// identifiers, such as "remote.s3". Components sharing a prefix must have
	// the same number of identifiers; it is valid to register "remote.s3" and
	// "remote.http" but not "remote".
	//
	// Components may not have more than 2 identifiers.
	//
	// Each identifier must start with a valid ASCII letter, and be followed by
	// any number of underscores or alphanumeric ASCII characters.
	Name string

	// A singleton component only supports one instance of itself across the
	// whole process. Normally, multiple components of the same type may be
	// created.
	//
	// The fully-qualified name of a component is the combination of River block
	// name and all of its labels. Fully-qualified names must be unique across
	// the process. Components which are *NOT* singletons automatically support
	// user-supplied identifiers:
	//
	//     // Fully-qualified names: remote.s3.object-a, remote.s3.object-b
	//     remote.s3 "object-a" { ... }
	//     remote.s3 "object-b" { ... }
	//
	// This allows for multiple instances of the same component to be defined.
	// However, components registered as a singleton do not support user-supplied
	// identifiers:
	//
	//     node_exporter { ... }
	//
	// This prevents the user from defining multiple instances of node_exporter
	// with different fully-qualified names.
	Singleton bool

	// An example Arguments value that the registered component expects to
	// receive as input. Components should provide the zero value of their
	// Arguments type here.
	Args Arguments

	// An example Exports value that the registered component may emit as output.
	// A component which does not expose exports must leave this set to nil.
	Exports Exports

	// Build should construct a new component from an initial Arguments and set
	// of options.
	Build func(opts Options, args Arguments) (Component, error)
}

Registration describes a single component.

func Get

func Get(name string) (Registration, bool)

Get finds a registered component by name.

func (Registration) CloneArguments

func (r Registration) CloneArguments() Arguments

CloneArguments returns a new zero value of the registered Arguments type.