Update go-metrics (#9704)
This commit is contained in:
parent
25137a1702
commit
1d6a5ae058
2
go.mod
2
go.mod
|
@ -20,7 +20,7 @@ require (
|
|||
github.com/aliyun/alibaba-cloud-sdk-go v0.0.0-20190620160927-9418d7b0cd0f
|
||||
github.com/aliyun/aliyun-oss-go-sdk v0.0.0-20190307165228-86c17b95fcd5
|
||||
github.com/apple/foundationdb/bindings/go v0.0.0-20190411004307-cd5c9d91fad2
|
||||
github.com/armon/go-metrics v0.3.3
|
||||
github.com/armon/go-metrics v0.3.4
|
||||
github.com/armon/go-proxyproto v0.0.0-20190211145416-68259f75880e
|
||||
github.com/armon/go-radix v1.0.0
|
||||
github.com/asaskevich/govalidator v0.0.0-20180720115003-f9ffefc3facf
|
||||
|
|
2
go.sum
2
go.sum
|
@ -127,6 +127,8 @@ github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQh
|
|||
github.com/armon/go-metrics v0.3.0/go.mod h1:zXjbSimjXTd7vOpY8B0/2LpvNvDoXBuplAD+gJD3GYs=
|
||||
github.com/armon/go-metrics v0.3.3 h1:a9F4rlj7EWWrbj7BYw8J8+x+ZZkJeqzNyRk8hdPF+ro=
|
||||
github.com/armon/go-metrics v0.3.3/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb4QAOwNTFc=
|
||||
github.com/armon/go-metrics v0.3.4 h1:Xqf+7f2Vhl9tsqDYmXhnXInUdcrtgpRNpIA15/uldSc=
|
||||
github.com/armon/go-metrics v0.3.4/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb4QAOwNTFc=
|
||||
github.com/armon/go-proxyproto v0.0.0-20190211145416-68259f75880e h1:h0gP0hBU6DsA5IQduhLWGOEfIUKzJS5hhXQBSgHuF/g=
|
||||
github.com/armon/go-proxyproto v0.0.0-20190211145416-68259f75880e/go.mod h1:QmP9hvJ91BbJmGVGSbutW19IC0Q9phDCLGaomwTJbgU=
|
||||
github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
|
||||
|
|
|
@ -22,3 +22,5 @@ _testmain.go
|
|||
*.exe
|
||||
|
||||
/metrics.out
|
||||
|
||||
.idea
|
||||
|
|
|
@ -5,12 +5,11 @@ package prometheus
|
|||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"regexp"
|
||||
|
||||
"github.com/armon/go-metrics"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/push"
|
||||
|
@ -36,10 +35,24 @@ type PrometheusSink struct {
|
|||
gauges sync.Map
|
||||
summaries sync.Map
|
||||
counters sync.Map
|
||||
updates sync.Map
|
||||
expiration time.Duration
|
||||
}
|
||||
|
||||
type PrometheusGauge struct {
|
||||
prometheus.Gauge
|
||||
updatedAt time.Time
|
||||
}
|
||||
|
||||
type PrometheusSummary struct {
|
||||
prometheus.Summary
|
||||
updatedAt time.Time
|
||||
}
|
||||
|
||||
type PrometheusCounter struct {
|
||||
prometheus.Counter
|
||||
updatedAt time.Time
|
||||
}
|
||||
|
||||
// NewPrometheusSink creates a new PrometheusSink using the default options.
|
||||
func NewPrometheusSink() (*PrometheusSink, error) {
|
||||
return NewPrometheusSinkFrom(DefaultPrometheusOpts)
|
||||
|
@ -51,7 +64,6 @@ func NewPrometheusSinkFrom(opts PrometheusOpts) (*PrometheusSink, error) {
|
|||
gauges: sync.Map{},
|
||||
summaries: sync.Map{},
|
||||
counters: sync.Map{},
|
||||
updates: sync.Map{},
|
||||
expiration: opts.Expiration,
|
||||
}
|
||||
|
||||
|
@ -72,32 +84,35 @@ func (p *PrometheusSink) Collect(c chan<- prometheus.Metric) {
|
|||
expire := p.expiration != 0
|
||||
now := time.Now()
|
||||
p.gauges.Range(func(k, v interface{}) bool {
|
||||
last, _ := p.updates.Load(k)
|
||||
if expire && last.(time.Time).Add(p.expiration).Before(now) {
|
||||
p.updates.Delete(k)
|
||||
p.gauges.Delete(k)
|
||||
} else {
|
||||
v.(prometheus.Gauge).Collect(c)
|
||||
if v != nil {
|
||||
lastUpdate := v.(*PrometheusGauge).updatedAt
|
||||
if expire && lastUpdate.Add(p.expiration).Before(now) {
|
||||
p.gauges.Delete(k)
|
||||
} else {
|
||||
v.(*PrometheusGauge).Collect(c)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
p.summaries.Range(func(k, v interface{}) bool {
|
||||
last, _ := p.updates.Load(k)
|
||||
if expire && last.(time.Time).Add(p.expiration).Before(now) {
|
||||
p.updates.Delete(k)
|
||||
p.summaries.Delete(k)
|
||||
} else {
|
||||
v.(prometheus.Summary).Collect(c)
|
||||
if v != nil {
|
||||
lastUpdate := v.(*PrometheusSummary).updatedAt
|
||||
if expire && lastUpdate.Add(p.expiration).Before(now) {
|
||||
p.summaries.Delete(k)
|
||||
} else {
|
||||
v.(*PrometheusSummary).Collect(c)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
p.counters.Range(func(k, v interface{}) bool {
|
||||
last, _ := p.updates.Load(k)
|
||||
if expire && last.(time.Time).Add(p.expiration).Before(now) {
|
||||
p.updates.Delete(k)
|
||||
p.counters.Delete(k)
|
||||
} else {
|
||||
v.(prometheus.Counter).Collect(c)
|
||||
if v != nil {
|
||||
lastUpdate := v.(*PrometheusCounter).updatedAt
|
||||
if expire && lastUpdate.Add(p.expiration).Before(now) {
|
||||
p.counters.Delete(k)
|
||||
} else {
|
||||
v.(*PrometheusCounter).Collect(c)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
@ -131,17 +146,31 @@ func (p *PrometheusSink) SetGauge(parts []string, val float32) {
|
|||
|
||||
func (p *PrometheusSink) SetGaugeWithLabels(parts []string, val float32, labels []metrics.Label) {
|
||||
key, hash := p.flattenKey(parts, labels)
|
||||
g, ok := p.gauges.Load(hash)
|
||||
if !ok {
|
||||
g = prometheus.NewGauge(prometheus.GaugeOpts{
|
||||
pg, ok := p.gauges.Load(hash)
|
||||
|
||||
// The sync.Map underlying gauges stores pointers to our structs. If we need to make updates,
|
||||
// rather than modifying the underlying value directly, which would be racy, we make a local
|
||||
// copy by dereferencing the pointer we get back, making the appropriate changes, and then
|
||||
// storing a pointer to our local copy. The underlying Prometheus types are threadsafe,
|
||||
// so there's no issues there. It's possible for racy updates to occur to the updatedAt
|
||||
// value, but since we're always setting it to time.Now(), it doesn't really matter.
|
||||
if ok {
|
||||
localGauge := *pg.(*PrometheusGauge)
|
||||
localGauge.Set(float64(val))
|
||||
localGauge.updatedAt = time.Now()
|
||||
p.gauges.Store(hash, &localGauge)
|
||||
} else {
|
||||
g := prometheus.NewGauge(prometheus.GaugeOpts{
|
||||
Name: key,
|
||||
Help: key,
|
||||
ConstLabels: prometheusLabels(labels),
|
||||
})
|
||||
p.gauges.Store(hash, g)
|
||||
g.Set(float64(val))
|
||||
pg = &PrometheusGauge{
|
||||
g, time.Now(),
|
||||
}
|
||||
p.gauges.Store(hash, pg)
|
||||
}
|
||||
g.(prometheus.Gauge).Set(float64(val))
|
||||
p.updates.Store(hash, time.Now())
|
||||
}
|
||||
|
||||
func (p *PrometheusSink) AddSample(parts []string, val float32) {
|
||||
|
@ -150,19 +179,27 @@ func (p *PrometheusSink) AddSample(parts []string, val float32) {
|
|||
|
||||
func (p *PrometheusSink) AddSampleWithLabels(parts []string, val float32, labels []metrics.Label) {
|
||||
key, hash := p.flattenKey(parts, labels)
|
||||
g, ok := p.summaries.Load(hash)
|
||||
if !ok {
|
||||
g = prometheus.NewSummary(prometheus.SummaryOpts{
|
||||
ps, ok := p.summaries.Load(hash)
|
||||
|
||||
if ok {
|
||||
localSummary := *ps.(*PrometheusSummary)
|
||||
localSummary.Observe(float64(val))
|
||||
localSummary.updatedAt = time.Now()
|
||||
p.summaries.Store(hash, &localSummary)
|
||||
} else {
|
||||
s := prometheus.NewSummary(prometheus.SummaryOpts{
|
||||
Name: key,
|
||||
Help: key,
|
||||
MaxAge: 10 * time.Second,
|
||||
ConstLabels: prometheusLabels(labels),
|
||||
Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
|
||||
})
|
||||
p.summaries.Store(hash, g)
|
||||
s.Observe(float64(val))
|
||||
ps = &PrometheusSummary{
|
||||
s, time.Now(),
|
||||
}
|
||||
p.summaries.Store(hash, ps)
|
||||
}
|
||||
g.(prometheus.Summary).Observe(float64(val))
|
||||
p.updates.Store(hash, time.Now())
|
||||
}
|
||||
|
||||
// EmitKey is not implemented. Prometheus doesn’t offer a type for which an
|
||||
|
@ -177,17 +214,25 @@ func (p *PrometheusSink) IncrCounter(parts []string, val float32) {
|
|||
|
||||
func (p *PrometheusSink) IncrCounterWithLabels(parts []string, val float32, labels []metrics.Label) {
|
||||
key, hash := p.flattenKey(parts, labels)
|
||||
g, ok := p.counters.Load(hash)
|
||||
if !ok {
|
||||
g = prometheus.NewCounter(prometheus.CounterOpts{
|
||||
pc, ok := p.counters.Load(hash)
|
||||
|
||||
if ok {
|
||||
localCounter := *pc.(*PrometheusCounter)
|
||||
localCounter.Add(float64(val))
|
||||
localCounter.updatedAt = time.Now()
|
||||
p.counters.Store(hash, &localCounter)
|
||||
} else {
|
||||
c := prometheus.NewCounter(prometheus.CounterOpts{
|
||||
Name: key,
|
||||
Help: key,
|
||||
ConstLabels: prometheusLabels(labels),
|
||||
})
|
||||
p.counters.Store(hash, g)
|
||||
c.Add(float64(val))
|
||||
pc = &PrometheusCounter{
|
||||
c, time.Now(),
|
||||
}
|
||||
p.counters.Store(hash, pc)
|
||||
}
|
||||
g.(prometheus.Counter).Add(float64(val))
|
||||
p.updates.Store(hash, time.Now())
|
||||
}
|
||||
|
||||
type PrometheusPushSink struct {
|
||||
|
@ -199,12 +244,10 @@ type PrometheusPushSink struct {
|
|||
}
|
||||
|
||||
func NewPrometheusPushSink(address string, pushIterval time.Duration, name string) (*PrometheusPushSink, error) {
|
||||
|
||||
promSink := &PrometheusSink{
|
||||
gauges: sync.Map{},
|
||||
summaries: sync.Map{},
|
||||
counters: sync.Map{},
|
||||
updates: sync.Map{},
|
||||
expiration: 60 * time.Second,
|
||||
}
|
||||
|
||||
|
|
|
@ -102,7 +102,7 @@ github.com/apple/foundationdb/bindings/go/src/fdb
|
|||
github.com/apple/foundationdb/bindings/go/src/fdb/directory
|
||||
github.com/apple/foundationdb/bindings/go/src/fdb/subspace
|
||||
github.com/apple/foundationdb/bindings/go/src/fdb/tuple
|
||||
# github.com/armon/go-metrics v0.3.3
|
||||
# github.com/armon/go-metrics v0.3.4
|
||||
github.com/armon/go-metrics
|
||||
github.com/armon/go-metrics/circonus
|
||||
github.com/armon/go-metrics/datadog
|
||||
|
|
Loading…
Reference in New Issue