add metrics for dns

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

.gitea/workflows/job_lint.yml

@@ -0,0 +1,26 @@
+name: lint
+
+on:
+  pull_request:
+    types: [opened, reopened, synchronize]
+    branches:
+    - master
+    - v3
+    - v4
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+    - name: setup-go
+      uses: actions/setup-go@v5
+      with:
+        go-version: 'stable'
+    - name: checkout
+      uses: actions/checkout@v3
+    - name: deps
+      run: go get -v -d ./...
+    - name: lint
+      uses: https://github.com/golangci/golangci-lint-action@v6
+      with:
+        version: 'latest'

.gitea/workflows/job_test.yml

@@ -1,29 +1,20 @@
-name: pipeline
+name: test
+
 on:
   pull_request:
+    types: [opened, reopened, synchronize]
     branches:
     - master
     - v3
     - v4
+  push:
+    branches:
+    - master
+    - v3
+    - v4
+
 jobs:
-  lint:
-    name: lint
-    runs-on: ubuntu-latest
-    steps:
-    - name: setup-go
-      uses: actions/setup-go@v5
-      with:
-        go-version: 'stable'
-    - name: checkout
-      uses: actions/checkout@v3
-    - name: deps
-      run: go get -v -d ./...
-    - name: lint
-      uses: https://github.com/golangci/golangci-lint-action@v6
-      with:
-        version: v1.62.2
   test:
-    name: test
     runs-on: ubuntu-latest
     steps:
     - name: setup-go

client/noop.go

@@ -222,7 +222,7 @@ func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opt
 	ts := time.Now()
 	n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
 	var sp tracer.Span
-	ctx, sp = n.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
+	ctx, sp = n.opts.Tracer.Start(ctx, "rpc-client",
 		tracer.WithSpanKind(tracer.SpanKindClient),
 		tracer.WithSpanLabels("endpoint", req.Endpoint()),
 	)
@@ -385,7 +385,7 @@ func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption
 	ts := time.Now()
 	n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
 	var sp tracer.Span
-	ctx, sp = n.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
+	ctx, sp = n.opts.Tracer.Start(ctx, "rpc-client",
 		tracer.WithSpanKind(tracer.SpanKindClient),
 		tracer.WithSpanLabels("endpoint", req.Endpoint()),
 	)

logger/slog/slog.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"log/slog"
 	"os"
+	"reflect"
 	"regexp"
 	"runtime"
 	"strconv"
@@ -171,7 +172,29 @@ func (s *slogLogger) Init(opts ...logger.Option) error {
 	}
 
 	attrs, _ := s.argsAttrs(s.opts.Fields)
-	s.handler = &wrapper{h: slog.NewJSONHandler(s.opts.Out, handleOpt).WithAttrs(attrs)}
+
+	var h slog.Handler
+	if s.opts.Context != nil {
+		if v, ok := s.opts.Context.Value(handlerKey{}).(slog.Handler); ok && v != nil {
+			h = v
+		}
+
+		if fn := s.opts.Context.Value(handlerFnKey{}); fn != nil {
+			if rfn := reflect.ValueOf(fn); rfn.Kind() == reflect.Func {
+				if ret := rfn.Call([]reflect.Value{reflect.ValueOf(s.opts.Out), reflect.ValueOf(handleOpt)}); len(ret) == 1 {
+					if iface, ok := ret[0].Interface().(slog.Handler); ok && iface != nil {
+						h = iface
+					}
+				}
+			}
+		}
+	}
+
+	if h == nil {
+		h = slog.NewJSONHandler(s.opts.Out, handleOpt)
+	}
+
+	s.handler = &wrapper{h: h.WithAttrs(attrs)}
 	s.handler.level.Store(int64(loggerToSlogLevel(s.opts.Level)))
 	s.mu.Unlock()
 
@@ -329,3 +352,15 @@ func (s *slogLogger) argsAttrs(args []interface{}) ([]slog.Attr, error) {
 
 	return attrs, err
 }
+
+type handlerKey struct{}
+
+func WithHandler(h slog.Handler) logger.Option {
+	return logger.SetOption(handlerKey{}, h)
+}
+
+type handlerFnKey struct{}
+
+func WithHandlerFunc(fn any) logger.Option {
+	return logger.SetOption(handlerFnKey{}, fn)
+}

logger/slog/slog_test.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"log/slog"
 	"strings"
 	"testing"
 
