added simple wrapper for grpc errors for getting nodeID and address

Author: asmyasnikov

internal/conn/conn.go

@@ -380,7 +380,7 @@ func invoke(
 		defer onTransportError(ctx, err)
 
 		if !useWrapping {
-			return opID, issues, err
+			return opID, issues, withConnInfo(err, nodeID, address)
 		}
 
 		if sentMark.canRetry() {
@@ -545,7 +545,7 @@ func (c *conn) NewStream(
 		}()
 
 		if !useWrapping {
-			return nil, err
+			return nil, withConnInfo(err, c.NodeID(), c.Address())
 		}
 
 		if sentMark.canRetry() {

internal/conn/errors.go

@@ -38,3 +38,34 @@ func IsBadConn(err error, goodConnCodes ...grpcCodes.Code) bool {
 
 	return true
 }
+
+type grpcError struct {
+	err error
+
+	nodeID  uint32
+	address string
+}
+
+func (e *grpcError) Error() string {
+	return e.err.Error()
+}
+
+func (e *grpcError) As(target any) bool {
+	return xerrors.As(e.err, target)
+}
+
+func (e *grpcError) NodeID() uint32 {
+	return e.nodeID
+}
+
+func (e *grpcError) Address() string {
+	return e.address
+}
+
+func withConnInfo(err error, nodeID uint32, address string) error {
+	return &grpcError{
+		err:     err,
+		nodeID:  nodeID,
+		address: address,
+	}
+}

internal/conn/errors_test.go

@@ -108,3 +108,21 @@ func TestIsBadConn(t *testing.T) {
 		})
 	}
 }
+
+func TestGrpcError(t *testing.T) {
+	err := withConnInfo(grpcStatus.Error(grpcCodes.Unavailable, "test"), 123, "test:123")
+	require.Equal(t, `rpc error: code = Unavailable desc = test`, err.Error())
+	var nodeID interface {
+		NodeID() uint32
+	}
+	require.ErrorAs(t, err, &nodeID)
+	require.Equal(t, uint32(123), nodeID.NodeID())
+	var address interface {
+		Address() string
+	}
+	require.ErrorAs(t, err, &address)
+	require.Equal(t, "test:123", address.Address())
+	s, has := grpcStatus.FromError(err)
+	require.True(t, has)
+	require.Equal(t, grpcCodes.Unavailable, s.Code())
+}

internal/conn/grpc_client_stream.go

@@ -58,7 +58,7 @@ func (s *grpcClientStream) CloseSend() (err error) {
 		}
 
 		if !s.wrapping {
-			return err
+			return withConnInfo(err, s.parentConn.NodeID(), s.parentConn.Address())
 		}
 
 		return xerrors.WithStackTrace(xerrors.Transport(
@@ -97,7 +97,7 @@ func (s *grpcClientStream) SendMsg(m interface{}) (err error) {
 		}()
 
 		if !s.wrapping {
-			return err
+			return withConnInfo(err, s.parentConn.NodeID(), s.parentConn.Address())
 		}
 
 		if s.sentMark.canRetry() {
@@ -156,7 +156,7 @@ func (s *grpcClientStream) RecvMsg(m interface{}) (err error) { //nolint:funlen
 		}()
 
 		if !s.wrapping {
-			return err
+			return withConnInfo(err, s.parentConn.NodeID(), s.parentConn.Address())
 		}
 
 		if s.sentMark.canRetry() {

internal/xerrors/operation.go

@@ -28,6 +28,14 @@ func (e *operationError) Code() int32 {
 	return int32(e.code)
 }
 
+func (e *operationError) NodeID() uint32 {
+	return e.nodeID
+}
+
+func (e *operationError) Address() string {
+	return e.address
+}
+
 func (e *operationError) Name() string {
 	return "operation/" + e.code.String()
 }

internal/xerrors/transport.go

@@ -27,6 +27,14 @@ func (e *transportError) GRPCStatus() *grpcStatus.Status {
 
 func (e *transportError) isYdbError() {}
 
+func (e *transportError) NodeID() uint32 {
+	return e.nodeID
+}
+
+func (e *transportError) Address() string {
+	return e.address
+}
+
 func (e *transportError) Code() int32 {
 	return int32(e.status.Code())
 }
@@ -134,8 +142,8 @@ func IsTransportError(err error, codes ...grpcCodes.Code) bool {
 	var status *grpcStatus.Status
 	if t := (*transportError)(nil); errors.As(err, &t) {
 		status = t.status
-	} else if t, has := grpcStatus.FromError(err); has {
-		status = t
+	} else if s, has := grpcStatus.FromError(err); has {
+		status = s
 	}
 	if status != nil {
 		if len(codes) == 0 {