Merge pull request #162 from ydb-platform/recursive

v3.14.5: minifixes from #160

Author: asmyasnikov

CHANGELOG.md

@@ -9,6 +9,7 @@
 * Refactored `table.CreateSession` as retry operation with options
 * Moved log level from root of repository to package `log`
 * Added details and address to transport error
+* Fixed `recursive` param in `ratelimiter.ListResource` 
 
 ## 3.14.4
 * Implemented auto-removing `conn.Conn` from `conn.Pool` with counting usages of `conn.Conn`

internal/balancer/multi/multi.go

@@ -44,11 +44,8 @@ func WithBalancer(b balancer.Balancer, filter func(cc conn.Conn) bool) Option {
 type Option func(*multi)
 
 func (m *multi) Contains(x balancer.Element) bool {
-	for i, x := range x.(multiHandle).elements {
-		if x == nil {
-			continue
-		}
-		if m.balancer[i].Contains(x) {
+	for i, h := range x.(multiHandle).elements {
+		if h != nil && m.balancer[i].Contains(h) {
 			return true
 		}
 	}
@@ -75,8 +72,7 @@ func (m *multi) Insert(conn conn.Conn) balancer.Element {
 
 	for i, f := range m.filter {
 		if f(conn) {
-			x := m.balancer[i].Insert(conn)
-			h.elements[i] = x
+			h.elements[i] = m.balancer[i].Insert(conn)
 			inserted = true
 		}
 	}
@@ -87,17 +83,17 @@ func (m *multi) Insert(conn conn.Conn) balancer.Element {
 }
 
 func (m *multi) Update(x balancer.Element, info info.Info) {
-	for i, x := range x.(multiHandle).elements {
-		if x != nil {
-			m.balancer[i].Update(x, info)
+	for i, h := range x.(multiHandle).elements {
+		if h != nil {
+			m.balancer[i].Update(h, info)
 		}
 	}
 }
 
 func (m *multi) Remove(x balancer.Element) (removed bool) {
-	for i, x := range x.(multiHandle).elements {
-		if x != nil {
-			if m.balancer[i].Remove(x) {
+	for i, h := range x.(multiHandle).elements {
+		if h != nil {
+			if m.balancer[i].Remove(h) {
 				removed = true
 			}
 		}

internal/cluster/cluster.go

@@ -437,15 +437,7 @@ func DiffEndpoints(curr, next []endpoint.Endpoint, eq, add, del func(i, j int))
 		func(i, j int) int {
 			return compareEndpoints(curr[i], next[j])
 		},
-		func(i, j int) {
-			eq(i, j)
-		},
-		func(i, j int) {
-			add(i, j)
-		},
-		func(i, j int) {
-			del(i, j)
-		},
+		eq, add, del,
 	)
 }
 

internal/conn/conn.go

@@ -210,13 +210,13 @@ func (c *conn) changeUsages(delta int32) int32 {
 func (c *conn) incUsages() {
 	c.Lock()
 	defer c.Unlock()
+	c.lastUsage = time.Now()
 	c.changeUsages(1)
 }
 
 func (c *conn) decUsages() int32 {
 	c.Lock()
 	defer c.Unlock()
-	c.lastUsage = time.Now()
 	return c.changeUsages(-1)
 }
 

internal/conn/pool.go

@@ -66,21 +66,19 @@ func (p *pool) Get(ctx context.Context, endpoint endpoint.Endpoint) Conn {
 		return cc
 	}
 
-	cc = newConn(
-		endpoint,
-		p.config,
-		withOnClose(func(c *conn) {
-			p.mtx.Lock()
-			defer p.mtx.Unlock()
-			delete(p.conns, c.Endpoint().Address())
-		}),
-	)
+	cc = newConn(endpoint, p.config, withOnClose(p.remove))
 
 	p.conns[address] = cc
 
 	return cc
 }
 
+func (p *pool) remove(c *conn) {
+	p.mtx.Lock()
+	defer p.mtx.Unlock()
+	delete(p.conns, c.Endpoint().Address())
+}
+
 func (p *pool) Pessimize(ctx context.Context, cc Conn, cause error) {
 	e := cc.Endpoint().Copy()
 
@@ -101,7 +99,7 @@ func (p *pool) Pessimize(ctx context.Context, cc Conn, cause error) {
 	)(cc.SetState(Banned))
 }
 
-func (p *pool) Take(ctx context.Context) error {
+func (p *pool) Take(context.Context) error {
 	atomic.AddInt64(&p.usages, 1)
 	return nil
 }
@@ -134,21 +132,14 @@ func (p *pool) Release(ctx context.Context) error {
 	return nil
 }
 
-func (p *pool) connParker(ctx context.Context, interval time.Duration) {
+func (p *pool) connParker(ctx context.Context, ttl, interval time.Duration) {
 	ticker := time.NewTicker(interval)
-	ttl := p.config.ConnectionTTL()
 	for {
 		select {
 		case <-p.done:
 			return
 		case <-ticker.C:
-			p.mtx.RLock()
-			conns := make([]*conn, 0, len(p.conns))
-			for _, c := range p.conns {
-				conns = append(conns, c)
-			}
-			p.mtx.RUnlock()
-			for _, c := range conns {
+			for _, c := range p.collectConns() {
 				if time.Since(c.LastUsage()) > ttl {
 					_ = c.park(ctx)
 				}
@@ -157,6 +148,16 @@ func (p *pool) connParker(ctx context.Context, interval time.Duration) {
 	}
 }
 
+func (p *pool) collectConns() []*conn {
+	p.mtx.RLock()
+	defer p.mtx.RUnlock()
+	conns := make([]*conn, 0, len(p.conns))
+	for _, c := range p.conns {
+		conns = append(conns, c)
+	}
+	return conns
+}
+
 func NewPool(
 	ctx context.Context,
 	config Config,
@@ -169,7 +170,7 @@ func NewPool(
 		done:   make(chan struct{}),
 	}
 	if ttl := config.ConnectionTTL(); ttl > 0 {
-		go p.connParker(ctx, ttl/2)
+		go p.connParker(ctx, ttl, ttl/2)
 	}
 	return p
 }

internal/discovery/discovery.go

@@ -75,8 +75,7 @@ func New(
 
 				cluster.DiffEndpoints(curr, next,
 					func(i, j int) {
-						// Endpoints are equal, but we still need to update meta
-						// data such that load factor and others.
+						// Endpoints are equal, but we still need to update metadata (e.g., load factor).
 						crudExplorer.Update(
 							ctx,
 							next[j],

internal/errors/transport.go

@@ -116,7 +116,12 @@ func (t *TransportError) Error() string {
 	}
 	if len(t.details) > 0 {
 		b.WriteString(", details: ")
-		b.WriteString(fmt.Sprintf("%v", t.details))
+		if len(t.details) > 0 {
+			b.WriteString(", details:")
+			for _, detail := range t.details {
+				b.WriteString(fmt.Sprintf("\n- %v", detail))
+			}
+		}
 	}
 	return b.String()
 }

internal/ratelimiter/ratelimiter.go

@@ -120,6 +120,7 @@ func (c *client) ListResource(
 	response, err = c.service.ListResources(ctx, &Ydb_RateLimiter.ListResourcesRequest{
 		CoordinationNodePath: coordinationNodePath,
 		ResourcePath:         resourcePath,
+		Recursive:            recursive,
 		OperationParams: operation.Params(
 			c.config.OperationTimeout(),
 			c.config.OperationCancelAfter(),