config

Documentation

Index

• func InferStreamID(dir, path string) (string, error)
• func LintBytes(opts LintOptions, rawBytes []byte) ([]docs.Lint, error)
• func MarshalYAML(v yaml.Node) ([]byte, error)
• func ReadFileEnvSwap(path string) (configBytes []byte, lints []docs.Lint, err error)
• func ReadFileLinted(path string, opts LintOptions, config *Type) ([]docs.Lint, error)
• func ReadStreamFile(path string) (conf stream.Config, lints []string, err error)
• func ReplaceEnvVariables(inBytes []byte) []byte
• func Spec() docs.FieldSpecs
• func SpecWithoutStream() docs.FieldSpecs
• type LintOptions
• type MainUpdateFunc
• type OptFunc
  • func OptAddOverrides(overrides ...string) OptFunc
  • func OptSetStreamPaths(streamsPaths ...string) OptFunc
  • func OptTestSuffix(suffix string) OptFunc
• type Reader
  • func NewReader(mainPath string, resourcePaths []string, opts ...OptFunc) *Reader
  • func (r *Reader) BeginFileWatching(mgr bundle.NewManagement, strict bool) error
  • func (r *Reader) Close(ctx context.Context) error
  • func (r *Reader) Read(conf *Type) (lints []string, err error)
  • func (r *Reader) ReadStreams(confs map[string]stream.Config) (lints []string, err error)
  • func (r *Reader) SubscribeConfigChanges(fn MainUpdateFunc) error
  • func (r *Reader) SubscribeStreamChanges(fn StreamUpdateFunc) error
• type StreamUpdateFunc
• type Type
  • func New() Type

Functions

func InferStreamID

func InferStreamID(dir, path string) (string, error)

InferStreamID attempts to infer a stream identifier from a file path and containing directory. If the dir field is non-empty then the identifier will include all sub-directories in the path as an id prefix, this means loading streams with the same file name from different branches are still given unique names.

func LintBytes

func LintBytes(opts LintOptions, rawBytes []byte) ([]docs.Lint, error)

LintBytes attempts to report errors within a user config. Returns a slice of lint results.

func MarshalYAML

func MarshalYAML(v yaml.Node) ([]byte, error)

MarshalYAML marshals a structure into YAML with consistent formatting across all Benthos components.

func ReadFileEnvSwap

func ReadFileEnvSwap(path string) (configBytes []byte, lints []docs.Lint, err error)

ReadFileEnvSwap reads a file and replaces any environment variable interpolations before returning the contents. Linting errors are returned if the file has an unexpected higher level format, such as invalid utf-8 encoding.

func ReadFileLinted

func ReadFileLinted(path string, opts LintOptions, config *Type) ([]docs.Lint, error)

ReadFileLinted will attempt to read a configuration file path into a structure. Returns an array of lint messages or an error.

func ReadStreamFile

func ReadStreamFile(path string) (conf stream.Config, lints []string, err error)

ReadStreamFile attempts to read a stream config and returns the result.

func ReplaceEnvVariables

func ReplaceEnvVariables(inBytes []byte) []byte

ReplaceEnvVariables will search a blob of data for the pattern `${FOO:bar}`, where `FOO` is an environment variable name and `bar` is a default value. The `bar` section (including the colon) can be left out if there is no appropriate default value for the field.

For each aforementioned pattern found in the blob the contents of the respective environment variable will be read and will replace the pattern. If the environment variable is empty or does not exist then either the default value is used or the field will be left empty.

func Spec

func Spec() docs.FieldSpecs

Spec returns a docs.FieldSpec for an entire Benthos configuration.

func SpecWithoutStream

func SpecWithoutStream() docs.FieldSpecs

SpecWithoutStream describes a stream config without the core stream fields.

Types

type LintOptions

type LintOptions struct {
	RejectDeprecated bool
	RequireLabels    bool
}

LintOptions specifies the linters that will be enabled.

type MainUpdateFunc

