Documentation
¶
Index ¶
- func LintBytes(lintConf docs.LintConfig, rawBytes []byte) ([]docs.Lint, error)
- func MarshalYAML(v yaml.Node) ([]byte, error)
- func ReadFileEnvSwap(store ifs.FS, path string, lookupEnvFn func(name string) (string, bool)) (configBytes []byte, lints []docs.Lint, modTime time.Time, err error)
- func ReadFileLinted(fs ifs.FS, path string, skipEnvVarCheck bool, lConf docs.LintConfig, ...) ([]docs.Lint, error)
- func ReplaceEnvVariables(inBytes []byte, lookupFn func(string) (string, bool)) (replaced []byte, err error)
- func ShouldReread(err error) bool
- func Spec() docs.FieldSpecs
- func SpecWithoutStream() docs.FieldSpecs
- type ErrMissingEnvVars
- type ErrNoReread
- type MainUpdateFunc
- type OptFunc
- type 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
- func (r *Reader) TriggerMainUpdate(mgr bundle.NewManagement, strict bool, newPath string) error
- func (r *Reader) TriggerResourceDelete(mgr bundle.NewManagement, path string) error
- func (r *Reader) TriggerResourceUpdate(mgr bundle.NewManagement, strict bool, path string) error
- func (r *Reader) TriggerStreamUpdate(mgr bundle.NewManagement, strict bool, path string) error
- type StreamUpdateFunc
- type Type
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func LintBytes ¶
LintBytes attempts to report errors within a user config. Returns a slice of lint results.
func MarshalYAML ¶
MarshalYAML marshals a structure into YAML with consistent formatting across all Benthos components.
func ReadFileEnvSwap ¶
func ReadFileEnvSwap(store ifs.FS, path string, lookupEnvFn func(name string) (string, bool)) (configBytes []byte, lints []docs.Lint, modTime time.Time, 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.
An modTime timestamp is returned if the modtime of the file is available.
func ReadFileLinted ¶
func ReadFileLinted(fs ifs.FS, path string, skipEnvVarCheck bool, lConf docs.LintConfig, 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 ReplaceEnvVariables ¶
func ReplaceEnvVariables(inBytes []byte, lookupFn func(string) (string, bool)) (replaced []byte, err error)
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 ShouldReread ¶
ShouldReread returns true if the error returned from an update trigger is non nil and also temporal, and therefore it is worth trying the update again even if the content has not changed.
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 ErrMissingEnvVars ¶ added in v1.1.1
type ErrMissingEnvVars struct {
Variables []string
// Our best attempt at parsing the config that's missing variables by simply
// inserting an empty string. There's a good chance this is still a valid
// config! :)
BestAttempt []byte
}
ErrMissingEnvVars is returned when attempting environment variable interpolations where the referenced environment variables are missing.
func (*ErrMissingEnvVars) Error ¶ added in v1.1.1
func (e *ErrMissingEnvVars) Error() string
Error returns a rather sweet error message.
type ErrNoReread ¶
type ErrNoReread struct {
// contains filtered or unexported fields
}
ErrNoReread is an error type returned from update triggers that indicates an attempt should not be re-made unless the source file has been modified.
func (*ErrNoReread) Error ¶
func (e *ErrNoReread) Error() string
Error returns a human readable error string.
type MainUpdateFunc ¶
MainUpdateFunc is a closure function called whenever a main config has been updated. If an error is returned 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 ¶
OptAddOverrides adds one or more override expressions to the config reader, each of the form `path=value`.
func OptSetBootstrapConfig ¶ added in v1.1.1
OptSetBootstrapConfig sets a config to be used as the default for each parse. This can be used to change the default behaviours of benthos configs.
func OptSetLintConfig ¶ added in v1.1.2
func OptSetLintConfig(lConf docs.LintConfig) OptFunc
OptSetLintConfig sets the config used for linting files.
func OptSetStreamPaths ¶
OptSetStreamPaths marks this config reader as operating in streams mode, and adds a list of paths to obtain individual stream configs from.
func OptTestSuffix ¶
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 (*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) ReadStreams ¶
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.
func (*Reader) TriggerMainUpdate ¶
TriggerMainUpdate attempts to re-read the main configuration file, trigger the provided main update func, and apply changes to resources to the provided manager as appropriate.
func (*Reader) TriggerResourceDelete ¶ added in v1.1.1
func (r *Reader) TriggerResourceDelete(mgr bundle.NewManagement, path string) error
TriggerResourceDelete attempts to remove all resources that originated from a given file.
func (*Reader) TriggerResourceUpdate ¶
TriggerResourceUpdate attempts to re-read a resource configuration file and apply changes to the provided manager as appropriate.
func (*Reader) TriggerStreamUpdate ¶
TriggerStreamUpdate attempts to re-read a stream configuration file, and trigger the provided stream update func.
type StreamUpdateFunc ¶
StreamUpdateFunc is a closure function called whenever a stream config has been updated. If an error is returned then the attempt will be made again after a grace period.
When the provided config is nil it is a signal that the stream has been deleted, and it is expected that the provided update func should shut that stream down.
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.
Source Files
Directories