influx

Documentation

Overview

Package influx provides an InfluxDB implementation for metrics. The model is similar to other push-based instrumentation systems. Observations are aggregated locally and emitted to the Influx server on regular intervals.

Index

• type BatchPointsWriter
• type Counter
  • func (c *Counter) Add(delta float64)
  • func (c *Counter) With(labelValues ...string) metrics.Counter
• type Gauge
  • func (g *Gauge) Add(delta float64)
  • func (g *Gauge) Set(value float64)
  • func (g *Gauge) With(labelValues ...string) metrics.Gauge
• type Histogram
  • func (h *Histogram) Observe(value float64)
  • func (h *Histogram) With(labelValues ...string) metrics.Histogram
• type Influx
  • func New(tags map[string]string, conf influxdb.BatchPointsConfig, logger log.Logger) *Influx
  • func (in *Influx) NewCounter(name string) *Counter
  • func (in *Influx) NewGauge(name string) *Gauge
  • func (in *Influx) NewHistogram(name string) *Histogram
  • func (in *Influx) WriteLoop(ctx context.Context, c <-chan time.Time, w BatchPointsWriter)
  • func (in *Influx) WriteTo(w BatchPointsWriter) (err error)

Examples

• Counter
• Gauge
• Histogram

Types

type BatchPointsWriter

type BatchPointsWriter interface {
	Write(influxdb.BatchPoints) error
}

BatchPointsWriter captures a subset of the influxdb.Client methods necessary for emitting metrics observations.

type Counter

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

Counter is an Influx counter. Observations are forwarded to an Influx object, and aggregated (summed) per timeseries.

Example

in := New(map[string]string{"a": "b"}, influxdb.BatchPointsConfig{}, log.NewNopLogger())
counter := in.NewCounter("influx_counter")
counter.Add(10)
counter.With("error", "true").Add(1)
counter.With("error", "false").Add(2)
counter.Add(50)

client := &bufWriter{}
in.WriteTo(client)

expectedLines := []string{
	`(influx_counter,a=b count=60) [0-9]{19}`,
	`(influx_counter,a=b,error=true count=1) [0-9]{19}`,
	`(influx_counter,a=b,error=false count=2) [0-9]{19}`,
}

if err := extractAndPrintMessage(expectedLines, client.buf.String()); err != nil {
	fmt.Println(err.Error())
}

Output:

influx_counter,a=b count=60
influx_counter,a=b,error=true count=1
influx_counter,a=b,error=false count=2

func (*Counter) Add

func (c *Counter) Add(delta float64)

Add implements metrics.Counter.

func (*Counter) With

func (c *Counter) With(labelValues ...string) metrics.Counter

With implements metrics.Counter.

type Gauge

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

Gauge is an Influx gauge. Observations are forwarded to a Dogstatsd object, and aggregated (the last observation selected) per timeseries.

Example

in := New(map[string]string{"a": "b"}, influxdb.BatchPointsConfig{}, log.NewNopLogger())
gauge := in.NewGauge("influx_gauge")
gauge.Set(10)
gauge.With("error", "true").Set(2)
gauge.With("error", "true").Set(1)
gauge.With("error", "false").Set(2)
gauge.Set(50)
gauge.With("test", "true").Set(1)
gauge.With("test", "true").Add(1)

client := &bufWriter{}
in.WriteTo(client)

expectedLines := []string{
	`(influx_gauge,a=b,test=true value=2) [0-9]{19}`,
	`(influx_gauge,a=b value=50) [0-9]{19}`,
	`(influx_gauge,a=b,error=true value=1) [0-9]{19}`,
	`(influx_gauge,a=b,error=false value=2) [0-9]{19}`,
}

if err := extractAndPrintMessage(expectedLines, client.buf.String()); err != nil {
	fmt.Println(err.Error())
}

Output:

influx_gauge,a=b,test=true value=2
influx_gauge,a=b value=50
influx_gauge,a=b,error=true value=1
influx_gauge,a=b,error=false value=2

func (*Gauge) Add

func (g *Gauge) Add(delta float64)

