add node info to error

Author: rekby

internal/conn/conn.go

@@ -66,10 +66,10 @@ func (c *conn) Address() string {
 func (c *conn) Ping(ctx context.Context) error {
 	cc, err := c.realConn(ctx)
 	if err != nil {
-		return xerrors.WithStackTrace(err)
+		return c.wrapError(err)
 	}
 	if !isAvailable(cc) {
-		return xerrors.WithStackTrace(errUnavailableConnection)
+		return c.wrapError(errUnavailableConnection)
 	}
 	return nil
 }
@@ -115,7 +115,7 @@ func (c *conn) park(ctx context.Context) (err error) {
 	err = c.close()
 
 	if err != nil {
-		return xerrors.WithStackTrace(err)
+		return c.wrapError(err)
 	}
 
 	return nil
@@ -169,7 +169,7 @@ func (c *conn) GetState() (s State) {
 
 func (c *conn) realConn(ctx context.Context) (cc *grpc.ClientConn, err error) {
 	if c.isClosed() {
-		return nil, xerrors.WithStackTrace(errClosedConnection)
+		return nil, c.wrapError(errClosedConnection)
 	}
 
 	c.mtx.Lock()
@@ -213,7 +213,7 @@ func (c *conn) realConn(ctx context.Context) (cc *grpc.ClientConn, err error) {
 			xerrors.WithAddress(address),
 		)
 
-		return nil, xerrors.WithStackTrace(
+		return nil, c.wrapError(
 			xerrors.Retryable(err,
 				xerrors.WithName("realConn"),
 			),
@@ -250,7 +250,7 @@ func (c *conn) close() (err error) {
 	err = c.cc.Close()
 	c.cc = nil
 	c.setState(Offline)
-	return xerrors.WithStackTrace(err)
+	return c.wrapError(err)
 }
 
 func (c *conn) isClosed() bool {
@@ -286,7 +286,7 @@ func (c *conn) Close(ctx context.Context) (err error) {
 		onClose(c)
 	}
 
-	return xerrors.WithStackTrace(err)
+	return c.wrapError(err)
 }
 
 func (c *conn) Invoke(
@@ -320,7 +320,7 @@ func (c *conn) Invoke(
 
 	cc, err = c.realConn(ctx)
 	if err != nil {
-		return xerrors.WithStackTrace(err)
+		return c.wrapError(err)
 	}
 
 	c.touchLastUsage()
@@ -339,9 +339,9 @@ func (c *conn) Invoke(
 				xerrors.WithAddress(c.Address()),
 			)
 			if sentMark.canRetry() {
-				return xerrors.WithStackTrace(xerrors.Retryable(err, xerrors.WithName("Invoke")))
+				return c.wrapError(xerrors.Retryable(err, xerrors.WithName("Invoke")))
 			}
-			return xerrors.WithStackTrace(err)
+			return c.wrapError(err)
 		}
 
 		return err
@@ -355,10 +355,10 @@ func (c *conn) Invoke(
 		if useWrapping {
 			switch {
 			case !o.GetOperation().GetReady():
-				return xerrors.WithStackTrace(errOperationNotReady)
+				return c.wrapError(errOperationNotReady)
 
 			case o.GetOperation().GetStatus() != Ydb.StatusIds_SUCCESS:
-				return xerrors.WithStackTrace(
+				return c.wrapError(
 					xerrors.Operation(
 						xerrors.FromOperation(o.GetOperation()),
 						xerrors.WithNodeAddress(c.Address()),
@@ -406,7 +406,7 @@ func (c *conn) NewStream(
 
 	cc, err = c.realConn(ctx)
 	if err != nil {
-		return nil, xerrors.WithStackTrace(err)
+		return nil, c.wrapError(err)
 	}
 
 	c.touchLastUsage()
@@ -425,9 +425,9 @@ func (c *conn) NewStream(
 				xerrors.WithAddress(c.Address()),
 			)
 			if sentMark.canRetry() {
-				return s, xerrors.WithStackTrace(xerrors.Retryable(err, xerrors.WithName("NewStream")))
+				return s, c.wrapError(xerrors.Retryable(err, xerrors.WithName("NewStream")))
 			}
-			return s, xerrors.WithStackTrace(err)
+			return s, c.wrapError(err)
 		}
 
 		return s, err
@@ -446,6 +446,11 @@ func (c *conn) NewStream(
 	}, nil
 }
 
+func (c *conn) wrapError(err error) error {
+	nodeErr := newNodeError(c.endpoint.NodeID(), c.endpoint.Address(), err)
+	return xerrors.WithStackTrace(nodeErr, xerrors.WithSkipDepth(1))
+}
+
 type option func(c *conn)
 
 func withOnClose(onClose func(*conn)) option {

internal/conn/error.go

@@ -0,0 +1,25 @@
+package conn
+
+import "fmt"
+
+type nodeError struct {
+	id       uint32
+	endpoint string
+	err      error
+}
+
+func newNodeError(id uint32, endpoint string, err error) nodeError {
+	return nodeError{
+		id:       id,
+		endpoint: endpoint,
+		err:      err,
+	}
+}
+
+func (n nodeError) Error() string {
+	return fmt.Sprintf("on node %v (%v): %v", n.id, n.endpoint, n.err)
+}
+
+func (n nodeError) Unwrap() error {
+	return n.err
+}

internal/conn/error_test.go

@@ -0,0 +1,23 @@
+package conn
+
+import (
+	"errors"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestNodeErrorError(t *testing.T) {
+	testErr := errors.New("test")
+	nodeErr := newNodeError(1, "localhost:1234", testErr)
+	message := nodeErr.Error()
+
+	require.Equal(t, "on node 1 (localhost:1234): test", message)
+}
+
+func TestNodeErrorUnwrap(t *testing.T) {
+	testErr := errors.New("test")
+	nodeErr := newNodeError(1, "asd", testErr)
+
+	unwrapped := errors.Unwrap(nodeErr)
+	require.Equal(t, testErr, unwrapped)
+}

internal/conn/grpc_client_stream.go

@@ -29,14 +29,14 @@ func (s *grpcClientStream) CloseSend() (err error) {
 
 	if err != nil {
 		if s.wrapping {
-			return xerrors.WithStackTrace(
+			return s.wrapError(
 				xerrors.Transport(
 					err,
 					xerrors.WithAddress(s.c.Address()),
 				),
 			)
 		}
-		return xerrors.WithStackTrace(err)
+		return s.wrapError(err)
 	}
 
 	return nil
@@ -58,11 +58,11 @@ func (s *grpcClientStream) SendMsg(m interface{}) (err error) {
 				xerrors.WithAddress(s.c.Address()),
 			)
 			if s.sentMark.canRetry() {
-				return xerrors.WithStackTrace(xerrors.Retryable(err,
+				return s.wrapError(xerrors.Retryable(err,
 					xerrors.WithName("SendMsg"),
 				))
 			}
-			return xerrors.WithStackTrace(err)
+			return s.wrapError(err)
 		}
 
 		return err
@@ -98,11 +98,11 @@ func (s *grpcClientStream) RecvMsg(m interface{}) (err error) {
 				xerrors.WithAddress(s.c.Address()),
 			)
 			if s.sentMark.canRetry() {
-				return xerrors.WithStackTrace(xerrors.Retryable(err,
+				return s.wrapError(xerrors.Retryable(err,
 					xerrors.WithName("RecvMsg"),
 				))
 			}
-			return xerrors.WithStackTrace(err)
+			return s.wrapError(err)
 		}
 
 		return err
@@ -111,7 +111,7 @@ func (s *grpcClientStream) RecvMsg(m interface{}) (err error) {
 	if s.wrapping {
 		if operation, ok := m.(wrap.StreamOperationResponse); ok {
 			if status := operation.GetStatus(); status != Ydb.StatusIds_SUCCESS {
-				return xerrors.WithStackTrace(
+				return s.wrapError(
 					xerrors.Operation(
 						xerrors.FromOperation(operation),
 						xerrors.WithNodeAddress(s.c.Address()),
@@ -124,6 +124,11 @@ func (s *grpcClientStream) RecvMsg(m interface{}) (err error) {
 	return nil
 }
 
+func (s *grpcClientStream) wrapError(err error) error {
+	nodeErr := newNodeError(s.c.endpoint.NodeID(), s.c.endpoint.Address(), err)
+	return xerrors.WithStackTrace(nodeErr, xerrors.WithSkipDepth(1))
+}
+
 func createPinger(c *conn) context.CancelFunc {
 	c.touchLastUsage()
 	ctx, cancel := xcontext.WithCancel(context.Background())