meter: improve Gauge

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

README.md

@@ -1,5 +1,5 @@
 # Micro
-![Coverage](https://img.shields.io/badge/Coverage-34.1%25-yellow)
+![Coverage](https://img.shields.io/badge/Coverage-33.8%25-yellow)
 [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0)
 [![Doc](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white&style=flat-square)](https://pkg.go.dev/go.unistack.org/micro/v4?tab=overview)
 [![Status](https://git.unistack.org/unistack-org/micro/actions/workflows/job_tests.yml/badge.svg?branch=v4)](https://git.unistack.org/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Av4+event%3Apush)

hooks/sql/stats.go

@@ -3,6 +3,7 @@ package sql
 import (
 	"context"
 	"database/sql"
+	"sync"
 	"time"
 )
 
@@ -11,31 +12,84 @@ type Statser interface {
 }
 
 func NewStatsMeter(ctx context.Context, db Statser, opts ...Option) {
-	options := NewOptions(opts...)
-
-	go func() {
-		ticker := time.NewTicker(options.MeterStatsInterval)
-		defer ticker.Stop()
-
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
 	if db == nil {
 		return
 	}
+
+	options := NewOptions(opts...)
+
+	var (
+		statsMu                                                     sync.Mutex
+		lastUpdated                                                 time.Time
+		maxOpenConnections, openConnections, inUse, idle, waitCount float64
+		maxIdleClosed, maxIdleTimeClosed, maxLifetimeClosed         float64
+		waitDuration                                                float64
+	)
+
+	updateFn := func() {
+		statsMu.Lock()
+		defer statsMu.Unlock()
+
+		if time.Since(lastUpdated) < options.MeterStatsInterval {
+			return
+		}
+
 		stats := db.Stats()
-				options.Meter.Counter(MaxOpenConnections).Set(uint64(stats.MaxOpenConnections))
-				options.Meter.Counter(OpenConnections).Set(uint64(stats.OpenConnections))
-				options.Meter.Counter(InuseConnections).Set(uint64(stats.InUse))
-				options.Meter.Counter(IdleConnections).Set(uint64(stats.Idle))
-				options.Meter.Counter(WaitConnections).Set(uint64(stats.WaitCount))
-				options.Meter.FloatCounter(BlockedSeconds).Set(stats.WaitDuration.Seconds())
-				options.Meter.Counter(MaxIdleClosed).Set(uint64(stats.MaxIdleClosed))
-				options.Meter.Counter(MaxIdletimeClosed).Set(uint64(stats.MaxIdleTimeClosed))
-				options.Meter.Counter(MaxLifetimeClosed).Set(uint64(stats.MaxLifetimeClosed))
+		maxOpenConnections = float64(stats.MaxOpenConnections)
+		openConnections = float64(stats.OpenConnections)
+		inUse = float64(stats.InUse)
+		idle = float64(stats.Idle)
+		waitCount = float64(stats.WaitCount)
+		maxIdleClosed = float64(stats.MaxIdleClosed)
+		maxIdleTimeClosed = float64(stats.MaxIdleTimeClosed)
+		maxLifetimeClosed = float64(stats.MaxLifetimeClosed)
+		waitDuration = float64(stats.WaitDuration.Seconds())
+
+		lastUpdated = time.Now()
 	}
-		}
-	}()
+
+	options.Meter.Gauge(MaxOpenConnections, func() float64 {
+		updateFn()
+		return maxOpenConnections
+	})
+
+	options.Meter.Gauge(OpenConnections, func() float64 {
+		updateFn()
+		return openConnections
+	})
+
+	options.Meter.Gauge(InuseConnections, func() float64 {
+		updateFn()
+		return inUse
+	})
+
+	options.Meter.Gauge(IdleConnections, func() float64 {
+		updateFn()
+		return idle
+	})
+
+	options.Meter.Gauge(WaitConnections, func() float64 {
+		updateFn()
+		return waitCount
+	})
+
+	options.Meter.Gauge(BlockedSeconds, func() float64 {
+		updateFn()
+		return waitDuration
+	})
+
+	options.Meter.Gauge(MaxIdleClosed, func() float64 {
+		updateFn()
+		return maxIdleClosed
+	})
+
+	options.Meter.Gauge(MaxIdletimeClosed, func() float64 {
+		updateFn()
+		return maxIdleTimeClosed
+	})
+
+	options.Meter.Gauge(MaxLifetimeClosed, func() float64 {
+		updateFn()
+		return maxLifetimeClosed
+	})
 }

logger/options.go

@@ -52,6 +52,12 @@ type Options struct {
 	AddStacktrace bool
 	// DedupKeys deduplicate keys in log output
 	DedupKeys bool
+	// FatalFinalizers runs in order in [logger.Fatal] method
+	FatalFinalizers []func(context.Context)
+}
+
+var DefaultFatalFinalizer = func(ctx context.Context) {
+	os.Exit(1)
 }
 
 // NewOptions creates new options struct
@@ -65,6 +71,7 @@ func NewOptions(opts ...Option) Options {
 		AddSource:        true,
 		TimeFunc:         time.Now,
 		Meter:            meter.DefaultMeter,
+		FatalFinalizers:  []func(context.Context){DefaultFatalFinalizer},
 	}
 
 	WithMicroKeys()(&options)
@@ -76,6 +83,13 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
+// WithFatalFinalizers set logger.Fatal finalizers
+func WithFatalFinalizers(fncs ...func(context.Context)) Option {
+	return func(o *Options) {
+		o.FatalFinalizers = fncs
+	}
+}
+
 // WithContextAttrFuncs appends default funcs for the context attrs filler
 func WithContextAttrFuncs(fncs ...ContextAttrFunc) Option {
 	return func(o *Options) {

logger/slog/slog.go

@@ -4,14 +4,12 @@ import (
 	"context"
 	"io"
 	"log/slog"
-	"os"
 	"reflect"
 	"regexp"
 	"runtime"
 	"strconv"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/semconv"
@@ -231,11 +229,12 @@ func (s *slogLogger) Error(ctx context.Context, msg string, attrs ...interface{}
 
 func (s *slogLogger) Fatal(ctx context.Context, msg string, attrs ...interface{}) {
 	s.printLog(ctx, logger.FatalLevel, msg, attrs...)
+	for _, fn := range s.opts.FatalFinalizers {
+		fn(ctx)
+	}
 	if closer, ok := s.opts.Out.(io.Closer); ok {
 		closer.Close()
 	}
-	time.Sleep(1 * time.Second)
-	os.Exit(1)
 }
 
 func (s *slogLogger) Warn(ctx context.Context, msg string, attrs ...interface{}) {

logger/slog/slog_test.go

@@ -469,3 +469,25 @@ func Test_WithContextAttrFunc(t *testing.T) {
 
 	// t.Logf("xxx %s", buf.Bytes())
 }
+
+func TestFatalFinalizers(t *testing.T) {
+	ctx := context.TODO()
+	buf := bytes.NewBuffer(nil)
+	l := NewLogger(
+		logger.WithLevel(logger.TraceLevel),
+		logger.WithOutput(buf),
+	)
+	if err := l.Init(
+		logger.WithFatalFinalizers(func(ctx context.Context) {
+			l.Info(ctx, "fatal finalizer")
+		})); err != nil {
+		t.Fatal(err)
+	}
+	l.Fatal(ctx, "info_msg1")
+	if !bytes.Contains(buf.Bytes(), []byte("fatal finalizer")) {
+		t.Fatalf("logger dont have fatal message, buf %s", buf.Bytes())
+	}
+	if !bytes.Contains(buf.Bytes(), []byte("info_msg1")) {
+		t.Fatalf("logger dont have info_msg1 message, buf %s", buf.Bytes())
+	}
+}

meter/meter.go

@@ -59,6 +59,8 @@ type Meter interface {
 	Options() Options
 	// String return meter type
 	String() string
+	// Unregister metric name and drop all data
+	Unregister(name string, labels ...string) bool
 }
 
 // Counter is a counter
@@ -80,7 +82,11 @@ type FloatCounter interface {
 
 // Gauge is a float64 gauge
 type Gauge interface {
+	Add(float64)
 	Get() float64
+	Set(float64)
+	Dec()
+	Inc()
 }
 
 // Histogram is a histogram for non-negative values with automatically created buckets

meter/noop.go

@@ -28,6 +28,10 @@ func (r *noopMeter) Name() string {
 	return r.opts.Name
 }
 
+func (r *noopMeter) Unregister(name string, labels ...string) bool {
+	return true
+}
+
 // Init initialize options
 func (r *noopMeter) Init(opts ...Option) error {
 	for _, o := range opts {
@@ -132,6 +136,18 @@ type noopGauge struct {
 	labels []string
 }
 
+func (r *noopGauge) Add(float64) {
+}
+
+func (r *noopGauge) Set(float64) {
+}
+
+func (r *noopGauge) Inc() {
+}
+
+func (r *noopGauge) Dec() {
+}
+
 func (r *noopGauge) Get() float64 {
 	return 0
 }

util/xpool/pool.go

@@ -6,18 +6,18 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"go.unistack.org/micro/v4/meter"
 	"go.unistack.org/micro/v4/semconv"
 )
 
-var (
-	pools   = make([]Statser, 0)
-	poolsMu sync.Mutex
-)
+func unregisterMetrics(size int) {
+	meter.DefaultMeter.Unregister(semconv.PoolGetTotal, "capacity", strconv.Itoa(size))
+	meter.DefaultMeter.Unregister(semconv.PoolPutTotal, "capacity", strconv.Itoa(size))
+	meter.DefaultMeter.Unregister(semconv.PoolMisTotal, "capacity", strconv.Itoa(size))
+	meter.DefaultMeter.Unregister(semconv.PoolRetTotal, "capacity", strconv.Itoa(size))
+}
 
-// Stats struct
 type Stats struct {
 	Get uint64
 	Put uint64
@@ -25,41 +25,13 @@ type Stats struct {
 	Ret uint64
 }
 
-// Statser provides buffer pool stats
-type Statser interface {
-	Stats() Stats
-	Cap() int
-}
-
-func init() {
-	go newStatsMeter()
-}
-
-func newStatsMeter() {
-	ticker := time.NewTicker(meter.DefaultMeterStatsInterval)
-	defer ticker.Stop()
-
-	for range ticker.C {
-		poolsMu.Lock()
-		for _, st := range pools {
-			stats := st.Stats()
-			meter.DefaultMeter.Counter(semconv.PoolGetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Get)
-			meter.DefaultMeter.Counter(semconv.PoolPutTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Put)
-			meter.DefaultMeter.Counter(semconv.PoolMisTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Mis)
-			meter.DefaultMeter.Counter(semconv.PoolRetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Ret)
-		}
-		poolsMu.Unlock()
-	}
-}
-
-var (
-	_ Statser = (*BytePool)(nil)
-	_ Statser = (*BytesPool)(nil)
-	_ Statser = (*StringsPool)(nil)
-)
-
 type Pool[T any] struct {
 	p   *sync.Pool
+	get *atomic.Uint64
+	put *atomic.Uint64
+	mis *atomic.Uint64
+	ret *atomic.Uint64
+	c   int
 }
 
 func (p Pool[T]) Put(t T) {
@@ -70,37 +42,82 @@ func (p Pool[T]) Get() T {
 	return p.p.Get().(T)
 }
 
-func NewPool[T any](fn func() T) Pool[T] {
-	return Pool[T]{
-		p: &sync.Pool{
+func NewPool[T any](fn func() T, size int) Pool[T] {
+	p := Pool[T]{
+		c:   size,
+		get: &atomic.Uint64{},
+		put: &atomic.Uint64{},
+		mis: &atomic.Uint64{},
+		ret: &atomic.Uint64{},
+	}
+
+	p.p = &sync.Pool{
 		New: func() interface{} {
+			p.mis.Add(1)
 			return fn()
 		},
-		},
 	}
+
+	meter.DefaultMeter.Gauge(semconv.PoolGetTotal, func() float64 {
+		return float64(p.get.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolPutTotal, func() float64 {
+		return float64(p.put.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolMisTotal, func() float64 {
+		return float64(p.mis.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolRetTotal, func() float64 {
+		return float64(p.ret.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	return p
 }
 
 type BytePool struct {
 	p   *sync.Pool
-	get uint64
-	put uint64
-	mis uint64
-	ret uint64
+	get *atomic.Uint64
+	put *atomic.Uint64
+	mis *atomic.Uint64
+	ret *atomic.Uint64
 	c   int
 }
 
 func NewBytePool(size int) *BytePool {
-	p := &BytePool{c: size}
+	p := &BytePool{
+		c:   size,
+		get: &atomic.Uint64{},
+		put: &atomic.Uint64{},
+		mis: &atomic.Uint64{},
+		ret: &atomic.Uint64{},
+	}
 	p.p = &sync.Pool{
 		New: func() interface{} {
-			atomic.AddUint64(&p.mis, 1)
+			p.mis.Add(1)
 			b := make([]byte, 0, size)
 			return &b
 		},
 	}
-	poolsMu.Lock()
-	pools = append(pools, p)
-	poolsMu.Unlock()
+
+	meter.DefaultMeter.Gauge(semconv.PoolGetTotal, func() float64 {
+		return float64(p.get.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolPutTotal, func() float64 {
+		return float64(p.put.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolMisTotal, func() float64 {
+		return float64(p.mis.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolRetTotal, func() float64 {
+		return float64(p.ret.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
 	return p
 }
 
@@ -110,49 +127,73 @@ func (p *BytePool) Cap() int {
 
 func (p *BytePool) Stats() Stats {
 	return Stats{
-		Put: atomic.LoadUint64(&p.put),
-		Get: atomic.LoadUint64(&p.get),
-		Mis: atomic.LoadUint64(&p.mis),
-		Ret: atomic.LoadUint64(&p.ret),
+		Put: p.put.Load(),
+		Get: p.get.Load(),
+		Mis: p.mis.Load(),
+		Ret: p.ret.Load(),
 	}
 }
 
 func (p *BytePool) Get() *[]byte {
-	atomic.AddUint64(&p.get, 1)
+	p.get.Add(1)
 	return p.p.Get().(*[]byte)
 }
 
 func (p *BytePool) Put(b *[]byte) {
-	atomic.AddUint64(&p.put, 1)
+	p.put.Add(1)
 	if cap(*b) > p.c {
-		atomic.AddUint64(&p.ret, 1)
+		p.ret.Add(1)
 		return
 	}
 	*b = (*b)[:0]
 	p.p.Put(b)
 }
 
+func (p *BytePool) Close() {
+	unregisterMetrics(p.c)
+}
+
 type BytesPool struct {
 	p   *sync.Pool
-	get uint64
-	put uint64
-	mis uint64
-	ret uint64
+	get *atomic.Uint64
+	put *atomic.Uint64
+	mis *atomic.Uint64
+	ret *atomic.Uint64
 	c   int
 }
 
 func NewBytesPool(size int) *BytesPool {
-	p := &BytesPool{c: size}
+	p := &BytesPool{
+		c:   size,
+		get: &atomic.Uint64{},
+		put: &atomic.Uint64{},
+		mis: &atomic.Uint64{},
+		ret: &atomic.Uint64{},
+	}
 	p.p = &sync.Pool{
 		New: func() interface{} {
-			atomic.AddUint64(&p.mis, 1)
+			p.mis.Add(1)
 			b := bytes.NewBuffer(make([]byte, 0, size))
 			return b
 		},
 	}
-	poolsMu.Lock()
-	pools = append(pools, p)
-	poolsMu.Unlock()
+
+	meter.DefaultMeter.Gauge(semconv.PoolGetTotal, func() float64 {
+		return float64(p.get.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolPutTotal, func() float64 {
+		return float64(p.put.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolMisTotal, func() float64 {
+		return float64(p.mis.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
+	meter.DefaultMeter.Gauge(semconv.PoolRetTotal, func() float64 {
+		return float64(p.ret.Load())
+	}, "capacity", strconv.Itoa(p.c))
+
 	return p
 }
 
@@ -162,10 +203,10 @@ func (p *BytesPool) Cap() int {
 
 func (p *BytesPool) Stats() Stats {
 	return Stats{
-		Put: atomic.LoadUint64(&p.put),
-		Get: atomic.LoadUint64(&p.get),
-		Mis: atomic.LoadUint64(&p.mis),
-		Ret: atomic.LoadUint64(&p.ret),
+		Put: p.put.Load(),
+		Get: p.get.Load(),
+		Mis: p.mis.Load(),
+		Ret: p.ret.Load(),
 	}
 }
 
@@ -174,34 +215,43 @@ func (p *BytesPool) Get() *bytes.Buffer {
 }
 
 func (p *BytesPool) Put(b *bytes.Buffer) {
+	p.put.Add(1)
 	if (*b).Cap() > p.c {
-		atomic.AddUint64(&p.ret, 1)
+		p.ret.Add(1)
 		return
 	}
 	b.Reset()
 	p.p.Put(b)
 }
 
+func (p *BytesPool) Close() {
+	unregisterMetrics(p.c)
+}
+
 type StringsPool struct {
 	p   *sync.Pool
-	get uint64
-	put uint64
-	mis uint64
-	ret uint64
+	get *atomic.Uint64
+	put *atomic.Uint64
+	mis *atomic.Uint64
+	ret *atomic.Uint64
 	c   int
 }
 
 func NewStringsPool(size int) *StringsPool {
-	p := &StringsPool{c: size}
+	p := &StringsPool{
+		c:   size,
+		get: &atomic.Uint64{},
+		put: &atomic.Uint64{},
+		mis: &atomic.Uint64{},
+		ret: &atomic.Uint64{},
+	}
 	p.p = &sync.Pool{
 		New: func() interface{} {
-			atomic.AddUint64(&p.mis, 1)
+			p.mis.Add(1)
 			return &strings.Builder{}
 		},
 	}
-	poolsMu.Lock()
-	pools = append(pools, p)
-	poolsMu.Unlock()
+
 	return p
 }
 
@@ -211,24 +261,28 @@ func (p *StringsPool) Cap() int {
 
 func (p *StringsPool) Stats() Stats {
 	return Stats{
-		Put: atomic.LoadUint64(&p.put),
-		Get: atomic.LoadUint64(&p.get),
-		Mis: atomic.LoadUint64(&p.mis),
-		Ret: atomic.LoadUint64(&p.ret),
+		Put: p.put.Load(),
+		Get: p.get.Load(),
+		Mis: p.mis.Load(),
+		Ret: p.ret.Load(),
 	}
 }
 
 func (p *StringsPool) Get() *strings.Builder {
-	atomic.AddUint64(&p.get, 1)
+	p.get.Add(1)
 	return p.p.Get().(*strings.Builder)
 }
 
 func (p *StringsPool) Put(b *strings.Builder) {
-	atomic.AddUint64(&p.put, 1)
+	p.put.Add(1)
 	if b.Cap() > p.c {
-		atomic.AddUint64(&p.ret, 1)
+		p.ret.Add(1)
 		return
 	}
 	b.Reset()
 	p.p.Put(b)
 }
+
+func (p *StringsPool) Close() {
+	unregisterMetrics(p.c)
+}