Add implements metrics.Gauge.

func (*Gauge) Set

func (g *Gauge) Set(value float64)

Set implements metrics.Gauge.

func (*Gauge) With

func (g *Gauge) With(labelValues ...string) metrics.Gauge

With implements metrics.Gauge.

type Histogram

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

Histogram is an Influx histrogram. Observations are aggregated into a generic.Histogram and emitted as per-quantile gauges to the Influx server.

Example

in := New(map[string]string{"foo": "alpha"}, influxdb.BatchPointsConfig{}, log.NewNopLogger())
histogram := in.NewHistogram("influx_histogram")
histogram.Observe(float64(10))
histogram.With("error", "true").Observe(float64(1))
histogram.With("error", "false").Observe(float64(2))
histogram.Observe(float64(50))

client := &bufWriter{}
in.WriteTo(client)

expectedLines := []string{
	`(influx_histogram,foo=alpha p50=10,p90=50,p95=50,p99=50) [0-9]{19}`,
	`(influx_histogram,error=true,foo=alpha p50=1,p90=1,p95=1,p99=1) [0-9]{19}`,
	`(influx_histogram,error=false,foo=alpha p50=2,p90=2,p95=2,p99=2) [0-9]{19}`,
}

if err := extractAndPrintMessage(expectedLines, client.buf.String()); err != nil {
	fmt.Println(err.Error())
}

Output:

influx_histogram,foo=alpha p50=10,p90=50,p95=50,p99=50
influx_histogram,error=true,foo=alpha p50=1,p90=1,p95=1,p99=1
influx_histogram,error=false,foo=alpha p50=2,p90=2,p95=2,p99=2

func (*Histogram) Observe

func (h *Histogram) Observe(value float64)

Observe implements metrics.Histogram.

func (*Histogram) With

func (h *Histogram) With(labelValues ...string) metrics.Histogram

With implements metrics.Histogram.

type Influx

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

Influx is a store for metrics that will be emitted to an Influx database.

Influx is a general purpose time-series database, and has no native concepts of counters, gauges, or histograms. Counters are modeled as a timeseries with one data point per flush, with a "count" field that reflects all adds since the last flush. Gauges are modeled as a timeseries with one data point per flush, with a "value" field that reflects the current state of the gauge. Histograms are modeled as a timeseries with one data point per combination of tags, with a set of quantile fields that reflects the p50, p90, p95 & p99.

Influx tags are attached to the Influx object, can be given to each metric at construction and can be updated anytime via With function. Influx fields are mapped to Go kit label values directly by this collector. Actual metric values are provided as fields with specific names depending on the metric.

All observations are collected in memory locally, and flushed on demand.

func New

func New(tags map[string]string, conf influxdb.BatchPointsConfig, logger log.Logger) *Influx

New returns an Influx, ready to create metrics and collect observations. Tags are applied to all metrics created from this object. The BatchPointsConfig is used during flushing.

func (*Influx) NewCounter

func (in *Influx) NewCounter(name string) *Counter

NewCounter returns an Influx counter.

func (*Influx) NewGauge

func (in *Influx) NewGauge(name string) *Gauge

NewGauge returns an Influx gauge.

func (*Influx) NewHistogram

func (in *Influx) NewHistogram(name string) *Histogram

NewHistogram returns an Influx histogram.

func (*Influx) WriteLoop

func (in *Influx) WriteLoop(ctx context.Context, c <-chan time.Time, w BatchPointsWriter)

WriteLoop is a helper method that invokes WriteTo to the passed writer every time the passed channel fires. This method blocks until the channel is closed, so clients probably want to run it in its own goroutine. For typical usage, create a time.Ticker and pass its C channel to this method.

func (*Influx) WriteTo

func (in *Influx) WriteTo(w BatchPointsWriter) (err error)

WriteTo flushes the buffered content of the metrics to the writer, in an Influx BatchPoints format. WriteTo abides best-effort semantics, so observations are lost if there is a problem with the write. Clients should be sure to call WriteTo regularly, ideally through the WriteLoop helper method.