Merge pull request #138 from ydb-platform/endpoint-info

## 3.11.8

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,13 @@
+## 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
+* Fixed `internal.errors.New` wrapping depth
+
 ## 3.11.7
 * Removed internal alias-type `errors.IssuesIterator`
 

internal/cluster/cluster.go

@@ -15,7 +15,6 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/cluster/entry"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/conn"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint/info"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/errors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/repeater"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
@@ -37,8 +36,10 @@ var (
 
 	// ErrNilBalancerElement returned when requested on a nil Balancer element.
 	ErrNilBalancerElement = errors.New("nil balancer element")
+
 	// ErrUnknownBalancerElement returned when requested on a unknown Balancer element.
 	ErrUnknownBalancerElement = errors.New("unknown balancer element")
+
 	// ErrUnknownTypeOfBalancerElement returned when requested on a unknown types of Balancer element.
 	ErrUnknownTypeOfBalancerElement = errors.New("unknown types of balancer element")
 )
@@ -210,7 +211,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)
 		}
 	}()
 
@@ -235,7 +236,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())
@@ -261,16 +262,14 @@ func (c *cluster) Insert(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		panic("ydb: can't insert already existing endpoint")
 	}
 
-	var wait chan struct{}
-	defer func() {
-		if wait != nil {
-			close(wait)
-		}
-	}()
+	cc.Endpoint().Touch()
 
 	entry := entry.Entry{Conn: cc}
+
 	entry.InsertInto(c.balancer)
+
 	c.index[e.Address()] = entry
+
 	if e.NodeID() > 0 {
 		c.endpoints[e.NodeID()] = cc
 	}
@@ -280,7 +279,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())
@@ -307,6 +306,8 @@ func (c *cluster) Update(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		panic("ydb: cluster entry with nil conn")
 	}
 
+	entry.Conn.Endpoint().Touch()
+
 	delete(c.endpoints, e.NodeID())
 	c.index[e.Address()] = entry
 
@@ -316,15 +317,15 @@ func (c *cluster) Update(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 
 	if entry.Handle != nil {
 		// entry.Handle may be nil when connection is being tracked.
-		c.balancer.Update(entry.Handle, info.Info{})
+		c.balancer.Update(entry.Handle, e.Info())
 	}
 
 	return entry.Conn
 }
 
 // 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

@@ -45,6 +45,7 @@ type conn struct {
 	state    State
 	locks    int32
 	ttl      timeutil.Timer
+	onClose  []func(Conn)
 }
 
 func (c *conn) IsState(states ...State) bool {
@@ -79,7 +80,7 @@ func (c *conn) Endpoint() endpoint.Endpoint {
 	if c != nil {
 		return c.endpoint
 	}
-	return endpoint.Endpoint{}
+	return nil
 }
 
 func (c *conn) SetState(ctx context.Context, s State) State {
@@ -92,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
@@ -123,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)
@@ -160,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)))
@@ -200,6 +201,9 @@ func (c *conn) Close(ctx context.Context) (err error) {
 	c.closed = true
 	err = c.close(ctx)
 	c.setState(ctx, Destroyed)
+	for _, f := range c.onClose {
+		f(c)
+	}
 	return err
 }
 
@@ -210,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))
@@ -249,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() {
@@ -316,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() {
@@ -347,13 +351,24 @@ func (c *conn) NewStream(
 	}, nil
 }
 
-func New(endpoint endpoint.Endpoint, config Config) Conn {
+type option func(c *conn)
+
+func withOnClose(onClose func(Conn)) option {
+	return func(c *conn) {
+		c.onClose = append(c.onClose, onClose)
+	}
+}
+
+func New(endpoint endpoint.Endpoint, config Config, opts ...option) Conn {
 	c := &conn{
 		state:    Created,
 		endpoint: endpoint,
 		config:   config,
 		done:     make(chan struct{}),
 	}
+	for _, o := range opts {
+		o(c)
+	}
 	if ttl := config.ConnectionTTL(); ttl > 0 {
 		c.ttl = timeutil.NewTimer(ttl)
 	}

internal/conn/pool.go

@@ -61,7 +61,11 @@ func (p *pool) GetConn(e endpoint.Endpoint) Conn {
 	if cc, ok := p.conns[e.Address()]; ok {
 		return cc
 	}
-	cc := New(e, p.config)
+	cc := New(e, p.config, withOnClose(func(c Conn) {
+		p.mtx.Lock()
+		defer p.mtx.Unlock()
+		delete(p.conns, c.Endpoint().Address())
+	}))
 	p.conns[e.Address()] = cc
 	return cc
 }
@@ -70,14 +74,17 @@ func (p *pool) Close(ctx context.Context) error {
 	close(p.done)
 
 	p.mtx.Lock()
-	defer p.mtx.Unlock()
+	conns := make([]Conn, 0, len(p.conns))
+	for _, c := range p.conns {
+		conns = append(conns, c)
+	}
+	p.mtx.Unlock()
 
 	var issues []error
-	for a, c := range p.conns {
+	for _, c := range conns {
 		if err := c.Close(ctx); err != nil {
 			issues = append(issues, err)
 		}
-		delete(p.conns, a)
 	}
 
 	if len(issues) > 0 {

internal/discovery/discovery.go

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