outflux

Documentation

Index

• Constants
• Variables
• func DefaultBackoff(retry, maxRetries int) time.Duration
• func SendMeasurements(ctx context.Context, url *url.URL, client *http.Client, ...) (err error)
• func SetWriter(w *Proxy)
• func Write(b []byte) (int, error)
• func WriteMeasurement(m dagr.Measurement) (int64, error)
• func WriteMeasurements(ms ...dagr.Measurement) (int64, error)
• func WritePoint(key string, when time.Time, tags dagr.Tags, fields dagr.Fields) (n int64, err error)
• type BackoffFunc
• type BadStatusError
  • func (e *BadStatusError) Error() string
  • func (e *BadStatusError) InfluxError() (*InfluxError, error)
• type Director
• type FixedBackoff
  • func (d FixedBackoff) Backoff(int, int) time.Duration
• type FlushSize
• type InfluxError
• type Logger
• type Option
  • func RequestLimit(n int) Option
• type Proxy
  • func New(client *http.Client, destURL string, opts ...Option) *Proxy
  • func NewURL(client *http.Client, destURL *url.URL, opts ...Option) *Proxy
  • func (w *Proxy) Flush(ctx context.Context) error
  • func (w *Proxy) Start(ctx context.Context, interval time.Duration)
  • func (w *Proxy) Transaction(fn WriteFunc) (err error)
  • func (w *Proxy) Write(b []byte) (int, error)
  • func (w *Proxy) WriteMeasurement(measurement dagr.Measurement) (n int64, err error)
  • func (w *Proxy) WriteMeasurements(measurements ...dagr.Measurement) (n int64, err error)
  • func (w *Proxy) WritePoint(key string, when time.Time, tags dagr.Tags, fields dagr.Fields) (n int64, err error)
  • func (w *Proxy) Writer() io.WriteCloser
• type RetryLimit
• type Timeout
• type WriteFunc

Constants

const DefaultRetries = RetryLimit(3)

DefaultRetries is the default number of retries to attempt per InfluxDB request. It is recommended you set this low enough that you don't end up with a significant backlog of InfluxDB requests in the event of an outage. If you have memory to spare, consider passing a RequestLimit to limit the number of in-flight requests during failures, otherwise you may end up with an undesirable number of open connections for higher retry counts.

Variables

var ErrNotInfluxError = errors.New("outflux: error is not an InfluxDB error")

ErrNotInfluxError is returned by (*BadStatusError).InfxluError when an error body does not appear to describe an InfluxDB JSON error message.

var Stdlog = stdlog{}

Stdlog is a Logger that uses the standard log package's logger.

Functions

func DefaultBackoff

func DefaultBackoff(retry, maxRetries int) time.Duration

DefaultBackoff returns the default backoff, which linearly increases a delay of 8 seconds by 3 seconds per retry. It has a maximum retry duration of 30 seconds.

func SendMeasurements

func SendMeasurements(ctx context.Context, url *url.URL, client *http.Client, measurements ...dagr.Measurement) (err error)

SendMeasurements sends the dagr Measurements to the given URL as a POST request. If an error occurs, that error is returned.

func SetWriter

func SetWriter(w *Proxy)

SetWriter replaces the current Proxy with w. w may be nil, causing all subsequent writes via shared Proxy functions to be silently discarded.

func Write

func Write(b []byte) (int, error)

Write writes an arbitrary sequence of bytes to the shared outflux Proxy. It returns the number of bytes written and any error that occurred. If b is nil or empty, the call is a no-op.

func WriteMeasurement

func WriteMeasurement(m dagr.Measurement) (int64, error)

WriteMeasurement writes a single dagr.Measurement to the shared outflux Proxy. It returns the number of bytes written and any error that occurred in writing the measurement. If m is nil, it is a no-op.

func WriteMeasurements

func WriteMeasurements(ms ...dagr.Measurement) (int64, error)

WriteMeasurements writes multiple dagr.Measurements to the shared outflux Proxy. It returns the number of bytes written and any error that occurred in writing the measurements. If ms is nil or empty, it is a no-op.

func WritePoint

func WritePoint(key string, when time.Time, tags dagr.Tags, fields dagr.Fields) (n int64, err error)

WritePoint writes a single point to the shared outflux Proxy. It returns the number of bytes written and any error that occurred in writing it.

Types

type BackoffFunc

type BackoffFunc func(retry, maxRetries int) time.Duration

BackoffFunc is a function used to compute retry backoff. Each retry occurs after the duration returned by a BackoffFunc. If the returned delay is <= 0, retries occur as soon as possible.

The argument retry is always >= 1 and maxRetries is always >= retry. If the BackoffFunc is nil, it uses DefaultBackoff.

var DefaultBackoffFunc BackoffFunc = DefaultBackoff

DefaultBackoffFunc is a BackoffFunc pointer to DefaultBackoff (i.e., the default backoff for retried requests in the event of a failure).

type BadStatusError

type BadStatusError struct {
	Code int
	Body []byte
	Err  error
}

BadStatusError is any error that occurs as a result of a request failing. It includes the response code, body, and any error that occurred as a result of reading the body (never EOF).

func (*BadStatusError) Error

func (e *BadStatusError) Error() string

func (*BadStatusError) InfluxError

func (e *BadStatusError) InfluxError() (*InfluxError, error)

InfluxError attempts to parse and return the BadStatusError's body as an InfluxDB error (i.e., JSON of {"error":"description"}). If there is an error parsing it as an InfluxError, it will return either a JSON-specific error or ErrNotInfluxError if the body was empty or not a JSON object.

type Director

type Director func(*http.Request) error

A Director is responsible for configuring an HTTP request as needed before sending it. If the Director returns an error, the request is discarded immediately.

