* Returned safe-thread copy of endpoint.Endpoint to trace callbacks

* Checked cluster close/empty on keeper goroutine

Author: asmyasnikov

CHANGELOG.md

@@ -1,9 +1,11 @@
 ## 3.11.8
 * Added `trace.EndpointInfo.LastUpdated()` timestamp
 * Refactored `endpoint.Endpoint` (split to struct `endopint` and interface `Endpoint`)
+* Returned safe-thread copy of `endpoint.Endpoint` to trace callbacks 
 * Added `endpoint.Endpoint.Touch()` func for refresh endpoint info
 * Added `conn.conn.onClose` slice for call optional funcs on close step
 * Added removing `conn.Conn` from `conn.Pool` on `conn.Conn.Close()` call
+* Checked cluster close/empty on keeper goroutine
 
 ## 3.11.7
 * Removed internal alias-type `errors.IssuesIterator`

internal/cluster/cluster.go

@@ -209,7 +209,7 @@ func (c *cluster) Get(ctx context.Context, opts ...crudOption) (cc conn.Conn, er
 		if err != nil {
 			onDone(nil, err)
 		} else {
-			onDone(cc.Endpoint(), nil)
+			onDone(cc.Endpoint().Copy(), nil)
 		}
 	}()
 
@@ -234,7 +234,7 @@ func (c *cluster) Get(ctx context.Context, opts ...crudOption) (cc conn.Conn, er
 
 // Insert inserts new connection into the cluster.
 func (c *cluster) Insert(ctx context.Context, e endpoint.Endpoint, opts ...crudOption) (cc conn.Conn) {
-	onDone := trace.DriverOnClusterInsert(c.config.Trace(), &ctx, e)
+	onDone := trace.DriverOnClusterInsert(c.config.Trace(), &ctx, e.Copy())
 	defer func() {
 		if cc != nil {
 			onDone(cc.GetState())
@@ -277,7 +277,7 @@ func (c *cluster) Insert(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 
 // Update updates existing connection's runtime stats such that load factor and others.
 func (c *cluster) Update(ctx context.Context, e endpoint.Endpoint, opts ...crudOption) (cc conn.Conn) {
-	onDone := trace.DriverOnClusterUpdate(c.config.Trace(), &ctx, e)
+	onDone := trace.DriverOnClusterUpdate(c.config.Trace(), &ctx, e.Copy())
 	defer func() {
 		if cc != nil {
 			onDone(cc.GetState())
@@ -323,7 +323,7 @@ func (c *cluster) Update(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 
 // Remove removes and closes previously inserted connection.
 func (c *cluster) Remove(ctx context.Context, e endpoint.Endpoint, opts ...crudOption) (cc conn.Conn) {
-	onDone := trace.DriverOnClusterRemove(c.config.Trace(), &ctx, e)
+	onDone := trace.DriverOnClusterRemove(c.config.Trace(), &ctx, e.Copy())
 	defer func() {
 		if cc != nil {
 			onDone(cc.GetState())

internal/conn/conn.go

@@ -93,7 +93,7 @@ func (c *conn) setState(ctx context.Context, s State) State {
 	onDone := trace.DriverOnConnStateChange(
 		trace.ContextDriver(ctx).Compose(c.config.Trace()),
 		&ctx,
-		c.endpoint,
+		c.endpoint.Copy(),
 		c.state,
 	)
 	c.state = s
@@ -124,7 +124,7 @@ func (c *conn) take(ctx context.Context) (cc *grpc.ClientConn, err error) {
 	onDone := trace.DriverOnConnTake(
 		trace.ContextDriver(ctx).Compose(c.config.Trace()),
 		&ctx,
-		c.endpoint,
+		c.endpoint.Copy(),
 	)
 	defer func() {
 		onDone(int(atomic.LoadInt32(&c.locks)), err)
@@ -161,7 +161,7 @@ func (c *conn) release(ctx context.Context) {
 	onDone := trace.DriverOnConnRelease(
 		trace.ContextDriver(ctx).Compose(c.config.Trace()),
 		&ctx,
-		c.endpoint,
+		c.endpoint.Copy(),
 	)
 	atomic.AddInt32(&c.locks, -1)
 	onDone(int(atomic.LoadInt32(&c.locks)))
@@ -214,7 +214,7 @@ func (c *conn) pessimize(ctx context.Context, err error) {
 	trace.DriverOnPessimizeNode(
 		trace.ContextDriver(ctx).Compose(c.config.Trace()),
 		&ctx,
-		c.endpoint,
+		c.endpoint.Copy(),
 		c.GetState(),
 		err,
 	)(c.SetState(ctx, Banned))
@@ -253,7 +253,7 @@ func (c *conn) Invoke(
 	onDone := trace.DriverOnConnInvoke(
 		trace.ContextDriver(ctx).Compose(c.config.Trace()),
 		&ctx,
-		c.endpoint,
+		c.endpoint.Copy(),
 		trace.Method(method),
 	)
 	defer func() {
@@ -320,7 +320,7 @@ func (c *conn) NewStream(
 	streamRecv := trace.DriverOnConnNewStream(
 		trace.ContextDriver(ctx).Compose(c.config.Trace()),
 		&ctx,
-		c.endpoint,
+		c.endpoint.Copy(),
 		trace.Method(method),
 	)
 	defer func() {

internal/discovery/discovery.go

@@ -125,7 +125,7 @@ func (d *client) Discover(ctx context.Context) (endpoints []endpoint.Endpoint, e
 	defer func() {
 		nodes := make([]trace.EndpointInfo, 0, len(endpoints))
 		for _, e := range endpoints {
-			nodes = append(nodes, e)
+			nodes = append(nodes, e.Copy())
 		}
 		onDone(location, nodes, err)
 	}()

internal/endpoint/endpoint.go

@@ -2,6 +2,7 @@ package endpoint
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint/info"
@@ -10,6 +11,7 @@ import (
 type Endpoint interface {
 	Info() info.Info
 	String() string
+	Copy() Endpoint
 
 	NodeID() uint32
 	Address() string
@@ -22,6 +24,7 @@ type Endpoint interface {
 }
 
 type endpoint struct {
+	mu       sync.RWMutex
 	id       uint32
 	address  string
 	location string
@@ -33,7 +36,23 @@ type endpoint struct {
 	lastUpdated time.Time
 }
 
+func (e *endpoint) Copy() Endpoint {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+	return &endpoint{
+		id:          e.id,
+		address:     e.address,
+		location:    e.location,
+		services:    append(make([]string, 0, len(e.services)), e.services...),
+		loadFactor:  e.loadFactor,
+		local:       e.local,
+		lastUpdated: e.lastUpdated,
+	}
+}
+
 func (e *endpoint) String() string {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return fmt.Sprintf(`{id:%d,address:"%s",local:%t,location:"%s",loadFactor:%f,lastUpdated:"%s"}`,
 		e.id,
 		e.address,
@@ -45,6 +64,8 @@ func (e *endpoint) String() string {
 }
 
 func (e *endpoint) Info() info.Info {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return info.Info{
 		ID:         e.id,
 		Address:    e.address,
@@ -54,30 +75,44 @@ func (e *endpoint) Info() info.Info {
 }
 
 func (e *endpoint) NodeID() uint32 {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.id
 }
 
 func (e *endpoint) Address() (address string) {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.address
 }
 
 func (e *endpoint) Location() string {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.location
 }
 
 func (e *endpoint) LocalDC() bool {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.local
 }
 
 func (e *endpoint) LoadFactor() float32 {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.loadFactor
 }
 
 func (e *endpoint) LastUpdated() time.Time {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.lastUpdated
 }
 
 func (e *endpoint) Touch(opts ...option) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.lastUpdated = time.Now()
 	for _, o := range opts {
 		o(e)
@@ -88,36 +123,48 @@ type option func(e *endpoint)
 
 func WithID(id uint32) option {
 	return func(e *endpoint) {
+		e.mu.Lock()
+		defer e.mu.Unlock()
 		e.id = id
 	}
 }
 
 func WithLocation(location string) option {
 	return func(e *endpoint) {
+		e.mu.Lock()
+		defer e.mu.Unlock()
 		e.location = location
 	}
 }
 
 func WithLocalDC(local bool) option {
 	return func(e *endpoint) {
+		e.mu.Lock()
+		defer e.mu.Unlock()
 		e.local = local
 	}
 }
 
 func WithLoadFactor(loadFactor float32) option {
 	return func(e *endpoint) {
+		e.mu.Lock()
+		defer e.mu.Unlock()
 		e.loadFactor = loadFactor
 	}
 }
 
 func WithServices(services []string) option {
 	return func(e *endpoint) {
+		e.mu.Lock()
+		defer e.mu.Unlock()
 		e.services = append(e.services, services...)
 	}
 }
 
 func WithLastUpdated(ts time.Time) option {
 	return func(e *endpoint) {
+		e.mu.Lock()
+		defer e.mu.Unlock()
 		e.lastUpdated = ts
 	}
 }

internal/table/client.go

@@ -9,6 +9,7 @@ import (
 
 	"google.golang.org/grpc"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/cluster"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/deadline"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/errors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table"
@@ -580,6 +581,8 @@ func (c *client) keeper() {
 				if err != nil {
 					switch {
 					case
+						errors.Is(err, cluster.ErrClusterClosed),
+						errors.Is(err, cluster.ErrClusterEmpty),
 						errors.IsOpError(err, errors.StatusBadSession),
 						errors.IsTransportError(err, errors.TransportErrorDeadlineExceeded):
 						toDelete = append(toDelete, s)

trace/driver.go

@@ -93,6 +93,7 @@ type EndpointInfo interface {
 	NodeID() uint32
 	Address() string
 	LocalDC() bool
+	Location() string
 	LastUpdated() time.Time
 }