Merge pull request #147 from ydb-platform/repeater-trace

refactoring of repeater package + add trace.Driver.OnRepeaterWakeUp e…

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 3.11.13
+* Added `trace.Driver.OnRepeaterWakeUp` event
+* Refactored package `repeater`
+
 ## 3.11.12
 * Added `trace.ClusterInsertDoneInfo.Inserted` boolean flag for notify about success of insert endpoint into balancer
 * Added `trace.ClusterRemoveDoneInfo.Removed` boolean flag for notify about success of remove endpoint from balancer

internal/db/database.go

@@ -81,6 +81,7 @@ func New(
 		ctx,
 		db.cluster.GetConn(endpoint.New(c.Endpoint(), endpoint.WithLocalDC(true))),
 		db.cluster,
+		db.config.Trace(),
 		opts...,
 	)
 	if err != nil {

internal/discovery/discovery.go

@@ -24,6 +24,7 @@ func New(
 	ctx context.Context,
 	cc conn.Conn,
 	crudExplorer cluster.CRUDExplorerLocker,
+	driverTrace trace.Driver,
 	opts ...config.Option,
 ) (_ discovery.Client, err error) {
 	c := &client{
@@ -57,9 +58,8 @@ func New(
 	}
 
 	crudExplorer.SetExplorer(
-		repeater.NewRepeater(
+		repeater.New(
 			deadline.ContextWithoutDeadline(ctx),
-			c.config.Interval(),
 			func(ctx context.Context) (err error) {
 				next, err = c.Discover(ctx)
 				if err != nil {
@@ -107,6 +107,9 @@ func New(
 
 				return nil
 			},
+			repeater.WithInterval(c.config.Interval()),
+			repeater.WithName("discovery"),
+			repeater.WithTrace(driverTrace),
 		),
 	)
 

internal/meta/version.go

@@ -1,5 +1,5 @@
 package meta
 
 const (
-	Version = "ydb-go-sdk/3.11.12"
+	Version = "ydb-go-sdk/3.11.13"
 )

internal/repeater/repeat.go

@@ -2,10 +2,9 @@ package repeater
 
 import (
 	"context"
-	"sync"
 	"time"
 
-	"github.com/ydb-platform/ydb-go-sdk/v3/testutil/timeutil"
+	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 type Repeater interface {
@@ -19,50 +18,73 @@ type repeater struct {
 	// Interval must be greater than zero; if not, Repeater will panic.
 	interval time.Duration
 
+	name  string
+	trace trace.Driver
+
 	// Task is a function that must be executed periodically.
 	task func(context.Context) error
 
-	timer    timeutil.Timer
-	stopOnce sync.Once
-	stop     chan struct{}
-	done     chan struct{}
-	ctx      context.Context
-	cancel   context.CancelFunc
-	force    chan struct{}
+	stop  chan struct{}
+	done  chan struct{}
+	force chan struct{}
+}
+
+type option func(r *repeater)
+
+func WithName(name string) option {
+	return func(r *repeater) {
+		r.name = name
+	}
+}
+
+func WithTrace(trace trace.Driver) option {
+	return func(r *repeater) {
+		r.trace = trace
+	}
+}
+
+func WithInterval(interval time.Duration) option {
+	return func(r *repeater) {
+		r.interval = interval
+	}
 }
 
-// NewRepeater creates and begins to execute task periodically.
-func NewRepeater(
+type event string
+
+const (
+	eventTick  = event("tick")
+	eventForce = event("force")
+)
+
+// New creates and begins to execute task periodically.
+func New(
 	ctx context.Context,
-	interval time.Duration,
 	task func(ctx context.Context) (err error),
+	opts ...option,
 ) Repeater {
-	if interval <= 0 {
-		return nil
-	}
-	ctx, cancel := context.WithCancel(ctx)
 	r := &repeater{
-		interval: interval,
-		task:     task,
-		timer:    timeutil.NewTimer(interval),
-		stopOnce: sync.Once{},
-		stop:     make(chan struct{}),
-		done:     make(chan struct{}),
-		ctx:      ctx,
-		cancel:   cancel,
-		force:    make(chan struct{}),
+		task:  task,
+		stop:  make(chan struct{}),
+		done:  make(chan struct{}),
+		force: make(chan struct{}),
+	}
+
+	for _, o := range opts {
+		o(r)
+	}
+
+	if r.interval <= 0 {
+		return nil
 	}
-	go r.worker()
+
+	go r.worker(ctx, r.interval)
+
 	return r
 }
 
 // Stop stops to execute its task.
 func (r *repeater) Stop() {
-	r.stopOnce.Do(func() {
-		close(r.stop)
-		r.cancel()
-		<-r.done
-	})
+	close(r.stop)
 }
 
 func (r *repeater) Force() {
@@ -72,30 +94,32 @@ func (r *repeater) Force() {
 	}
 }
 
-func (r *repeater) singleTask() {
-	if err := r.task(r.ctx); err != nil {
-		r.timer.Reset(time.Second)
-	} else {
-		r.timer.Reset(r.interval)
-	}
+func (r *repeater) wakeUp(ctx context.Context, e event) {
+	var cancel context.CancelFunc
+	ctx, cancel = context.WithCancel(ctx)
+	defer cancel()
+
+	trace.DriverOnRepeaterWakeUp(
+		r.trace,
+		&ctx,
+		r.name,
+		string(e),
+	)(r.task(ctx))
 }
 
-func (r *repeater) worker() {
-	defer func() {
-		close(r.done)
-	}()
-	r.singleTask()
+func (r *repeater) worker(ctx context.Context, interval time.Duration) {
+	defer close(r.done)
+
 	for {
 		select {
+		case <-ctx.Done():
+			return
 		case <-r.stop:
 			return
-		case <-r.timer.C():
-			r.singleTask()
+		case <-time.After(interval):
+			r.wakeUp(ctx, eventTick)
 		case <-r.force:
-			if !r.timer.Stop() {
-				<-r.timer.C()
-			}
-			r.timer.Reset(r.interval)
+			r.wakeUp(ctx, eventForce)
 		}
 	}
 }

internal/repeater/test/repeat_test.go

@@ -300,6 +300,31 @@ func Driver(log Logger, details trace.Details) (t trace.Driver) {
 				}
 			}
 		}
+		t.OnRepeaterWakeUp = func(info trace.RepeaterTickStartInfo) func(trace.RepeaterTickDoneInfo) {
+			name := info.Name
+			event := info.Event
+			log.Tracef(`repeater wake up {name:"%s",event:"%s"}`,
+				name,
+				event,
+			)
+			start := time.Now()
+			return func(info trace.RepeaterTickDoneInfo) {
+				if info.Error == nil {
+					log.Tracef(`repeater wake up done {name:"%s",event:"%s",latency:"%v"}`,
+						name,
+						event,
+						time.Since(start),
+					)
+				} else {
+					log.Errorf(`repeater wake up fail {name:"%s",event:"%s",latency:"%v",error:"%v"}`,
+						name,
+						event,
+						time.Since(start),
+						info.Error,
+					)
+				}
+			}
+		}
 	}
 	if details&trace.DriverClusterEvents != 0 {
 		// nolint:govet