Package influx

type BatchPointsWriter ¶

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

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

type Counter ¶

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

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

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

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 ¶

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

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

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

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 ¶

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

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

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

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 ¶

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.

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

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