metrics: split Gauge interface, fix tests, fix race on Sample

This commit is contained in:
Martin Holst Swende 2023-08-31 11:50:17 +02:00
parent 9591faf40d
commit b7f8d9b4ce
No known key found for this signature in database
GPG key ID: 683B438C05A5DDF0
19 changed files with 118 additions and 203 deletions

View file

@ -1103,12 +1103,10 @@ func (s *StateDB) deleteStorage(addr common.Address, addrHash common.Hash, root
slotDeletionSkip.Inc(1) slotDeletionSkip.Inc(1)
} }
n := int64(len(slots)) n := int64(len(slots))
if n > slotDeletionMaxCount.Value() {
slotDeletionMaxCount.Update(n) slotDeletionMaxCount.UpdateIfGt(int64(len(slots)))
} slotDeletionMaxSize.UpdateIfGt(int64(size))
if int64(size) > slotDeletionMaxSize.Value() {
slotDeletionMaxSize.Update(int64(size))
}
slotDeletionTimer.UpdateSince(start) slotDeletionTimer.UpdateSince(start)
slotDeletionCount.Mark(n) slotDeletionCount.Mark(n)
slotDeletionSize.Mark(int64(size)) slotDeletionSize.Mark(int64(size))

View file

@ -119,7 +119,7 @@ func (exp *exp) publishCounterFloat64(name string, metric metrics.CounterFloat64
v.Set(metric.Count()) v.Set(metric.Count())
} }
func (exp *exp) publishGauge(name string, metric metrics.Gauge) { func (exp *exp) publishGauge(name string, metric metrics.GaugeSnapshot) {
v := exp.getInt(name) v := exp.getInt(name)
v.Set(metric.Value()) v.Set(metric.Value())
} }
@ -193,7 +193,7 @@ func (exp *exp) syncToExpvar() {
case metrics.CounterFloat64: case metrics.CounterFloat64:
exp.publishCounterFloat64(name, i) exp.publishCounterFloat64(name, i)
case metrics.Gauge: case metrics.Gauge:
exp.publishGauge(name, i) exp.publishGauge(name, i.Snapshot())
case metrics.GaugeFloat64: case metrics.GaugeFloat64:
exp.publishGaugeFloat64(name, i) exp.publishGaugeFloat64(name, i)
case metrics.GaugeInfo: case metrics.GaugeInfo:

View file

@ -2,13 +2,18 @@ package metrics
import "sync/atomic" import "sync/atomic"
// gaugeSnapshot contains a readonly int64.
type GaugeSnapshot interface {
Value() int64
}
// Gauges hold an int64 value that can be set arbitrarily. // Gauges hold an int64 value that can be set arbitrarily.
type Gauge interface { type Gauge interface {
Snapshot() Gauge Snapshot() GaugeSnapshot
Update(int64) Update(int64)
UpdateIfGt(int64)
Dec(int64) Dec(int64)
Inc(int64) Inc(int64)
Value() int64
} }
// GetOrRegisterGauge returns an existing Gauge or constructs and registers a // GetOrRegisterGauge returns an existing Gauge or constructs and registers a
@ -38,57 +43,23 @@ func NewRegisteredGauge(name string, r Registry) Gauge {
return c return c
} }
// NewFunctionalGauge constructs a new FunctionalGauge. // gaugeSnapshot is a read-only copy of another Gauge.
func NewFunctionalGauge(f func() int64) Gauge { type gaugeSnapshot int64
if !Enabled {
return NilGauge{}
}
return &FunctionalGauge{value: f}
}
// NewRegisteredFunctionalGauge constructs and registers a new StandardGauge.
func NewRegisteredFunctionalGauge(name string, r Registry, f func() int64) Gauge {
c := NewFunctionalGauge(f)
if nil == r {
r = DefaultRegistry
}
r.Register(name, c)
return c
}
// GaugeSnapshot is a read-only copy of another Gauge.
type GaugeSnapshot int64
// Snapshot returns the snapshot.
func (g GaugeSnapshot) Snapshot() Gauge { return g }
// Update panics.
func (GaugeSnapshot) Update(int64) {
panic("Update called on a GaugeSnapshot")
}
// Dec panics.
func (GaugeSnapshot) Dec(int64) {
panic("Dec called on a GaugeSnapshot")
}
// Inc panics.
func (GaugeSnapshot) Inc(int64) {
panic("Inc called on a GaugeSnapshot")
}
// Value returns the value at the time the snapshot was taken. // Value returns the value at the time the snapshot was taken.
func (g GaugeSnapshot) Value() int64 { return int64(g) } func (g gaugeSnapshot) Value() int64 { return int64(g) }
// NilGauge is a no-op Gauge. // NilGauge is a no-op Gauge.
type NilGauge struct{} type NilGauge struct{}
// Snapshot is a no-op. // Snapshot is a no-op.
func (NilGauge) Snapshot() Gauge { return NilGauge{} } func (NilGauge) Snapshot() GaugeSnapshot { return NilGauge{} }
// Update is a no-op. // Update is a no-op.
func (NilGauge) Update(v int64) {} func (NilGauge) Update(v int64) {}
func (NilGauge) UpdateIfGt(v int64) {}
// Dec is a no-op. // Dec is a no-op.
func (NilGauge) Dec(i int64) {} func (NilGauge) Dec(i int64) {}
@ -105,8 +76,8 @@ type StandardGauge struct {
} }
// Snapshot returns a read-only copy of the gauge. // Snapshot returns a read-only copy of the gauge.
func (g *StandardGauge) Snapshot() Gauge { func (g *StandardGauge) Snapshot() GaugeSnapshot {
return GaugeSnapshot(g.Value()) return gaugeSnapshot(g.value.Load())
} }
// Update updates the gauge's value. // Update updates the gauge's value.
@ -114,9 +85,11 @@ func (g *StandardGauge) Update(v int64) {
g.value.Store(v) g.value.Store(v)
} }
// Value returns the gauge's current value. // Update updates the gauge's value if v is larger then the current valie.
func (g *StandardGauge) Value() int64 { func (g *StandardGauge) UpdateIfGt(v int64) {
return g.value.Load() if g.value.Load() < v {
g.value.Store(v)
}
} }
// Dec decrements the gauge's current value by the given amount. // Dec decrements the gauge's current value by the given amount.
@ -128,31 +101,3 @@ func (g *StandardGauge) Dec(i int64) {
func (g *StandardGauge) Inc(i int64) { func (g *StandardGauge) Inc(i int64) {
g.value.Add(i) g.value.Add(i)
} }
// FunctionalGauge returns value from given function
type FunctionalGauge struct {
value func() int64
}
// Value returns the gauge's current value.
func (g FunctionalGauge) Value() int64 {
return g.value()
}
// Snapshot returns the snapshot.
func (g FunctionalGauge) Snapshot() Gauge { return GaugeSnapshot(g.Value()) }
// Update panics.
func (FunctionalGauge) Update(int64) {
panic("Update called on a FunctionalGauge")
}
// Dec panics.
func (FunctionalGauge) Dec(int64) {
panic("Dec called on a FunctionalGauge")
}
// Inc panics.
func (FunctionalGauge) Inc(int64) {
panic("Inc called on a FunctionalGauge")
}

View file

@ -1,7 +1,6 @@
package metrics package metrics
import ( import (
"fmt"
"testing" "testing"
) )
@ -13,14 +12,6 @@ func BenchmarkGauge(b *testing.B) {
} }
} }
func TestGauge(t *testing.T) {
g := NewGauge()
g.Update(int64(47))
if v := g.Value(); v != 47 {
t.Errorf("g.Value(): 47 != %v\n", v)
}
}
func TestGaugeSnapshot(t *testing.T) { func TestGaugeSnapshot(t *testing.T) {
g := NewGauge() g := NewGauge()
g.Update(int64(47)) g.Update(int64(47))
@ -34,35 +25,7 @@ func TestGaugeSnapshot(t *testing.T) {
func TestGetOrRegisterGauge(t *testing.T) { func TestGetOrRegisterGauge(t *testing.T) {
r := NewRegistry() r := NewRegistry()
NewRegisteredGauge("foo", r).Update(47) NewRegisteredGauge("foo", r).Update(47)
if g := GetOrRegisterGauge("foo", r); g.Value() != 47 { if g := GetOrRegisterGauge("foo", r); g.Snapshot().Value() != 47 {
t.Fatal(g) t.Fatal(g)
} }
} }
func TestFunctionalGauge(t *testing.T) {
var counter int64
fg := NewFunctionalGauge(func() int64 {
counter++
return counter
})
fg.Value()
fg.Value()
if counter != 2 {
t.Error("counter != 2")
}
}
func TestGetOrRegisterFunctionalGauge(t *testing.T) {
r := NewRegistry()
NewRegisteredFunctionalGauge("foo", r, func() int64 { return 47 })
if g := GetOrRegisterGauge("foo", r); g.Value() != 47 {
t.Fatal(g)
}
}
func ExampleGetOrRegisterGauge() {
m := "server.bytes_sent"
g := GetOrRegisterGauge(m, nil)
g.Update(47)
fmt.Println(g.Value()) // Output: 47
}

View file

@ -70,7 +70,7 @@ func graphite(c *GraphiteConfig) error {
case CounterFloat64: case CounterFloat64:
fmt.Fprintf(w, "%s.%s.count %f %d\n", c.Prefix, name, metric.Count(), now) fmt.Fprintf(w, "%s.%s.count %f %d\n", c.Prefix, name, metric.Count(), now)
case Gauge: case Gauge:
fmt.Fprintf(w, "%s.%s.value %d %d\n", c.Prefix, name, metric.Value(), now) fmt.Fprintf(w, "%s.%s.value %d %d\n", c.Prefix, name, metric.Snapshot().Value(), now)
case GaugeFloat64: case GaugeFloat64:
fmt.Fprintf(w, "%s.%s.value %f %d\n", c.Prefix, name, metric.Value(), now) fmt.Fprintf(w, "%s.%s.value %f %d\n", c.Prefix, name, metric.Value(), now)
case GaugeInfo: case GaugeInfo:

View file

@ -14,7 +14,7 @@ func TestGetOrRegisterHistogram(t *testing.T) {
r := NewRegistry() r := NewRegistry()
s := NewUniformSample(100) s := NewUniformSample(100)
NewRegisteredHistogram("foo", r, s).Update(47) NewRegisteredHistogram("foo", r, s).Update(47)
if h := GetOrRegisterHistogram("foo", r, s); h.Count() != 1 { if h := GetOrRegisterHistogram("foo", r, s).Snapshot(); h.Count() != 1 {
t.Fatal(h) t.Fatal(h)
} }
} }
@ -24,11 +24,11 @@ func TestHistogram10000(t *testing.T) {
for i := 1; i <= 10000; i++ { for i := 1; i <= 10000; i++ {
h.Update(int64(i)) h.Update(int64(i))
} }
testHistogram10000(t, h) testHistogram10000(t, h.Snapshot())
} }
func TestHistogramEmpty(t *testing.T) { func TestHistogramEmpty(t *testing.T) {
h := NewHistogram(NewUniformSample(100)) h := NewHistogram(NewUniformSample(100)).Snapshot()
if count := h.Count(); count != 0 { if count := h.Count(); count != 0 {
t.Errorf("h.Count(): 0 != %v\n", count) t.Errorf("h.Count(): 0 != %v\n", count)
} }
@ -66,7 +66,7 @@ func TestHistogramSnapshot(t *testing.T) {
testHistogram10000(t, snapshot) testHistogram10000(t, snapshot)
} }
func testHistogram10000(t *testing.T, h Histogram) { func testHistogram10000(t *testing.T, h HistogramSnapshot) {
if count := h.Count(); count != 10000 { if count := h.Count(); count != 10000 {
t.Errorf("h.Count(): 10000 != %v\n", count) t.Errorf("h.Count(): 10000 != %v\n", count)
} }

View file

@ -61,7 +61,7 @@ func (rep *Reporter) Run() {
// calculate sum of squares from data provided by metrics.Histogram // calculate sum of squares from data provided by metrics.Histogram
// see http://en.wikipedia.org/wiki/Standard_deviation#Rapid_calculation_methods // see http://en.wikipedia.org/wiki/Standard_deviation#Rapid_calculation_methods
func sumSquares(s metrics.Sample) float64 { func sumSquares(s metrics.SampleSnapshot) float64 {
count := float64(s.Count()) count := float64(s.Count())
sumSquared := math.Pow(count*s.Mean(), 2) sumSquared := math.Pow(count*s.Mean(), 2)
sumSquares := math.Pow(count*s.StdDev(), 2) + sumSquared/count sumSquares := math.Pow(count*s.StdDev(), 2) + sumSquared/count
@ -70,7 +70,7 @@ func sumSquares(s metrics.Sample) float64 {
} }
return sumSquares return sumSquares
} }
func sumSquaresTimer(t metrics.Timer) float64 { func sumSquaresTimer(t metrics.TimerSnapshot) float64 {
count := float64(t.Count()) count := float64(t.Count())
sumSquared := math.Pow(count*t.Mean(), 2) sumSquared := math.Pow(count*t.Mean(), 2)
sumSquares := math.Pow(count*t.StdDev(), 2) + sumSquared/count sumSquares := math.Pow(count*t.StdDev(), 2) + sumSquared/count
@ -97,9 +97,10 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
measurement[Period] = rep.Interval.Seconds() measurement[Period] = rep.Interval.Seconds()
switch m := metric.(type) { switch m := metric.(type) {
case metrics.Counter: case metrics.Counter:
ms := m.Snapshot()
if m.Count() > 0 { if m.Count() > 0 {
measurement[Name] = fmt.Sprintf("%s.%s", name, "count") measurement[Name] = fmt.Sprintf("%s.%s", name, "count")
measurement[Value] = float64(m.Count()) measurement[Value] = float64(ms.Count())
measurement[Attributes] = map[string]interface{}{ measurement[Attributes] = map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,
DisplayUnitsShort: OperationsShort, DisplayUnitsShort: OperationsShort,
@ -108,9 +109,9 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
snapshot.Counters = append(snapshot.Counters, measurement) snapshot.Counters = append(snapshot.Counters, measurement)
} }
case metrics.CounterFloat64: case metrics.CounterFloat64:
if m.Count() > 0 { if count := m.Snapshot().Count(); count > 0 {
measurement[Name] = fmt.Sprintf("%s.%s", name, "count") measurement[Name] = fmt.Sprintf("%s.%s", name, "count")
measurement[Value] = m.Count() measurement[Value] = count
measurement[Attributes] = map[string]interface{}{ measurement[Attributes] = map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,
DisplayUnitsShort: OperationsShort, DisplayUnitsShort: OperationsShort,
@ -120,20 +121,21 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
} }
case metrics.Gauge: case metrics.Gauge:
measurement[Name] = name measurement[Name] = name
measurement[Value] = float64(m.Value()) measurement[Value] = float64(m.Snapshot().Value())
snapshot.Gauges = append(snapshot.Gauges, measurement) snapshot.Gauges = append(snapshot.Gauges, measurement)
case metrics.GaugeFloat64: case metrics.GaugeFloat64:
measurement[Name] = name measurement[Name] = name
measurement[Value] = m.Value() measurement[Value] = m.Snapshot().Value()
snapshot.Gauges = append(snapshot.Gauges, measurement) snapshot.Gauges = append(snapshot.Gauges, measurement)
case metrics.GaugeInfo: case metrics.GaugeInfo:
measurement[Name] = name measurement[Name] = name
measurement[Value] = m.Value() measurement[Value] = m.Value()
snapshot.Gauges = append(snapshot.Gauges, measurement) snapshot.Gauges = append(snapshot.Gauges, measurement)
case metrics.Histogram: case metrics.Histogram:
if m.Count() > 0 { ms := m.Snapshot()
if ms.Count() > 0 {
gauges := make([]Measurement, histogramGaugeCount) gauges := make([]Measurement, histogramGaugeCount)
s := m.Sample() s := ms.Sample()
measurement[Name] = fmt.Sprintf("%s.%s", name, "hist") measurement[Name] = fmt.Sprintf("%s.%s", name, "hist")
measurement[Count] = uint64(s.Count()) measurement[Count] = uint64(s.Count())
measurement[Max] = float64(s.Max()) measurement[Max] = float64(s.Max())
@ -151,13 +153,14 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
snapshot.Gauges = append(snapshot.Gauges, gauges...) snapshot.Gauges = append(snapshot.Gauges, gauges...)
} }
case metrics.Meter: case metrics.Meter:
ms := m.Snapshot()
measurement[Name] = name measurement[Name] = name
measurement[Value] = float64(m.Count()) measurement[Value] = float64(ms.Count())
snapshot.Counters = append(snapshot.Counters, measurement) snapshot.Counters = append(snapshot.Counters, measurement)
snapshot.Gauges = append(snapshot.Gauges, snapshot.Gauges = append(snapshot.Gauges,
Measurement{ Measurement{
Name: fmt.Sprintf("%s.%s", name, "1min"), Name: fmt.Sprintf("%s.%s", name, "1min"),
Value: m.Rate1(), Value: ms.Rate1(),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: map[string]interface{}{ Attributes: map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,
@ -167,7 +170,7 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
}, },
Measurement{ Measurement{
Name: fmt.Sprintf("%s.%s", name, "5min"), Name: fmt.Sprintf("%s.%s", name, "5min"),
Value: m.Rate5(), Value: ms.Rate5(),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: map[string]interface{}{ Attributes: map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,
@ -177,7 +180,7 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
}, },
Measurement{ Measurement{
Name: fmt.Sprintf("%s.%s", name, "15min"), Name: fmt.Sprintf("%s.%s", name, "15min"),
Value: m.Rate15(), Value: ms.Rate15(),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: map[string]interface{}{ Attributes: map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,
@ -187,26 +190,27 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
}, },
) )
case metrics.Timer: case metrics.Timer:
ms := m.Snapshot()
measurement[Name] = name measurement[Name] = name
measurement[Value] = float64(m.Count()) measurement[Value] = float64(ms.Count())
snapshot.Counters = append(snapshot.Counters, measurement) snapshot.Counters = append(snapshot.Counters, measurement)
if m.Count() > 0 { if ms.Count() > 0 {
libratoName := fmt.Sprintf("%s.%s", name, "timer.mean") libratoName := fmt.Sprintf("%s.%s", name, "timer.mean")
gauges := make([]Measurement, histogramGaugeCount) gauges := make([]Measurement, histogramGaugeCount)
gauges[0] = Measurement{ gauges[0] = Measurement{
Name: libratoName, Name: libratoName,
Count: uint64(m.Count()), Count: uint64(ms.Count()),
Sum: m.Mean() * float64(m.Count()), Sum: ms.Mean() * float64(ms.Count()),
Max: float64(m.Max()), Max: float64(ms.Max()),
Min: float64(m.Min()), Min: float64(ms.Min()),
SumSquares: sumSquaresTimer(m), SumSquares: sumSquaresTimer(ms),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: rep.TimerAttributes, Attributes: rep.TimerAttributes,
} }
for i, p := range rep.Percentiles { for i, p := range rep.Percentiles {
gauges[i+1] = Measurement{ gauges[i+1] = Measurement{
Name: fmt.Sprintf("%s.timer.%2.0f", name, p*100), Name: fmt.Sprintf("%s.timer.%2.0f", name, p*100),
Value: m.Percentile(p), Value: ms.Percentile(p),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: rep.TimerAttributes, Attributes: rep.TimerAttributes,
} }
@ -215,7 +219,7 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
snapshot.Gauges = append(snapshot.Gauges, snapshot.Gauges = append(snapshot.Gauges,
Measurement{ Measurement{
Name: fmt.Sprintf("%s.%s", name, "rate.1min"), Name: fmt.Sprintf("%s.%s", name, "rate.1min"),
Value: m.Rate1(), Value: ms.Rate1(),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: map[string]interface{}{ Attributes: map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,
@ -225,7 +229,7 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
}, },
Measurement{ Measurement{
Name: fmt.Sprintf("%s.%s", name, "rate.5min"), Name: fmt.Sprintf("%s.%s", name, "rate.5min"),
Value: m.Rate5(), Value: ms.Rate5(),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: map[string]interface{}{ Attributes: map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,
@ -235,7 +239,7 @@ func (rep *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot B
}, },
Measurement{ Measurement{
Name: fmt.Sprintf("%s.%s", name, "rate.15min"), Name: fmt.Sprintf("%s.%s", name, "rate.15min"),
Value: m.Rate15(), Value: ms.Rate15(),
Period: int64(rep.Interval.Seconds()), Period: int64(rep.Interval.Seconds()),
Attributes: map[string]interface{}{ Attributes: map[string]interface{}{
DisplayUnitsLong: Operations, DisplayUnitsLong: Operations,

View file

@ -29,7 +29,7 @@ func LogScaled(r Registry, freq time.Duration, scale time.Duration, l Logger) {
l.Printf(" count: %f\n", metric.Count()) l.Printf(" count: %f\n", metric.Count())
case Gauge: case Gauge:
l.Printf("gauge %s\n", name) l.Printf("gauge %s\n", name)
l.Printf(" value: %9d\n", metric.Value()) l.Printf(" value: %9d\n", metric.Snapshot().Value())
case GaugeFloat64: case GaugeFloat64:
l.Printf("gauge %s\n", name) l.Printf("gauge %s\n", name)
l.Printf(" value: %f\n", metric.Value()) l.Printf(" value: %f\n", metric.Value())

View file

@ -124,7 +124,7 @@ func (NilMeter) Stop() {}
// StandardMeter is the standard implementation of a Meter. // StandardMeter is the standard implementation of a Meter.
type StandardMeter struct { type StandardMeter struct {
count atomic.Int64 count atomic.Int64
temp atomic.Int64 uncounted atomic.Int64 // not yet added to the EWMAs
rateMean atomic.Uint64 rateMean atomic.Uint64
a1, a5, a15 EWMA a1, a5, a15 EWMA
@ -152,13 +152,13 @@ func (m *StandardMeter) Stop() {
// Mark records the occurrence of n events. // Mark records the occurrence of n events.
func (m *StandardMeter) Mark(n int64) { func (m *StandardMeter) Mark(n int64) {
m.temp.Add(n) m.uncounted.Add(n)
} }
// Snapshot returns a read-only copy of the meter. // Snapshot returns a read-only copy of the meter.
func (m *StandardMeter) Snapshot() MeterSnapshot { func (m *StandardMeter) Snapshot() MeterSnapshot {
return &meterSnapshot{ return &meterSnapshot{
count: m.count.Load(), count: m.count.Load() + m.uncounted.Load(),
rate1: m.a1.Rate(), rate1: m.a1.Rate(),
rate5: m.a5.Rate(), rate5: m.a5.Rate(),
rate15: m.a15.Rate(), rate15: m.a15.Rate(),
@ -168,7 +168,7 @@ func (m *StandardMeter) Snapshot() MeterSnapshot {
func (m *StandardMeter) tick() { func (m *StandardMeter) tick() {
// Take the uncounted values, add to count // Take the uncounted values, add to count
n := m.temp.Swap(0) n := m.uncounted.Swap(0)
count := m.count.Add(n) count := m.count.Add(n)
m.rateMean.Store(math.Float64bits(float64(count) / time.Since(m.startTime).Seconds())) m.rateMean.Store(math.Float64bits(float64(count) / time.Since(m.startTime).Seconds()))
// Update the EWMA's internal state // Update the EWMA's internal state

View file

@ -12,11 +12,17 @@ func BenchmarkMeter(b *testing.B) {
m.Mark(1) m.Mark(1)
} }
} }
func TestMeter(t *testing.T) {
m := NewMeter()
m.Mark(47)
if v := m.Snapshot().Count(); v != 47 {
t.Fatalf("have %d want %d", v, 47)
}
}
func TestGetOrRegisterMeter(t *testing.T) { func TestGetOrRegisterMeter(t *testing.T) {
r := NewRegistry() r := NewRegistry()
NewRegisteredMeter("foo", r).Mark(47) NewRegisteredMeter("foo", r).Mark(47)
if m := GetOrRegisterMeter("foo", r); m.Count() != 47 { if m := GetOrRegisterMeter("foo", r).Snapshot(); m.Count() != 47 {
t.Fatal(m.Count()) t.Fatal(m.Count())
} }
} }
@ -31,10 +37,10 @@ func TestMeterDecay(t *testing.T) {
ma.meters[m] = struct{}{} ma.meters[m] = struct{}{}
m.Mark(1) m.Mark(1)
ma.tickMeters() ma.tickMeters()
rateMean := m.RateMean() rateMean := m.Snapshot().RateMean()
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
ma.tickMeters() ma.tickMeters()
if m.RateMean() >= rateMean { if m.Snapshot().RateMean() >= rateMean {
t.Error("m.RateMean() didn't decrease") t.Error("m.RateMean() didn't decrease")
} }
} }
@ -42,7 +48,7 @@ func TestMeterDecay(t *testing.T) {
func TestMeterNonzero(t *testing.T) { func TestMeterNonzero(t *testing.T) {
m := NewMeter() m := NewMeter()
m.Mark(3) m.Mark(3)
if count := m.Count(); count != 3 { if count := m.Snapshot().Count(); count != 3 {
t.Errorf("m.Count(): 3 != %v\n", count) t.Errorf("m.Count(): 3 != %v\n", count)
} }
} }
@ -59,16 +65,8 @@ func TestMeterStop(t *testing.T) {
} }
} }
func TestMeterSnapshot(t *testing.T) {
m := NewMeter()
m.Mark(1)
if snapshot := m.Snapshot(); m.RateMean() != snapshot.RateMean() {
t.Fatal(snapshot)
}
}
func TestMeterZero(t *testing.T) { func TestMeterZero(t *testing.T) {
m := NewMeter() m := NewMeter().Snapshot()
if count := m.Count(); count != 0 { if count := m.Count(); count != 0 {
t.Errorf("m.Count(): 0 != %v\n", count) t.Errorf("m.Count(): 0 != %v\n", count)
} }
@ -79,13 +77,13 @@ func TestMeterRepeat(t *testing.T) {
for i := 0; i < 101; i++ { for i := 0; i < 101; i++ {
m.Mark(int64(i)) m.Mark(int64(i))
} }
if count := m.Count(); count != 5050 { if count := m.Snapshot().Count(); count != 5050 {
t.Errorf("m.Count(): 5050 != %v\n", count) t.Errorf("m.Count(): 5050 != %v\n", count)
} }
for i := 0; i < 101; i++ { for i := 0; i < 101; i++ {
m.Mark(int64(i)) m.Mark(int64(i))
} }
if count := m.Count(); count != 10100 { if count := m.Snapshot().Count(); count != 10100 {
t.Errorf("m.Count(): 10100 != %v\n", count) t.Errorf("m.Count(): 10100 != %v\n", count)
} }
} }

View file

@ -99,7 +99,7 @@ func Example() {
t.Update(1) t.Update(1)
fmt.Println(c.Count()) fmt.Println(c.Count())
fmt.Println(t.Min()) fmt.Println(t.Snapshot().Min())
// Output: 17 // Output: 17
// 1 // 1
} }

View file

@ -69,7 +69,7 @@ func (c *OpenTSDBConfig) writeRegistry(w io.Writer, now int64, shortHostname str
case CounterFloat64: case CounterFloat64:
fmt.Fprintf(w, "put %s.%s.count %d %f host=%s\n", c.Prefix, name, now, metric.Count(), shortHostname) fmt.Fprintf(w, "put %s.%s.count %d %f host=%s\n", c.Prefix, name, now, metric.Count(), shortHostname)
case Gauge: case Gauge:
fmt.Fprintf(w, "put %s.%s.value %d %d host=%s\n", c.Prefix, name, now, metric.Value(), shortHostname) fmt.Fprintf(w, "put %s.%s.value %d %d host=%s\n", c.Prefix, name, now, metric.Snapshot().Value(), shortHostname)
case GaugeFloat64: case GaugeFloat64:
fmt.Fprintf(w, "put %s.%s.value %d %f host=%s\n", c.Prefix, name, now, metric.Value(), shortHostname) fmt.Fprintf(w, "put %s.%s.value %d %f host=%s\n", c.Prefix, name, now, metric.Value(), shortHostname)
case GaugeInfo: case GaugeInfo:

View file

@ -83,7 +83,7 @@ func (c *collector) addCounterFloat64(name string, m metrics.CounterFloat64) {
c.writeGaugeCounter(name, m.Count()) c.writeGaugeCounter(name, m.Count())
} }
func (c *collector) addGauge(name string, m metrics.Gauge) { func (c *collector) addGauge(name string, m metrics.GaugeSnapshot) {
c.writeGaugeCounter(name, m.Value()) c.writeGaugeCounter(name, m.Value())
} }

View file

@ -154,7 +154,7 @@ func (r *StandardRegistry) GetAll() map[string]map[string]interface{} {
case CounterFloat64: case CounterFloat64:
values["count"] = metric.Count() values["count"] = metric.Count()
case Gauge: case Gauge:
values["value"] = metric.Value() values["value"] = metric.Snapshot().Value()
case GaugeFloat64: case GaugeFloat64:
values["value"] = metric.Value() values["value"] = metric.Value()
case Healthcheck: case Healthcheck:

View file

@ -353,8 +353,9 @@ func (s *UniformSample) Snapshot() SampleSnapshot {
s.mutex.Lock() s.mutex.Lock()
values := make([]int64, len(s.values)) values := make([]int64, len(s.values))
copy(values, s.values) copy(values, s.values)
count := s.count
s.mutex.Unlock() s.mutex.Unlock()
return newSampleSnapshot(s.count, s.values) return newSampleSnapshot(count, values)
} }
// Update samples a new value. // Update samples a new value.

View file

@ -87,10 +87,11 @@ func BenchmarkUniformSample1028(b *testing.B) {
} }
func TestExpDecaySample10(t *testing.T) { func TestExpDecaySample10(t *testing.T) {
s := NewExpDecaySample(100, 0.99) sw := NewExpDecaySample(100, 0.99)
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
s.Update(int64(i)) sw.Update(int64(i))
} }
s := sw.Snapshot()
if size := s.Count(); size != 10 { if size := s.Count(); size != 10 {
t.Errorf("s.Count(): 10 != %v\n", size) t.Errorf("s.Count(): 10 != %v\n", size)
} }
@ -108,10 +109,11 @@ func TestExpDecaySample10(t *testing.T) {
} }
func TestExpDecaySample100(t *testing.T) { func TestExpDecaySample100(t *testing.T) {
s := NewExpDecaySample(1000, 0.01) sw := NewExpDecaySample(1000, 0.01)
for i := 0; i < 100; i++ { for i := 0; i < 100; i++ {
s.Update(int64(i)) sw.Update(int64(i))
} }
s := sw.Snapshot()
if size := s.Count(); size != 100 { if size := s.Count(); size != 100 {
t.Errorf("s.Count(): 100 != %v\n", size) t.Errorf("s.Count(): 100 != %v\n", size)
} }
@ -129,10 +131,11 @@ func TestExpDecaySample100(t *testing.T) {
} }
func TestExpDecaySample1000(t *testing.T) { func TestExpDecaySample1000(t *testing.T) {
s := NewExpDecaySample(100, 0.99) sw := NewExpDecaySample(100, 0.99)
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
s.Update(int64(i)) sw.Update(int64(i))
} }
s := sw.Snapshot()
if size := s.Count(); size != 1000 { if size := s.Count(); size != 1000 {
t.Errorf("s.Count(): 1000 != %v\n", size) t.Errorf("s.Count(): 1000 != %v\n", size)
} }
@ -154,14 +157,15 @@ func TestExpDecaySample1000(t *testing.T) {
// The priority becomes +Inf quickly after starting if this is done, // The priority becomes +Inf quickly after starting if this is done,
// effectively freezing the set of samples until a rescale step happens. // effectively freezing the set of samples until a rescale step happens.
func TestExpDecaySampleNanosecondRegression(t *testing.T) { func TestExpDecaySampleNanosecondRegression(t *testing.T) {
s := NewExpDecaySample(100, 0.99) sw := NewExpDecaySample(100, 0.99)
for i := 0; i < 100; i++ { for i := 0; i < 100; i++ {
s.Update(10) sw.Update(10)
} }
time.Sleep(1 * time.Millisecond) time.Sleep(1 * time.Millisecond)
for i := 0; i < 100; i++ { for i := 0; i < 100; i++ {
s.Update(20) sw.Update(20)
} }
s := sw.Snapshot()
v := s.Values() v := s.Values()
avg := float64(0) avg := float64(0)
for i := 0; i < len(v); i++ { for i := 0; i < len(v); i++ {
@ -201,14 +205,15 @@ func TestExpDecaySampleStatistics(t *testing.T) {
for i := 1; i <= 10000; i++ { for i := 1; i <= 10000; i++ {
s.(*ExpDecaySample).update(now.Add(time.Duration(i)), int64(i)) s.(*ExpDecaySample).update(now.Add(time.Duration(i)), int64(i))
} }
testExpDecaySampleStatistics(t, s) testExpDecaySampleStatistics(t, s.Snapshot())
} }
func TestUniformSample(t *testing.T) { func TestUniformSample(t *testing.T) {
s := NewUniformSample(100) sw := NewUniformSample(100)
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
s.Update(int64(i)) sw.Update(int64(i))
} }
s := sw.Snapshot()
if size := s.Count(); size != 1000 { if size := s.Count(); size != 1000 {
t.Errorf("s.Count(): 1000 != %v\n", size) t.Errorf("s.Count(): 1000 != %v\n", size)
} }
@ -226,11 +231,12 @@ func TestUniformSample(t *testing.T) {
} }
func TestUniformSampleIncludesTail(t *testing.T) { func TestUniformSampleIncludesTail(t *testing.T) {
s := NewUniformSample(100) sw := NewUniformSample(100)
max := 100 max := 100
for i := 0; i < max; i++ { for i := 0; i < max; i++ {
s.Update(int64(i)) sw.Update(int64(i))
} }
s := sw.Snapshot()
v := s.Values() v := s.Values()
sum := 0 sum := 0
exp := (max - 1) * max / 2 exp := (max - 1) * max / 2
@ -257,7 +263,7 @@ func TestUniformSampleStatistics(t *testing.T) {
for i := 1; i <= 10000; i++ { for i := 1; i <= 10000; i++ {
s.Update(int64(i)) s.Update(int64(i))
} }
testUniformSampleStatistics(t, s) testUniformSampleStatistics(t, s.Snapshot())
} }
func benchmarkSample(b *testing.B, s Sample) { func benchmarkSample(b *testing.B, s Sample) {
@ -274,7 +280,7 @@ func benchmarkSample(b *testing.B, s Sample) {
b.Logf("GC cost: %d ns/op", int(memStats.PauseTotalNs-pauseTotalNs)/b.N) b.Logf("GC cost: %d ns/op", int(memStats.PauseTotalNs-pauseTotalNs)/b.N)
} }
func testExpDecaySampleStatistics(t *testing.T, s Sample) { func testExpDecaySampleStatistics(t *testing.T, s SampleSnapshot) {
if count := s.Count(); count != 10000 { if count := s.Count(); count != 10000 {
t.Errorf("s.Count(): 10000 != %v\n", count) t.Errorf("s.Count(): 10000 != %v\n", count)
} }
@ -302,7 +308,7 @@ func testExpDecaySampleStatistics(t *testing.T, s Sample) {
} }
} }
func testUniformSampleStatistics(t *testing.T, s Sample) { func testUniformSampleStatistics(t *testing.T, s SampleSnapshot) {
if count := s.Count(); count != 10000 { if count := s.Count(); count != 10000 {
t.Errorf("s.Count(): 10000 != %v\n", count) t.Errorf("s.Count(): 10000 != %v\n", count)
} }
@ -356,7 +362,7 @@ func TestUniformSampleConcurrentUpdateCount(t *testing.T) {
} }
}() }()
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
s.Count() s.Snapshot().Count()
time.Sleep(5 * time.Millisecond) time.Sleep(5 * time.Millisecond)
} }
quit <- struct{}{} quit <- struct{}{}

View file

@ -20,7 +20,7 @@ func Syslog(r Registry, d time.Duration, w *syslog.Writer) {
case CounterFloat64: case CounterFloat64:
w.Info(fmt.Sprintf("counter %s: count: %f", name, metric.Count())) w.Info(fmt.Sprintf("counter %s: count: %f", name, metric.Count()))
case Gauge: case Gauge:
w.Info(fmt.Sprintf("gauge %s: value: %d", name, metric.Value())) w.Info(fmt.Sprintf("gauge %s: value: %d", name, metric.Snapshot().Value()))
case GaugeFloat64: case GaugeFloat64:
w.Info(fmt.Sprintf("gauge %s: value: %f", name, metric.Value())) w.Info(fmt.Sprintf("gauge %s: value: %f", name, metric.Value()))
case GaugeInfo: case GaugeInfo:

View file

@ -18,7 +18,7 @@ func BenchmarkTimer(b *testing.B) {
func TestGetOrRegisterTimer(t *testing.T) { func TestGetOrRegisterTimer(t *testing.T) {
r := NewRegistry() r := NewRegistry()
NewRegisteredTimer("foo", r).Update(47) NewRegisteredTimer("foo", r).Update(47)
if tm := GetOrRegisterTimer("foo", r); tm.Count() != 1 { if tm := GetOrRegisterTimer("foo", r).Snapshot(); tm.Count() != 1 {
t.Fatal(tm) t.Fatal(tm)
} }
} }
@ -27,7 +27,7 @@ func TestTimerExtremes(t *testing.T) {
tm := NewTimer() tm := NewTimer()
tm.Update(math.MaxInt64) tm.Update(math.MaxInt64)
tm.Update(0) tm.Update(0)
if stdDev := tm.StdDev(); stdDev != 4.611686018427388e+18 { if stdDev := tm.Snapshot().StdDev(); stdDev != 4.611686018427388e+18 {
t.Errorf("tm.StdDev(): 4.611686018427388e+18 != %v\n", stdDev) t.Errorf("tm.StdDev(): 4.611686018427388e+18 != %v\n", stdDev)
} }
} }
@ -56,7 +56,7 @@ func TestTimerFunc(t *testing.T) {
}) })
var ( var (
drift = time.Millisecond * 2 drift = time.Millisecond * 2
measured = time.Duration(tm.Max()) measured = time.Duration(tm.Snapshot().Max())
ceil = actualTime + drift ceil = actualTime + drift
floor = actualTime - drift floor = actualTime - drift
) )
@ -66,7 +66,7 @@ func TestTimerFunc(t *testing.T) {
} }
func TestTimerZero(t *testing.T) { func TestTimerZero(t *testing.T) {
tm := NewTimer() tm := NewTimer().Snapshot()
if count := tm.Count(); count != 0 { if count := tm.Count(); count != 0 {
t.Errorf("tm.Count(): 0 != %v\n", count) t.Errorf("tm.Count(): 0 != %v\n", count)
} }
@ -110,5 +110,5 @@ func ExampleGetOrRegisterTimer() {
m := "account.create.latency" m := "account.create.latency"
t := GetOrRegisterTimer(m, nil) t := GetOrRegisterTimer(m, nil)
t.Update(47) t.Update(47)
fmt.Println(t.Max()) // Output: 47 fmt.Println(t.Snapshot().Max()) // Output: 47
} }

View file

@ -35,7 +35,7 @@ func WriteOnce(r Registry, w io.Writer) {
fmt.Fprintf(w, " count: %f\n", metric.Count()) fmt.Fprintf(w, " count: %f\n", metric.Count())
case Gauge: case Gauge:
fmt.Fprintf(w, "gauge %s\n", namedMetric.name) fmt.Fprintf(w, "gauge %s\n", namedMetric.name)
fmt.Fprintf(w, " value: %9d\n", metric.Value()) fmt.Fprintf(w, " value: %9d\n", metric.Snapshot().Value())
case GaugeFloat64: case GaugeFloat64:
fmt.Fprintf(w, "gauge %s\n", namedMetric.name) fmt.Fprintf(w, "gauge %s\n", namedMetric.name)
fmt.Fprintf(w, " value: %f\n", metric.Value()) fmt.Fprintf(w, " value: %f\n", metric.Value())