Add prometheus serializer and use it in prometheus output (#6703)
This commit is contained in:
Daniel Nelson
GitHub
@@ -1,6 +1,7 @@
|
||||
# Prometheus Client Service Output Plugin
|
||||
# Prometheus Output Plugin
|
||||
|
||||
This plugin starts a [Prometheus](https://prometheus.io/) Client, it exposes all metrics on `/metrics` (default) to be polled by a Prometheus server.
|
||||
This plugin starts a [Prometheus](https://prometheus.io/) Client, it exposes
|
||||
all metrics on `/metrics` (default) to be polled by a Prometheus server.
|
||||
|
||||
## Configuration
|
||||
|
||||
@@ -10,6 +11,14 @@ This plugin starts a [Prometheus](https://prometheus.io/) Client, it exposes all
|
||||
## Address to listen on.
|
||||
listen = ":9273"
|
||||
|
||||
## Metric version controls the mapping from Telegraf metrics into
|
||||
## Prometheus format. When using the prometheus input, use the same value in
|
||||
## both plugins to ensure metrics are round-tripped without modification.
|
||||
##
|
||||
## example: metric_version = 1; deprecated in 1.13
|
||||
## metric_version = 2; recommended version
|
||||
# metric_version = 1
|
||||
|
||||
## Use HTTP Basic Authentication.
|
||||
# basic_username = "Foo"
|
||||
# basic_password = "Bar"
|
||||
|
||||
@@ -1,18 +1,12 @@
|
||||
package prometheus_client
|
||||
package prometheus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -20,73 +14,30 @@ import (
|
||||
"github.com/influxdata/telegraf/internal"
|
||||
tlsint "github.com/influxdata/telegraf/internal/tls"
|
||||
"github.com/influxdata/telegraf/plugins/outputs"
|
||||
"github.com/influxdata/telegraf/plugins/outputs/prometheus_client/v1"
|
||||
"github.com/influxdata/telegraf/plugins/outputs/prometheus_client/v2"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
)
|
||||
|
||||
var (
|
||||
invalidNameCharRE = regexp.MustCompile(`[^a-zA-Z0-9_:]`)
|
||||
validNameCharRE = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*`)
|
||||
defaultListen = ":9273"
|
||||
defaultPath = "/metrics"
|
||||
defaultExpirationInterval = internal.Duration{Duration: 60 * time.Second}
|
||||
)
|
||||
|
||||
// SampleID uniquely identifies a Sample
|
||||
type SampleID string
|
||||
|
||||
// Sample represents the current value of a series.
|
||||
type Sample struct {
|
||||
// Labels are the Prometheus labels.
|
||||
Labels map[string]string
|
||||
// Value is the value in the Prometheus output. Only one of these will populated.
|
||||
Value float64
|
||||
HistogramValue map[float64]uint64
|
||||
SummaryValue map[float64]float64
|
||||
// Histograms and Summaries need a count and a sum
|
||||
Count uint64
|
||||
Sum float64
|
||||
// Metric timestamp
|
||||
Timestamp time.Time
|
||||
// Expiration is the deadline that this Sample is valid until.
|
||||
Expiration time.Time
|
||||
}
|
||||
|
||||
// MetricFamily contains the data required to build valid prometheus Metrics.
|
||||
type MetricFamily struct {
|
||||
// Samples are the Sample belonging to this MetricFamily.
|
||||
Samples map[SampleID]*Sample
|
||||
// Need the telegraf ValueType because there isn't a Prometheus ValueType
|
||||
// representing Histogram or Summary
|
||||
TelegrafValueType telegraf.ValueType
|
||||
// LabelSet is the label counts for all Samples.
|
||||
LabelSet map[string]int
|
||||
}
|
||||
|
||||
type PrometheusClient struct {
|
||||
Listen string
|
||||
BasicUsername string `toml:"basic_username"`
|
||||
BasicPassword string `toml:"basic_password"`
|
||||
IPRange []string `toml:"ip_range"`
|
||||
ExpirationInterval internal.Duration `toml:"expiration_interval"`
|
||||
Path string `toml:"path"`
|
||||
CollectorsExclude []string `toml:"collectors_exclude"`
|
||||
StringAsLabel bool `toml:"string_as_label"`
|
||||
ExportTimestamp bool `toml:"export_timestamp"`
|
||||
|
||||
tlsint.ServerConfig
|
||||
|
||||
server *http.Server
|
||||
url string
|
||||
|
||||
sync.Mutex
|
||||
// fam is the non-expired MetricFamily by Prometheus metric name.
|
||||
fam map[string]*MetricFamily
|
||||
// now returns the current time.
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
var sampleConfig = `
|
||||
## Address to listen on
|
||||
listen = ":9273"
|
||||
|
||||
## Metric version controls the mapping from Telegraf metrics into
|
||||
## Prometheus format. When using the prometheus input, use the same value in
|
||||
## both plugins to ensure metrics are round-tripped without modification.
|
||||
##
|
||||
## example: metric_version = 1; deprecated in 1.13
|
||||
## metric_version = 2; recommended version
|
||||
# metric_version = 1
|
||||
|
||||
## Use HTTP Basic Authentication.
|
||||
# basic_username = "Foo"
|
||||
# basic_password = "Bar"
|
||||
@@ -121,46 +72,42 @@ var sampleConfig = `
|
||||
# export_timestamp = false
|
||||
`
|
||||
|
||||
func (p *PrometheusClient) auth(h http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if p.BasicUsername != "" && p.BasicPassword != "" {
|
||||
w.Header().Set("WWW-Authenticate", `Basic realm="Restricted"`)
|
||||
|
||||
username, password, ok := r.BasicAuth()
|
||||
if !ok ||
|
||||
subtle.ConstantTimeCompare([]byte(username), []byte(p.BasicUsername)) != 1 ||
|
||||
subtle.ConstantTimeCompare([]byte(password), []byte(p.BasicPassword)) != 1 {
|
||||
http.Error(w, "Not authorized", 401)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if len(p.IPRange) > 0 {
|
||||
matched := false
|
||||
remoteIPs, _, _ := net.SplitHostPort(r.RemoteAddr)
|
||||
remoteIP := net.ParseIP(remoteIPs)
|
||||
for _, iprange := range p.IPRange {
|
||||
_, ipNet, err := net.ParseCIDR(iprange)
|
||||
if err != nil {
|
||||
http.Error(w, "Config Error in ip_range setting", 500)
|
||||
return
|
||||
}
|
||||
if ipNet.Contains(remoteIP) {
|
||||
matched = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !matched {
|
||||
http.Error(w, "Not authorized", 401)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
h.ServeHTTP(w, r)
|
||||
})
|
||||
type Collector interface {
|
||||
Describe(ch chan<- *prometheus.Desc)
|
||||
Collect(ch chan<- prometheus.Metric)
|
||||
Add(metrics []telegraf.Metric) error
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) Connect() error {
|
||||
type PrometheusClient struct {
|
||||
Listen string `toml:"listen"`
|
||||
MetricVersion int `toml:"metric_version"`
|
||||
BasicUsername string `toml:"basic_username"`
|
||||
BasicPassword string `toml:"basic_password"`
|
||||
IPRange []string `toml:"ip_range"`
|
||||
ExpirationInterval internal.Duration `toml:"expiration_interval"`
|
||||
Path string `toml:"path"`
|
||||
CollectorsExclude []string `toml:"collectors_exclude"`
|
||||
StringAsLabel bool `toml:"string_as_label"`
|
||||
ExportTimestamp bool `toml:"export_timestamp"`
|
||||
tlsint.ServerConfig
|
||||
|
||||
Log telegraf.Logger `toml:"-"`
|
||||
|
||||
server *http.Server
|
||||
url *url.URL
|
||||
collector Collector
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) Description() string {
|
||||
return "Configuration for the Prometheus client to spawn"
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) SampleConfig() string {
|
||||
return sampleConfig
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) Init() error {
|
||||
defaultCollectors := map[string]bool{
|
||||
"gocollector": true,
|
||||
"process": true,
|
||||
@@ -181,421 +128,137 @@ func (p *PrometheusClient) Connect() error {
|
||||
}
|
||||
}
|
||||
|
||||
err := registry.Register(p)
|
||||
if err != nil {
|
||||
return err
|
||||
switch p.MetricVersion {
|
||||
default:
|
||||
fallthrough
|
||||
case 1:
|
||||
p.Log.Warnf("Use of deprecated configuration: metric_version = 1; please update to metric_version = 2")
|
||||
p.collector = v1.NewCollector(p.ExpirationInterval.Duration, p.StringAsLabel, p.Log)
|
||||
err := registry.Register(p.collector)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
case 2:
|
||||
p.collector = v2.NewCollector(p.ExpirationInterval.Duration, p.StringAsLabel)
|
||||
err := registry.Register(p.collector)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if p.Listen == "" {
|
||||
p.Listen = "localhost:9273"
|
||||
ipRange := make([]*net.IPNet, 0, len(p.IPRange))
|
||||
for _, cidr := range p.IPRange {
|
||||
_, ipNet, err := net.ParseCIDR(cidr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error parsing ip_range: %v", err)
|
||||
}
|
||||
|
||||
ipRange = append(ipRange, ipNet)
|
||||
}
|
||||
|
||||
if p.Path == "" {
|
||||
p.Path = "/metrics"
|
||||
}
|
||||
authHandler := internal.AuthHandler(p.BasicUsername, p.BasicPassword, onAuthError)
|
||||
rangeHandler := internal.IPRangeHandler(ipRange, onError)
|
||||
promHandler := promhttp.HandlerFor(registry, promhttp.HandlerOpts{ErrorHandling: promhttp.ContinueOnError})
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle(p.Path, p.auth(promhttp.HandlerFor(
|
||||
registry, promhttp.HandlerOpts{ErrorHandling: promhttp.ContinueOnError})))
|
||||
if p.Path == "" {
|
||||
p.Path = "/"
|
||||
}
|
||||
mux.Handle(p.Path, authHandler(rangeHandler(promHandler)))
|
||||
|
||||
tlsConfig, err := p.TLSConfig()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
p.server = &http.Server{
|
||||
Addr: p.Listen,
|
||||
Handler: mux,
|
||||
TLSConfig: tlsConfig,
|
||||
}
|
||||
|
||||
var listener net.Listener
|
||||
if tlsConfig != nil {
|
||||
listener, err = tls.Listen("tcp", p.Listen, tlsConfig)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) listen() (net.Listener, error) {
|
||||
if p.server.TLSConfig != nil {
|
||||
return tls.Listen("tcp", p.Listen, p.server.TLSConfig)
|
||||
} else {
|
||||
listener, err = net.Listen("tcp", p.Listen)
|
||||
return net.Listen("tcp", p.Listen)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) Connect() error {
|
||||
listener, err := p.listen()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
p.url = createURL(tlsConfig, listener, p.Path)
|
||||
scheme := "http"
|
||||
if p.server.TLSConfig != nil {
|
||||
scheme = "https"
|
||||
}
|
||||
|
||||
p.url = &url.URL{
|
||||
Scheme: scheme,
|
||||
Host: listener.Addr().String(),
|
||||
Path: p.Path,
|
||||
}
|
||||
|
||||
p.Log.Infof("Listening on %s", p.URL())
|
||||
|
||||
p.wg.Add(1)
|
||||
go func() {
|
||||
defer p.wg.Done()
|
||||
err := p.server.Serve(listener)
|
||||
if err != nil && err != http.ErrServerClosed {
|
||||
log.Printf("E! Error creating prometheus metric endpoint, err: %s\n",
|
||||
err.Error())
|
||||
p.Log.Errorf("Server error: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func onAuthError(rw http.ResponseWriter, code int) {
|
||||
rw.Header().Set("WWW-Authenticate", `Basic realm="Restricted"`)
|
||||
http.Error(rw, http.StatusText(code), code)
|
||||
}
|
||||
|
||||
func onError(rw http.ResponseWriter, code int) {
|
||||
http.Error(rw, http.StatusText(code), code)
|
||||
}
|
||||
|
||||
// Address returns the address the plugin is listening on. If not listening
|
||||
// an empty string is returned.
|
||||
func (p *PrometheusClient) URL() string {
|
||||
return p.url
|
||||
}
|
||||
|
||||
func createURL(tlsConfig *tls.Config, listener net.Listener, path string) string {
|
||||
u := url.URL{
|
||||
Scheme: "http",
|
||||
Host: listener.Addr().String(),
|
||||
Path: path,
|
||||
if p.url != nil {
|
||||
return p.url.String()
|
||||
}
|
||||
|
||||
if tlsConfig != nil {
|
||||
u.Scheme = "https"
|
||||
}
|
||||
return u.String()
|
||||
return ""
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) Close() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
err := p.server.Shutdown(ctx)
|
||||
prometheus.Unregister(p)
|
||||
p.url = ""
|
||||
p.wg.Wait()
|
||||
p.url = nil
|
||||
prometheus.Unregister(p.collector)
|
||||
return err
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) SampleConfig() string {
|
||||
return sampleConfig
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) Description() string {
|
||||
return "Configuration for the Prometheus client to spawn"
|
||||
}
|
||||
|
||||
// Implements prometheus.Collector
|
||||
func (p *PrometheusClient) Describe(ch chan<- *prometheus.Desc) {
|
||||
prometheus.NewGauge(prometheus.GaugeOpts{Name: "Dummy", Help: "Dummy"}).Describe(ch)
|
||||
}
|
||||
|
||||
// Expire removes Samples that have expired.
|
||||
func (p *PrometheusClient) Expire() {
|
||||
now := p.now()
|
||||
for name, family := range p.fam {
|
||||
for key, sample := range family.Samples {
|
||||
if p.ExpirationInterval.Duration != 0 && now.After(sample.Expiration) {
|
||||
for k := range sample.Labels {
|
||||
family.LabelSet[k]--
|
||||
}
|
||||
delete(family.Samples, key)
|
||||
|
||||
if len(family.Samples) == 0 {
|
||||
delete(p.fam, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Collect implements prometheus.Collector
|
||||
func (p *PrometheusClient) Collect(ch chan<- prometheus.Metric) {
|
||||
p.Lock()
|
||||
defer p.Unlock()
|
||||
|
||||
p.Expire()
|
||||
|
||||
for name, family := range p.fam {
|
||||
// Get list of all labels on MetricFamily
|
||||
var labelNames []string
|
||||
for k, v := range family.LabelSet {
|
||||
if v > 0 {
|
||||
labelNames = append(labelNames, k)
|
||||
}
|
||||
}
|
||||
desc := prometheus.NewDesc(name, "Telegraf collected metric", labelNames, nil)
|
||||
|
||||
for _, sample := range family.Samples {
|
||||
// Get labels for this sample; unset labels will be set to the
|
||||
// empty string
|
||||
var labels []string
|
||||
for _, label := range labelNames {
|
||||
v := sample.Labels[label]
|
||||
labels = append(labels, v)
|
||||
}
|
||||
|
||||
var metric prometheus.Metric
|
||||
var err error
|
||||
switch family.TelegrafValueType {
|
||||
case telegraf.Summary:
|
||||
metric, err = prometheus.NewConstSummary(desc, sample.Count, sample.Sum, sample.SummaryValue, labels...)
|
||||
case telegraf.Histogram:
|
||||
metric, err = prometheus.NewConstHistogram(desc, sample.Count, sample.Sum, sample.HistogramValue, labels...)
|
||||
default:
|
||||
metric, err = prometheus.NewConstMetric(desc, getPromValueType(family.TelegrafValueType), sample.Value, labels...)
|
||||
}
|
||||
if err != nil {
|
||||
log.Printf("E! Error creating prometheus metric, "+
|
||||
"key: %s, labels: %v,\nerr: %s\n",
|
||||
name, labels, err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
if p.ExportTimestamp {
|
||||
metric = prometheus.NewMetricWithTimestamp(sample.Timestamp, metric)
|
||||
}
|
||||
ch <- metric
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func sanitize(value string) string {
|
||||
return invalidNameCharRE.ReplaceAllString(value, "_")
|
||||
}
|
||||
|
||||
func isValidTagName(tag string) bool {
|
||||
return validNameCharRE.MatchString(tag)
|
||||
}
|
||||
|
||||
func getPromValueType(tt telegraf.ValueType) prometheus.ValueType {
|
||||
switch tt {
|
||||
case telegraf.Counter:
|
||||
return prometheus.CounterValue
|
||||
case telegraf.Gauge:
|
||||
return prometheus.GaugeValue
|
||||
default:
|
||||
return prometheus.UntypedValue
|
||||
}
|
||||
}
|
||||
|
||||
// CreateSampleID creates a SampleID based on the tags of a telegraf.Metric.
|
||||
func CreateSampleID(tags map[string]string) SampleID {
|
||||
pairs := make([]string, 0, len(tags))
|
||||
for k, v := range tags {
|
||||
pairs = append(pairs, fmt.Sprintf("%s=%s", k, v))
|
||||
}
|
||||
sort.Strings(pairs)
|
||||
return SampleID(strings.Join(pairs, ","))
|
||||
}
|
||||
|
||||
func addSample(fam *MetricFamily, sample *Sample, sampleID SampleID) {
|
||||
|
||||
for k := range sample.Labels {
|
||||
fam.LabelSet[k]++
|
||||
}
|
||||
|
||||
fam.Samples[sampleID] = sample
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) addMetricFamily(point telegraf.Metric, sample *Sample, mname string, sampleID SampleID) {
|
||||
var fam *MetricFamily
|
||||
var ok bool
|
||||
if fam, ok = p.fam[mname]; !ok {
|
||||
fam = &MetricFamily{
|
||||
Samples: make(map[SampleID]*Sample),
|
||||
TelegrafValueType: point.Type(),
|
||||
LabelSet: make(map[string]int),
|
||||
}
|
||||
p.fam[mname] = fam
|
||||
}
|
||||
|
||||
addSample(fam, sample, sampleID)
|
||||
}
|
||||
|
||||
// Sorted returns a copy of the metrics in time ascending order. A copy is
|
||||
// made to avoid modifying the input metric slice since doing so is not
|
||||
// allowed.
|
||||
func sorted(metrics []telegraf.Metric) []telegraf.Metric {
|
||||
batch := make([]telegraf.Metric, 0, len(metrics))
|
||||
for i := len(metrics) - 1; i >= 0; i-- {
|
||||
batch = append(batch, metrics[i])
|
||||
}
|
||||
sort.Slice(batch, func(i, j int) bool {
|
||||
return batch[i].Time().Before(batch[j].Time())
|
||||
})
|
||||
return batch
|
||||
}
|
||||
|
||||
func (p *PrometheusClient) Write(metrics []telegraf.Metric) error {
|
||||
p.Lock()
|
||||
defer p.Unlock()
|
||||
|
||||
now := p.now()
|
||||
|
||||
for _, point := range sorted(metrics) {
|
||||
tags := point.Tags()
|
||||
sampleID := CreateSampleID(tags)
|
||||
|
||||
labels := make(map[string]string)
|
||||
for k, v := range tags {
|
||||
tName := sanitize(k)
|
||||
if !isValidTagName(tName) {
|
||||
continue
|
||||
}
|
||||
labels[tName] = v
|
||||
}
|
||||
|
||||
// Prometheus doesn't have a string value type, so convert string
|
||||
// fields to labels if enabled.
|
||||
if p.StringAsLabel {
|
||||
for fn, fv := range point.Fields() {
|
||||
switch fv := fv.(type) {
|
||||
case string:
|
||||
tName := sanitize(fn)
|
||||
if !isValidTagName(tName) {
|
||||
continue
|
||||
}
|
||||
labels[tName] = fv
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
switch point.Type() {
|
||||
case telegraf.Summary:
|
||||
var mname string
|
||||
var sum float64
|
||||
var count uint64
|
||||
summaryvalue := make(map[float64]float64)
|
||||
for fn, fv := range point.Fields() {
|
||||
var value float64
|
||||
switch fv := fv.(type) {
|
||||
case int64:
|
||||
value = float64(fv)
|
||||
case uint64:
|
||||
value = float64(fv)
|
||||
case float64:
|
||||
value = fv
|
||||
default:
|
||||
continue
|
||||
}
|
||||
|
||||
switch fn {
|
||||
case "sum":
|
||||
sum = value
|
||||
case "count":
|
||||
count = uint64(value)
|
||||
default:
|
||||
limit, err := strconv.ParseFloat(fn, 64)
|
||||
if err == nil {
|
||||
summaryvalue[limit] = value
|
||||
}
|
||||
}
|
||||
}
|
||||
sample := &Sample{
|
||||
Labels: labels,
|
||||
SummaryValue: summaryvalue,
|
||||
Count: count,
|
||||
Sum: sum,
|
||||
Timestamp: point.Time(),
|
||||
Expiration: now.Add(p.ExpirationInterval.Duration),
|
||||
}
|
||||
mname = sanitize(point.Name())
|
||||
|
||||
if !isValidTagName(mname) {
|
||||
continue
|
||||
}
|
||||
|
||||
p.addMetricFamily(point, sample, mname, sampleID)
|
||||
|
||||
case telegraf.Histogram:
|
||||
var mname string
|
||||
var sum float64
|
||||
var count uint64
|
||||
histogramvalue := make(map[float64]uint64)
|
||||
for fn, fv := range point.Fields() {
|
||||
var value float64
|
||||
switch fv := fv.(type) {
|
||||
case int64:
|
||||
value = float64(fv)
|
||||
case uint64:
|
||||
value = float64(fv)
|
||||
case float64:
|
||||
value = fv
|
||||
default:
|
||||
continue
|
||||
}
|
||||
|
||||
switch fn {
|
||||
case "sum":
|
||||
sum = value
|
||||
case "count":
|
||||
count = uint64(value)
|
||||
default:
|
||||
limit, err := strconv.ParseFloat(fn, 64)
|
||||
if err == nil {
|
||||
histogramvalue[limit] = uint64(value)
|
||||
}
|
||||
}
|
||||
}
|
||||
sample := &Sample{
|
||||
Labels: labels,
|
||||
HistogramValue: histogramvalue,
|
||||
Count: count,
|
||||
Sum: sum,
|
||||
Timestamp: point.Time(),
|
||||
Expiration: now.Add(p.ExpirationInterval.Duration),
|
||||
}
|
||||
mname = sanitize(point.Name())
|
||||
|
||||
if !isValidTagName(mname) {
|
||||
continue
|
||||
}
|
||||
|
||||
p.addMetricFamily(point, sample, mname, sampleID)
|
||||
|
||||
default:
|
||||
for fn, fv := range point.Fields() {
|
||||
// Ignore string and bool fields.
|
||||
var value float64
|
||||
switch fv := fv.(type) {
|
||||
case int64:
|
||||
value = float64(fv)
|
||||
case uint64:
|
||||
value = float64(fv)
|
||||
case float64:
|
||||
value = fv
|
||||
default:
|
||||
continue
|
||||
}
|
||||
|
||||
sample := &Sample{
|
||||
Labels: labels,
|
||||
Value: value,
|
||||
Timestamp: point.Time(),
|
||||
Expiration: now.Add(p.ExpirationInterval.Duration),
|
||||
}
|
||||
|
||||
// Special handling of value field; supports passthrough from
|
||||
// the prometheus input.
|
||||
var mname string
|
||||
switch point.Type() {
|
||||
case telegraf.Counter:
|
||||
if fn == "counter" {
|
||||
mname = sanitize(point.Name())
|
||||
}
|
||||
case telegraf.Gauge:
|
||||
if fn == "gauge" {
|
||||
mname = sanitize(point.Name())
|
||||
}
|
||||
}
|
||||
if mname == "" {
|
||||
if fn == "value" {
|
||||
mname = sanitize(point.Name())
|
||||
} else {
|
||||
mname = sanitize(fmt.Sprintf("%s_%s", point.Name(), fn))
|
||||
}
|
||||
}
|
||||
if !isValidTagName(mname) {
|
||||
continue
|
||||
}
|
||||
p.addMetricFamily(point, sample, mname, sampleID)
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return p.collector.Add(metrics)
|
||||
}
|
||||
|
||||
func init() {
|
||||
outputs.Add("prometheus_client", func() telegraf.Output {
|
||||
return &PrometheusClient{
|
||||
ExpirationInterval: internal.Duration{Duration: time.Second * 60},
|
||||
Listen: defaultListen,
|
||||
Path: defaultPath,
|
||||
ExpirationInterval: defaultExpirationInterval,
|
||||
StringAsLabel: true,
|
||||
fam: make(map[string]*MetricFamily),
|
||||
now: time.Now,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1,693 +1,304 @@
|
||||
package prometheus_client
|
||||
package prometheus
|
||||
|
||||
import (
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/internal"
|
||||
"github.com/influxdata/telegraf/metric"
|
||||
prometheus_input "github.com/influxdata/telegraf/plugins/inputs/prometheus"
|
||||
"github.com/influxdata/telegraf/testutil"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func setUnixTime(client *PrometheusClient, sec int64) {
|
||||
client.now = func() time.Time {
|
||||
return time.Unix(sec, 0)
|
||||
}
|
||||
}
|
||||
|
||||
// NewClient initializes a PrometheusClient.
|
||||
func NewClient() *PrometheusClient {
|
||||
return &PrometheusClient{
|
||||
ExpirationInterval: internal.Duration{Duration: time.Second * 60},
|
||||
StringAsLabel: true,
|
||||
fam: make(map[string]*MetricFamily),
|
||||
now: time.Now,
|
||||
}
|
||||
}
|
||||
|
||||
func TestWrite_Basic(t *testing.T) {
|
||||
now := time.Now()
|
||||
pt1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 0.0},
|
||||
now)
|
||||
var metrics = []telegraf.Metric{
|
||||
pt1,
|
||||
}
|
||||
|
||||
client := NewClient()
|
||||
err = client.Write(metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, telegraf.Untyped, fam.TelegrafValueType)
|
||||
require.Equal(t, map[string]int{}, fam.LabelSet)
|
||||
|
||||
sample, ok := fam.Samples[CreateSampleID(pt1.Tags())]
|
||||
require.True(t, ok)
|
||||
|
||||
require.Equal(t, 0.0, sample.Value)
|
||||
require.True(t, now.Before(sample.Expiration))
|
||||
}
|
||||
|
||||
func TestWrite_IntField(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 42},
|
||||
time.Now())
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
for _, v := range fam.Samples {
|
||||
require.Equal(t, 42.0, v.Value)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func TestWrite_FieldNotValue(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"howdy": 0.0},
|
||||
time.Now())
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo_howdy"]
|
||||
require.True(t, ok)
|
||||
for _, v := range fam.Samples {
|
||||
require.Equal(t, 0.0, v.Value)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWrite_SkipNonNumberField(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": "howdy"},
|
||||
time.Now())
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
_, ok := client.fam["foo"]
|
||||
require.False(t, ok)
|
||||
}
|
||||
|
||||
func TestWrite_Counters(t *testing.T) {
|
||||
type args struct {
|
||||
measurement string
|
||||
tags map[string]string
|
||||
fields map[string]interface{}
|
||||
valueType telegraf.ValueType
|
||||
}
|
||||
var tests = []struct {
|
||||
name string
|
||||
args args
|
||||
err error
|
||||
metricName string
|
||||
valueType telegraf.ValueType
|
||||
func TestMetricVersion1(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
output *PrometheusClient
|
||||
metrics []telegraf.Metric
|
||||
expected []byte
|
||||
}{
|
||||
{
|
||||
name: "field named value is not added to metric name",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"value": 42},
|
||||
valueType: telegraf.Counter,
|
||||
name: "simple",
|
||||
output: &PrometheusClient{
|
||||
Listen: ":0",
|
||||
MetricVersion: 1,
|
||||
CollectorsExclude: []string{"gocollector", "process"},
|
||||
Path: "/metrics",
|
||||
Log: testutil.Logger{},
|
||||
},
|
||||
metricName: "foo",
|
||||
valueType: telegraf.Counter,
|
||||
metrics: []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"cpu",
|
||||
map[string]string{
|
||||
"host": "example.org",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"time_idle": 42.0,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
},
|
||||
expected: []byte(`
|
||||
# HELP cpu_time_idle Telegraf collected metric
|
||||
# TYPE cpu_time_idle untyped
|
||||
cpu_time_idle{host="example.org"} 42
|
||||
`),
|
||||
},
|
||||
{
|
||||
name: "field named counter is not added to metric name",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"counter": 42},
|
||||
valueType: telegraf.Counter,
|
||||
name: "prometheus untyped",
|
||||
output: &PrometheusClient{
|
||||
Listen: ":0",
|
||||
MetricVersion: 1,
|
||||
CollectorsExclude: []string{"gocollector", "process"},
|
||||
Path: "/metrics",
|
||||
Log: testutil.Logger{},
|
||||
},
|
||||
metricName: "foo",
|
||||
valueType: telegraf.Counter,
|
||||
metrics: []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"cpu_time_idle",
|
||||
map[string]string{
|
||||
"host": "example.org",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"value": 42.0,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
},
|
||||
expected: []byte(`
|
||||
# HELP cpu_time_idle Telegraf collected metric
|
||||
# TYPE cpu_time_idle untyped
|
||||
cpu_time_idle{host="example.org"} 42
|
||||
`),
|
||||
},
|
||||
{
|
||||
name: "field with any other name is added to metric name",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"other": 42},
|
||||
valueType: telegraf.Counter,
|
||||
name: "prometheus counter",
|
||||
output: &PrometheusClient{
|
||||
Listen: ":0",
|
||||
MetricVersion: 1,
|
||||
CollectorsExclude: []string{"gocollector", "process"},
|
||||
Path: "/metrics",
|
||||
Log: testutil.Logger{},
|
||||
},
|
||||
metricName: "foo_other",
|
||||
valueType: telegraf.Counter,
|
||||
metrics: []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"cpu_time_idle",
|
||||
map[string]string{
|
||||
"host": "example.org",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"counter": 42.0,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
telegraf.Counter,
|
||||
),
|
||||
},
|
||||
expected: []byte(`
|
||||
# HELP cpu_time_idle Telegraf collected metric
|
||||
# TYPE cpu_time_idle counter
|
||||
cpu_time_idle{host="example.org"} 42
|
||||
`),
|
||||
},
|
||||
{
|
||||
name: "uint64 fields are output",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"value": uint64(42)},
|
||||
valueType: telegraf.Counter,
|
||||
name: "prometheus gauge",
|
||||
output: &PrometheusClient{
|
||||
Listen: ":0",
|
||||
MetricVersion: 1,
|
||||
CollectorsExclude: []string{"gocollector", "process"},
|
||||
Path: "/metrics",
|
||||
Log: testutil.Logger{},
|
||||
},
|
||||
metricName: "foo",
|
||||
valueType: telegraf.Counter,
|
||||
metrics: []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"cpu_time_idle",
|
||||
map[string]string{
|
||||
"host": "example.org",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"gauge": 42.0,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
telegraf.Gauge,
|
||||
),
|
||||
},
|
||||
expected: []byte(`
|
||||
# HELP cpu_time_idle Telegraf collected metric
|
||||
# TYPE cpu_time_idle gauge
|
||||
cpu_time_idle{host="example.org"} 42
|
||||
`),
|
||||
},
|
||||
{
|
||||
name: "prometheus histogram",
|
||||
output: &PrometheusClient{
|
||||
Listen: ":0",
|
||||
MetricVersion: 1,
|
||||
CollectorsExclude: []string{"gocollector", "process"},
|
||||
Path: "/metrics",
|
||||
Log: testutil.Logger{},
|
||||
},
|
||||
metrics: []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"http_request_duration_seconds",
|
||||
map[string]string{},
|
||||
map[string]interface{}{
|
||||
"sum": 53423,
|
||||
"0.05": 24054,
|
||||
"0.1": 33444,
|
||||
"0.2": 100392,
|
||||
"0.5": 129389,
|
||||
"1": 133988,
|
||||
"+Inf": 144320,
|
||||
"count": 144320,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
telegraf.Histogram,
|
||||
),
|
||||
},
|
||||
expected: []byte(`
|
||||
# HELP http_request_duration_seconds Telegraf collected metric
|
||||
# TYPE http_request_duration_seconds histogram
|
||||
http_request_duration_seconds_bucket{le="0.05"} 24054
|
||||
http_request_duration_seconds_bucket{le="0.1"} 33444
|
||||
http_request_duration_seconds_bucket{le="0.2"} 100392
|
||||
http_request_duration_seconds_bucket{le="0.5"} 129389
|
||||
http_request_duration_seconds_bucket{le="1"} 133988
|
||||
http_request_duration_seconds_bucket{le="+Inf"} 144320
|
||||
http_request_duration_seconds_sum 53423
|
||||
http_request_duration_seconds_count 144320
|
||||
`),
|
||||
},
|
||||
{
|
||||
name: "prometheus summary",
|
||||
output: &PrometheusClient{
|
||||
Listen: ":0",
|
||||
MetricVersion: 1,
|
||||
CollectorsExclude: []string{"gocollector", "process"},
|
||||
Path: "/metrics",
|
||||
Log: testutil.Logger{},
|
||||
},
|
||||
metrics: []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"rpc_duration_seconds",
|
||||
map[string]string{},
|
||||
map[string]interface{}{
|
||||
"0.01": 3102,
|
||||
"0.05": 3272,
|
||||
"0.5": 4773,
|
||||
"0.9": 9001,
|
||||
"0.99": 76656,
|
||||
"count": 2693,
|
||||
"sum": 17560473,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
telegraf.Summary,
|
||||
),
|
||||
},
|
||||
expected: []byte(`
|
||||
# HELP rpc_duration_seconds Telegraf collected metric
|
||||
# TYPE rpc_duration_seconds summary
|
||||
rpc_duration_seconds{quantile="0.01"} 3102
|
||||
rpc_duration_seconds{quantile="0.05"} 3272
|
||||
rpc_duration_seconds{quantile="0.5"} 4773
|
||||
rpc_duration_seconds{quantile="0.9"} 9001
|
||||
rpc_duration_seconds{quantile="0.99"} 76656
|
||||
rpc_duration_seconds_sum 1.7560473e+07
|
||||
rpc_duration_seconds_count 2693
|
||||
`),
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
m, err := metric.New(
|
||||
tt.args.measurement,
|
||||
tt.args.tags,
|
||||
tt.args.fields,
|
||||
time.Now(),
|
||||
tt.args.valueType,
|
||||
)
|
||||
client := NewClient()
|
||||
err = client.Write([]telegraf.Metric{m})
|
||||
require.Equal(t, tt.err, err)
|
||||
err := tt.output.Init()
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam[tt.metricName]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, tt.valueType, fam.TelegrafValueType)
|
||||
err = tt.output.Connect()
|
||||
require.NoError(t, err)
|
||||
|
||||
defer func() {
|
||||
err := tt.output.Close()
|
||||
require.NoError(t, err)
|
||||
}()
|
||||
|
||||
err = tt.output.Write(tt.metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
resp, err := http.Get(tt.output.URL())
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
defer resp.Body.Close()
|
||||
body, err := ioutil.ReadAll(resp.Body)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t,
|
||||
strings.TrimSpace(string(tt.expected)),
|
||||
strings.TrimSpace(string(body)))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestWrite_Sanitize(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo.bar:colon",
|
||||
map[string]string{"tag-with-dash": "localhost.local"},
|
||||
map[string]interface{}{"field-with-dash-and:colon": 42},
|
||||
time.Now(),
|
||||
telegraf.Counter)
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo_bar:colon_field_with_dash_and:colon"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, map[string]int{"tag_with_dash": 1}, fam.LabelSet)
|
||||
|
||||
sample1, ok := fam.Samples[CreateSampleID(p1.Tags())]
|
||||
require.True(t, ok)
|
||||
|
||||
require.Equal(t, map[string]string{
|
||||
"tag_with_dash": "localhost.local"}, sample1.Labels)
|
||||
}
|
||||
|
||||
func TestWrite_Gauge(t *testing.T) {
|
||||
type args struct {
|
||||
measurement string
|
||||
tags map[string]string
|
||||
fields map[string]interface{}
|
||||
valueType telegraf.ValueType
|
||||
}
|
||||
var tests = []struct {
|
||||
name string
|
||||
args args
|
||||
err error
|
||||
metricName string
|
||||
valueType telegraf.ValueType
|
||||
func TestMetricVersion2(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
output *PrometheusClient
|
||||
metrics []telegraf.Metric
|
||||
expected []byte
|
||||
}{
|
||||
{
|
||||
name: "field named value is not added to metric name",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"value": 42},
|
||||
valueType: telegraf.Gauge,
|
||||
name: "simple",
|
||||
output: &PrometheusClient{
|
||||
Listen: ":0",
|
||||
MetricVersion: 2,
|
||||
CollectorsExclude: []string{"gocollector", "process"},
|
||||
Path: "/metrics",
|
||||
Log: testutil.Logger{},
|
||||
},
|
||||
metricName: "foo",
|
||||
valueType: telegraf.Gauge,
|
||||
},
|
||||
{
|
||||
name: "field named gauge is not added to metric name",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"gauge": 42},
|
||||
valueType: telegraf.Gauge,
|
||||
metrics: []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"cpu",
|
||||
map[string]string{
|
||||
"host": "example.org",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"time_idle": 42.0,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
},
|
||||
metricName: "foo",
|
||||
valueType: telegraf.Gauge,
|
||||
},
|
||||
{
|
||||
name: "field with any other name is added to metric name",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"other": 42},
|
||||
valueType: telegraf.Gauge,
|
||||
},
|
||||
metricName: "foo_other",
|
||||
valueType: telegraf.Gauge,
|
||||
},
|
||||
{
|
||||
name: "uint64 fields are output",
|
||||
args: args{
|
||||
measurement: "foo",
|
||||
fields: map[string]interface{}{"value": uint64(42)},
|
||||
valueType: telegraf.Counter,
|
||||
},
|
||||
metricName: "foo",
|
||||
valueType: telegraf.Counter,
|
||||
expected: []byte(`
|
||||
# HELP cpu_time_idle Telegraf collected metric
|
||||
# TYPE cpu_time_idle untyped
|
||||
cpu_time_idle{host="example.org"} 42
|
||||
`),
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
m, err := metric.New(
|
||||
tt.args.measurement,
|
||||
tt.args.tags,
|
||||
tt.args.fields,
|
||||
time.Now(),
|
||||
tt.args.valueType,
|
||||
)
|
||||
client := NewClient()
|
||||
err = client.Write([]telegraf.Metric{m})
|
||||
require.Equal(t, tt.err, err)
|
||||
err := tt.output.Init()
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam[tt.metricName]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, tt.valueType, fam.TelegrafValueType)
|
||||
err = tt.output.Connect()
|
||||
require.NoError(t, err)
|
||||
|
||||
defer func() {
|
||||
err := tt.output.Close()
|
||||
require.NoError(t, err)
|
||||
}()
|
||||
|
||||
err = tt.output.Write(tt.metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
resp, err := http.Get(tt.output.URL())
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
defer resp.Body.Close()
|
||||
body, err := ioutil.ReadAll(resp.Body)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t,
|
||||
strings.TrimSpace(string(tt.expected)),
|
||||
strings.TrimSpace(string(body)))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestWrite_Summary(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"sum": 84, "count": 42, "0": 2, "0.5": 3, "1": 4},
|
||||
time.Now(),
|
||||
telegraf.Summary)
|
||||
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 1, len(fam.Samples))
|
||||
|
||||
sample1, ok := fam.Samples[CreateSampleID(p1.Tags())]
|
||||
require.True(t, ok)
|
||||
|
||||
require.Equal(t, 84.0, sample1.Sum)
|
||||
require.Equal(t, uint64(42), sample1.Count)
|
||||
require.Equal(t, 3, len(sample1.SummaryValue))
|
||||
}
|
||||
|
||||
func TestWrite_Histogram(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"sum": 84, "count": 42, "0": 2, "0.5": 3, "1": 4},
|
||||
time.Now(),
|
||||
telegraf.Histogram)
|
||||
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 1, len(fam.Samples))
|
||||
|
||||
sample1, ok := fam.Samples[CreateSampleID(p1.Tags())]
|
||||
require.True(t, ok)
|
||||
|
||||
require.Equal(t, 84.0, sample1.Sum)
|
||||
require.Equal(t, uint64(42), sample1.Count)
|
||||
require.Equal(t, 3, len(sample1.HistogramValue))
|
||||
}
|
||||
|
||||
func TestWrite_MixedValueType(t *testing.T) {
|
||||
now := time.Now()
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 1.0},
|
||||
now,
|
||||
telegraf.Counter)
|
||||
p2, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 2.0},
|
||||
now,
|
||||
telegraf.Gauge)
|
||||
var metrics = []telegraf.Metric{p1, p2}
|
||||
|
||||
client := NewClient()
|
||||
err = client.Write(metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 1, len(fam.Samples))
|
||||
}
|
||||
|
||||
func TestWrite_MixedValueTypeUpgrade(t *testing.T) {
|
||||
now := time.Now()
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
map[string]string{"a": "x"},
|
||||
map[string]interface{}{"value": 1.0},
|
||||
now,
|
||||
telegraf.Untyped)
|
||||
p2, err := metric.New(
|
||||
"foo",
|
||||
map[string]string{"a": "y"},
|
||||
map[string]interface{}{"value": 2.0},
|
||||
now,
|
||||
telegraf.Gauge)
|
||||
var metrics = []telegraf.Metric{p1, p2}
|
||||
|
||||
client := NewClient()
|
||||
err = client.Write(metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 2, len(fam.Samples))
|
||||
}
|
||||
|
||||
func TestWrite_MixedValueTypeDowngrade(t *testing.T) {
|
||||
now := time.Now()
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
map[string]string{"a": "x"},
|
||||
map[string]interface{}{"value": 1.0},
|
||||
now,
|
||||
telegraf.Gauge)
|
||||
p2, err := metric.New(
|
||||
"foo",
|
||||
map[string]string{"a": "y"},
|
||||
map[string]interface{}{"value": 2.0},
|
||||
now,
|
||||
telegraf.Untyped)
|
||||
var metrics = []telegraf.Metric{p1, p2}
|
||||
|
||||
client := NewClient()
|
||||
err = client.Write(metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 2, len(fam.Samples))
|
||||
}
|
||||
|
||||
func TestWrite_Tags(t *testing.T) {
|
||||
now := time.Now()
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 1.0},
|
||||
now)
|
||||
p2, err := metric.New(
|
||||
"foo",
|
||||
map[string]string{"host": "localhost"},
|
||||
map[string]interface{}{"value": 2.0},
|
||||
now)
|
||||
var metrics = []telegraf.Metric{p1, p2}
|
||||
|
||||
client := NewClient()
|
||||
err = client.Write(metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, telegraf.Untyped, fam.TelegrafValueType)
|
||||
|
||||
require.Equal(t, map[string]int{"host": 1}, fam.LabelSet)
|
||||
|
||||
sample1, ok := fam.Samples[CreateSampleID(p1.Tags())]
|
||||
require.True(t, ok)
|
||||
|
||||
require.Equal(t, 1.0, sample1.Value)
|
||||
require.True(t, now.Before(sample1.Expiration))
|
||||
|
||||
sample2, ok := fam.Samples[CreateSampleID(p2.Tags())]
|
||||
require.True(t, ok)
|
||||
|
||||
require.Equal(t, 2.0, sample2.Value)
|
||||
require.True(t, now.Before(sample2.Expiration))
|
||||
}
|
||||
|
||||
func TestWrite_StringFields(t *testing.T) {
|
||||
now := time.Now()
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 1.0, "status": "good"},
|
||||
now,
|
||||
telegraf.Counter)
|
||||
p2, err := metric.New(
|
||||
"bar",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"status": "needs numeric field"},
|
||||
now,
|
||||
telegraf.Gauge)
|
||||
var metrics = []telegraf.Metric{p1, p2}
|
||||
|
||||
client := NewClient()
|
||||
err = client.Write(metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 1, fam.LabelSet["status"])
|
||||
|
||||
fam, ok = client.fam["bar"]
|
||||
require.False(t, ok)
|
||||
}
|
||||
|
||||
func TestDoNotWrite_StringFields(t *testing.T) {
|
||||
now := time.Now()
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 1.0, "status": "good"},
|
||||
now,
|
||||
telegraf.Counter)
|
||||
p2, err := metric.New(
|
||||
"bar",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"status": "needs numeric field"},
|
||||
now,
|
||||
telegraf.Gauge)
|
||||
var metrics = []telegraf.Metric{p1, p2}
|
||||
|
||||
client := &PrometheusClient{
|
||||
ExpirationInterval: internal.Duration{Duration: time.Second * 60},
|
||||
StringAsLabel: false,
|
||||
fam: make(map[string]*MetricFamily),
|
||||
now: time.Now,
|
||||
}
|
||||
|
||||
err = client.Write(metrics)
|
||||
require.NoError(t, err)
|
||||
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 0, fam.LabelSet["status"])
|
||||
|
||||
fam, ok = client.fam["bar"]
|
||||
require.False(t, ok)
|
||||
}
|
||||
|
||||
func TestExpire(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 1.0},
|
||||
time.Now())
|
||||
setUnixTime(client, 0)
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
p2, err := metric.New(
|
||||
"bar",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 2.0},
|
||||
time.Now())
|
||||
setUnixTime(client, 1)
|
||||
err = client.Write([]telegraf.Metric{p2})
|
||||
|
||||
setUnixTime(client, 61)
|
||||
require.Equal(t, 2, len(client.fam))
|
||||
client.Expire()
|
||||
require.Equal(t, 1, len(client.fam))
|
||||
}
|
||||
|
||||
func TestExpire_TagsNoDecrement(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 1.0},
|
||||
time.Now())
|
||||
setUnixTime(client, 0)
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
p2, err := metric.New(
|
||||
"foo",
|
||||
map[string]string{"host": "localhost"},
|
||||
map[string]interface{}{"value": 2.0},
|
||||
time.Now())
|
||||
setUnixTime(client, 1)
|
||||
err = client.Write([]telegraf.Metric{p2})
|
||||
|
||||
setUnixTime(client, 61)
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 2, len(fam.Samples))
|
||||
client.Expire()
|
||||
require.Equal(t, 1, len(fam.Samples))
|
||||
|
||||
require.Equal(t, map[string]int{"host": 1}, fam.LabelSet)
|
||||
}
|
||||
|
||||
func TestExpire_TagsWithDecrement(t *testing.T) {
|
||||
client := NewClient()
|
||||
|
||||
p1, err := metric.New(
|
||||
"foo",
|
||||
map[string]string{"host": "localhost"},
|
||||
map[string]interface{}{"value": 1.0},
|
||||
time.Now())
|
||||
setUnixTime(client, 0)
|
||||
err = client.Write([]telegraf.Metric{p1})
|
||||
require.NoError(t, err)
|
||||
|
||||
p2, err := metric.New(
|
||||
"foo",
|
||||
make(map[string]string),
|
||||
map[string]interface{}{"value": 2.0},
|
||||
time.Now())
|
||||
setUnixTime(client, 1)
|
||||
err = client.Write([]telegraf.Metric{p2})
|
||||
|
||||
setUnixTime(client, 61)
|
||||
fam, ok := client.fam["foo"]
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 2, len(fam.Samples))
|
||||
client.Expire()
|
||||
require.Equal(t, 1, len(fam.Samples))
|
||||
|
||||
require.Equal(t, map[string]int{"host": 0}, fam.LabelSet)
|
||||
}
|
||||
|
||||
var pTesting *PrometheusClient
|
||||
|
||||
func TestPrometheusWritePointEmptyTag(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
pClient, p, err := setupPrometheus()
|
||||
require.NoError(t, err)
|
||||
defer pClient.Close()
|
||||
|
||||
now := time.Now()
|
||||
tags := make(map[string]string)
|
||||
pt1, _ := metric.New(
|
||||
"test_point_1",
|
||||
tags,
|
||||
map[string]interface{}{"value": 0.0},
|
||||
now)
|
||||
pt2, _ := metric.New(
|
||||
"test_point_2",
|
||||
tags,
|
||||
map[string]interface{}{"value": 1.0},
|
||||
now)
|
||||
var metrics = []telegraf.Metric{
|
||||
pt1,
|
||||
pt2,
|
||||
}
|
||||
require.NoError(t, pClient.Write(metrics))
|
||||
|
||||
expected := []struct {
|
||||
name string
|
||||
value float64
|
||||
tags map[string]string
|
||||
}{
|
||||
{"test_point_1", 0.0, tags},
|
||||
{"test_point_2", 1.0, tags},
|
||||
}
|
||||
|
||||
var acc testutil.Accumulator
|
||||
|
||||
require.NoError(t, p.Gather(&acc))
|
||||
for _, e := range expected {
|
||||
acc.AssertContainsFields(t, e.name,
|
||||
map[string]interface{}{"value": e.value})
|
||||
}
|
||||
|
||||
tags = make(map[string]string)
|
||||
tags["testtag"] = "testvalue"
|
||||
pt3, _ := metric.New(
|
||||
"test_point_3",
|
||||
tags,
|
||||
map[string]interface{}{"value": 0.0},
|
||||
now)
|
||||
pt4, _ := metric.New(
|
||||
"test_point_4",
|
||||
tags,
|
||||
map[string]interface{}{"value": 1.0},
|
||||
now)
|
||||
metrics = []telegraf.Metric{
|
||||
pt3,
|
||||
pt4,
|
||||
}
|
||||
require.NoError(t, pClient.Write(metrics))
|
||||
|
||||
expected2 := []struct {
|
||||
name string
|
||||
value float64
|
||||
}{
|
||||
{"test_point_3", 0.0},
|
||||
{"test_point_4", 1.0},
|
||||
}
|
||||
|
||||
require.NoError(t, p.Gather(&acc))
|
||||
for _, e := range expected2 {
|
||||
acc.AssertContainsFields(t, e.name,
|
||||
map[string]interface{}{"value": e.value})
|
||||
}
|
||||
}
|
||||
|
||||
func setupPrometheus() (*PrometheusClient, *prometheus_input.Prometheus, error) {
|
||||
if pTesting == nil {
|
||||
pTesting = NewClient()
|
||||
pTesting.Listen = "localhost:9127"
|
||||
pTesting.Path = "/metrics"
|
||||
err := pTesting.Connect()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
} else {
|
||||
pTesting.fam = make(map[string]*MetricFamily)
|
||||
}
|
||||
|
||||
time.Sleep(time.Millisecond * 200)
|
||||
|
||||
p := &prometheus_input.Prometheus{
|
||||
URLs: []string{"http://localhost:9127/metrics"},
|
||||
}
|
||||
|
||||
return pTesting, p, nil
|
||||
}
|
||||
|
||||
391
plugins/outputs/prometheus_client/v1/collector.go
Normal file
391
plugins/outputs/prometheus_client/v1/collector.go
Normal file
@@ -0,0 +1,391 @@
|
||||
package v1
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
var (
|
||||
invalidNameCharRE = regexp.MustCompile(`[^a-zA-Z0-9_:]`)
|
||||
validNameCharRE = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*`)
|
||||
)
|
||||
|
||||
// SampleID uniquely identifies a Sample
|
||||
type SampleID string
|
||||
|
||||
// Sample represents the current value of a series.
|
||||
type Sample struct {
|
||||
// Labels are the Prometheus labels.
|
||||
Labels map[string]string
|
||||
// Value is the value in the Prometheus output. Only one of these will populated.
|
||||
Value float64
|
||||
HistogramValue map[float64]uint64
|
||||
SummaryValue map[float64]float64
|
||||
// Histograms and Summaries need a count and a sum
|
||||
Count uint64
|
||||
Sum float64
|
||||
// Metric timestamp
|
||||
Timestamp time.Time
|
||||
// Expiration is the deadline that this Sample is valid until.
|
||||
Expiration time.Time
|
||||
}
|
||||
|
||||
// MetricFamily contains the data required to build valid prometheus Metrics.
|
||||
type MetricFamily struct {
|
||||
// Samples are the Sample belonging to this MetricFamily.
|
||||
Samples map[SampleID]*Sample
|
||||
// Need the telegraf ValueType because there isn't a Prometheus ValueType
|
||||
// representing Histogram or Summary
|
||||
TelegrafValueType telegraf.ValueType
|
||||
// LabelSet is the label counts for all Samples.
|
||||
LabelSet map[string]int
|
||||
}
|
||||
|
||||
type Collector struct {
|
||||
ExpirationInterval time.Duration
|
||||
StringAsLabel bool
|
||||
ExportTimestamp bool
|
||||
Log telegraf.Logger
|
||||
|
||||
sync.Mutex
|
||||
fam map[string]*MetricFamily
|
||||
}
|
||||
|
||||
func NewCollector(expire time.Duration, stringsAsLabel bool, logger telegraf.Logger) *Collector {
|
||||
return &Collector{
|
||||
ExpirationInterval: expire,
|
||||
StringAsLabel: stringsAsLabel,
|
||||
Log: logger,
|
||||
fam: make(map[string]*MetricFamily),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Collector) Describe(ch chan<- *prometheus.Desc) {
|
||||
prometheus.NewGauge(prometheus.GaugeOpts{Name: "Dummy", Help: "Dummy"}).Describe(ch)
|
||||
}
|
||||
|
||||
func (c *Collector) Collect(ch chan<- prometheus.Metric) {
|
||||
c.Lock()
|
||||
defer c.Unlock()
|
||||
|
||||
c.Expire(time.Now(), c.ExpirationInterval)
|
||||
|
||||
for name, family := range c.fam {
|
||||
// Get list of all labels on MetricFamily
|
||||
var labelNames []string
|
||||
for k, v := range family.LabelSet {
|
||||
if v > 0 {
|
||||
labelNames = append(labelNames, k)
|
||||
}
|
||||
}
|
||||
desc := prometheus.NewDesc(name, "Telegraf collected metric", labelNames, nil)
|
||||
|
||||
for _, sample := range family.Samples {
|
||||
// Get labels for this sample; unset labels will be set to the
|
||||
// empty string
|
||||
var labels []string
|
||||
for _, label := range labelNames {
|
||||
v := sample.Labels[label]
|
||||
labels = append(labels, v)
|
||||
}
|
||||
|
||||
var metric prometheus.Metric
|
||||
var err error
|
||||
switch family.TelegrafValueType {
|
||||
case telegraf.Summary:
|
||||
metric, err = prometheus.NewConstSummary(desc, sample.Count, sample.Sum, sample.SummaryValue, labels...)
|
||||
case telegraf.Histogram:
|
||||
metric, err = prometheus.NewConstHistogram(desc, sample.Count, sample.Sum, sample.HistogramValue, labels...)
|
||||
default:
|
||||
metric, err = prometheus.NewConstMetric(desc, getPromValueType(family.TelegrafValueType), sample.Value, labels...)
|
||||
}
|
||||
if err != nil {
|
||||
c.Log.Errorf("Error creating prometheus metric: "+
|
||||
"key: %s, labels: %v, err: %v",
|
||||
name, labels, err)
|
||||
continue
|
||||
}
|
||||
|
||||
if c.ExportTimestamp {
|
||||
metric = prometheus.NewMetricWithTimestamp(sample.Timestamp, metric)
|
||||
}
|
||||
ch <- metric
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func sanitize(value string) string {
|
||||
return invalidNameCharRE.ReplaceAllString(value, "_")
|
||||
}
|
||||
|
||||
func isValidTagName(tag string) bool {
|
||||
return validNameCharRE.MatchString(tag)
|
||||
}
|
||||
|
||||
func getPromValueType(tt telegraf.ValueType) prometheus.ValueType {
|
||||
switch tt {
|
||||
case telegraf.Counter:
|
||||
return prometheus.CounterValue
|
||||
case telegraf.Gauge:
|
||||
return prometheus.GaugeValue
|
||||
default:
|
||||
return prometheus.UntypedValue
|
||||
}
|
||||
}
|
||||
|
||||
// CreateSampleID creates a SampleID based on the tags of a telegraf.Metric.
|
||||
func CreateSampleID(tags map[string]string) SampleID {
|
||||
pairs := make([]string, 0, len(tags))
|
||||
for k, v := range tags {
|
||||
pairs = append(pairs, fmt.Sprintf("%s=%s", k, v))
|
||||
}
|
||||
sort.Strings(pairs)
|
||||
return SampleID(strings.Join(pairs, ","))
|
||||
}
|
||||
|
||||
func addSample(fam *MetricFamily, sample *Sample, sampleID SampleID) {
|
||||
|
||||
for k := range sample.Labels {
|
||||
fam.LabelSet[k]++
|
||||
}
|
||||
|
||||
fam.Samples[sampleID] = sample
|
||||
}
|
||||
|
||||
func (c *Collector) addMetricFamily(point telegraf.Metric, sample *Sample, mname string, sampleID SampleID) {
|
||||
var fam *MetricFamily
|
||||
var ok bool
|
||||
if fam, ok = c.fam[mname]; !ok {
|
||||
fam = &MetricFamily{
|
||||
Samples: make(map[SampleID]*Sample),
|
||||
TelegrafValueType: point.Type(),
|
||||
LabelSet: make(map[string]int),
|
||||
}
|
||||
c.fam[mname] = fam
|
||||
}
|
||||
|
||||
addSample(fam, sample, sampleID)
|
||||
}
|
||||
|
||||
// Sorted returns a copy of the metrics in time ascending order. A copy is
|
||||
// made to avoid modifying the input metric slice since doing so is not
|
||||
// allowed.
|
||||
func sorted(metrics []telegraf.Metric) []telegraf.Metric {
|
||||
batch := make([]telegraf.Metric, 0, len(metrics))
|
||||
for i := len(metrics) - 1; i >= 0; i-- {
|
||||
batch = append(batch, metrics[i])
|
||||
}
|
||||
sort.Slice(batch, func(i, j int) bool {
|
||||
return batch[i].Time().Before(batch[j].Time())
|
||||
})
|
||||
return batch
|
||||
}
|
||||
|
||||
func (c *Collector) Add(metrics []telegraf.Metric) error {
|
||||
c.Lock()
|
||||
defer c.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
|
||||
for _, point := range sorted(metrics) {
|
||||
tags := point.Tags()
|
||||
sampleID := CreateSampleID(tags)
|
||||
|
||||
labels := make(map[string]string)
|
||||
for k, v := range tags {
|
||||
tName := sanitize(k)
|
||||
if !isValidTagName(tName) {
|
||||
continue
|
||||
}
|
||||
labels[tName] = v
|
||||
}
|
||||
|
||||
// Prometheus doesn't have a string value type, so convert string
|
||||
// fields to labels if enabled.
|
||||
if c.StringAsLabel {
|
||||
for fn, fv := range point.Fields() {
|
||||
switch fv := fv.(type) {
|
||||
case string:
|
||||
tName := sanitize(fn)
|
||||
if !isValidTagName(tName) {
|
||||
continue
|
||||
}
|
||||
labels[tName] = fv
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
switch point.Type() {
|
||||
case telegraf.Summary:
|
||||
var mname string
|
||||
var sum float64
|
||||
var count uint64
|
||||
summaryvalue := make(map[float64]float64)
|
||||
for fn, fv := range point.Fields() {
|
||||
var value float64
|
||||
switch fv := fv.(type) {
|
||||
case int64:
|
||||
value = float64(fv)
|
||||
case uint64:
|
||||
value = float64(fv)
|
||||
case float64:
|
||||
value = fv
|
||||
default:
|
||||
continue
|
||||
}
|
||||
|
||||
switch fn {
|
||||
case "sum":
|
||||
sum = value
|
||||
case "count":
|
||||
count = uint64(value)
|
||||
default:
|
||||
limit, err := strconv.ParseFloat(fn, 64)
|
||||
if err == nil {
|
||||
summaryvalue[limit] = value
|
||||
}
|
||||
}
|
||||
}
|
||||
sample := &Sample{
|
||||
Labels: labels,
|
||||
SummaryValue: summaryvalue,
|
||||
Count: count,
|
||||
Sum: sum,
|
||||
Timestamp: point.Time(),
|
||||
Expiration: now.Add(c.ExpirationInterval),
|
||||
}
|
||||
mname = sanitize(point.Name())
|
||||
|
||||
if !isValidTagName(mname) {
|
||||
continue
|
||||
}
|
||||
|
||||
c.addMetricFamily(point, sample, mname, sampleID)
|
||||
|
||||
case telegraf.Histogram:
|
||||
var mname string
|
||||
var sum float64
|
||||
var count uint64
|
||||
histogramvalue := make(map[float64]uint64)
|
||||
for fn, fv := range point.Fields() {
|
||||
var value float64
|
||||
switch fv := fv.(type) {
|
||||
case int64:
|
||||
value = float64(fv)
|
||||
case uint64:
|
||||
value = float64(fv)
|
||||
case float64:
|
||||
value = fv
|
||||
default:
|
||||
continue
|
||||
}
|
||||
|
||||
switch fn {
|
||||
case "sum":
|
||||
sum = value
|
||||
case "count":
|
||||
count = uint64(value)
|
||||
default:
|
||||
limit, err := strconv.ParseFloat(fn, 64)
|
||||
if err == nil {
|
||||
histogramvalue[limit] = uint64(value)
|
||||
}
|
||||
}
|
||||
}
|
||||
sample := &Sample{
|
||||
Labels: labels,
|
||||
HistogramValue: histogramvalue,
|
||||
Count: count,
|
||||
Sum: sum,
|
||||
Timestamp: point.Time(),
|
||||
Expiration: now.Add(c.ExpirationInterval),
|
||||
}
|
||||
mname = sanitize(point.Name())
|
||||
|
||||
if !isValidTagName(mname) {
|
||||
continue
|
||||
}
|
||||
|
||||
c.addMetricFamily(point, sample, mname, sampleID)
|
||||
|
||||
default:
|
||||
for fn, fv := range point.Fields() {
|
||||
// Ignore string and bool fields.
|
||||
var value float64
|
||||
switch fv := fv.(type) {
|
||||
case int64:
|
||||
value = float64(fv)
|
||||
case uint64:
|
||||
value = float64(fv)
|
||||
case float64:
|
||||
value = fv
|
||||
default:
|
||||
continue
|
||||
}
|
||||
|
||||
sample := &Sample{
|
||||
Labels: labels,
|
||||
Value: value,
|
||||
Timestamp: point.Time(),
|
||||
Expiration: now.Add(c.ExpirationInterval),
|
||||
}
|
||||
|
||||
// Special handling of value field; supports passthrough from
|
||||
// the prometheus input.
|
||||
var mname string
|
||||
switch point.Type() {
|
||||
case telegraf.Counter:
|
||||
if fn == "counter" {
|
||||
mname = sanitize(point.Name())
|
||||
}
|
||||
case telegraf.Gauge:
|
||||
if fn == "gauge" {
|
||||
mname = sanitize(point.Name())
|
||||
}
|
||||
}
|
||||
if mname == "" {
|
||||
if fn == "value" {
|
||||
mname = sanitize(point.Name())
|
||||
} else {
|
||||
mname = sanitize(fmt.Sprintf("%s_%s", point.Name(), fn))
|
||||
}
|
||||
}
|
||||
if !isValidTagName(mname) {
|
||||
continue
|
||||
}
|
||||
c.addMetricFamily(point, sample, mname, sampleID)
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Collector) Expire(now time.Time, age time.Duration) {
|
||||
if age == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
for name, family := range c.fam {
|
||||
for key, sample := range family.Samples {
|
||||
if age != 0 && now.After(sample.Expiration) {
|
||||
for k := range sample.Labels {
|
||||
family.LabelSet[k]--
|
||||
}
|
||||
delete(family.Samples, key)
|
||||
|
||||
if len(family.Samples) == 0 {
|
||||
delete(c.fam, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
87
plugins/outputs/prometheus_client/v2/collector.go
Normal file
87
plugins/outputs/prometheus_client/v2/collector.go
Normal file
@@ -0,0 +1,87 @@
|
||||
package v2
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
serializer "github.com/influxdata/telegraf/plugins/serializers/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
)
|
||||
|
||||
type Metric struct {
|
||||
family *dto.MetricFamily
|
||||
metric *dto.Metric
|
||||
}
|
||||
|
||||
func (m *Metric) Desc() *prometheus.Desc {
|
||||
labelNames := make([]string, 0, len(m.metric.Label))
|
||||
for _, label := range m.metric.Label {
|
||||
labelNames = append(labelNames, *label.Name)
|
||||
}
|
||||
|
||||
desc := prometheus.NewDesc(*m.family.Name, *m.family.Help, labelNames, nil)
|
||||
|
||||
return desc
|
||||
}
|
||||
|
||||
func (m *Metric) Write(out *dto.Metric) error {
|
||||
out.Label = m.metric.Label
|
||||
out.Counter = m.metric.Counter
|
||||
out.Untyped = m.metric.Untyped
|
||||
out.Gauge = m.metric.Gauge
|
||||
out.Histogram = m.metric.Histogram
|
||||
out.Summary = m.metric.Summary
|
||||
out.TimestampMs = m.metric.TimestampMs
|
||||
return nil
|
||||
}
|
||||
|
||||
type Collector struct {
|
||||
sync.Mutex
|
||||
expireDuration time.Duration
|
||||
coll *serializer.Collection
|
||||
}
|
||||
|
||||
func NewCollector(expire time.Duration, stringsAsLabel bool) *Collector {
|
||||
config := serializer.FormatConfig{}
|
||||
if stringsAsLabel {
|
||||
config.StringHandling = serializer.StringAsLabel
|
||||
}
|
||||
return &Collector{
|
||||
expireDuration: expire,
|
||||
coll: serializer.NewCollection(config),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Collector) Describe(ch chan<- *prometheus.Desc) {
|
||||
// Sending no descriptor at all marks the Collector as "unchecked",
|
||||
// i.e. no checks will be performed at registration time, and the
|
||||
// Collector may yield any Metric it sees fit in its Collect method.
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Collector) Collect(ch chan<- prometheus.Metric) {
|
||||
c.Lock()
|
||||
defer c.Unlock()
|
||||
|
||||
for _, family := range c.coll.GetProto() {
|
||||
for _, metric := range family.Metric {
|
||||
ch <- &Metric{family: family, metric: metric}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Collector) Add(metrics []telegraf.Metric) error {
|
||||
c.Lock()
|
||||
defer c.Unlock()
|
||||
|
||||
for _, metric := range metrics {
|
||||
c.coll.Add(metric)
|
||||
}
|
||||
|
||||
if c.expireDuration != 0 {
|
||||
c.coll.Expire(time.Now(), c.expireDuration)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user