Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

feat(prometheus): add transformers and codecs

pull/11347/head
Chris Goller 2019-01-20 12:26:22 -06:00
parent e5c8e545b9
commit a8a48ac5d7
5 changed files with 243 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
prometheus/codec.go Normal file
View File

@ -0,0 +1,66 @@
package prometheus
import (
"bytes"
"encoding/json"
"fmt"
"io"
dto "github.com/prometheus/client_model/go"
"github.com/prometheus/common/expfmt"
)
// DecodeExpfmt decodes the reader of format into metric families.
func DecodeExpfmt(r io.Reader, format expfmt.Format) ([]*dto.MetricFamily, error) {
dec := expfmt.NewDecoder(r, format)
mfs := []*dto.MetricFamily{}
for {
var mf dto.MetricFamily
if err := dec.Decode(&mf); err != nil {
if err == io.EOF {
break
}
if err != nil {
fmt.Println("decode error")
return nil, err
}
}
mfs = append(mfs, &mf)
}
return mfs, nil
}
// EncodeExpfmt encodes the metrics family with delimited
// protobuf (expt.FmtProtoDelim).
func EncodeExpfmt(mfs []*dto.MetricFamily) ([]byte, error) {
buf := &bytes.Buffer{}
enc := expfmt.NewEncoder(buf, expfmt.FmtProtoDelim)
for _, mf := range mfs {
if err := enc.Encode(mf); err != nil {
return nil, err
}
}
return buf.Bytes(), nil
}
// DecodeJSON decodes a JSON array of metrics families.
func DecodeJSON(r io.Reader) ([]*dto.MetricFamily, error) {
dec := json.NewDecoder(r)
families := []*dto.MetricFamily{}
for {
mfs := []*dto.MetricFamily{}
if err := dec.Decode(&mfs); err == io.EOF {
break
} else if err != nil {
return nil, err
}
families = append(families, mfs...)
}
return families, nil
}
// EncodeJSON encodes the metric families to JSON.
func EncodeJSON(mfs []*dto.MetricFamily) ([]byte, error) {
return json.Marshal(mfs)
}

125
prometheus/codec_test.go Normal file
View File

@ -0,0 +1,125 @@
package prometheus_test
import (
"bytes"
"testing"
pr "github.com/influxdata/influxdb/prometheus"
dto "github.com/prometheus/client_model/go"
"github.com/prometheus/common/expfmt"
)
func Test_CodecExpfmt(t *testing.T) {
mf1 := []*dto.MetricFamily{NewCounter("mf1", 1.0, pr.L("n1", "v1"))}
mf2 := []*dto.MetricFamily{NewCounter("mf2", 1.0, pr.L("n2", "v2"))}
b1, err := pr.EncodeExpfmt(mf1)
if err != nil {
t.Fatalf("encodeExpfmt() error = %v", err)
}
got1, err := pr.DecodeExpfmt(bytes.NewBuffer(b1), expfmt.FmtProtoDelim)
if err != nil {
t.Fatalf("decodeExpfmt() error = %v", err)
}
for i := range got1 {
if got1[i].String() != mf1[i].String() {
t.Errorf("codec1() = %v, want %v", got1[i].String(), mf1[i].String())
}
}
b2, err := pr.EncodeExpfmt(mf2)
if err != nil {
t.Fatalf("encodeExpfmt() error = %v", err)
}
got2, err := pr.DecodeExpfmt(bytes.NewBuffer(b2), expfmt.FmtProtoDelim)
if err != nil {
t.Fatalf("decodeExpfmt() error = %v", err)
}
for i := range got2 {
if got2[i].String() != mf2[i].String() {
t.Errorf("codec2() = %v, want %v", got2[i].String(), mf2[i].String())
}
}
b3 := append(b2, b1...)
b3 = append(b3, b2...)
mf3 := []*dto.MetricFamily{
NewCounter("mf2", 1.0, pr.L("n2", "v2")),
NewCounter("mf1", 1.0, pr.L("n1", "v1")),
NewCounter("mf2", 1.0, pr.L("n2", "v2")),
}
got3, err := pr.DecodeExpfmt(bytes.NewBuffer(b3), expfmt.FmtProtoDelim)
if err != nil {
t.Fatalf("decodeExpfmt() error = %v", err)
}
for i := range got3 {
if got3[i].String() != mf3[i].String() {
t.Errorf("codec3() = %v, want %v", got3[i].String(), mf3[i].String())
}
}
}
func Test_CodecJSON(t *testing.T) {
mf1 := []*dto.MetricFamily{NewCounter("mf1", 1.0, pr.L("n1", "v1")), NewCounter("mf1", 1.0, pr.L("n1", "v1"))}
mf2 := []*dto.MetricFamily{NewCounter("mf2", 1.0, pr.L("n2", "v2"))}
b1, err := pr.EncodeJSON(mf1)
if err != nil {
t.Fatalf("encodeJSON() error = %v", err)
}
got1, err := pr.DecodeJSON(bytes.NewBuffer(b1))
if err != nil {
t.Fatalf("decodeJSON() error = %v", err)
}
for i := range got1 {
if got1[i].String() != mf1[i].String() {
t.Errorf("codec1() = %v, want %v", got1[i].String(), mf1[i].String())
}
}
b2, err := pr.EncodeJSON(mf2)
if err != nil {
t.Fatalf("encodeJSON() error = %v", err)
}
got2, err := pr.DecodeJSON(bytes.NewBuffer(b2))
if err != nil {
t.Fatalf("decodeJSON() error = %v", err)
}
for i := range got2 {
if got2[i].String() != mf2[i].String() {
t.Errorf("codec2() = %v, want %v", got2[i].String(), mf2[i].String())
}
}
b3 := append(b2, b1...)
b3 = append(b3, b2...)
mf3 := []*dto.MetricFamily{
NewCounter("mf2", 1.0, pr.L("n2", "v2")),
NewCounter("mf1", 1.0, pr.L("n1", "v1")),
NewCounter("mf1", 1.0, pr.L("n1", "v1")),
NewCounter("mf2", 1.0, pr.L("n2", "v2")),
}
got3, err := pr.DecodeJSON(bytes.NewBuffer(b3))
if err != nil {
t.Fatalf("decodeJSON() error = %v", err)
}
for i := range got3 {
if got3[i].String() != mf3[i].String() {
t.Errorf("codec3() = %v, want %v", got3[i].String(), mf3[i].String())
}
}
}