@@ -15,6 +16,23 @@ import (
 	"go.unistack.org/micro/v3/logger"
 )
 
+func TestWithHandlerFunc(t *testing.T) {
+	ctx := context.TODO()
+	buf := bytes.NewBuffer(nil)
+	l := NewLogger(logger.WithLevel(logger.InfoLevel), logger.WithOutput(buf),
+		WithHandlerFunc(slog.NewTextHandler),
+	)
+	if err := l.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	l.Info(ctx, "msg1")
+
+	if !bytes.Contains(buf.Bytes(), []byte(`msg=msg1`)) {
+		t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
+	}
+}
+
 func TestWithAddFields(t *testing.T) {
 	ctx := context.TODO()
 	buf := bytes.NewBuffer(nil)

semconv/cache.go

@@ -0,0 +1,14 @@
+package semconv
+
+var (
+	// CacheRequestDurationSeconds specifies meter metric name
+	CacheRequestDurationSeconds = "micro_cache_request_duration_seconds"
+	// CacheRequestLatencyMicroseconds specifies meter metric name
+	CacheRequestLatencyMicroseconds = "micro_cache_request_latency_microseconds"
+	// CacheRequestTotal specifies meter metric name
+	CacheRequestTotal = "micro_cache_request_total"
+	// CacheRequestInflight specifies meter metric name
+	CacheRequestInflight = "micro_cache_request_inflight"
+	// CacheItemsTotal specifies total cache items
+	CacheItemsTotal = "micro_cache_items_total"
+)

semconv/store.go

@@ -3,7 +3,7 @@ package semconv
 var (
 	// StoreRequestDurationSeconds specifies meter metric name
 	StoreRequestDurationSeconds = "micro_store_request_duration_seconds"
-	// ClientRequestLatencyMicroseconds specifies meter metric name
+	// StoreRequestLatencyMicroseconds specifies meter metric name
 	StoreRequestLatencyMicroseconds = "micro_store_request_latency_microseconds"
 	// StoreRequestTotal specifies meter metric name
 	StoreRequestTotal = "micro_store_request_total"

service.go

@@ -34,7 +34,10 @@ func init() {
 		),
 	)
 
-	net.DefaultResolver = utildns.NewNetResolver(utildns.Timeout(1 * time.Second))
+	net.DefaultResolver = utildns.NewNetResolver(
+		utildns.Timeout(1*time.Second),
+		utildns.MinCacheTTL(5*time.Second),
+	)
 }
 
 // Service is an interface that wraps the lower level components.

store/store.go

