dpbuffered

Documentation

Index

• Variables
• type BufferedForwarder
  • func NewBufferedForwarder(ctx context.Context, config *Config, sendTo Sink, ...) *BufferedForwarder
  • func (forwarder *BufferedForwarder) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) error
  • func (forwarder *BufferedForwarder) AddEvents(ctx context.Context, events []*event.Event) error
  • func (forwarder *BufferedForwarder) AddSpans(ctx context.Context, traces []*trace.Span) error
  • func (forwarder *BufferedForwarder) Close() error
  • func (forwarder *BufferedForwarder) Datapoints() []*datapoint.Datapoint
  • func (forwarder *BufferedForwarder) DebugDatapoints() []*datapoint.Datapoint
  • func (forwarder *BufferedForwarder) DebugEndpoints() map[string]http.Handler
  • func (forwarder *BufferedForwarder) DefaultDatapoints() []*datapoint.Datapoint
  • func (forwarder *BufferedForwarder) Pipeline() int64
  • func (forwarder *BufferedForwarder) StartupFinished() error
• type Config
  • func (c *Config) String() string
• type Sink

Variables

var DefaultConfig = &Config{
	BufferSize:         pointer.Int64(1000000),
	MaxTotalDatapoints: pointer.Int64(1000000),
	MaxTotalEvents:     pointer.Int64(1000000),
	MaxTotalSpans:      pointer.Int64(1000000),
	MaxDrainSize:       pointer.Int64(30000),
	NumDrainingThreads: pointer.Int64(int64(runtime.NumCPU())),
	Name:               pointer.String(""),
	UseAuthFromRequest: pointer.Bool(false),
}

DefaultConfig are default values for buffered forwarders

Types

type BufferedForwarder

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

BufferedForwarder abstracts out datapoint buffering. Points put on its channel are buffered and sent in large groups to a waiting sink

func NewBufferedForwarder

func NewBufferedForwarder(ctx context.Context, config *Config, sendTo Sink, closeIt, afterStartup func() error, logger log.Logger, debugEnpoints func() map[string]http.Handler) *BufferedForwarder

NewBufferedForwarder is used only by this package to create a forwarder that buffers its datapoint channel

func (*BufferedForwarder) AddDatapoints

func (forwarder *BufferedForwarder) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) error

AddDatapoints sends the datapoints to a chan buffer that eventually is flushed in big groups

func (*BufferedForwarder) AddEvents

func (forwarder *BufferedForwarder) AddEvents(ctx context.Context, events []*event.Event) error

AddEvents sends the events to a chan buffer that eventually is flushed in big groups

func (*BufferedForwarder) AddSpans

func (forwarder *BufferedForwarder) AddSpans(ctx context.Context, traces []*trace.Span) error

AddSpans sends the traces to a chan buffer that traceually is flushed in big groups

func (*BufferedForwarder) Close

func (forwarder *BufferedForwarder) Close() error

Close stops the threads that are flushing channel points to the next forwarder

func (*BufferedForwarder) Datapoints

func (forwarder *BufferedForwarder) Datapoints() []*datapoint.Datapoint

Datapoints implements the sfxclient.Collector interface and returns all datapoints about the buffered forwarder

func (*BufferedForwarder) DebugDatapoints

func (forwarder *BufferedForwarder) DebugDatapoints() []*datapoint.Datapoint

DebugDatapoints returns debug level datapoints about this forwarder, including errors processing datapoints

func (*BufferedForwarder) DebugEndpoints

func (forwarder *BufferedForwarder) DebugEndpoints() map[string]http.Handler

DebugEndpoints returns any configured http handlers and the point they can be reached at

func (*BufferedForwarder) DefaultDatapoints

func (forwarder *BufferedForwarder) DefaultDatapoints() []*datapoint.Datapoint

DefaultDatapoints does nothing and exists to satisfy the protocol.forwarder interface

func (*BufferedForwarder) Pipeline

func (forwarder *BufferedForwarder) Pipeline() int64

Pipeline for a BufferedForwarder is the total of all buffers and what is in flight

func (*BufferedForwarder) StartupFinished

func (forwarder *BufferedForwarder) StartupFinished() error

StartupFinished runs the afterStartup method on the forwarder

type Config

type Config struct {
	BufferSize         *int64
	MaxTotalDatapoints *int64
	MaxTotalEvents     *int64
	MaxTotalSpans      *int64
	MaxDrainSize       *int64
	NumDrainingThreads *int64
	Checker            *dpsink.ItemFlagger
	Cdim               *log.CtxDimensions
	Name               *string
	UseAuthFromRequest *bool
}

Config controls BufferedForwarder limits

func (*Config) String

func (c *Config) String() string

type Sink

type Sink interface {
	dpsink.Sink
	trace.Sink
}

Sink is a dpsink and trace.sink