add using lazy connect (#361)

#357

Co-authored-by: Василий Толстов <v.tolstov@unistack.org>
Reviewed-on: #361
Reviewed-by: Василий Толстов <v.tolstov@unistack.org>
Co-authored-by: Evstigneev Denis <danteevstigneev@yandex.ru>
Co-committed-by: Evstigneev Denis <danteevstigneev@yandex.ru>

logger/slog/slog.go

@@ -8,6 +8,7 @@ import (
 	"runtime"
 	"strconv"
 	"sync"
+	"sync/atomic"
 
 	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/semconv"
@@ -31,6 +32,27 @@ var (
 	fatalValue = slog.StringValue("fatal")
 )
 
+type wrapper struct {
+	h     slog.Handler
+	level atomic.Int64
+}
+
+func (h *wrapper) Enabled(ctx context.Context, level slog.Level) bool {
+	return level >= slog.Level(int(h.level.Load()))
+}
+
+func (h *wrapper) Handle(ctx context.Context, rec slog.Record) error {
+	return h.h.Handle(ctx, rec)
+}
+
+func (h *wrapper) WithAttrs(attrs []slog.Attr) slog.Handler {
+	return h.WithAttrs(attrs)
+}
+
+func (h *wrapper) WithGroup(name string) slog.Handler {
+	return h.WithGroup(name)
+}
+
 func (s *slogLogger) renameAttr(_ []string, a slog.Attr) slog.Attr {
 	switch a.Key {
 	case slog.SourceKey:
@@ -68,7 +90,7 @@ func (s *slogLogger) renameAttr(_ []string, a slog.Attr) slog.Attr {
 
 type slogLogger struct {
 	leveler *slog.LevelVar
-	handler slog.Handler
+	handler *wrapper
 	opts    logger.Options
 	mu      sync.RWMutex
 }
@@ -82,51 +104,52 @@ func (s *slogLogger) Clone(opts ...logger.Option) logger.Logger {
 		o(&options)
 	}
 
-	l := &slogLogger{
-		opts: options,
+	if len(options.ContextAttrFuncs) == 0 {
+		options.ContextAttrFuncs = logger.DefaultContextAttrFuncs
 	}
 
-	l.leveler = new(slog.LevelVar)
-	handleOpt := &slog.HandlerOptions{
-		ReplaceAttr: l.renameAttr,
-		Level:       l.leveler,
-		AddSource:   l.opts.AddSource,
+	attrs, _ := s.argsAttrs(options.Fields)
+	l := &slogLogger{
+		handler: &wrapper{h: s.handler.h.WithAttrs(attrs)},
+		opts:    options,
 	}
-	l.leveler.Set(loggerToSlogLevel(l.opts.Level))
-	l.handler = slog.New(slog.NewJSONHandler(options.Out, handleOpt)).With(options.Fields...).Handler()
+	l.handler.level.Store(int64(loggerToSlogLevel(options.Level)))
 
 	return l
 }
 
 func (s *slogLogger) V(level logger.Level) bool {
-	return s.opts.Level.Enabled(level)
+	s.mu.Lock()
+	v := s.opts.Level.Enabled(level)
+	s.mu.Unlock()
+	return v
 }
 
 func (s *slogLogger) Level(level logger.Level) {
-	s.leveler.Set(loggerToSlogLevel(level))
+	s.mu.Lock()
+	s.opts.Level = level
+	s.handler.level.Store(int64(loggerToSlogLevel(level)))
+	s.mu.Unlock()
 }
 
 func (s *slogLogger) Options() logger.Options {
 	return s.opts
 }
 
-func (s *slogLogger) Fields(attrs ...interface{}) logger.Logger {
+func (s *slogLogger) Fields(fields ...interface{}) logger.Logger {
 	s.mu.RLock()
-	level := s.leveler.Level()
 	options := s.opts
 	s.mu.RUnlock()
 
 	l := &slogLogger{opts: options}
-	l.leveler = new(slog.LevelVar)
-	l.leveler.Set(level)
 
-	handleOpt := &slog.HandlerOptions{
-		ReplaceAttr: l.renameAttr,
-		Level:       l.leveler,
-		AddSource:   l.opts.AddSource,
+	if len(options.ContextAttrFuncs) == 0 {
+		options.ContextAttrFuncs = logger.DefaultContextAttrFuncs
 	}
 
-	l.handler = slog.New(slog.NewJSONHandler(l.opts.Out, handleOpt)).With(attrs...).Handler()
+	attrs, _ := s.argsAttrs(fields)
+	l.handler = &wrapper{h: s.handler.h.WithAttrs(attrs)}
+	l.handler.level.Store(int64(loggerToSlogLevel(l.opts.Level)))
 
 	return l
 }
@@ -134,22 +157,23 @@ func (s *slogLogger) Fields(attrs ...interface{}) logger.Logger {
 func (s *slogLogger) Init(opts ...logger.Option) error {
 	s.mu.Lock()
 
-	if len(s.opts.ContextAttrFuncs) == 0 {
-		s.opts.ContextAttrFuncs = logger.DefaultContextAttrFuncs
-	}
-
 	for _, o := range opts {
 		o(&s.opts)
 	}
 
-	s.leveler = new(slog.LevelVar)
+	if len(s.opts.ContextAttrFuncs) == 0 {
+		s.opts.ContextAttrFuncs = logger.DefaultContextAttrFuncs
+	}
+
 	handleOpt := &slog.HandlerOptions{
 		ReplaceAttr: s.renameAttr,
-		Level:       s.leveler,
+		Level:       loggerToSlogLevel(logger.TraceLevel),
 		AddSource:   s.opts.AddSource,
 	}
-	s.leveler.Set(loggerToSlogLevel(s.opts.Level))
-	s.handler = slog.New(slog.NewJSONHandler(s.opts.Out, handleOpt)).With(s.opts.Fields...).Handler()
+
+	attrs, _ := s.argsAttrs(s.opts.Fields)
+	s.handler = &wrapper{h: slog.NewJSONHandler(s.opts.Out, handleOpt).WithAttrs(attrs)}
+	s.handler.level.Store(int64(loggerToSlogLevel(s.opts.Level)))
 	s.mu.Unlock()
 
 	return nil
@@ -192,27 +216,34 @@ func (s *slogLogger) String() string {
 	return "slog"
 }
 
-func (s *slogLogger) printLog(ctx context.Context, lvl logger.Level, msg string, attrs ...interface{}) {
+func (s *slogLogger) printLog(ctx context.Context, lvl logger.Level, msg string, args ...interface{}) {
 	if !s.V(lvl) {
 		return
 	}
+	var argError error
 
 	s.opts.Meter.Counter(semconv.LoggerMessageTotal, "level", lvl.String()).Inc()
 
-	attrs = prepareAttributes(attrs)
+	attrs, err := s.argsAttrs(args)
+	if err != nil {
+		argError = err
+	}
+	if argError != nil {
+		if span, ok := tracer.SpanFromContext(ctx); ok {
+			span.SetStatus(tracer.SpanStatusError, argError.Error())
+		}
+	}
 
 	for _, fn := range s.opts.ContextAttrFuncs {
-		a := prepareAttributes(fn(ctx))
-		attrs = append(attrs, a...)
+		ctxAttrs, err := s.argsAttrs(fn(ctx))
+		if err != nil {
+			argError = err
 		}
-
-	for _, attr := range attrs {
-		if ve, hasErr := attr.(error); hasErr && ve != nil {
-			attrs = append(attrs, slog.String(s.opts.ErrorKey, ve.Error()))
+		attrs = append(attrs, ctxAttrs...)
+	}
+	if argError != nil {
 		if span, ok := tracer.SpanFromContext(ctx); ok {
-				span.SetStatus(tracer.SpanStatusError, ve.Error())
-			}
-			break
+			span.SetStatus(tracer.SpanStatusError, argError.Error())
 		}
 	}
 
@@ -229,7 +260,7 @@ func (s *slogLogger) printLog(ctx context.Context, lvl logger.Level, msg string,
 	var pcs [1]uintptr
 	runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, printLog, LogLvlMethod]
 	r := slog.NewRecord(s.opts.TimeFunc(), loggerToSlogLevel(lvl), msg, pcs[0])
-	r.Add(attrs...)
+	r.AddAttrs(attrs...)
 	_ = s.handler.Handle(ctx, r)
 }
 
@@ -276,11 +307,26 @@ func slogToLoggerLevel(level slog.Level) logger.Level {
 	}
 }
 
-func prepareAttributes(attrs []interface{}) []interface{} {
-	if len(attrs)%2 == 1 {
-		attrs = append(attrs, badKey)
-		attrs[len(attrs)-1], attrs[len(attrs)-2] = attrs[len(attrs)-2], attrs[len(attrs)-1]
+func (s *slogLogger) argsAttrs(args []interface{}) ([]slog.Attr, error) {
+	attrs := make([]slog.Attr, 0, len(args))
+	var err error
+
+	for idx := 0; idx < len(args); idx++ {
+		switch arg := args[idx].(type) {
+		case slog.Attr:
+			attrs = append(attrs, arg)
+		case string:
+			if idx+1 < len(args) {
+				attrs = append(attrs, slog.Any(arg, args[idx+1]))
+				idx += 1
+			} else {
+				attrs = append(attrs, slog.String(badKey, arg))
+			}
+		case error:
+			attrs = append(attrs, slog.String(s.opts.ErrorKey, arg.Error()))
+			err = arg
+		}
 	}
 
-	return attrs
+	return attrs, err
 }

logger/slog/slog_test.go

@@ -5,15 +5,66 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/google/uuid"
-	"go.unistack.org/micro/v3/metadata"
 	"log"
 	"strings"
 	"testing"
 
+	"github.com/google/uuid"
+	"go.unistack.org/micro/v3/metadata"
+
 	"go.unistack.org/micro/v3/logger"
 )
 
+func TestMultipleFieldsWithLevel(t *testing.T) {
+	ctx := context.TODO()
+	buf := bytes.NewBuffer(nil)
+	l := NewLogger(logger.WithLevel(logger.InfoLevel), logger.WithOutput(buf))
+	if err := l.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	l = l.Fields("key", "val")
+
+	l.Info(ctx, "msg1")
+	nl := l.Clone(logger.WithLevel(logger.DebugLevel))
+	nl.Debug(ctx, "msg2")
+	l.Debug(ctx, "msg3")
+	if !bytes.Contains(buf.Bytes(), []byte(`"key":"val"`)) {
+		t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"msg1"`)) {
+		t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"msg2"`)) {
+		t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
+	}
+	if bytes.Contains(buf.Bytes(), []byte(`"msg3"`)) {
+		t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
+	}
+}
+
+func TestMultipleFields(t *testing.T) {
+	ctx := context.TODO()
+	buf := bytes.NewBuffer(nil)
+	l := NewLogger(logger.WithLevel(logger.InfoLevel), logger.WithOutput(buf))
+	if err := l.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	l = l.Fields("key", "val")
+
+	l = l.Fields("key1", "val1")
+
+	l.Info(ctx, "msg")
+
+	if !bytes.Contains(buf.Bytes(), []byte(`"key":"val"`)) {
+		t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"key1":"val1"`)) {
+		t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
+	}
+}
+
 func TestError(t *testing.T) {
 	ctx := context.TODO()
 	buf := bytes.NewBuffer(nil)
@@ -43,9 +94,6 @@ func TestErrorf(t *testing.T) {
 	}
 
 	l.Log(ctx, logger.ErrorLevel, "message", errors.New("error msg"))
-	if !bytes.Contains(buf.Bytes(), []byte(`"!BADKEY":"`)) {
-		t.Fatalf("logger BADKEY not works, buf contains: %s", buf.Bytes())
-	}
 
 	l.Log(ctx, logger.ErrorLevel, "", errors.New("error msg"))
 	if !bytes.Contains(buf.Bytes(), []byte(`"error":"error msg"`)) {
@@ -236,5 +284,11 @@ func Test_WithContextAttrFunc(t *testing.T) {
 	if !(bytes.Contains(buf.Bytes(), []byte(`"source-service":"Test-System"`))) {
 		t.Fatalf("logger info, buf %s", buf.Bytes())
 	}
+	buf.Reset()
+	imd, _ := metadata.FromIncomingContext(ctx)
+	l.Info(ctx, "test message1")
+	imd.Set("Source-Service", "Test-System2")
+	l.Info(ctx, "test message2")
 
+	// t.Logf("xxx %s", buf.Bytes())
 }

service.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 
 	"github.com/KimMachineGun/automemlimit/memlimit"
-	_ "go.uber.org/automaxprocs"
+	"go.uber.org/automaxprocs/maxprocs"
 	"go.unistack.org/micro/v3/broker"
 	"go.unistack.org/micro/v3/client"
 	"go.unistack.org/micro/v3/config"
@@ -20,6 +20,7 @@ import (
 )
 
 func init() {
+	maxprocs.Set()
 	memlimit.SetGoMemLimitWithOpts(
 		memlimit.WithRatio(0.9),
 		memlimit.WithProvider(

store/memory/memory.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	cache "github.com/patrickmn/go-cache"
@@ -20,8 +21,11 @@ func NewStore(opts ...store.Option) store.Store {
 }
 
 func (m *memoryStore) Connect(ctx context.Context) error {
+	if m.opts.LazyConnect {
 		return nil
 	}
+	return m.connect(ctx)
+}
 
 func (m *memoryStore) Disconnect(ctx context.Context) error {
 	m.store.Flush()
@@ -36,6 +40,7 @@ type memoryStore struct {
 	funcDelete  store.FuncDelete
 	store       *cache.Cache
 	opts        store.Options
+	isConnected atomic.Int32
 }
 
 func (m *memoryStore) key(prefix, key string) string {
@@ -145,6 +150,11 @@ func (m *memoryStore) Name() string {
 }
 
 func (m *memoryStore) Exists(ctx context.Context, key string, opts ...store.ExistsOption) error {
+	if m.opts.LazyConnect {
+		if err := m.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return m.funcExists(ctx, key, opts...)
 }
 
@@ -157,6 +167,11 @@ func (m *memoryStore) fnExists(ctx context.Context, key string, opts ...store.Ex
 }
 
 func (m *memoryStore) Read(ctx context.Context, key string, val interface{}, opts ...store.ReadOption) error {
+	if m.opts.LazyConnect {
+		if err := m.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return m.funcRead(ctx, key, val, opts...)
 }
 
@@ -169,6 +184,11 @@ func (m *memoryStore) fnRead(ctx context.Context, key string, val interface{}, o
 }
 
 func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, opts ...store.WriteOption) error {
+	if m.opts.LazyConnect {
+		if err := m.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return m.funcWrite(ctx, key, val, opts...)
 }
 
@@ -193,6 +213,11 @@ func (m *memoryStore) fnWrite(ctx context.Context, key string, val interface{},
 }
 
 func (m *memoryStore) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error {
+	if m.opts.LazyConnect {
+		if err := m.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return m.funcDelete(ctx, key, opts...)
 }
 
@@ -211,6 +236,11 @@ func (m *memoryStore) Options() store.Options {
 }
 
 func (m *memoryStore) List(ctx context.Context, opts ...store.ListOption) ([]string, error) {
+	if m.opts.LazyConnect {
+		if err := m.connect(ctx); err != nil {
+			return nil, err
+		}
+	}
 	return m.funcList(ctx, opts...)
 }
 
@@ -244,3 +274,8 @@ func (m *memoryStore) fnList(ctx context.Context, opts ...store.ListOption) ([]s
 
 	return keys, nil
 }
+
+func (m *memoryStore) connect(ctx context.Context) error {
+	m.isConnected.CompareAndSwap(0, 1)
+	return nil
+}

store/noop.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"context"
+	"sync/atomic"
 
 	"go.unistack.org/micro/v3/options"
 )
@@ -15,6 +16,7 @@ type noopStore struct {
 	funcList    FuncList
 	funcDelete  FuncDelete
 	opts        Options
+	isConnected atomic.Int32
 }
 
 func NewStore(opts ...Option) *noopStore {
@@ -52,13 +54,11 @@ func (n *noopStore) Init(opts ...Option) error {
 }
 
 func (n *noopStore) Connect(ctx context.Context) error {
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	default:
-	}
+	if n.opts.LazyConnect {
 		return nil
 	}
+	return n.connect(ctx)
+}
 
 func (n *noopStore) Disconnect(ctx context.Context) error {
 	select {
@@ -70,6 +70,11 @@ func (n *noopStore) Disconnect(ctx context.Context) error {
 }
 
 func (n *noopStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
+	if n.opts.LazyConnect {
+		if err := n.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return n.funcRead(ctx, key, val, opts...)
 }
 
@@ -83,6 +88,11 @@ func (n *noopStore) fnRead(ctx context.Context, key string, val interface{}, opt
 }
 
 func (n *noopStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
+	if n.opts.LazyConnect {
+		if err := n.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return n.funcDelete(ctx, key, opts...)
 }
 
@@ -96,6 +106,11 @@ func (n *noopStore) fnDelete(ctx context.Context, key string, opts ...DeleteOpti
 }
 
 func (n *noopStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
+	if n.opts.LazyConnect {
+		if err := n.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return n.funcExists(ctx, key, opts...)
 }
 
@@ -109,6 +124,11 @@ func (n *noopStore) fnExists(ctx context.Context, key string, opts ...ExistsOpti
 }
 
 func (n *noopStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
+	if n.opts.LazyConnect {
+		if err := n.connect(ctx); err != nil {
+			return err
+		}
+	}
 	return n.funcWrite(ctx, key, val, opts...)
 }
 
@@ -122,6 +142,11 @@ func (n *noopStore) fnWrite(ctx context.Context, key string, val interface{}, op
 }
 
 func (n *noopStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
+	if n.opts.LazyConnect {
+		if err := n.connect(ctx); err != nil {
+			return nil, err
+		}
+	}
 	return n.funcList(ctx, opts...)
 }
 
@@ -145,3 +170,15 @@ func (n *noopStore) String() string {
 func (n *noopStore) Options() Options {
 	return n.opts
 }
+
+func (n *noopStore) connect(ctx context.Context) error {
+	if n.isConnected.CompareAndSwap(0, 1) {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+	}
+
+	return nil
+}

store/options.go

@@ -41,6 +41,8 @@ type Options struct {
 	Timeout time.Duration
 	// Hooks can be run before/after store Read/List/Write/Exists/Delete
 	Hooks options.Hooks
+	// LazyConnect creates a connection when using store
+	LazyConnect bool
 }
 
 // NewOptions creates options struct

util/time/duration.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"strconv"
 	"time"
+
+	"gopkg.in/yaml.v3"
 )
 
 type Duration int64
@@ -53,6 +55,31 @@ loop:
 	return time.ParseDuration(fmt.Sprintf("%dh%s", hours, s[p:]))
 }
 
+func (d Duration) MarshalYAML() (interface{}, error) {
+	return time.Duration(d).String(), nil
+}
+
+func (d *Duration) UnmarshalYAML(n *yaml.Node) error {
+	var v interface{}
+	if err := yaml.Unmarshal([]byte(n.Value), &v); err != nil {
+		return err
+	}
+	switch value := v.(type) {
+	case float64:
+		*d = Duration(time.Duration(value))
+		return nil
+	case string:
+		dv, err := ParseDuration(value)
+		if err != nil {
+			return err
+		}
+		*d = Duration(dv)
+		return nil
+	default:
+		return fmt.Errorf("invalid duration")
+	}
+}
+
 func (d Duration) MarshalJSON() ([]byte, error) {
 	return json.Marshal(time.Duration(d).String())
 }

util/time/duration_test.go

@@ -5,8 +5,44 @@ import (
 	"encoding/json"
 	"testing"
 	"time"
+
+	"gopkg.in/yaml.v3"
 )
 
+func TestMarshalYAML(t *testing.T) {
+	d := Duration(10000000)
+	buf, err := yaml.Marshal(d)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(buf, []byte(`10ms
+`)) {
+		t.Fatalf("invalid duration: %s != %s", buf, `10ms`)
+	}
+}
+
+func TestUnmarshalYAML(t *testing.T) {
+	type str struct {
+		TTL Duration `yaml:"ttl"`
+	}
+	v := &str{}
+	var err error
+
+	err = yaml.Unmarshal([]byte(`{"ttl":"10ms"}`), v)
+	if err != nil {
+		t.Fatal(err)
+	} else if v.TTL != 10000000 {
+		t.Fatalf("invalid duration %v != 10000000", v.TTL)
+	}
+
+	err = yaml.Unmarshal([]byte(`{"ttl":"1y"}`), v)
+	if err != nil {
+		t.Fatal(err)
+	} else if v.TTL != 31622400000000000 {
+		t.Fatalf("invalid duration %v != 31622400000000000", v.TTL)
+	}
+}
+
 func TestMarshalJSON(t *testing.T) {
 	d := Duration(10000000)
 	buf, err := json.Marshal(d)