Documentation
¶
Overview ¶
Package manager implements the types.Manager interface used for creating and sharing resources across a Benthos service.
Index ¶
- func Spec() docs.FieldSpecs
- type APIReg
- type ErrResourceNotFound
- type InputWrapper
- func (w *InputWrapper) CloseExistingInput(ctx context.Context, forSwap bool) error
- func (w *InputWrapper) ConnectionStatus() (s component.ConnectionStatuses)
- func (w *InputWrapper) SwapInput(i input.Streamed)
- func (w *InputWrapper) TransactionChan() <-chan message.Transaction
- func (w *InputWrapper) TriggerCloseNow()
- func (w *InputWrapper) TriggerStopConsuming()
- func (w *InputWrapper) WaitForClose(ctx context.Context) error
- type OptFunc
- func OptSetAPIReg(r APIReg) OptFunc
- func OptSetBloblangEnvironment(env *bloblang.Environment) OptFunc
- func OptSetEngineVersion(v string) OptFunc
- func OptSetEnvironment(e *bundle.Environment) OptFunc
- func OptSetFS(fs ifs.FS) OptFunc
- func OptSetLogger(logger log.Modular) OptFunc
- func OptSetMetrics(stats *metrics.Namespaced) OptFunc
- func OptSetStreamHTTPNamespacing(enabled bool) OptFunc
- func OptSetStreamsMode(b bool) OptFunc
- func OptSetTracer(tracer trace.TracerProvider) OptFunc
- type ResourceConfig
- type Type
- func (t *Type) AccessCache(ctx context.Context, name string, fn func(cache.V1)) (err error)
- func (t *Type) AccessInput(ctx context.Context, name string, fn func(input.Streamed)) (err error)
- func (t *Type) AccessOutput(ctx context.Context, name string, fn func(output.Sync)) (err error)
- func (t *Type) AccessProcessor(ctx context.Context, name string, fn func(processor.V1)) (err error)
- func (t *Type) AccessRateLimit(ctx context.Context, name string, fn func(ratelimit.V1)) (err error)
- func (t *Type) BloblEnvironment() *bloblang.Environment
- func (t *Type) CloseObservability(ctx context.Context) error
- func (t *Type) EngineVersion() string
- func (t *Type) Environment() *bundle.Environment
- func (t *Type) FS() ifs.FS
- func (t *Type) ForStream(id string) bundle.NewManagement
- func (t *Type) GetDocs(name string, ctype docs.Type) (docs.ComponentSpec, bool)
- func (t *Type) GetGeneric(key any) (any, bool)
- func (t *Type) GetOrSetGeneric(key, value any) (actual any, loaded bool)
- func (t *Type) GetPipe(name string) (<-chan message.Transaction, error)
- func (t *Type) IntoPath(segments ...string) bundle.NewManagement
- func (t *Type) Label() string
- func (t *Type) Logger() log.Modular
- func (t *Type) Metrics() metrics.Type
- func (t *Type) NewBuffer(conf buffer.Config) (buffer.Streamed, error)
- func (t *Type) NewCache(conf cache.Config) (cache.V1, error)
- func (t *Type) NewInput(conf input.Config) (input.Streamed, error)
- func (t *Type) NewOutput(conf output.Config, pipelines ...processor.PipelineConstructorFunc) (output.Streamed, error)
- func (t *Type) NewProcessor(conf processor.Config) (processor.V1, error)
- func (t *Type) NewRateLimit(conf ratelimit.Config) (ratelimit.V1, error)
- func (t *Type) NewScanner(conf scanner.Config) (scanner.Creator, error)
- func (t *Type) Path() []string
- func (t *Type) ProbeCache(name string) bool
- func (t *Type) ProbeInput(name string) bool
- func (t *Type) ProbeOutput(name string) bool
- func (t *Type) ProbeProcessor(name string) bool
- func (t *Type) ProbeRateLimit(name string) bool
- func (t *Type) RegisterEndpoint(apiPath, desc string, h http.HandlerFunc)
- func (t *Type) RemoveCache(ctx context.Context, name string) error
- func (t *Type) RemoveInput(ctx context.Context, name string) error
- func (t *Type) RemoveOutput(ctx context.Context, name string) error
- func (t *Type) RemoveProcessor(ctx context.Context, name string) error
- func (t *Type) RemoveRateLimit(ctx context.Context, name string) error
- func (t *Type) SetGeneric(key, value any)
- func (t *Type) SetPipe(name string, tran <-chan message.Transaction)
- func (t *Type) StoreCache(ctx context.Context, name string, conf cache.Config) error
- func (t *Type) StoreInput(ctx context.Context, name string, conf input.Config) error
- func (t *Type) StoreOutput(ctx context.Context, name string, conf output.Config) error
- func (t *Type) StoreProcessor(ctx context.Context, name string, conf processor.Config) error
- func (t *Type) StoreRateLimit(ctx context.Context, name string, conf ratelimit.Config) error
- func (t *Type) Tracer() trace.TracerProvider
- func (t *Type) TriggerCloseNow()
- func (t *Type) TriggerStopConsuming()
- func (t *Type) UnsetPipe(name string, tran <-chan message.Transaction)
- func (t *Type) WaitForClose(ctx context.Context) error
- func (t *Type) WithAddedMetrics(m metrics.Type) bundle.NewManagement
- func (t *Type) WithMetricsMapping(m *metrics.Mapping) *Type
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type APIReg ¶
type APIReg interface {
RegisterEndpoint(path, desc string, h http.HandlerFunc)
}
APIReg is an interface representing an API builder.
type ErrResourceNotFound ¶
type ErrResourceNotFound string
ErrResourceNotFound represents an error where a named resource could not be accessed because it was not found by the manager.
func (ErrResourceNotFound) Error ¶
func (e ErrResourceNotFound) Error() string
Error implements the standard error interface.
type InputWrapper ¶
type InputWrapper struct {
// contains filtered or unexported fields
}
InputWrapper is a wrapper for a streamed input.
func WrapInput ¶
func WrapInput(i input.Streamed) *InputWrapper
WrapInput wraps a streamed input and starts the transaction processing loop.
func (*InputWrapper) CloseExistingInput ¶
func (w *InputWrapper) CloseExistingInput(ctx context.Context, forSwap bool) error
CloseExistingInput instructs the wrapped input to stop consuming messages and waits for it to shut down.
func (*InputWrapper) ConnectionStatus ¶ added in v4.31.0
func (w *InputWrapper) ConnectionStatus() (s component.ConnectionStatuses)
ConnectionStatus returns the current status of the given component connection. The result is a slice in order to accommodate higher order components that wrap several others.
func (*InputWrapper) SwapInput ¶
func (w *InputWrapper) SwapInput(i input.Streamed)
SwapInput swaps the wrapped input with another one.
func (*InputWrapper) TransactionChan ¶
func (w *InputWrapper) TransactionChan() <-chan message.Transaction
TransactionChan returns a transactions channel for consuming messages from the wrapped input\.
func (*InputWrapper) TriggerCloseNow ¶
func (w *InputWrapper) TriggerCloseNow()
TriggerCloseNow triggers the shut down of the wrapped input but should not block the calling goroutine.
func (*InputWrapper) TriggerStopConsuming ¶
func (w *InputWrapper) TriggerStopConsuming()
TriggerStopConsuming instructs the wrapped input to start shutting down resources once all pending messages are delivered and acknowledged. This call does not block.
func (*InputWrapper) WaitForClose ¶
func (w *InputWrapper) WaitForClose(ctx context.Context) error
WaitForClose is a blocking call to wait until the wrapped input has finished shutting down and cleaning up resources.
type OptFunc ¶
type OptFunc func(*Type)
OptFunc is an opt setting for a manager type.
func OptSetAPIReg ¶
OptSetAPIReg sets the multiplexer used by components of this manager for registering their own HTTP endpoints.
func OptSetBloblangEnvironment ¶
func OptSetBloblangEnvironment(env *bloblang.Environment) OptFunc
OptSetBloblangEnvironment determines the environment from which the manager parses bloblang functions and methods. This option is for internal use only.
func OptSetEngineVersion ¶
OptSetEngineVersion sets the engine version reported to components. This can be any scheme, or no scheme at all.
func OptSetEnvironment ¶
func OptSetEnvironment(e *bundle.Environment) OptFunc
OptSetEnvironment determines the environment from which the manager initializes components and resources. This option is for internal use only.
func OptSetFS ¶
OptSetFS determines which ifs.FS implementation to use for its filesystem. This can be used to override the default os based filesystem implementation.
func OptSetLogger ¶
OptSetLogger sets the logger from which the manager emits log events for components.
func OptSetMetrics ¶
func OptSetMetrics(stats *metrics.Namespaced) OptFunc
OptSetMetrics sets the metrics exporter from which the manager creates metrics for components.
func OptSetStreamHTTPNamespacing ¶
OptSetStreamHTTPNamespacing determines whether HTTP endpoints registered from within a stream should be prefixed with the stream name.
func OptSetStreamsMode ¶
OptSetStreamsMode marks the manager as being created for running streams mode resources. This ensures that a label "stream" is added to metrics.
func OptSetTracer ¶
func OptSetTracer(tracer trace.TracerProvider) OptFunc
OptSetTracer sets the tracer provider from which the manager creates tracing spans.
type ResourceConfig ¶
type ResourceConfig struct {
ResourceInputs []input.Config `yaml:"input_resources,omitempty"`
ResourceProcessors []processor.Config `yaml:"processor_resources,omitempty"`
ResourceOutputs []output.Config `yaml:"output_resources,omitempty"`
ResourceCaches []cache.Config `yaml:"cache_resources,omitempty"`
ResourceRateLimits []ratelimit.Config `yaml:"rate_limit_resources,omitempty"`
}
ResourceConfig contains fields for specifying resource components at the root of a Benthos config.
func FromAny ¶
func FromAny(prov docs.Provider, v any) (conf ResourceConfig, err error)
FromAny returns a resource config from a parsed config, yaml node or any value.
func FromParsed ¶
func FromParsed(prov docs.Provider, pConf *docs.ParsedConfig) (conf ResourceConfig, err error)
FromParsed extracts a resource config from a parsed config.
func NewResourceConfig ¶
func NewResourceConfig() ResourceConfig
NewResourceConfig creates a ResourceConfig with default values.
func (*ResourceConfig) AddFrom ¶
func (r *ResourceConfig) AddFrom(extra *ResourceConfig) error
AddFrom takes another Config and adds all of its resources to itself. If there are any resource name collisions an error is returned.
type Type ¶
type Type struct {
// contains filtered or unexported fields
}
Type is an implementation of types.Manager, which is expected by Benthos components that need to register service wide behaviours such as HTTP endpoints and event listeners, and obtain service wide shared resources such as caches and other resources.
func New ¶
func New(conf ResourceConfig, opts ...OptFunc) (*Type, error)
New returns an instance of manager.Type, which can be shared amongst components and logical threads of a Benthos service.
func (*Type) AccessCache ¶
AccessCache attempts to access a cache resource by a unique identifier and executes a closure function with the cache as an argument. Returns an error if the cache does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessInput ¶
AccessInput attempts to access an input resource by a unique identifier and executes a closure function with the input as an argument. Returns an error if the input does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessOutput ¶
AccessOutput attempts to access an output resource by a unique identifier and executes a closure function with the output as an argument. Returns an error if the output does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessProcessor ¶
AccessProcessor attempts to access a processor resource by a unique identifier and executes a closure function with the processor as an argument. Returns an error if the processor does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessRateLimit ¶
AccessRateLimit attempts to access a rate limit resource by a unique identifier and executes a closure function with the rate limit as an argument. Returns an error if the rate limit does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) BloblEnvironment ¶
func (t *Type) BloblEnvironment() *bloblang.Environment
BloblEnvironment returns a Bloblang environment used by the manager. This is for internal use only.
func (*Type) CloseObservability ¶
CloseObservability attempts to clean up observability (metrics, tracing, etc) components owned by the manager. This should only be called when the manager itself has finished shutting down and when it is the sole owner of the observability components.
func (*Type) EngineVersion ¶
EngineVersion returns the stored version string for the engine. This version string could be any format.
func (*Type) Environment ¶
func (t *Type) Environment() *bundle.Environment
Environment returns a bundle environment used by the manager. This is for internal use only.
func (*Type) FS ¶
FS returns an ifs.FS implementation that provides access to a filesystem. By default this simply access the os package, with relative paths resolved from the directory that the process is running from.
func (*Type) ForStream ¶
func (t *Type) ForStream(id string) bundle.NewManagement
ForStream returns a variant of this manager to be used by a particular stream identifier, where APIs registered will be namespaced by that id.
func (*Type) GetGeneric ¶ added in v4.31.0
GetGeneric attempts to obtain and return a generic resource value by key.
func (*Type) GetOrSetGeneric ¶ added in v4.31.0
GetOrSetGeneric attempts to obtain an existing value for a given key if present. Otherwise, it stores and returns the provided value. The loaded result is true if the value was loaded, false if stored.
func (*Type) GetPipe ¶
func (t *Type) GetPipe(name string) (<-chan message.Transaction, error)
GetPipe attempts to obtain and return a named output Pipe.
func (*Type) IntoPath ¶
func (t *Type) IntoPath(segments ...string) bundle.NewManagement
IntoPath returns a variant of this manager to be used by a particular component path, which is a child of the current component, where observability components will be automatically tagged with the new path.
func (*Type) NewOutput ¶
func (t *Type) NewOutput(conf output.Config, pipelines ...processor.PipelineConstructorFunc) (output.Streamed, error)
NewOutput attempts to create a new output component from a config.
func (*Type) NewProcessor ¶
NewProcessor attempts to create a new processor component from a config.
func (*Type) NewRateLimit ¶
NewRateLimit attempts to create a new rate limit component from a config.
func (*Type) NewScanner ¶
NewScanner attempts to create a new scanner component from a config.
func (*Type) ProbeCache ¶
ProbeCache returns true if a cache resource exists under the provided name.
func (*Type) ProbeInput ¶
ProbeInput returns true if an input resource exists under the provided name.
func (*Type) ProbeOutput ¶
ProbeOutput returns true if an output resource exists under the provided name.
func (*Type) ProbeProcessor ¶
ProbeProcessor returns true if a processor resource exists under the provided name.
func (*Type) ProbeRateLimit ¶
ProbeRateLimit returns true if a rate limit resource exists under the provided name.
func (*Type) RegisterEndpoint ¶
func (t *Type) RegisterEndpoint(apiPath, desc string, h http.HandlerFunc)
RegisterEndpoint registers a server wide HTTP endpoint.
func (*Type) RemoveCache ¶
RemoveCache attempts to close and remove an existing cache resource.
func (*Type) RemoveInput ¶
RemoveInput attempts to close and remove an existing input resource.
func (*Type) RemoveOutput ¶
RemoveOutput attempts to close and remove an existing output resource.
func (*Type) RemoveProcessor ¶
RemoveProcessor attempts to close and remove an existing processor resource.
func (*Type) RemoveRateLimit ¶
RemoveRateLimit attempts to close and remove an existing rate limit resource.
func (*Type) SetGeneric ¶ added in v4.31.0
SetGeneric attempts to set a generic resource to a given value by key.
func (*Type) SetPipe ¶
func (t *Type) SetPipe(name string, tran <-chan message.Transaction)
SetPipe registers a new transaction chan to a named pipe.
func (*Type) StoreCache ¶
StoreCache attempts to store a new cache resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreInput ¶
StoreInput attempts to store a new input resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreOutput ¶
StoreOutput attempts to store a new output resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreProcessor ¶
StoreProcessor attempts to store a new processor resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreRateLimit ¶
StoreRateLimit attempts to store a new rate limit resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) Tracer ¶
func (t *Type) Tracer() trace.TracerProvider
Tracer returns a tracer provider with the current component context.
func (*Type) TriggerCloseNow ¶
func (t *Type) TriggerCloseNow()
TriggerCloseNow triggers the absolute shut down of this component but should not block the calling goroutine.
func (*Type) TriggerStopConsuming ¶
func (t *Type) TriggerStopConsuming()
TriggerStopConsuming instructs the manager to stop resource inputs and outputs from consuming data. This call does not block.
func (*Type) UnsetPipe ¶
func (t *Type) UnsetPipe(name string, tran <-chan message.Transaction)
UnsetPipe removes a named pipe transaction chan.
func (*Type) WaitForClose ¶
WaitForClose is a blocking call to wait until the component has finished shutting down and cleaning up resources.
func (*Type) WithAddedMetrics ¶
func (t *Type) WithAddedMetrics(m metrics.Type) bundle.NewManagement
WithAddedMetrics returns a modified version of the manager where metrics are registered to both the current metrics target as well as the provided one.
Source Files
Directories