manager

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
  • func (e ErrResourceNotFound) Error() string
• type OptFunc
  • func OptSetAPIReg(r APIReg) OptFunc
  • func OptSetBloblangEnvironment(env *bloblang.Environment) 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
  • func NewResourceConfig() ResourceConfig
  • func (r *ResourceConfig) AddFrom(extra *ResourceConfig) error
• type Type
  • func New(conf ResourceConfig, opts ...OptFunc) (*Type, error)
  • func (t *Type) AccessCache(ctx context.Context, name string, fn func(cache.V1)) error
  • func (t *Type) AccessInput(ctx context.Context, name string, fn func(input.Streamed)) error
  • func (t *Type) AccessOutput(ctx context.Context, name string, fn func(output.Sync)) error
  • func (t *Type) AccessProcessor(ctx context.Context, name string, fn func(processor.V1)) error
  • func (t *Type) AccessRateLimit(ctx context.Context, name string, fn func(ratelimit.V1)) error
  • func (t *Type) BloblEnvironment() *bloblang.Environment
  • func (t *Type) CloseObservability(ctx context.Context) error
  • 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) 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) 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) 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

Functions

func Spec

func Spec() docs.FieldSpecs

Spec returns a field spec for the manager configuration.

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 OptFunc

type OptFunc func(*Type)

OptFunc is an opt setting for a manager type.

func OptSetAPIReg

func OptSetAPIReg(r APIReg) OptFunc

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 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

func OptSetFS(fs ifs.FS) OptFunc

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

func OptSetLogger(logger log.Modular) OptFunc

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

func OptSetStreamHTTPNamespacing(enabled bool) OptFunc

OptSetStreamHTTPNamespacing determines whether HTTP endpoints registered from within a stream should be prefixed with the stream name.

func OptSetStreamsMode

func OptSetStreamsMode(b bool) OptFunc

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     `json:"input_resources,omitempty" yaml:"input_resources,omitempty"`
	ResourceProcessors []processor.Config `json:"processor_resources,omitempty" yaml:"processor_resources,omitempty"`
	ResourceOutputs    []output.Config    `json:"output_resources,omitempty" yaml:"output_resources,omitempty"`
	ResourceCaches     []cache.Config     `json:"cache_resources,omitempty" yaml:"cache_resources,omitempty"`
	ResourceRateLimits []ratelimit.Config `json:"rate_limit_resources,omitempty" yaml:"rate_limit_resources,omitempty"`
}

ResourceConfig contains fields for specifying resource components at the root of a Benthos 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

func (t *Type) AccessCache(ctx context.Context, name string, fn func(cache.V1)) error

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

func (t *Type) AccessInput(ctx context.Context, name string, fn func(input.Streamed)) error

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

func (t *Type) AccessOutput(ctx context.Context, name string, fn func(output.Sync)) error

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

func (t *Type) AccessProcessor(ctx context.Context, name string, fn func(processor.V1)) error

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

func (t *Type) AccessRateLimit(ctx context.Context, name string, fn func(ratelimit.V1)) error

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

func (t *Type) CloseObservability(ctx context.Context) error

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) 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

func (t *Type) FS() ifs.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) GetDocs

func (t *Type) GetDocs(name string, ctype docs.Type) (docs.ComponentSpec, bool)

GetDocs returns a documentation spec for an implementation of a component.

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) Label

func (t *Type) Label() string

Label returns the current component label held by a manager.

func (*Type) Logger

func (t *Type) Logger() log.Modular

Logger returns a logger preset with the current component context.

func (*Type) Metrics

func (t *Type) Metrics() metrics.Type

Metrics returns an aggregator preset with the current component context.

func (*Type) NewBuffer

func (t *Type) NewBuffer(conf buffer.Config) (buffer.Streamed, error)

NewBuffer attempts to create a new buffer component from a config.

func (*Type) NewCache

func (t *Type) NewCache(conf cache.Config) (cache.V1, error)

NewCache attempts to create a new cache component from a config.

func (*Type) NewInput

func (t *Type) NewInput(conf input.Config) (input.Streamed, error)

NewInput attempts to create a new input component from a config.

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

func (t *Type) NewProcessor(conf processor.Config) (processor.V1, error)

NewProcessor attempts to create a new processor component from a config.

func (*Type) NewRateLimit

func (t *Type) NewRateLimit(conf ratelimit.Config) (ratelimit.V1, error)

NewRateLimit attempts to create a new rate limit component from a config.

func (*Type) Path

func (t *Type) Path() []string

Path returns the current component path held by a manager.

func (*Type) ProbeCache

func (t *Type) ProbeCache(name string) bool

ProbeCache returns true if a cache resource exists under the provided name.

func (*Type) ProbeInput

func (t *Type) ProbeInput(name string) bool

ProbeInput returns true if an input resource exists under the provided name.

func (*Type) ProbeOutput

func (t *Type) ProbeOutput(name string) bool

ProbeOutput returns true if an output resource exists under the provided name.

func (*Type) ProbeProcessor

func (t *Type) ProbeProcessor(name string) bool

ProbeProcessor returns true if a processor resource exists under the provided name.

func (*Type) ProbeRateLimit

func (t *Type) ProbeRateLimit(name string) bool

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

func (t *Type) RemoveCache(ctx context.Context, name string) error

RemoveCache attempts to close and remove an existing cache resource.

func (*Type) RemoveInput

func (t *Type) RemoveInput(ctx context.Context, name string) error

RemoveInput attempts to close and remove an existing input resource.

func (*Type) RemoveOutput

func (t *Type) RemoveOutput(ctx context.Context, name string) error

RemoveOutput attempts to close and remove an existing output resource.

func (*Type) RemoveProcessor

func (t *Type) RemoveProcessor(ctx context.Context, name string) error

RemoveProcessor attempts to close and remove an existing processor resource.

func (*Type) RemoveRateLimit

func (t *Type) RemoveRateLimit(ctx context.Context, name string) error

RemoveRateLimit attempts to close and remove an existing rate limit resource.

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

func (t *Type) StoreCache(ctx context.Context, name string, conf cache.Config) error

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

func (t *Type) StoreInput(ctx context.Context, name string, conf input.Config) error

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

func (t *Type) StoreOutput(ctx context.Context, name string, conf output.Config) error

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

func (t *Type) StoreProcessor(ctx context.Context, name string, conf processor.Config) error

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

func (t *Type) StoreRateLimit(ctx context.Context, name string, conf ratelimit.Config) error

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

func (t *Type) WaitForClose(ctx context.Context) error

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.

func (*Type) WithMetricsMapping

func (t *Type) WithMetricsMapping(m *metrics.Mapping) *Type

WithMetricsMapping returns a manager with the stored metrics exporter wrapped with a mapping.