feat: add grpc client interceptor to parse error (#598)

Signed-off-by: jyjiangkai <[email protected]>
vanus-labs · Apr 20, 2023 · 7f8f352 · 7f8f352
1 parent 0604c09
commit 7f8f352
Show file tree

Hide file tree

Showing 11 changed files with 62 additions and 38 deletions.
diff --git a/client/internal/vanus/net/connection/connect.go b/client/internal/vanus/net/connection/connect.go
@@ -15,10 +15,16 @@
 package connection
 
 import (
+	// standard libraries.
 	"context"
+
+	// third-party libraries.
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+
+	// first-party libraries.
+	"github.com/vanus-labs/vanus/internal/primitive/interceptor/errinterceptor"
 )
 
 func Connect(ctx context.Context, endpoint string) (*grpc.ClientConn, error) {
@@ -27,6 +33,7 @@ func Connect(ctx context.Context, endpoint string) (*grpc.ClientConn, error) {
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
 		grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor()),
+		grpc.WithUnaryInterceptor(errinterceptor.UnaryClientInterceptor()),
 	}
 	conn, err := grpc.DialContext(ctx, endpoint, opts...)
 	if err != nil {

diff --git a/client/pkg/eventlog/eventlog_impl.go b/client/pkg/eventlog/eventlog_impl.go
@@ -278,7 +278,7 @@ func (l *eventlog) updateReadableSegments(ctx context.Context, rs []*record.Segm
 		if err != nil {
 			// FIXME: create or update segment failed
 			log.Debug().
-				Uint64("segment", segment.id).
+				Interface("segment", segment).
 				Msg("update readable segment failed")
 			continue
 		}
@@ -342,22 +342,20 @@ type logWriter struct {
 }
 
 func (w *logWriter) Append(ctx context.Context, events *cloudevents.CloudEventBatch) (offs []int64, err error) {
-	retryTimes := defaultRetryTimes
-	for i := 1; i <= retryTimes; i++ {
-		offsets, err := w.doAppend(ctx, events)
+	for i := 1; i <= defaultRetryTimes; i++ {
+		offs, err = w.doAppend(ctx, events)
 		if err == nil {
-			return offsets, nil
+			return offs, nil
 		}
 		if !errors.Is(err, errors.ErrSegmentFull) {
 			log.Error(ctx).Err(err).Msg("log-writer append failed")
 			return nil, err
 		}
 		log.Debug(ctx).
 			Int("retry_time", i).
-			Ints64("retry_time", offsets).
 			Err(err).Msg("log-writer append failed cause segment full")
 	}
-	return nil, errors.ErrUnknown
+	return nil, err
 }
 
 func (w *logWriter) Log() Eventlog {

diff --git a/internal/gateway/proxy/proxy.go b/internal/gateway/proxy/proxy.go
@@ -189,7 +189,7 @@ func (cp *ControllerProxy) Publish(ctx context.Context, req *proxypb.PublishRequ
 		err := checkExtension(e.Attributes)
 		if err != nil {
 			responseCode = http.StatusBadRequest
-			return nil, v2.NewHTTPResult(http.StatusBadRequest, err.Error())
+			return nil, errors.ErrInvalidArgument.WithMessage(err.Error())
 		}
 		if e.Attributes == nil {
 			e.Attributes = make(map[string]*cloudevents.CloudEvent_CloudEventAttributeValue, 0)
@@ -208,14 +208,14 @@ func (cp *ControllerProxy) Publish(ctx context.Context, req *proxypb.PublishRequ
 						Stringer("eventTime", eventTime).
 						Msg("invalid format of event time")
 					responseCode = http.StatusBadRequest
-					return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid delivery time")
+					return nil, errors.ErrInvalidArgument.WithMessage("invalid delivery time")
 				}
 				eventTime.Attr = &cloudevents.CloudEvent_CloudEventAttributeValue_CeTimestamp{
 					CeTimestamp: timestamppb.New(t),
 				}
 			default:
 				responseCode = http.StatusBadRequest
-				return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid delivery time")
+				return nil, errors.ErrInvalidArgument.WithMessage("invalid delivery time")
 			}
 			tID, err := cp.ctrl.EventbusService().GetSystemEventbusByName(ctx, primitive.TimerEventbusName)
 			if err != nil {
@@ -246,7 +246,7 @@ func (cp *ControllerProxy) writeEvents(
 		log.Warn(ctx).Err(err).
 			Stringer("eventbus", eventbusID).
 			Msg("append to failed")
-		return v2.NewHTTPResult(http.StatusInternalServerError, err.Error())
+		return err
 	}
 	return nil
 }

diff --git a/internal/primitive/interceptor/errinterceptor/error.go b/internal/primitive/interceptor/errinterceptor/error.go
@@ -41,3 +41,21 @@ func UnaryServerInterceptor() grpc.UnaryServerInterceptor {
 		return res, errors.ConvertToGRPCError(err)
 	}
 }
+
+func UnaryClientInterceptor() grpc.UnaryClientInterceptor {
+	return func(
+		ctx context.Context,
+		method string,
+		req interface{},
+		reply interface{},
+		cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		opts ...grpc.CallOption,
+	) error {
+		err := invoker(ctx, method, req, reply, cc, opts...)
+		if et, ok := errors.FromError(err); ok {
+			return et
+		}
+		return err
+	}
+}
diff --git a/pkg/cluster/auth.go b/pkg/cluster/auth.go
@@ -20,7 +20,6 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	"github.com/vanus-labs/vanus/pkg/cluster/raw_client"
-	"github.com/vanus-labs/vanus/pkg/errors"
 	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
 	metapb "github.com/vanus-labs/vanus/proto/pkg/meta"
 )
@@ -32,15 +31,15 @@ type authService struct {
 func (a *authService) GetUserRole(ctx context.Context, user string) ([]*metapb.UserRole, error) {
 	resp, err := a.client.GetUserRole(ctx, &ctrlpb.GetUserRoleRequest{UserIdentifier: user})
 	if err != nil {
-		return nil, errors.UnwrapOrUnknown(err)
+		return nil, err
 	}
 	return resp.GetUserRole(), nil
 }
 
 func (a *authService) GetUserByToken(ctx context.Context, token string) (string, error) {
 	user, err := a.client.GetUserByToken(ctx, wrapperspb.String(token))
 	if err != nil {
-		return "", errors.UnwrapOrUnknown(err)
+		return "", err
 	}
 	return user.GetValue(), nil
 }

diff --git a/pkg/cluster/eventbus.go b/pkg/cluster/eventbus.go
@@ -44,14 +44,14 @@ func (es *eventbusService) GetEventbusByName(ctx context.Context, ns, name strin
 
 	pb, err := es.nsSvc.GetNamespaceByName(ctx, ns)
 	if err != nil {
-		return nil, errors.UnwrapOrUnknown(err)
+		return nil, err
 	}
 	eb, err := es.client.GetEventbusWithHumanFriendly(ctx, &ctrlpb.GetEventbusWithHumanFriendlyRequest{
 		NamespaceId:  pb.Id,
 		EventbusName: name,
 	})
 	if err != nil {
-		return nil, errors.UnwrapOrUnknown(err)
+		return nil, err
 	}
 
 	// es.cache.Store(key, eb) unmask when dirty cache is resolved
@@ -66,7 +66,7 @@ func (es *eventbusService) GetEventbus(ctx context.Context, id uint64) (*meta.Ev
 
 	eb, err := es.client.GetEventbus(ctx, wrapperspb.UInt64(id))
 	if err != nil {
-		return nil, errors.UnwrapOrUnknown(err)
+		return nil, err
 	}
 
 	// es.cache.Store(id, eb) unmask when dirty cache is resolved
@@ -93,21 +93,20 @@ func (es *eventbusService) CreateSystemEventbusIfNotExist(ctx context.Context, n
 
 	nsPb, err := es.nsSvc.GetSystemNamespace(ctx)
 	if err != nil {
-		return nil, errors.UnwrapOrUnknown(err)
+		return nil, err
 	}
 
-	eventbus, err := es.client.CreateSystemEventbus(ctx, &ctrlpb.CreateEventbusRequest{
+	return es.client.CreateSystemEventbus(ctx, &ctrlpb.CreateEventbusRequest{
 		Name:        name,
 		LogNumber:   int32(defaultSystemEventbusEventlog),
 		Description: desc,
 		NamespaceId: nsPb.Id,
 	})
-	return eventbus, errors.UnwrapOrUnknown(err)
 }
 
 func (es *eventbusService) Delete(ctx context.Context, id uint64) error {
 	_, err := es.client.DeleteEventbus(ctx, wrapperspb.UInt64(id))
-	return errors.UnwrapOrUnknown(err)
+	return err
 }
 
 func (es *eventbusService) RawClient() ctrlpb.EventbusControllerClient {

diff --git a/pkg/cluster/namespace.go b/pkg/cluster/namespace.go
@@ -21,7 +21,6 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	"github.com/vanus-labs/vanus/pkg/cluster/raw_client"
-	"github.com/vanus-labs/vanus/pkg/errors"
 	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
 	metapb "github.com/vanus-labs/vanus/proto/pkg/meta"
 )
@@ -43,7 +42,7 @@ func (ns *namespaceService) GetNamespace(ctx context.Context, id uint64) (*metap
 	}
 	n, err := ns.client.GetNamespace(ctx, &ctrlpb.GetNamespaceRequest{Id: id})
 	if err != nil {
-		return nil, errors.UnwrapOrUnknown(err)
+		return nil, err
 	}
 	// ns.cache.Store(id, n) unmask when dirty cache is resolved
 	return n, nil
@@ -64,7 +63,7 @@ func (ns *namespaceService) GetNamespaceByName(ctx context.Context, name string)
 	}
 	n, err := ns.client.GetNamespaceWithHumanFriendly(ctx, wrapperspb.String(name))
 	if err != nil {
-		return nil, errors.UnwrapOrUnknown(err)
+		return nil, err
 	}
 	// ns.cache.Store(name, n) unmask when dirty cache is resolved
 	return n, nil

diff --git a/pkg/cluster/raw_client/conn.go b/pkg/cluster/raw_client/conn.go
@@ -17,7 +17,6 @@ package raw_client
 import (
 	"context"
 	"fmt"
-	"github.com/vanus-labs/vanus/observability/log"
 	"os"
 	"strconv"
 	"sync"
@@ -30,9 +29,10 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
 
-	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
-
+	"github.com/vanus-labs/vanus/internal/primitive/interceptor/errinterceptor"
+	"github.com/vanus-labs/vanus/observability/log"
 	"github.com/vanus-labs/vanus/pkg/errors"
+	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
 )
 
 const (
@@ -159,9 +159,11 @@ func (c *Conn) getGRPCConn(ctx context.Context, addr string) *grpc.ClientConn {
 		_ = conn.Close() // make sure it's closed
 	}
 
-	var opts []grpc.DialOption
-	opts = append(opts, grpc.WithTransportCredentials(c.credentials))
-	opts = append(opts, grpc.WithBlock())
+	opts := []grpc.DialOption{
+		grpc.WithBlock(),
+		grpc.WithTransportCredentials(c.credentials),
+		grpc.WithUnaryInterceptor(errinterceptor.UnaryClientInterceptor()),
+	}
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
 	conn, err = grpc.DialContext(ctx, addr, opts...)

diff --git a/pkg/cluster/trigger.go b/pkg/cluster/trigger.go
@@ -8,7 +8,6 @@ import (
 	metapb "github.com/vanus-labs/vanus/proto/pkg/meta"
 
 	"github.com/vanus-labs/vanus/pkg/cluster/raw_client"
-	"github.com/vanus-labs/vanus/pkg/errors"
 )
 
 type triggerService struct {
@@ -28,6 +27,5 @@ func (es *triggerService) RegisterHeartbeat(ctx context.Context, interval time.D
 }
 
 func (es *triggerService) GetSubscription(ctx context.Context, id uint64) (*metapb.Subscription, error) {
-	subscription, err := es.client.GetSubscription(ctx, &ctrlpb.GetSubscriptionRequest{Id: id})
-	return subscription, errors.UnwrapOrUnknown(err)
+	return es.client.GetSubscription(ctx, &ctrlpb.GetSubscriptionRequest{Id: id})
 }
diff --git a/pkg/errors/utils.go b/pkg/errors/utils.go
@@ -15,6 +15,8 @@
 package errors
 
 import (
+	stderr "errors"
+
 	"github.com/pkg/errors"
 	"google.golang.org/grpc/status"
 )
@@ -96,7 +98,7 @@ func ConvertToGRPCError(err error) error {
 	}
 	e, ok := err.(*ErrorType)
 	if ok {
-		return errors.New(e.JSON())
+		return stderr.New(e.JSON())
 	}
 	return err
 }
diff --git a/vsctl/command/util.go b/vsctl/command/util.go
@@ -130,11 +130,13 @@ func Error(err error) string {
 	if err == nil {
 		return ""
 	}
-	e, _ := errors.FromError(err)
-	if e.Message == "" {
-		return e.Description
+	if et, ok := errors.FromError(err); ok {
+		if et.Message == "" {
+			return et.Description
+		}
+		return et.Message
 	}
-	return e.Message
+	return err.Error()
 }
 
 func formatID(id uint64) string {