Merge pull request #151 from ydb-platform/cluster-empty

v3.13.1: improve error messages

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,13 @@
+## 3.13.1
+* Improved error messages
+* Defended `cluster.balancer` with `sync.RWMutex` on `cluster.Insert`, `cluster.Update`, `cluster.Remove` and `cluster.Get`
+* Excluded `Close` and `Park` methods from `conn.Conn` interface
+* Fixed bug with `Multi` balancer `Create()` 
+* Improved `errors.IsTransportError` (check a few transport error codes instead check single transport error code)
+* Improved `errors.Is` (check a few errors instead check single error)
+* Refactored YDB errors checking API on client-side
+* Implemented of scripting traces 
+
 ## 3.13.0
 * Refactored `Connection` interface
 * Removed `CustomOption` and taking client with custom options

errors.go

@@ -28,29 +28,24 @@ func IsTimeoutError(err error) bool {
 	return errors.IsTimeoutError(err)
 }
 
-func IsTransportError(err error) bool {
-	return TransportErrorDescription(err) != nil
-}
-
-func IsTransportErrorCode(err error, codes ...int32) bool {
-	d := TransportErrorDescription(err)
-	if d == nil {
-		return false
-	}
-	for _, code := range codes {
-		if d.Code() == code {
-			return true
-		}
-	}
-	return false
+func IsTransportError(err error, codes ...int32) bool {
+	return errors.IsTransportError(
+		err,
+		func() (cc []errors.TransportErrorCode) {
+			for _, code := range codes {
+				cc = append(cc, errors.TransportErrorCode(code))
+			}
+			return cc
+		}()...,
+	)
 }
 
 func IsTransportErrorCancelled(err error) bool {
-	return IsTransportErrorCode(err, int32(errors.TransportErrorCanceled))
+	return IsTransportError(err, int32(errors.TransportErrorCanceled))
 }
 
 func IsTransportErrorResourceExhausted(err error) bool {
-	return IsTransportErrorCode(err, int32(errors.TransportErrorResourceExhausted))
+	return IsTransportError(err, int32(errors.TransportErrorResourceExhausted))
 }
 
 type Error interface {
@@ -72,21 +67,16 @@ func IsYdbError(err error) bool {
 	return IsTransportError(err) || IsOperationError(err)
 }
 
-func IsOperationError(err error) bool {
-	return OperationErrorDescription(err) != nil
-}
-
-func IsOperationErrorCode(err error, codes ...int32) bool {
-	d := OperationErrorDescription(err)
-	if d == nil {
-		return false
-	}
-	for _, code := range codes {
-		if d.Code() == code {
-			return true
-		}
-	}
-	return false
+func IsOperationError(err error, codes ...int32) bool {
+	return errors.IsOpError(
+		err,
+		func() (cc []errors.StatusCode) {
+			for _, code := range codes {
+				cc = append(cc, errors.StatusCode(code))
+			}
+			return cc
+		}()...,
+	)
 }
 
 func OperationErrorDescription(err error) Error {
@@ -98,23 +88,23 @@ func OperationErrorDescription(err error) Error {
 }
 
 func IsOperationErrorOverloaded(err error) bool {
-	return IsOperationErrorCode(err, int32(errors.StatusOverloaded))
+	return IsOperationError(err, int32(errors.StatusOverloaded))
 }
 
 func IsOperationErrorUnavailable(err error) bool {
-	return IsOperationErrorCode(err, int32(errors.StatusUnavailable))
+	return IsOperationError(err, int32(errors.StatusUnavailable))
 }
 
 func IsOperationErrorAlreadyExistsError(err error) bool {
-	return IsOperationErrorCode(err, int32(errors.StatusAlreadyExists))
+	return IsOperationError(err, int32(errors.StatusAlreadyExists))
 }
 
 func IsOperationErrorNotFoundError(err error) bool {
-	return IsOperationErrorCode(err, int32(errors.StatusNotFound))
+	return IsOperationError(err, int32(errors.StatusNotFound))
 }
 
 func IsOperationErrorSchemeError(err error) bool {
-	return IsOperationErrorCode(err, int32(errors.StatusSchemeError))
+	return IsOperationError(err, int32(errors.StatusSchemeError))
 }
 
 func IsRatelimiterAcquireError(err error) bool {

internal/balancer/multi/multi.go

@@ -24,9 +24,9 @@ type multi struct {
 }
 
 func (m *multi) Create() balancer.Balancer {
-	bb := m.balancer
-	for i := range bb {
-		bb[i] = bb[i].(balancer.Creator).Create()
+	bb := make([]balancer.Balancer, len(m.balancer))
+	for i, b := range m.balancer {
+		bb[i] = b.(balancer.Creator).Create()
 	}
 	return &multi{
 		balancer: bb,

internal/cluster/cluster.go

@@ -2,6 +2,7 @@ package cluster
 
 import (
 	"context"
+	"fmt"
 	"sort"
 	"strings"
 	"sync"
@@ -26,29 +27,20 @@ const (
 
 var (
 	// ErrClusterClosed returned when requested on a closed cluster.
-	ErrClusterClosed = errors.New("cluster closed")
+	ErrClusterClosed = fmt.Errorf("cluster closed")
 
 	// ErrClusterEmpty returned when no connections left in cluster.
-	ErrClusterEmpty = errors.New("cluster empty")
-
-	// ErrUnknownEndpoint returned when no connections left in cluster.
-	ErrUnknownEndpoint = errors.New("unknown endpoint")
-
-	// 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")
+	ErrClusterEmpty = fmt.Errorf("cluster empty")
 )
 
 type cluster struct {
-	config   config.Config
-	pool     conn.Pool
-	dial     func(context.Context, string) (*grpc.ClientConn, error)
-	balancer balancer.Balancer
+	config config.Config
+	pool   conn.Pool
+	dial   func(context.Context, string) (*grpc.ClientConn, error)
+
+	balancerMtx sync.RWMutex
+	balancer    balancer.Balancer
+
 	explorer repeater.Repeater
 
 	index     map[string]entry.Entry
@@ -77,6 +69,9 @@ func (c *cluster) Pessimize(ctx context.Context, cc conn.Conn, cause error) {
 		return
 	}
 
+	c.balancerMtx.Lock()
+	defer c.balancerMtx.Unlock()
+
 	if !c.balancer.Contains(entry.Handle) {
 		return
 	}
@@ -235,7 +230,7 @@ func (c *cluster) Get(ctx context.Context, opts ...crudOption) (cc conn.Conn, er
 	defer c.mu.RUnlock()
 
 	if c.closed {
-		return nil, ErrClusterClosed
+		return nil, errors.Error(ErrClusterClosed)
 	}
 
 	onDone := trace.DriverOnClusterGet(c.config.Trace(), &ctx)
@@ -258,9 +253,12 @@ func (c *cluster) Get(ctx context.Context, opts ...crudOption) (cc conn.Conn, er
 		}
 	}
 
+	c.balancerMtx.RLock()
+	defer c.balancerMtx.RUnlock()
+
 	cc = c.balancer.Next()
 	if cc == nil {
-		return nil, ErrClusterEmpty
+		return nil, errors.Error(ErrClusterEmpty)
 	}
 
 	return cc, nil
@@ -297,7 +295,7 @@ func (c *cluster) Insert(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 
 	entry := entry.Entry{Conn: cc}
 
-	inserted = entry.InsertInto(c.balancer)
+	inserted = entry.InsertInto(c.balancer, &c.balancerMtx)
 
 	c.index[e.Address()] = entry
 
@@ -346,10 +344,10 @@ func (c *cluster) Update(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		c.endpoints[e.NodeID()] = entry.Conn
 	}
 
-	if entry.Handle != nil {
-		// entry.Handle may be nil when connection is being tracked.
-		c.balancer.Update(entry.Handle, e.Info())
-	}
+	c.balancerMtx.Lock()
+	defer c.balancerMtx.Unlock()
+
+	c.balancer.Update(entry.Handle, e.Info())
 
 	return entry.Conn
 }
@@ -379,16 +377,11 @@ func (c *cluster) Remove(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		panic("ydb: can't remove not-existing endpoint")
 	}
 
-	removed = entry.RemoveFrom(c.balancer)
+	removed = entry.RemoveFrom(c.balancer, &c.balancerMtx)
 
 	delete(c.index, e.Address())
 	delete(c.endpoints, e.NodeID())
 
-	if entry.Conn != nil {
-		// entry.Conn may be nil when connection is being tracked after unsuccessful dial().
-		_ = entry.Conn.Close(ctx)
-	}
-
 	return entry.Conn
 }
 

internal/cluster/cluster_test.go

@@ -46,10 +46,10 @@ func TestClusterFastRedial(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
 			for i := 0; i < size*10; i++ {
-				c, err := c.Get(context.Background())
+				cc, err := c.Get(context.Background())
 				// enforce close bad connects to track them
-				if err == nil && c != nil && c.Endpoint().Address() == "bad:0" {
-					_ = c.Close(ctx)
+				if err == nil && c != nil && cc.Endpoint().Address() == "bad:0" {
+					_ = c.Remove(ctx, cc.Endpoint())
 				}
 			}
 			close(done)
@@ -214,7 +214,7 @@ func (ln *stubListener) Accept() (net.Conn, error) {
 	select {
 	case ln.C <- c:
 	case <-ln.exit:
-		return nil, errors.Errorf(0, "closed")
+		return nil, errors.Errorf("closed")
 	}
 	select {
 	case ln.S <- s:
@@ -239,7 +239,7 @@ func (ln *stubListener) Dial(ctx context.Context) (*grpc.ClientConn, error) {
 		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
 			select {
 			case <-ln.exit:
-				return nil, errors.Errorf(0, "refused")
+				return nil, errors.Errorf("refused")
 			case c := <-ln.C:
 				return c, nil
 			case <-ctx.Done():

internal/cluster/entry/entry.go

@@ -1,6 +1,8 @@
 package entry
 
 import (
+	"sync"
+
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/balancer"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/conn"
 )
@@ -11,22 +13,26 @@ type Entry struct {
 	Handle balancer.Element
 }
 
-func (c *Entry) InsertInto(b balancer.Balancer) bool {
+func (c *Entry) InsertInto(b balancer.Balancer, mtx *sync.RWMutex) bool {
 	if c.Handle != nil {
 		panic("ydb: Handle already exists")
 	}
 	if c.Conn == nil {
 		panic("ydb: can't insert nil Conn into balancer")
 	}
+	mtx.Lock()
 	c.Handle = b.Insert(c.Conn)
+	mtx.Unlock()
 	return c.Handle != nil
 }
 
-func (c *Entry) RemoveFrom(b balancer.Balancer) bool {
+func (c *Entry) RemoveFrom(b balancer.Balancer, mtx *sync.RWMutex) bool {
 	if c.Handle == nil {
 		return false
 	}
+	mtx.Lock()
 	b.Remove(c.Handle)
+	mtx.Unlock()
 	c.Handle = nil
 	return true
 }