type FixedBackoff

type FixedBackoff time.Duration

FixedBackoff defines a fixed backoff for a Proxy.

func (FixedBackoff) Backoff

func (d FixedBackoff) Backoff(int, int) time.Duration

Backoff returns the concrete duration defined by the receiver regardless of retries attempted.

type FlushSize

type FlushSize int

FlushSize controls the minimum size to exceed before the Proxy will auto-flush itself.

type InfluxError

type InfluxError struct {
	Error string `json:"error"`
}

InfluxError is a generic error message from InfluxDB.

type Logger

type Logger interface {
	Print(...interface{})
}

Logger is a basic logging interface that outflux uses to handle its logging behavior. You can set the logger by modifying the Log package variable. This is not safe for concurrent modification by virtue of it being a package variable, so you should only set it once at program startup, before you've given outflux a reason to log anything.

var Log Logger

Log is the Logger used by outflux. If nil, outflux will not log anything. All outflux log messages are prefixed with "outflux: " to identify them.

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option is any configuration option capable of configuring a Proxy on creation.

func RequestLimit

func RequestLimit(n int) Option

RequestLimit controls the maximum number of concurrent requests a proxy may send at a time by assigning it a task queue. It returns an Option usable when creating a Proxy.

If the same Option is used to create multiple proxies, each Proxy will share the same task queue. This can be used to ensure that multiple proxies (e.g., to different DBs or retention policies) can compete for a small set of shared resources. By default, proxies do not have request limits. A limit where n <= 0 removes the limit.

type Proxy

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

Proxy is a basic InfluxDB line protocol proxy. You may write measurements to it either using dagr or just via functions such as fmt.Fprintf. Writes are accumulated for a given duration then POST-ed to the URL the Proxy was configured with. It is safe to write to the Proxy while it is sending. Concurrent writes to the Proxy are safe, but you should ensure that all writes are atomic and contain all necessary data or occur inside of a Transaction call to ensure that nothing slips in between writes.

func New

func New(client *http.Client, destURL string, opts ...Option) *Proxy

New allocates a new Proxy with the given context, HTTP client, and URL. Unlike NewURL, this will parse the URL first. If the URL is empty, New panics. See NewURL for further information.

func NewURL

func NewURL(client *http.Client, destURL *url.URL, opts ...Option) *Proxy

NewURL allocates a new Proxy with a given context, HTTP client, and URL. If the URL is nil, NewURL panics. If the context is nil, a new background context is allocated specifically for the Proxy.

Additional configuration can be provided by passing Option values, such as Timeout and FlushSize.

If the HTTP client given is nil, NewURL will use http.DefaultClient.

func (*Proxy) Flush

func (w *Proxy) Flush(ctx context.Context) error

Flush forces the Proxy to send out all buffered measurement data as soon as possible. It returns once the flush has been received by the Proxy or the Proxy is closed. This only works after Start() has been called.

Flush will block until the write completes and return any relevant error that occurred during the send.

The context may not be nil.

func (*Proxy) Start

func (w *Proxy) Start(ctx context.Context, interval time.Duration)

Start creates a goroutine that POSTs buffered data at the given interval. If interval is not a positive duration, the Proxy will only send data when you call Flush or if the Proxy has been configured to send when exceeding a certain buffer size. The context passed may be used to signal cancellation or provide a hard deadline for the proxy to stop by.

The context may not be nil.

func (*Proxy) Transaction

func (w *Proxy) Transaction(fn WriteFunc) (err error)

Transaction locks the Proxy's write buffer and passes it to fn. Once fn completes, the lock is released. This is shorthand for just doing that yourself, in the event that you have a function to pass a writer to but want to avoid writing extra code. For example:

err := proxy.Transaction(func(w io.Writer) {
        _, err := buf.WriteTo(w)
        return err
})

if err != nil {
        // ...
}

The WriteFunc given may return an error. This has no effect on the outcome of the transaction and is entirely for convenience. If the Proxy is closed, it will return the context error for its closure.

func (*Proxy) Write

func (w *Proxy) Write(b []byte) (int, error)

Write writes the byte slice b to the write buffer of the Proxy. WriteMeasurements should be preferred to ensure that the writer is correctly sending InfluxDB line protocol messages, but may be used as a raw writer to the underlying Proxy buffers.

func (*Proxy) WriteMeasurement

func (w *Proxy) WriteMeasurement(measurement dagr.Measurement) (n int64, err error)

WriteMeasurement writes a single measurement to the Proxy.

func (*Proxy) WriteMeasurements

func (w *Proxy) WriteMeasurements(measurements ...dagr.Measurement) (n int64, err error)

WriteMeasurements writes all measurements in measurements to the Proxy, effectively queueing them for delivery.

func (*Proxy) WritePoint

func (w *Proxy) WritePoint(key string, when time.Time, tags dagr.Tags, fields dagr.Fields) (n int64, err error)

WritePoint writes a single point to the Proxy.

func (*Proxy) Writer

func (w *Proxy) Writer() io.WriteCloser

Writer returns a locked writer for the Proxy's write buffer. It must be closed to release the lock. Changes to the Writer are not counted against the flush size, as the writer is not tracked by the Proxy.

type RetryLimit

type RetryLimit int

RetryLimit controls the number of retries a proxy is allowed to make before giving up on sending a request.

type Timeout

type Timeout time.Duration

Timeout controls the timeout for InfluxDB requests. If the timeout is <= 0, soft timeouts are disabled. This does not affect client / transport and server timeouts, the former of which must be provided by way of an HTTP client on creation.

type WriteFunc

type WriteFunc func(io.Writer) error

WriteFunc is any transactional function that accepts an io.Writer. The writer received by the WriteFunc is only valid until it returns.