@@ -10,7 +10,7 @@ import (
 var (
 	ErrWatcherStopped = errors.New("watcher stopped")
 	// ErrNotConnected is returned when a store is not connected
-	ErrNotConnected = errors.New("not conected")
+	ErrNotConnected = errors.New("not connected")
 	// ErrNotFound is returned when a key doesn't exist
 	ErrNotFound = errors.New("not found")
 	// ErrInvalidKey is returned when a key has empty or have invalid format

util/dns/cache.go

@@ -6,6 +6,9 @@ import (
 	"net"
 	"sync"
 	"time"
+
+	"go.unistack.org/micro/v3/meter"
+	"go.unistack.org/micro/v3/semconv"
 )
 
 // DialFunc is a [net.Resolver.Dial] function.
@@ -19,6 +22,11 @@ func NewNetResolver(opts ...Option) *net.Resolver {
 		o(&options)
 	}
 
+	if options.Meter == nil {
+		options.Meter = meter.DefaultMeter
+		opts = append(opts, Meter(options.Meter))
+	}
+
 	return &net.Resolver{
 		PreferGo:     true,
 		StrictErrors: options.Resolver.StrictErrors,
@@ -56,6 +64,7 @@ type Options struct {
 	PreferIPV4      bool
 	PreferIPV6      bool
 	Timeout         time.Duration
+	Meter           meter.Meter
 }
 
 // MaxCacheEntries sets the maximum number of entries to cache.
@@ -87,6 +96,13 @@ func NegativeCache(b bool) Option {
 	}
 }
 
+// Meter sets meter.Meter
+func Meter(m meter.Meter) Option {
+	return func(o *Options) {
+		o.Meter = m
+	}
+}
+
 // Timeout sets upstream *net.Resolver timeout
 func Timeout(td time.Duration) Option {
 	return func(o *Options) {
@@ -156,7 +172,6 @@ func (c *cache) put(req string, res string) {
 	}
 
 	c.Lock()
-	defer c.Unlock()
 	if c.entries == nil {
 		c.entries = make(map[string]cacheEntry)
 	}
@@ -165,6 +180,8 @@ func (c *cache) put(req string, res string) {
 	var tested, evicted int
 	for k, e := range c.entries {
 		if time.Until(e.deadline) <= 0 {
+			c.opts.Meter.Counter(semconv.CacheItemsTotal, "type", "dns").Dec()
+			c.opts.Meter.Counter(semconv.CacheRequestTotal, "type", "dns", "method", "evict").Inc()
 			// delete expired entry
 			delete(c.entries, k)
 			evicted++
@@ -175,6 +192,8 @@ func (c *cache) put(req string, res string) {
 			continue
 		}
 		if evicted == 0 && c.opts.MaxCacheEntries > 0 && len(c.entries) >= c.opts.MaxCacheEntries {
+			c.opts.Meter.Counter(semconv.CacheItemsTotal, "type", "dns").Dec()
+			c.opts.Meter.Counter(semconv.CacheRequestTotal, "type", "dns", "method", "evict").Inc()
 			// delete at least one entry
 			delete(c.entries, k)
 		}
@@ -186,6 +205,9 @@ func (c *cache) put(req string, res string) {
 		deadline: time.Now().Add(ttl),
 		value:    res[2:],
 	}
+
+	c.opts.Meter.Counter(semconv.CacheItemsTotal, "type", "dns").Inc()
+	c.Unlock()
 }
 
 func (c *cache) get(req string) (res string) {
@@ -210,6 +232,7 @@ func (c *cache) get(req string) (res string) {
 		// prepend correct ID
 		return req[:2] + entry.value
 	}
+
 	return ""
 }
 
@@ -310,10 +333,19 @@ func getUint32(s string) int {
 
 func cachingRoundTrip(cache *cache, network, address string) roundTripper {
 	return func(ctx context.Context, req string) (res string, err error) {
+		cache.opts.Meter.Counter(semconv.CacheRequestInflight, "type", "dns").Inc()
+		defer cache.opts.Meter.Counter(semconv.CacheRequestInflight, "type", "dns").Dec()
 		// check cache
 		if res := cache.get(req); res != "" {
+			cache.opts.Meter.Counter(semconv.CacheRequestTotal, "type", "dns", "method", "get", "status", "hit").Inc()
 			return res, nil
 		}
+		cache.opts.Meter.Counter(semconv.CacheRequestTotal, "type", "dns", "method", "get", "status", "miss").Inc()
+		ts := time.Now()
+		defer func() {
+			cache.opts.Meter.Summary(semconv.CacheRequestLatencyMicroseconds, "type", "dns", "method", "get").UpdateDuration(ts)
+			cache.opts.Meter.Histogram(semconv.CacheRequestDurationSeconds, "type", "dns", "method", "get").UpdateDuration(ts)
+		}()
 
 		switch {
 		case cache.opts.PreferIPV4 && cache.opts.PreferIPV6:
@@ -340,6 +372,7 @@ func cachingRoundTrip(cache *cache, network, address string) roundTripper {
 			var d net.Dialer
 			conn, err = d.DialContext(ctx, network, address)
 		}
+
 		if err != nil {
 			return "", err
 		}

util/dns/cache_test.go

@@ -12,5 +12,11 @@ func TestCache(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Logf("addrs %v", addrs)
+
+	addrs, err = net.LookupHost("unistack.org")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_ = addrs
 }

util/dns/conn.go

@@ -11,15 +11,13 @@ import (
 )
 
 type dnsConn struct {
-	sync.Mutex
-
-	ibuf bytes.Buffer
-	obuf bytes.Buffer
-
+	deadline  time.Time
 	ctx       context.Context
 	cancel    context.CancelFunc
-	deadline  time.Time
 	roundTrip roundTripper
+	ibuf      bytes.Buffer
+	obuf      bytes.Buffer
+	sync.Mutex
 }
 
 type roundTripper func(ctx context.Context, req string) (res string, err error)
@@ -78,8 +76,8 @@ func (c *dnsConn) SetDeadline(t time.Time) error {
 
 func (c *dnsConn) SetReadDeadline(t time.Time) error {
 	c.Lock()
-	defer c.Unlock()
 	c.deadline = t
+	c.Unlock()
 	return nil
 }