type MainUpdateFunc func(conf stream.Config) bool

MainUpdateFunc is a closure function called whenever a main config has been updated. A boolean should be returned indicating whether the stream was successfully updated, if false then the attempt will be made again after a grace period.

type OptFunc

type OptFunc func(*Reader)

OptFunc is an opt function that changes the behaviour of a config reader.

func OptAddOverrides

func OptAddOverrides(overrides ...string) OptFunc

OptAddOverrides adds one or more override expressions to the config reader, each of the form `path=value`.

func OptSetStreamPaths

func OptSetStreamPaths(streamsPaths ...string) OptFunc

OptSetStreamPaths marks this config reader as operating in streams mode, and adds a list of paths to obtain individual stream configs from.

func OptTestSuffix

func OptTestSuffix(suffix string) OptFunc

OptTestSuffix configures the suffix given to unit test definition files, this is used in order to exclude unit tests from being run in streams mode with arbitrary directory walking.

type Reader

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

Reader provides utilities for parsing a Benthos config as a main file with a collection of resource files, and options such as overrides.

func NewReader

func NewReader(mainPath string, resourcePaths []string, opts ...OptFunc) *Reader

NewReader creates a new config reader.

func (*Reader) BeginFileWatching

func (r *Reader) BeginFileWatching(mgr bundle.NewManagement, strict bool) error

BeginFileWatching creates a goroutine that watches all active configuration files for changes. If a resource is changed then it is swapped out automatically through the provided manager. If a main config or stream config changes then the closures registered with either SubscribeConfigChanges or SubscribeStreamChanges will be called.

WARNING: Either SubscribeConfigChanges or SubscribeStreamChanges must be called before this, as otherwise it is unsafe to register them during watching.

func (*Reader) Close

func (r *Reader) Close(ctx context.Context) error

Close the reader, when this method exits all reloading will be stopped.

func (*Reader) Read

func (r *Reader) Read(conf *Type) (lints []string, err error)

Read a Benthos config from the files and options specified.

func (*Reader) ReadStreams

func (r *Reader) ReadStreams(confs map[string]stream.Config) (lints []string, err error)

ReadStreams attempts to read Benthos stream configs from one or more paths. Stream configs are extracted and added to a provided map, where the id is derived from the path of the stream config file.

func (*Reader) SubscribeConfigChanges

func (r *Reader) SubscribeConfigChanges(fn MainUpdateFunc) error

SubscribeConfigChanges registers a closure function to be called whenever the main configuration file is updated.

The provided closure should return true if the stream was successfully replaced.

func (*Reader) SubscribeStreamChanges

func (r *Reader) SubscribeStreamChanges(fn StreamUpdateFunc) error

SubscribeStreamChanges registers a closure to be called whenever the configuration of a stream is updated.

The provided closure should return true if the stream was successfully replaced.

type StreamUpdateFunc

type StreamUpdateFunc func(id string, conf stream.Config) bool

StreamUpdateFunc is a closure function called whenever a stream config has been updated. A boolean should be returned indicating whether the stream was successfully updated, if false then the attempt will be made again after a grace period.

type Type

type Type struct {
	HTTP                   api.Config `json:"http" yaml:"http"`
	stream.Config          `json:",inline" yaml:",inline"`
	manager.ResourceConfig `json:",inline" yaml:",inline"`
	Logger                 log.Config     `json:"logger" yaml:"logger"`
	Metrics                metrics.Config `json:"metrics" yaml:"metrics"`
	Tracer                 tracer.Config  `json:"tracer" yaml:"tracer"`
	SystemCloseDelay       string         `json:"shutdown_delay" yaml:"shutdown_delay"`
	SystemCloseTimeout     string         `json:"shutdown_timeout" yaml:"shutdown_timeout"`
	Tests                  []any          `json:"tests,omitempty" yaml:"tests,omitempty"`
}

Type is the Benthos service configuration struct.

func New

func New() Type

New returns a new configuration with default values.