Merge pull request 'tracer: append labels' (#326) from tracerfix into v3

Reviewed-on: #326

logger/slog/slog.go

@@ -61,8 +61,8 @@ func (s *slogLogger) renameAttr(_ []string, a slog.Attr) slog.Attr {
 }
 
 type slogLogger struct {
-	slog    *slog.Logger
 	leveler *slog.LevelVar
+	handler slog.Handler
 	opts    logger.Options
 	mu      sync.RWMutex
 }
@@ -70,6 +70,7 @@ type slogLogger struct {
 func (s *slogLogger) Clone(opts ...logger.Option) logger.Logger {
 	s.mu.RLock()
 	options := s.opts
+	s.mu.RUnlock()
 
 	for _, o := range opts {
 		o(&options)
@@ -86,10 +87,7 @@ func (s *slogLogger) Clone(opts ...logger.Option) logger.Logger {
 		AddSource:   l.opts.AddSource,
 	}
 	l.leveler.Set(loggerToSlogLevel(l.opts.Level))
-	handler := slog.NewJSONHandler(options.Out, handleOpt)
-	l.slog = slog.New(handler).With(options.Fields...)
-
-	s.mu.RUnlock()
+	l.handler = slog.New(slog.NewJSONHandler(options.Out, handleOpt)).With(options.Fields...).Handler()
 
 	return l
 }
@@ -108,9 +106,13 @@ func (s *slogLogger) Options() logger.Options {
 
 func (s *slogLogger) Fields(attrs ...interface{}) logger.Logger {
 	s.mu.RLock()
-	l := &slogLogger{opts: s.opts}
+	level := s.leveler.Level()
+	options := s.opts
+	s.mu.RUnlock()
+
+	l := &slogLogger{opts: options}
 	l.leveler = new(slog.LevelVar)
-	l.leveler.Set(s.leveler.Level())
+	l.leveler.Set(level)
 
 	handleOpt := &slog.HandlerOptions{
 		ReplaceAttr: l.renameAttr,
@@ -118,10 +120,7 @@ func (s *slogLogger) Fields(attrs ...interface{}) logger.Logger {
 		AddSource:   l.opts.AddSource,
 	}
 
-	handler := slog.NewJSONHandler(s.opts.Out, handleOpt)
-	l.slog = slog.New(handler).With(attrs...)
-
-	s.mu.RUnlock()
+	l.handler = slog.New(slog.NewJSONHandler(l.opts.Out, handleOpt)).With(attrs...).Handler()
 
 	return l
 }
@@ -144,9 +143,7 @@ func (s *slogLogger) Init(opts ...logger.Option) error {
 		AddSource:   s.opts.AddSource,
 	}
 	s.leveler.Set(loggerToSlogLevel(s.opts.Level))
-	handler := slog.NewJSONHandler(s.opts.Out, handleOpt)
-	s.slog = slog.New(handler).With(s.opts.Fields...)
-
+	s.handler = slog.New(slog.NewJSONHandler(s.opts.Out, handleOpt)).With(s.opts.Fields...).Handler()
 	s.mu.Unlock()
 
 	return nil
@@ -188,7 +185,7 @@ func (s *slogLogger) Log(ctx context.Context, lvl logger.Level, attrs ...interfa
 		}
 		return true
 	})
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Logf(ctx context.Context, lvl logger.Level, msg string, attrs ...interface{}) {
@@ -227,7 +224,7 @@ func (s *slogLogger) Logf(ctx context.Context, lvl logger.Level, msg string, att
 		}
 		return true
 	})
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Info(ctx context.Context, attrs ...interface{}) {
@@ -248,7 +245,7 @@ func (s *slogLogger) Info(ctx context.Context, attrs ...interface{}) {
 		}
 	}
 	r.Add(attrs[1:]...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Infof(ctx context.Context, msg string, attrs ...interface{}) {
@@ -269,7 +266,7 @@ func (s *slogLogger) Infof(ctx context.Context, msg string, attrs ...interface{}
 		}
 	}
 	r.Add(attrs...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Debug(ctx context.Context, attrs ...interface{}) {
@@ -290,7 +287,7 @@ func (s *slogLogger) Debug(ctx context.Context, attrs ...interface{}) {
 		}
 	}
 	r.Add(attrs[1:]...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Debugf(ctx context.Context, msg string, attrs ...interface{}) {
@@ -311,7 +308,7 @@ func (s *slogLogger) Debugf(ctx context.Context, msg string, attrs ...interface{
 		}
 	}
 	r.Add(attrs...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Trace(ctx context.Context, attrs ...interface{}) {
@@ -332,7 +329,7 @@ func (s *slogLogger) Trace(ctx context.Context, attrs ...interface{}) {
 		}
 	}
 	r.Add(attrs[1:]...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Tracef(ctx context.Context, msg string, attrs ...interface{}) {
@@ -353,7 +350,7 @@ func (s *slogLogger) Tracef(ctx context.Context, msg string, attrs ...interface{
 		}
 	}
 	r.Add(attrs[1:]...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Error(ctx context.Context, attrs ...interface{}) {
@@ -392,7 +389,7 @@ func (s *slogLogger) Error(ctx context.Context, attrs ...interface{}) {
 		}
 		return true
 	})
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Errorf(ctx context.Context, msg string, attrs ...interface{}) {
@@ -431,7 +428,7 @@ func (s *slogLogger) Errorf(ctx context.Context, msg string, attrs ...interface{
 		}
 		return true
 	})
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Fatal(ctx context.Context, attrs ...interface{}) {
@@ -452,7 +449,7 @@ func (s *slogLogger) Fatal(ctx context.Context, attrs ...interface{}) {
 		}
 	}
 	r.Add(attrs[1:]...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 	os.Exit(1)
 }
 
@@ -474,7 +471,7 @@ func (s *slogLogger) Fatalf(ctx context.Context, msg string, attrs ...interface{
 		}
 	}
 	r.Add(attrs...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 	os.Exit(1)
 }
 
@@ -496,7 +493,7 @@ func (s *slogLogger) Warn(ctx context.Context, attrs ...interface{}) {
 		}
 	}
 	r.Add(attrs[1:]...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Warnf(ctx context.Context, msg string, attrs ...interface{}) {
@@ -517,7 +514,7 @@ func (s *slogLogger) Warnf(ctx context.Context, msg string, attrs ...interface{}
 		}
 	}
 	r.Add(attrs[1:]...)
-	_ = s.slog.Handler().Handle(ctx, r)
+	_ = s.handler.Handle(ctx, r)
 }
 
 func (s *slogLogger) Name() string {

metadata/metadata.go

@@ -98,11 +98,12 @@ func (md Metadata) Del(keys ...string) {
 }
 
 // Copy makes a copy of the metadata
-func Copy(md Metadata) Metadata {
+func Copy(md Metadata, exclude ...string) Metadata {
 	nmd := New(len(md))
 	for key, val := range md {
 		nmd.Set(key, val)
 	}
+	nmd.Del(exclude...)
 	return nmd
 }
 

metadata/metadata_test.go

@@ -190,3 +190,14 @@ func TestMetadataContext(t *testing.T) {
 		t.Errorf("Expected metadata length 1 got %d", i)
 	}
 }
+
+func TestCopy(t *testing.T) {
+	md := New(2)
+	md.Set("key1", "val1", "key2", "val2")
+	nmd := Copy(md, "key2")
+	if len(nmd) != 1 {
+		t.Fatal("Copy exclude not works")
+	} else if nmd["Key1"] != "val1" {
+		t.Fatal("Copy exclude not works")
+	}
+}

server/options.go

@@ -15,6 +15,7 @@ import (
 	"go.unistack.org/micro/v3/network/transport"
 	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/register"
+	msync "go.unistack.org/micro/v3/sync"
 	"go.unistack.org/micro/v3/tracer"
 	"go.unistack.org/micro/v3/util/id"
 )
@@ -47,7 +48,7 @@ type Options struct {
 	// Listener may be passed if already created
 	Listener net.Listener
 	// Wait group
-	Wait *sync.WaitGroup
+	Wait *msync.WaitGroup
 	// TLSConfig specifies tls.Config for secure serving
 	TLSConfig *tls.Config
 	// Metadata holds the server metadata
@@ -282,7 +283,7 @@ func Wait(wg *sync.WaitGroup) Option {
 		if wg == nil {
 			wg = new(sync.WaitGroup)
 		}
-		o.Wait = wg
+		o.Wait = msync.WrapWaitGroup(wg)
 	}
 }
 
@@ -331,7 +332,6 @@ func GracefulTimeout(td time.Duration) Option {
 	}
 }
 
-
 // HandlerOptions struct
 type HandlerOptions struct {
 	// Context holds external options

sync/waitgroup.go

@@ -0,0 +1,69 @@
+package sync
+
+import (
+	"context"
+	"sync"
+)
+
+type WaitGroup struct {
+	wg *sync.WaitGroup
+	c  int
+	mu sync.Mutex
+}
+
+func WrapWaitGroup(wg *sync.WaitGroup) *WaitGroup {
+	g := &WaitGroup{
+		wg: wg,
+	}
+	return g
+}
+
+func NewWaitGroup() *WaitGroup {
+	var wg sync.WaitGroup
+	return WrapWaitGroup(&wg)
+}
+
+func (g *WaitGroup) Add(n int) {
+	g.mu.Lock()
+	g.c += n
+	g.wg.Add(n)
+	g.mu.Unlock()
+}
+
+func (g *WaitGroup) Done() {
+	g.mu.Lock()
+	g.c += -1
+	g.wg.Add(-1)
+	g.mu.Unlock()
+}
+
+func (g *WaitGroup) Wait() {
+	g.wg.Wait()
+}
+
+func (g *WaitGroup) WaitContext(ctx context.Context) {
+	done := make(chan struct{})
+	go func() {
+		g.wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-ctx.Done():
+		g.mu.Lock()
+		g.wg.Add(-g.c)
+		<-done
+		g.wg.Add(g.c)
+		g.mu.Unlock()
+		return
+	case <-done:
+		return
+	}
+}
+
+func (g *WaitGroup) Waiters() int {
+	g.mu.Lock()
+	c := g.c
+	g.mu.Unlock()
+	return c
+}

sync/waitgroup_test.go

@@ -0,0 +1,37 @@
+package sync
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestWaitGroupContext(t *testing.T) {
+	wg := NewWaitGroup()
+	_ = t
+	wg.Add(1)
+	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
+	defer cancel()
+	wg.WaitContext(ctx)
+}
+
+func TestWaitGroupReuse(t *testing.T) {
+	wg := NewWaitGroup()
+	defer func() {
+		if wg.Waiters() != 0 {
+			t.Fatal("lost goroutines")
+		}
+	}()
+
+	wg.Add(1)
+	defer wg.Done()
+	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
+	defer cancel()
+	wg.WaitContext(ctx)
+
+	wg.Add(1)
+	defer wg.Done()
+	ctx, cancel = context.WithTimeout(context.TODO(), 1*time.Second)
+	defer cancel()
+	wg.WaitContext(ctx)
+}

tracer/options.go

@@ -100,13 +100,13 @@ type EventOption func(o *EventOptions)
 
 func WithEventLabels(kv ...interface{}) EventOption {
 	return func(o *EventOptions) {
-		o.Labels = kv
+		o.Labels = append(o.Labels, kv...)
 	}
 }
 
 func WithSpanLabels(kv ...interface{}) SpanOption {
 	return func(o *SpanOptions) {
-		o.Labels = kv
+		o.Labels = append(o.Labels, kv...)
 	}
 }
 
@@ -159,7 +159,8 @@ func NewSpanOptions(opts ...SpanOption) SpanOptions {
 // NewOptions returns default options
 func NewOptions(opts ...Option) Options {
 	options := Options{
-		Logger: logger.DefaultLogger,
+		Logger:  logger.DefaultLogger,
+		Context: context.Background(),
 	}
 	for _, o := range opts {
 		o(&options)