3 Commits

Author SHA1 Message Date
d4472e1ab2 partially fix race cond
Some checks failed
codeql / analyze (go) (push) Failing after 49s
build / test (push) Failing after 4m54s
build / lint (push) Successful in 9m28s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-11-08 15:27:05 +03:00
6f6d362c20 experimental race free labels
Some checks failed
codeql / analyze (go) (push) Failing after 44s
build / test (push) Failing after 4m52s
build / lint (push) Successful in 9m27s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-11-07 16:49:16 +03:00
dd71d9ec59 add locking
Some checks failed
codeql / analyze (go) (push) Failing after 44s
build / test (push) Failing after 4m56s
build / lint (push) Successful in 9m35s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-11-06 00:00:17 +03:00
5 changed files with 165 additions and 105 deletions

View File

@@ -11,55 +11,57 @@ import (
type prometheusCounter struct {
name string
c *dto.Metric
n float64
}
func (c *prometheusCounter) Add(n int) {
addFloat64(c.c.Gauge.Value, float64(n))
addFloat64(&(c.n), float64(n))
}
func (c *prometheusCounter) Dec() {
addFloat64(c.c.Gauge.Value, float64(-1))
addFloat64(&(c.n), float64(-1))
}
func (c *prometheusCounter) Inc() {
addFloat64(c.c.Gauge.Value, float64(1))
addFloat64(&(c.n), float64(1))
}
func (c *prometheusCounter) Get() uint64 {
return uint64(getFloat64(c.c.Gauge.Value))
return uint64(getFloat64(&(c.n)))
}
func (c *prometheusCounter) Set(n uint64) {
setFloat64(c.c.Gauge.Value, math.Float64frombits(n))
setFloat64(&(c.n), math.Float64frombits(n))
}
type prometheusFloatCounter struct {
name string
c *dto.Metric
n float64
}
func (c *prometheusFloatCounter) Add(n float64) {
addFloat64(c.c.Gauge.Value, n)
addFloat64(&(c.n), n)
}
func (c *prometheusFloatCounter) Dec() {
addFloat64(c.c.Gauge.Value, float64(-1))
addFloat64(&(c.n), float64(-1))
}
func (c *prometheusFloatCounter) Inc() {
addFloat64(c.c.Gauge.Value, float64(1))
addFloat64(&(c.n), float64(1))
}
func (c *prometheusFloatCounter) Get() float64 {
return getFloat64(c.c.Gauge.Value)
return getFloat64(&(c.n))
}
func (c *prometheusFloatCounter) Set(n float64) {
setFloat64(c.c.Gauge.Value, n)
setFloat64(&(c.n), n)
}
func (c *prometheusFloatCounter) Sub(n float64) {
addFloat64(c.c.Gauge.Value, -n)
addFloat64(&(c.n), -n)
}
func setFloat64(_addr *float64, value float64) float64 {

View File

@@ -5,8 +5,9 @@ import dto "github.com/prometheus/client_model/go"
type prometheusGauge struct {
name string
c *dto.Metric
n float64
}
func (c prometheusGauge) Get() float64 {
return getFloat64(c.c.Gauge.Value)
func (c *prometheusGauge) Get() float64 {
return getFloat64(&(c.n))
}

View File

@@ -5,6 +5,7 @@ import (
"io"
"regexp"
"sync"
"sync/atomic"
"time"
"github.com/prometheus/client_golang/prometheus"
@@ -21,23 +22,24 @@ var _ meter.Meter = (*prometheusMeter)(nil)
type prometheusMeter struct {
opts meter.Options
set prometheus.Registerer
counter *sync.Map
floatCounter *sync.Map
gauge *sync.Map
histogram *sync.Map
summary *sync.Map
counter map[uint64]*prometheusCounter
floatCounter map[uint64]*prometheusFloatCounter
gauge map[uint64]*prometheusGauge
histogram map[uint64]*prometheusHistogram
summary map[uint64]*prometheusSummary
mfPool xpool.Pool[*dto.MetricFamily]
mu sync.Mutex
}
func NewMeter(opts ...meter.Option) *prometheusMeter {
return &prometheusMeter{
set: prometheus.NewRegistry(), // prometheus.DefaultRegisterer,
opts: meter.NewOptions(opts...),
counter: &sync.Map{},
floatCounter: &sync.Map{},
gauge: &sync.Map{},
histogram: &sync.Map{},
summary: &sync.Map{},
counter: make(map[uint64]*prometheusCounter),
floatCounter: make(map[uint64]*prometheusFloatCounter),
gauge: make(map[uint64]*prometheusGauge),
histogram: make(map[uint64]*prometheusHistogram),
summary: make(map[uint64]*prometheusSummary),
mfPool: xpool.NewPool[*dto.MetricFamily](func() *dto.MetricFamily {
return &dto.MetricFamily{}
}),
@@ -51,69 +53,84 @@ func (m *prometheusMeter) Name() string {
func (m *prometheusMeter) Counter(name string, labels ...string) meter.Counter {
clabels := meter.BuildLabels(append(m.opts.Labels, labels...)...)
h := newHash(name, clabels)
mc, ok := m.counter.Load(h)
m.mu.Lock()
c, ok := m.counter[h]
// fmt.Printf("counter name %s hash %v labels %v\n", name, h, labels)
m.mu.Unlock()
if !ok {
var v float64
mc = &prometheusCounter{
var n float64
c = &prometheusCounter{
name: name,
c: &dto.Metric{
Gauge: &dto.Gauge{Value: &v},
Gauge: &dto.Gauge{Value: &n},
Label: labelMetric(clabels),
},
}
m.counter.Store(h, mc)
m.mu.Lock()
m.counter[h] = c
m.mu.Unlock()
}
return mc.(*prometheusCounter)
return c
}
func (m *prometheusMeter) FloatCounter(name string, labels ...string) meter.FloatCounter {
clabels := meter.BuildLabels(append(m.opts.Labels, labels...)...)
h := newHash(name, clabels)
mc, ok := m.floatCounter.Load(h)
m.mu.Lock()
c, ok := m.floatCounter[h]
m.mu.Unlock()
if !ok {
var v float64
mc = &prometheusFloatCounter{
var n float64
c = &prometheusFloatCounter{
name: name,
c: &dto.Metric{
Gauge: &dto.Gauge{Value: &v},
Gauge: &dto.Gauge{Value: &n},
Label: labelMetric(clabels),
},
}
m.floatCounter.Store(h, mc)
m.mu.Lock()
m.floatCounter[h] = c
m.mu.Unlock()
}
return mc.(*prometheusFloatCounter)
return c
}
func (m *prometheusMeter) Gauge(name string, fn func() float64, labels ...string) meter.Gauge {
clabels := meter.BuildLabels(append(m.opts.Labels, labels...)...)
h := newHash(name, clabels)
mc, ok := m.gauge.Load(h)
m.mu.Lock()
c, ok := m.gauge[h]
m.mu.Unlock()
if !ok {
var v float64
mc = &prometheusGauge{
var n float64
c = &prometheusGauge{
name: name,
c: &dto.Metric{
Gauge: &dto.Gauge{Value: &v},
Gauge: &dto.Gauge{Value: &n},
Label: labelMetric(clabels),
},
}
m.gauge.Store(h, mc)
m.mu.Lock()
m.gauge[h] = c
m.mu.Unlock()
}
return mc.(*prometheusGauge)
return c
}
func (m *prometheusMeter) Histogram(name string, labels ...string) meter.Histogram {
clabels := meter.BuildLabels(append(m.opts.Labels, labels...)...)
h := newHash(name, clabels)
mc, ok := m.histogram.Load(h)
m.mu.Lock()
c, ok := m.histogram[h]
m.mu.Unlock()
if !ok {
var c uint64
var n uint64
var s float64
buckets := make([]float64, len(prometheus.DefBuckets))
copy(buckets, prometheus.DefBuckets)
mdto := &dto.Metric{
Histogram: &dto.Histogram{
SampleCount: &c,
SampleCount: &n,
SampleSum: &s,
CreatedTimestamp: timestamppb.Now(),
Bucket: make([]*dto.Bucket, len(buckets)),
@@ -124,59 +141,68 @@ func (m *prometheusMeter) Histogram(name string, labels ...string) meter.Histogr
var cc uint64
mdto.Histogram.Bucket[idx] = &dto.Bucket{CumulativeCount: &cc, UpperBound: &b}
}
mc = &prometheusHistogram{
c = &prometheusHistogram{
name: name,
c: mdto,
}
m.histogram.Store(h, mc)
m.mu.Lock()
m.histogram[h] = c
m.mu.Unlock()
}
return mc.(*prometheusHistogram)
return c
}
func (m *prometheusMeter) Summary(name string, labels ...string) meter.Summary {
clabels := meter.BuildLabels(append(m.opts.Labels, labels...)...)
h := newHash(name, clabels)
mc, ok := m.summary.Load(h)
m.mu.Lock()
c, ok := m.summary[h]
m.mu.Unlock()
if !ok {
var c uint64
var n uint64
var s float64
mc = &prometheusSummary{
c = &prometheusSummary{
name: name,
c: &dto.Metric{
Summary: &dto.Summary{
SampleCount: &c,
SampleCount: &n,
SampleSum: &s,
CreatedTimestamp: timestamppb.Now(),
},
Label: labelMetric(clabels),
},
}
m.summary.Store(h, mc)
m.mu.Lock()
m.summary[h] = c
m.mu.Unlock()
}
return mc.(*prometheusSummary)
return c
}
func (m *prometheusMeter) SummaryExt(name string, window time.Duration, quantiles []float64, labels ...string) meter.Summary {
clabels := meter.BuildLabels(append(m.opts.Labels, labels...)...)
h := newHash(name, clabels)
mc, ok := m.summary.Load(h)
m.mu.Lock()
c, ok := m.summary[h]
m.mu.Lock()
if !ok {
var c uint64
var n uint64
var s float64
mc = &prometheusSummary{
c = &prometheusSummary{
name: name,
c: &dto.Metric{
Summary: &dto.Summary{
SampleCount: &c,
SampleCount: &n,
SampleSum: &s,
},
Label: labelMetric(clabels),
},
}
m.summary.Store(h, mc)
m.mu.Lock()
m.summary[h] = c
m.mu.Unlock()
}
return mc.(*prometheusSummary)
return c
}
func (m *prometheusMeter) Init(opts ...meter.Option) error {
@@ -212,55 +238,59 @@ func (m *prometheusMeter) Write(w io.Writer, opts ...meter.Option) error {
enc := expfmt.NewEncoder(w, expfmt.NewFormat(expfmt.TypeTextPlain))
m.counter.Range(func(k, v any) bool {
c := v.(*prometheusCounter)
mf := m.mfPool.Get()
mf.Name = &c.name
mf.Type = dto.MetricType_GAUGE.Enum()
mf.Metric = append(mf.Metric, c.c)
mfs = append(mfs, mf)
return true
})
m.mu.Lock()
m.floatCounter.Range(func(k, v any) bool {
c := v.(*prometheusFloatCounter)
for _, mc := range m.counter {
mf := m.mfPool.Get()
mf.Name = &c.name
mf.Name = &mc.name
mf.Type = dto.MetricType_GAUGE.Enum()
mf.Metric = append(mf.Metric, c.c)
n := getFloat64(&(mc.n))
mc.c.Gauge.Value = &n
mf.Metric = append(mf.Metric, mc.c)
mfs = append(mfs, mf)
return true
})
}
m.gauge.Range(func(k, v any) bool {
c := v.(*prometheusGauge)
for _, mc := range m.floatCounter {
mf := m.mfPool.Get()
mf.Name = &c.name
mf.Name = &mc.name
mf.Type = dto.MetricType_GAUGE.Enum()
mf.Metric = append(mf.Metric, c.c)
n := getFloat64(&(mc.n))
mc.c.Gauge.Value = &n
mf.Metric = append(mf.Metric, mc.c)
mfs = append(mfs, mf)
return true
})
}
m.histogram.Range(func(k, v any) bool {
c := v.(*prometheusHistogram)
for _, mc := range m.gauge {
mf := m.mfPool.Get()
mf.Name = &mc.name
mf.Type = dto.MetricType_GAUGE.Enum()
n := getFloat64(&(mc.n))
mc.c.Gauge.Value = &n
mf.Metric = append(mf.Metric, mc.c)
mfs = append(mfs, mf)
}
for _, c := range m.histogram {
mf := m.mfPool.Get()
mf.Name = &c.name
mf.Type = dto.MetricType_HISTOGRAM.Enum()
mf.Metric = append(mf.Metric, c.c)
mfs = append(mfs, mf)
return true
})
}
m.summary.Range(func(k, v any) bool {
c := v.(*prometheusSummary)
for _, mc := range m.summary {
mf := m.mfPool.Get()
mf.Name = &c.name
mf.Name = &mc.name
mf.Type = dto.MetricType_SUMMARY.Enum()
mf.Metric = append(mf.Metric, c.c)
sc := atomic.LoadUint64(&(mc.sampleCount))
mc.c.Summary.SampleCount = &sc
ss := getFloat64(&(mc.SampleSum))
mc.c.Summary.SampleSum = &ss
mf.Metric = append(mf.Metric, mc.c)
mfs = append(mfs, mf)
return true
})
}
m.mu.Unlock()
for _, mf := range mfs {
_ = enc.Encode(mf)
@@ -310,11 +340,13 @@ func (m *prometheusMeter) Set(opts ...meter.Option) meter.Meter {
}
func labelMetric(labels []string) []*dto.LabelPair {
dtoLabels := make([]*dto.LabelPair, 0, len(labels)/2)
for idx := 0; idx < len(labels); idx += 2 {
nl := make([]string, len(labels))
copy(nl, labels)
dtoLabels := make([]*dto.LabelPair, 0, len(nl)/2)
for idx := 0; idx < len(nl); idx += 2 {
dtoLabels = append(dtoLabels, &dto.LabelPair{
Name: &(labels[idx]),
Value: &(labels[idx+1]),
Name: &(nl[idx]),
Value: &(nl[idx+1]),
})
}
return dtoLabels

View File

@@ -14,6 +14,28 @@ import (
"go.unistack.org/micro/v3/meter"
)
func TestHash(t *testing.T) {
m := NewMeter() // meter.Labels("test_key", "test_val"))
buf := bytes.NewBuffer(nil)
for i := 0; i < 100000; i++ {
go func() {
m.Counter("micro_server_request_total", "code", "16",
"endpoint", "/clientprofile.ClientProfileService/GetClientProfile",
"status", "failure").Inc()
m.Counter("micro_server_request_total", "code", "16",
"endpoint", "/clientproduct.ClientProductService/GetDepositProducts",
"status", "failure").Inc()
m.Counter("micro_server_request_total", "code", "16",
"endpoint", "/operationsinfo.OperationsInfoService/GetOperations",
"status", "failure").Inc()
}()
}
_ = m.Write(buf)
t.Logf("h1: %s\n", buf.Bytes())
}
func TestHistogram(t *testing.T) {
m := NewMeter()
name := "test"
@@ -56,6 +78,7 @@ func TestHistogram(t *testing.T) {
}
func TestSummary(t *testing.T) {
t.Skip()
name := "micro_server"
m := NewMeter()
m.Summary("micro_server").Update(1)
@@ -87,7 +110,7 @@ func TestSummary(t *testing.T) {
p.Observe(10)
p.Observe(30)
mdto := &dto.Metric{}
p.Write(mdto)
_ = p.Write(mdto)
pbuf := bytes.NewBuffer(nil)
enc := expfmt.NewEncoder(pbuf, expfmt.NewFormat(expfmt.TypeTextPlain))
mf := &dto.MetricFamily{Name: &name, Type: dto.MetricType_SUMMARY.Enum(), Metric: []*dto.Metric{mdto}}

View File

@@ -8,17 +8,19 @@ import (
)
type prometheusSummary struct {
name string
c *dto.Metric
name string
c *dto.Metric
sampleCount uint64
SampleSum float64
}
func (c prometheusSummary) Update(n float64) {
atomic.AddUint64(c.c.Summary.SampleCount, 1)
addFloat64(c.c.Summary.SampleSum, n)
func (c *prometheusSummary) Update(n float64) {
atomic.AddUint64(&(c.sampleCount), 1)
addFloat64(&(c.SampleSum), n)
}
func (c prometheusSummary) UpdateDuration(n time.Time) {
x := time.Since(n).Seconds()
atomic.AddUint64(c.c.Summary.SampleCount, 1)
addFloat64(c.c.Summary.SampleSum, x)
func (c *prometheusSummary) UpdateDuration(t time.Time) {
n := time.Since(t).Seconds()
atomic.AddUint64(&(c.sampleCount), 1)
addFloat64(&(c.SampleSum), n)
}