## 3.11.8

* Added `trace.EndpointInfo.LastUpdated()` timestamp
* Refactored `endpoint.Endpoint` (split to struct `endopint` and interface `Endpoint`)
* 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

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,10 @@
+## 3.11.8
+* Added `trace.EndpointInfo.LastUpdated()` timestamp
+* Refactored `endpoint.Endpoint` (split to struct `endopint` and interface `Endpoint`)
+* 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
+
 ## 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"
@@ -261,16 +260,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
 	}
@@ -307,6 +304,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,7 +315,7 @@ 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

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 {
@@ -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
 }
 
@@ -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
 }

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)
 		}
 		onDone(location, nodes, err)
 	}()

internal/endpoint/endpoint.go

@@ -1,71 +1,134 @@
 package endpoint
 
-type Endpoint struct {
+import (
+	"fmt"
+	"time"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint/info"
+)
+
+type Endpoint interface {
+	Info() info.Info
+	String() string
+
+	NodeID() uint32
+	Address() string
+	Location() string
+	LocalDC() bool
+	LoadFactor() float32
+	LastUpdated() time.Time
+
+	Touch(opts ...option)
+}
+
+type endpoint struct {
 	id       uint32
 	address  string
 	location string
 	services []string
 
 	loadFactor float32
 	local      bool
+
+	lastUpdated time.Time
+}
+
+func (e *endpoint) String() string {
+	return fmt.Sprintf(`{id:%d,address:"%s",local:%t,location:"%s",loadFactor:%f,lastUpdated:"%s"}`,
+		e.id,
+		e.address,
+		e.local,
+		e.location,
+		e.loadFactor,
+		e.lastUpdated.Format(time.RFC3339),
+	)
+}
+
+func (e *endpoint) Info() info.Info {
+	return info.Info{
+		ID:         e.id,
+		Address:    e.address,
+		LoadFactor: e.loadFactor,
+		Local:      e.local,
+	}
 }
 
-func (e Endpoint) NodeID() uint32 {
+func (e *endpoint) NodeID() uint32 {
 	return e.id
 }
 
-func (e Endpoint) Address() (address string) {
+func (e *endpoint) Address() (address string) {
 	return e.address
 }
 
-func (e Endpoint) Location() string {
+func (e *endpoint) Location() string {
 	return e.location
 }
 
-func (e Endpoint) LocalDC() bool {
+func (e *endpoint) LocalDC() bool {
 	return e.local
 }
 
-func (e Endpoint) LoadFactor() float32 {
+func (e *endpoint) LoadFactor() float32 {
 	return e.loadFactor
 }
 
-type option func(e *Endpoint)
+func (e *endpoint) LastUpdated() time.Time {
+	return e.lastUpdated
+}
+
+func (e *endpoint) Touch(opts ...option) {
+	e.lastUpdated = time.Now()
+	for _, o := range opts {
+		o(e)
+	}
+}
+
+type option func(e *endpoint)
 
 func WithID(id uint32) option {
-	return func(e *Endpoint) {
+	return func(e *endpoint) {
 		e.id = id
 	}
 }
 
 func WithLocation(location string) option {
-	return func(e *Endpoint) {
+	return func(e *endpoint) {
 		e.location = location
 	}
 }
 
 func WithLocalDC(local bool) option {
-	return func(e *Endpoint) {
+	return func(e *endpoint) {
 		e.local = local
 	}
 }
 
 func WithLoadFactor(loadFactor float32) option {
-	return func(e *Endpoint) {
+	return func(e *endpoint) {
 		e.loadFactor = loadFactor
 	}
 }
 
 func WithServices(services []string) option {
-	return func(e *Endpoint) {
+	return func(e *endpoint) {
 		e.services = append(e.services, services...)
 	}
 }
 
-func New(address string, opts ...option) (e Endpoint) {
-	e.address = address
+func WithLastUpdated(ts time.Time) option {
+	return func(e *endpoint) {
+		e.lastUpdated = ts
+	}
+}
+
+func New(address string, opts ...option) Endpoint {
+	e := &endpoint{
+		address:     address,
+		lastUpdated: time.Now(),
+	}
 	for _, o := range opts {
-		o(&e)
+		o(e)
 	}
-	return
+	return e
 }

internal/meta/version.go

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

log/discovery.go

@@ -17,12 +17,16 @@ func Discovery(log Logger, details trace.Details) (t trace.Discovery) {
 			start := time.Now()
 			return func(info trace.DiscoverDoneInfo) {
 				if info.Error == nil {
-					log.Debugf(`discover done {latency:"%s",endpoints:%v}`,
+					endpoints := make([]string, 0, len(info.Endpoints))
+					for _, e := range info.Endpoints {
+						endpoints = append(endpoints, e.String())
+					}
+					log.Debugf(`discover done {latency:"%v",endpoints:%v}`,
 						time.Since(start),
-						info.Endpoints,
+						endpoints,
 					)
 				} else {
-					log.Errorf(`discover failed {latency:"%s",error:"%s"}`,
+					log.Errorf(`discover failed {latency:"%v",error:"%s"}`,
 						time.Since(start),
 						info.Error,
 					)
@@ -34,13 +38,13 @@ func Discovery(log Logger, details trace.Details) (t trace.Discovery) {
 			start := time.Now()
 			return func(info trace.WhoAmIDoneInfo) {
 				if info.Error == nil {
-					log.Debugf(`whoAmI done {latency:"%s",user:%v,groups:%v}`,
+					log.Debugf(`whoAmI done {latency:"%v",user:%v,groups:%v}`,
 						time.Since(start),
 						info.User,
 						info.Groups,
 					)
 				} else {
-					log.Errorf(`whoAmI failed {latency:"%s",error:"%s"}`,
+					log.Errorf(`whoAmI failed {latency:"%v",error:"%s"}`,
 						time.Since(start),
 						info.Error,
 					)