From ebce157ddecaa4c14fc118ca19019e54c14db4c8 Mon Sep 17 00:00:00 2001 From: Aliaksandr Valialkin Date: Sat, 1 Jun 2019 23:18:41 +0300 Subject: [PATCH] Add Set for controlling metric sets to be exported via WritePrometheus call --- counter.go | 31 +---- gauge.go | 41 +----- metrics.go | 42 +----- set.go | 308 ++++++++++++++++++++++++++++++++++++++++++++ set_example_test.go | 28 ++++ set_test.go | 32 +++++ summary.go | 123 +++++------------- 7 files changed, 410 insertions(+), 195 deletions(-) create mode 100644 set.go create mode 100644 set_example_test.go create mode 100644 set_test.go diff --git a/counter.go b/counter.go index 60a62b3..e683729 100644 --- a/counter.go +++ b/counter.go @@ -17,9 +17,7 @@ import ( // // The returned counter is safe to use from concurrent goroutines. func NewCounter(name string) *Counter { - c := &Counter{} - registerMetric(name, c) - return c + return defaultSet.NewCounter(name) } // Counter is a counter. @@ -75,30 +73,5 @@ func (c *Counter) marshalTo(prefix string, w io.Writer) { // // Performance tip: prefer NewCounter instead of GetOrCreateCounter. func GetOrCreateCounter(name string) *Counter { - metricsMapLock.Lock() - nm := metricsMap[name] - metricsMapLock.Unlock() - if nm == nil { - // Slow path - create and register missing counter. - if err := validateMetric(name); err != nil { - panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) - } - nmNew := &namedMetric{ - name: name, - metric: &Counter{}, - } - metricsMapLock.Lock() - nm = metricsMap[name] - if nm == nil { - nm = nmNew - metricsMap[name] = nm - metricsList = append(metricsList, nm) - } - metricsMapLock.Unlock() - } - c, ok := nm.metric.(*Counter) - if !ok { - panic(fmt.Errorf("BUG: metric %q isn't a Counter. It is %T", name, nm.metric)) - } - return c + return defaultSet.GetOrCreateCounter(name) } diff --git a/gauge.go b/gauge.go index a71886d..4b93da1 100644 --- a/gauge.go +++ b/gauge.go @@ -19,14 +19,7 @@ import ( // // The returned gauge is safe to use from concurrent goroutines. func NewGauge(name string, f func() float64) *Gauge { - if f == nil { - panic(fmt.Errorf("BUG: f cannot be nil")) - } - g := &Gauge{ - f: f, - } - registerMetric(name, g) - return g + return defaultSet.NewGauge(name, f) } // Gauge is a float64 gauge. @@ -66,35 +59,5 @@ func (g *Gauge) marshalTo(prefix string, w io.Writer) { // // Performance tip: prefer NewGauge instead of GetOrCreateGauge. func GetOrCreateGauge(name string, f func() float64) *Gauge { - metricsMapLock.Lock() - nm := metricsMap[name] - metricsMapLock.Unlock() - if nm == nil { - // Slow path - create and register missing gauge. - if f == nil { - panic(fmt.Errorf("BUG: f cannot be nil")) - } - if err := validateMetric(name); err != nil { - panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) - } - nmNew := &namedMetric{ - name: name, - metric: &Gauge{ - f: f, - }, - } - metricsMapLock.Lock() - nm = metricsMap[name] - if nm == nil { - nm = nmNew - metricsMap[name] = nm - metricsList = append(metricsList, nm) - } - metricsMapLock.Unlock() - } - g, ok := nm.metric.(*Gauge) - if !ok { - panic(fmt.Errorf("BUG: metric %q isn't a Gauge. It is %T", name, nm.metric)) - } - return g + return defaultSet.GetOrCreateGauge(name, f) } diff --git a/metrics.go b/metrics.go index 19f8f8a..c28a08a 100644 --- a/metrics.go +++ b/metrics.go @@ -16,18 +16,10 @@ import ( "fmt" "io" "runtime" - "sort" - "sync" "github.com/valyala/histogram" ) -var ( - metricsMapLock sync.Mutex - metricsList []*namedMetric - metricsMap = make(map[string]*namedMetric) -) - type namedMetric struct { name string metric metric @@ -37,6 +29,8 @@ type metric interface { marshalTo(prefix string, w io.Writer) } +var defaultSet = NewSet() + // WritePrometheus writes all the registered metrics in Prometheus format to w. // // If exposeProcessMetrics is true, then various `go_*` metrics are exposed @@ -49,17 +43,7 @@ type metric interface { // }) // func WritePrometheus(w io.Writer, exposeProcessMetrics bool) { - lessFunc := func(i, j int) bool { - return metricsList[i].name < metricsList[j].name - } - metricsMapLock.Lock() - if !sort.SliceIsSorted(metricsList, lessFunc) { - sort.Slice(metricsList, lessFunc) - } - for _, nm := range metricsList { - nm.metric.marshalTo(nm.name, w) - } - metricsMapLock.Unlock() + defaultSet.WritePrometheus(w) if exposeProcessMetrics { writeProcessMetrics(w) } @@ -119,23 +103,3 @@ func writeProcessMetrics(w io.Writer) { fmt.Fprintf(w, "go_info_ext{compiler=%q, GOARCH=%q, GOOS=%q, GOROOT=%q} 1\n", runtime.Compiler, runtime.GOARCH, runtime.GOOS, runtime.GOROOT()) } - -func registerMetric(name string, m metric) { - if err := validateMetric(name); err != nil { - panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) - } - metricsMapLock.Lock() - nm, ok := metricsMap[name] - if !ok { - nm = &namedMetric{ - name: name, - metric: m, - } - metricsMap[name] = nm - metricsList = append(metricsList, nm) - } - metricsMapLock.Unlock() - if ok { - panic(fmt.Errorf("BUG: metric %q is already registered", name)) - } -} diff --git a/set.go b/set.go new file mode 100644 index 0000000..293a46b --- /dev/null +++ b/set.go @@ -0,0 +1,308 @@ +package metrics + +import ( + "fmt" + "io" + "sort" + "sync" + "time" +) + +// Set is a set of metrics. +// +// Metrics belonging to a set are exported separately from global metrics. +// +// Set.WritePrometheus must be called for exporting metrics from the set. +type Set struct { + mu sync.Mutex + a []*namedMetric + m map[string]*namedMetric +} + +// NewSet creates new set of metrics. +func NewSet() *Set { + return &Set{ + m: make(map[string]*namedMetric), + } +} + +// WritePrometheus writes all the metrics from s to w in Prometheus format. +func (s *Set) WritePrometheus(w io.Writer) { + lessFunc := func(i, j int) bool { + return s.a[i].name < s.a[j].name + } + s.mu.Lock() + if !sort.SliceIsSorted(s.a, lessFunc) { + sort.Slice(s.a, lessFunc) + } + for _, nm := range s.a { + nm.metric.marshalTo(nm.name, w) + } + s.mu.Unlock() +} + +// NewCounter registers and returns new counter with the given name in the s. +// +// name must be valid Prometheus-compatible metric with possible lables. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// The returned counter is safe to use from concurrent goroutines. +func (s *Set) NewCounter(name string) *Counter { + c := &Counter{} + s.registerMetric(name, c) + return c +} + +// GetOrCreateCounter returns registered counter in s with the given name +// or creates new counter if s doesn't contain counter with the given name. +// +// name must be valid Prometheus-compatible metric with possible lables. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// The returned counter is safe to use from concurrent goroutines. +// +// Performance tip: prefer NewCounter instead of GetOrCreateCounter. +func (s *Set) GetOrCreateCounter(name string) *Counter { + s.mu.Lock() + nm := s.m[name] + s.mu.Unlock() + if nm == nil { + // Slow path - create and register missing counter. + if err := validateMetric(name); err != nil { + panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) + } + nmNew := &namedMetric{ + name: name, + metric: &Counter{}, + } + s.mu.Lock() + nm = s.m[name] + if nm == nil { + nm = nmNew + s.m[name] = nm + s.a = append(s.a, nm) + } + s.mu.Unlock() + } + c, ok := nm.metric.(*Counter) + if !ok { + panic(fmt.Errorf("BUG: metric %q isn't a Counter. It is %T", name, nm.metric)) + } + return c +} + +// NewGauge registers and returns gauge with the given name in s, which calls f +// to obtain gauge value. +// +// name must be valid Prometheus-compatible metric with possible labels. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// f must be safe for concurrent calls. +// +// The returned gauge is safe to use from concurrent goroutines. +func (s *Set) NewGauge(name string, f func() float64) *Gauge { + if f == nil { + panic(fmt.Errorf("BUG: f cannot be nil")) + } + g := &Gauge{ + f: f, + } + s.registerMetric(name, g) + return g +} + +// GetOrCreateGauge returns registered gauge with the given name in s +// or creates new gauge if s doesn't contain gauge with the given name. +// +// name must be valid Prometheus-compatible metric with possible lables. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// The returned gauge is safe to use from concurrent goroutines. +// +// Performance tip: prefer NewGauge instead of GetOrCreateGauge. +func (s *Set) GetOrCreateGauge(name string, f func() float64) *Gauge { + s.mu.Lock() + nm := s.m[name] + s.mu.Unlock() + if nm == nil { + // Slow path - create and register missing gauge. + if f == nil { + panic(fmt.Errorf("BUG: f cannot be nil")) + } + if err := validateMetric(name); err != nil { + panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) + } + nmNew := &namedMetric{ + name: name, + metric: &Gauge{ + f: f, + }, + } + s.mu.Lock() + nm = s.m[name] + if nm == nil { + nm = nmNew + s.m[name] = nm + s.a = append(s.a, nm) + } + s.mu.Unlock() + } + g, ok := nm.metric.(*Gauge) + if !ok { + panic(fmt.Errorf("BUG: metric %q isn't a Gauge. It is %T", name, nm.metric)) + } + return g +} + +// NewSummary creates and returns new summary with the given name in s. +// +// name must be valid Prometheus-compatible metric with possible lables. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// The returned summary is safe to use from concurrent goroutines. +func (s *Set) NewSummary(name string) *Summary { + return s.NewSummaryExt(name, defaultSummaryWindow, defaultSummaryQuantiles) +} + +// NewSummaryExt creates and returns new summary in s with the given name, +// window and quantiles. +// +// name must be valid Prometheus-compatible metric with possible lables. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// The returned summary is safe to use from concurrent goroutines. +func (s *Set) NewSummaryExt(name string, window time.Duration, quantiles []float64) *Summary { + sm := newSummary(window, quantiles) + s.registerMetric(name, sm) + registerSummary(sm) + s.registerSummaryQuantiles(name, sm) + return sm +} + +// GetOrCreateSummary returns registered summary with the given name in s +// or creates new summary if s doesn't contain summary with the given name. +// +// name must be valid Prometheus-compatible metric with possible lables. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// The returned summary is safe to use from concurrent goroutines. +// +// Performance tip: prefer NewSummary instead of GetOrCreateSummary. +func (s *Set) GetOrCreateSummary(name string) *Summary { + return s.GetOrCreateSummaryExt(name, defaultSummaryWindow, defaultSummaryQuantiles) +} + +// GetOrCreateSummaryExt returns registered summary with the given name, +// window and quantiles in s or creates new summary if s doesn't +// contain summary with the given name. +// +// name must be valid Prometheus-compatible metric with possible lables. +// For instance, +// +// * foo +// * foo{bar="baz"} +// * foo{bar="baz",aaa="b"} +// +// The returned summary is safe to use from concurrent goroutines. +// +// Performance tip: prefer NewSummaryExt instead of GetOrCreateSummaryExt. +func (s *Set) GetOrCreateSummaryExt(name string, window time.Duration, quantiles []float64) *Summary { + s.mu.Lock() + nm := s.m[name] + s.mu.Unlock() + if nm == nil { + // Slow path - create and register missing summary. + if err := validateMetric(name); err != nil { + panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) + } + sm := newSummary(window, quantiles) + nmNew := &namedMetric{ + name: name, + metric: sm, + } + mustRegisterQuantiles := false + s.mu.Lock() + nm = s.m[name] + if nm == nil { + nm = nmNew + s.m[name] = nm + s.a = append(s.a, nm) + registerSummary(sm) + mustRegisterQuantiles = true + } + s.mu.Unlock() + if mustRegisterQuantiles { + s.registerSummaryQuantiles(name, sm) + } + } + sm, ok := nm.metric.(*Summary) + if !ok { + panic(fmt.Errorf("BUG: metric %q isn't a Summary. It is %T", name, nm.metric)) + } + if sm.window != window { + panic(fmt.Errorf("BUG: invalid window requested for the summary %q; requested %s; need %s", name, window, sm.window)) + } + if !isEqualQuantiles(sm.quantiles, quantiles) { + panic(fmt.Errorf("BUG: invalid quantiles requested from the summary %q; requested %v; need %v", name, quantiles, sm.quantiles)) + } + return sm +} + +func (s *Set) registerSummaryQuantiles(name string, sm *Summary) { + for i, q := range sm.quantiles { + quantileValueName := addTag(name, fmt.Sprintf(`quantile="%g"`, q)) + qv := &quantileValue{ + sm: sm, + idx: i, + } + s.registerMetric(quantileValueName, qv) + } +} + +func (s *Set) registerMetric(name string, m metric) { + if err := validateMetric(name); err != nil { + panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) + } + s.mu.Lock() + nm, ok := s.m[name] + if !ok { + nm = &namedMetric{ + name: name, + metric: m, + } + s.m[name] = nm + s.a = append(s.a, nm) + } + s.mu.Unlock() + if ok { + panic(fmt.Errorf("BUG: metric %q is already registered", name)) + } +} diff --git a/set_example_test.go b/set_example_test.go new file mode 100644 index 0000000..f9b1ab0 --- /dev/null +++ b/set_example_test.go @@ -0,0 +1,28 @@ +package metrics_test + +import ( + "bytes" + "fmt" + "github.com/VictoriaMetrics/metrics" +) + +func ExampleSet() { + // Create a set with a counter + s := metrics.NewSet() + sc := s.NewCounter("set_counter") + sc.Inc() + s.NewGauge(`set_gauge{foo="bar"}`, func() float64 { return 42 }) + + // Dump global metrics + var bb bytes.Buffer + + // Dump metrics from s. + bb.Reset() + s.WritePrometheus(&bb) + fmt.Printf("set metrics:\n%s\n", bb.String()) + + // Output: + // set metrics: + // set_counter 1 + // set_gauge{foo="bar"} 42 +} diff --git a/set_test.go b/set_test.go new file mode 100644 index 0000000..7e9fda9 --- /dev/null +++ b/set_test.go @@ -0,0 +1,32 @@ +package metrics + +import ( + "fmt" + "testing" +) + +func TestNewSet(t *testing.T) { + var ss []*Set + for i := 0; i < 10; i++ { + s := NewSet() + ss = append(ss, s) + } + for i := 0; i < 10; i++ { + s := ss[i] + for j := 0; j < 10; j++ { + c := s.NewCounter(fmt.Sprintf("counter_%d", j)) + c.Inc() + if n := c.Get(); n != 1 { + t.Fatalf("unexpected counter value; got %d; want %d", n, 1) + } + g := s.NewGauge(fmt.Sprintf("gauge_%d", j), func() float64 { return 123 }) + if v := g.Get(); v != 123 { + t.Fatalf("unexpected gauge value; got %v; want %v", v, 123) + } + sm := s.NewSummary(fmt.Sprintf("summary_%d", j)) + if sm == nil { + t.Fatalf("NewSummary returned nil") + } + } + } +} diff --git a/summary.go b/summary.go index e6179d8..7075aab 100644 --- a/summary.go +++ b/summary.go @@ -38,7 +38,7 @@ type Summary struct { // // The returned summary is safe to use from concurrent goroutines. func NewSummary(name string) *Summary { - return NewSummaryExt(name, defaultSummaryWindow, defaultSummaryQuantiles) + return defaultSet.NewSummary(name) } // NewSummaryExt creates and returns new summary with the given name, @@ -53,36 +53,21 @@ func NewSummary(name string) *Summary { // // The returned summary is safe to use from concurrent goroutines. func NewSummaryExt(name string, window time.Duration, quantiles []float64) *Summary { - s := newSummary(window, quantiles) - registerMetric(name, s) - registerSummary(s) - registerSummaryQuantiles(name, s) - return s + return defaultSet.NewSummaryExt(name, window, quantiles) } func newSummary(window time.Duration, quantiles []float64) *Summary { // Make a copy of quantiles in order to prevent from their modification by the caller. quantiles = append([]float64{}, quantiles...) validateQuantiles(quantiles) - s := &Summary{ + sm := &Summary{ curr: histogram.NewFast(), next: histogram.NewFast(), quantiles: quantiles, quantileValues: make([]float64, len(quantiles)), window: window, } - return s -} - -func registerSummaryQuantiles(name string, s *Summary) { - for i, q := range s.quantiles { - quantileValueName := addTag(name, fmt.Sprintf(`quantile="%g"`, q)) - qv := &quantileValue{ - s: s, - idx: i, - } - registerMetric(quantileValueName, qv) - } + return sm } func validateQuantiles(quantiles []float64) { @@ -94,29 +79,29 @@ func validateQuantiles(quantiles []float64) { } // Update updates the summary. -func (s *Summary) Update(v float64) { - s.mu.Lock() - s.curr.Update(v) - s.next.Update(v) - s.mu.Unlock() +func (sm *Summary) Update(v float64) { + sm.mu.Lock() + sm.curr.Update(v) + sm.next.Update(v) + sm.mu.Unlock() } // UpdateDuration updates request duration based on the given startTime. -func (s *Summary) UpdateDuration(startTime time.Time) { +func (sm *Summary) UpdateDuration(startTime time.Time) { d := time.Since(startTime).Seconds() - s.Update(d) + sm.Update(d) } -func (s *Summary) marshalTo(prefix string, w io.Writer) { - // Just update s.quantileValues and don't write anything to w. - // s.quantileValues will be marshaled later via quantileValue.marshalTo. - s.updateQuantiles() +func (sm *Summary) marshalTo(prefix string, w io.Writer) { + // Just update sm.quantileValues and don't write anything to w. + // sm.quantileValues will be marshaled later via quantileValue.marshalTo. + sm.updateQuantiles() } -func (s *Summary) updateQuantiles() { - s.mu.Lock() - s.quantileValues = s.curr.Quantiles(s.quantileValues[:0], s.quantiles) - s.mu.Unlock() +func (sm *Summary) updateQuantiles() { + sm.mu.Lock() + sm.quantileValues = sm.curr.Quantiles(sm.quantileValues[:0], sm.quantiles) + sm.mu.Unlock() } // GetOrCreateSummary returns registered summary with the given name @@ -134,7 +119,7 @@ func (s *Summary) updateQuantiles() { // // Performance tip: prefer NewSummary instead of GetOrCreateSummary. func GetOrCreateSummary(name string) *Summary { - return GetOrCreateSummaryExt(name, defaultSummaryWindow, defaultSummaryQuantiles) + return defaultSet.GetOrCreateSummary(name) } // GetOrCreateSummaryExt returns registered summary with the given name, @@ -152,45 +137,7 @@ func GetOrCreateSummary(name string) *Summary { // // Performance tip: prefer NewSummaryExt instead of GetOrCreateSummaryExt. func GetOrCreateSummaryExt(name string, window time.Duration, quantiles []float64) *Summary { - metricsMapLock.Lock() - nm := metricsMap[name] - metricsMapLock.Unlock() - if nm == nil { - // Slow path - create and register missing summary. - if err := validateMetric(name); err != nil { - panic(fmt.Errorf("BUG: invalid metric name %q: %s", name, err)) - } - s := newSummary(window, quantiles) - nmNew := &namedMetric{ - name: name, - metric: s, - } - mustRegisterQuantiles := false - metricsMapLock.Lock() - nm = metricsMap[name] - if nm == nil { - nm = nmNew - metricsMap[name] = nm - metricsList = append(metricsList, nm) - registerSummary(s) - mustRegisterQuantiles = true - } - metricsMapLock.Unlock() - if mustRegisterQuantiles { - registerSummaryQuantiles(name, s) - } - } - s, ok := nm.metric.(*Summary) - if !ok { - panic(fmt.Errorf("BUG: metric %q isn't a Summary. It is %T", name, nm.metric)) - } - if s.window != window { - panic(fmt.Errorf("BUG: invalid window requested for the summary %q; requested %s; need %s", name, window, s.window)) - } - if !isEqualQuantiles(s.quantiles, quantiles) { - panic(fmt.Errorf("BUG: invalid quantiles requested from the summary %q; requested %v; need %v", name, quantiles, s.quantiles)) - } - return s + return defaultSet.GetOrCreateSummaryExt(name, window, quantiles) } func isEqualQuantiles(a, b []float64) bool { @@ -207,14 +154,14 @@ func isEqualQuantiles(a, b []float64) bool { } type quantileValue struct { - s *Summary + sm *Summary idx int } func (qv *quantileValue) marshalTo(prefix string, w io.Writer) { - qv.s.mu.Lock() - v := qv.s.quantileValues[qv.idx] - qv.s.mu.Unlock() + qv.sm.mu.Lock() + v := qv.sm.quantileValues[qv.idx] + qv.sm.mu.Unlock() if !math.IsNaN(v) { fmt.Fprintf(w, "%s %g\n", prefix, v) } @@ -227,10 +174,10 @@ func addTag(name, tag string) string { return fmt.Sprintf("%s,%s}", name[:len(name)-1], tag) } -func registerSummary(s *Summary) { - window := s.window +func registerSummary(sm *Summary) { + window := sm.window summariesLock.Lock() - summaries[window] = append(summaries[window], s) + summaries[window] = append(summaries[window], sm) if len(summaries[window]) == 1 { go summariesSwapCron(window) } @@ -241,13 +188,13 @@ func summariesSwapCron(window time.Duration) { for { time.Sleep(window / 2) summariesLock.Lock() - for _, s := range summaries[window] { - s.mu.Lock() - tmp := s.curr - s.curr = s.next - s.next = tmp - s.next.Reset() - s.mu.Unlock() + for _, sm := range summaries[window] { + sm.mu.Lock() + tmp := sm.curr + sm.curr = sm.next + sm.next = tmp + sm.next.Reset() + sm.mu.Unlock() } summariesLock.Unlock() }