59
prometheus/filter_test.go
View File

@ -1,4 +1,4 @@
package prometheus
package prometheus_test
import (
"fmt"
@ -6,6 +6,7 @@ import (
"testing"
proto "github.com/golang/protobuf/proto"
pr "github.com/influxdata/influxdb/prometheus"
"github.com/prometheus/client_golang/prometheus"
dto "github.com/prometheus/client_model/go"
)
@ -13,7 +14,7 @@ import (
func TestFilter_Gather(t *testing.T) {
type fields struct {
Gatherer prometheus.Gatherer
Matcher Matcher
Matcher pr.Matcher
}
tests := []struct {
name string
@ -27,12 +28,12 @@ func TestFilter_Gather(t *testing.T) {
Gatherer: prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) {
return nil, nil
}),
Matcher: NewMatcher().
Matcher: pr.NewMatcher().
Family("http_api_requests_total",
L("handler", "platform"),
L("method", "GET"),
L("path", "/api/v2"),
L("status", "2XX"),
pr.L("handler", "platform"),
pr.L("method", "GET"),
pr.L("path", "/api/v2"),
pr.L("status", "2XX"),
),
},
},
@ -55,12 +56,12 @@ func TestFilter_Gather(t *testing.T) {
}
return []*dto.MetricFamily{mf}, nil
}),
Matcher: NewMatcher().
Matcher: pr.NewMatcher().
Family("http_api_requests_total",
L("handler", "platform"),
L("method", "GET"),
L("path", "/api/v2"),
L("status", "2XX"),
pr.L("handler", "platform"),
pr.L("method", "GET"),
pr.L("path", "/api/v2"),
pr.L("status", "2XX"),
),
},
want: []*dto.MetricFamily{},
@ -74,7 +75,7 @@ func TestFilter_Gather(t *testing.T) {
}
return []*dto.MetricFamily{mf}, nil
}),
Matcher: NewMatcher().
Matcher: pr.NewMatcher().
Family("go_memstats_frees_total"),
},
want: []*dto.MetricFamily{},
@ -85,7 +86,7 @@ func TestFilter_Gather(t *testing.T) {
Gatherer: prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) {
return []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0)}, nil
}),
Matcher: NewMatcher().
Matcher: pr.NewMatcher().
Family("go_memstats_frees_total"),
},
want: []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0)},
@ -94,28 +95,28 @@ func TestFilter_Gather(t *testing.T) {
name: "matching with labels a family with labels matches",
fields: fields{
Gatherer: prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) {
return []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, L("n1", "v1"))}, nil
return []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, pr.L("n1", "v1"))}, nil
}),
Matcher: NewMatcher().
Family("go_memstats_frees_total", L("n1", "v1")),
Matcher: pr.NewMatcher().
Family("go_memstats_frees_total", pr.L("n1", "v1")),
},
want: []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, L("n1", "v1"))},
want: []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, pr.L("n1", "v1"))},
},
{
name: "matching a family that has no labels with labels matches",
fields: fields{
Gatherer: prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) {
return []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, L("n1", "v1"))}, nil
return []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, pr.L("n1", "v1"))}, nil
}),
Matcher: NewMatcher().
Matcher: pr.NewMatcher().
Family("go_memstats_frees_total"),
},
want: []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, L("n1", "v1"))},
want: []*dto.MetricFamily{NewCounter("go_memstats_frees_total", 1.0, pr.L("n1", "v1"))},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
f := &Filter{
f := &pr.Filter{
Gatherer: tt.fields.Gatherer,
Matcher: tt.fields.Matcher,
}
@ -130,17 +131,3 @@ func TestFilter_Gather(t *testing.T) {
})
}
}
func NewCounter(name string, v float64, ls ...*dto.LabelPair) *dto.MetricFamily {
m := &dto.Metric{
Label: ls,
Counter: &dto.Counter{
Value: &v,
},
}
return &dto.MetricFamily{
Name: proto.String(name),
Type: dto.MetricType_COUNTER.Enum(),
Metric: []*dto.Metric{m},
}
}

20
prometheus/prometheus_test.go Normal file
View File

@ -0,0 +1,20 @@
package prometheus_test
import (
proto "github.com/golang/protobuf/proto"
dto "github.com/prometheus/client_model/go"
)
func NewCounter(name string, v float64, ls ...*dto.LabelPair) *dto.MetricFamily {
m := &dto.Metric{
Label: ls,
Counter: &dto.Counter{
Value: &v,
},
}
return &dto.MetricFamily{
Name: proto.String(name),
Type: dto.MetricType_COUNTER.Enum(),
Metric: []*dto.Metric{m},
}
}

9
prometheus/transformer.go Normal file
View File

@ -0,0 +1,9 @@
package prometheus
import dto "github.com/prometheus/client_model/go"
// Transformer modifies prometheus metrics families.
type Transformer interface {
// Transform updates the metrics family
Transform(mfs []*dto.MetricFamily) []*dto.MetricFamily
}