Support proxy interceptors visiting Payloads and Failures in gRPC errors (#213)

* Support visiting user payloads inside GRPC errors

* Add support for failure visitor

* Only visit payload/failures for multiop and query failures

* Dynamically generate errordetails types to visit

* still process responses when non-status error,

* fix test to actually mutate the underlying payload/failure

* Add message about failure converts coming before payload converters, add parenthesis

* Slice of protoadapt.MessageV1 fixes the type error

Author: yuandrew

cmd/proxygenerator/interceptor.go

@@ -53,7 +53,9 @@ import (
 	"fmt"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
     "google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/protoadapt"
 )
 
 // VisitPayloadsContext provides Payload context for visitor functions.
@@ -93,6 +95,9 @@ type PayloadVisitorInterceptorOptions struct {
 }
 
 // NewPayloadVisitorInterceptor creates a new gRPC interceptor for workflowservice messages.
+//
+// Note: Failure converters should come before payload codec converts, to allow the 
+// payloads generated by the failure convert to be intercepted by the payload codec converters.
 func NewPayloadVisitorInterceptor(options PayloadVisitorInterceptorOptions) (grpc.UnaryClientInterceptor, error) {
 	return func(ctx context.Context, method string, req, response interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
 		if reqMsg, ok := req.(proto.Message); ok && options.Outbound != nil {
@@ -104,6 +109,30 @@ func NewPayloadVisitorInterceptor(options PayloadVisitorInterceptorOptions) (grp
 
 		err := invoker(ctx, method, req, response, cc, opts...)
 		if err != nil {
+			if options.Inbound != nil {
+				stat, ok := status.FromError(err)
+				if ok {
+					// user provided payloads can sometimes end up in the status details of
+					// gRPC errors, make sure to visit those as well
+					newStatus := status.New(stat.Code(), stat.Message())
+					var newDetails []protoadapt.MessageV1
+					for _, detail := range stat.Details() {
+						detailAny, ok := detail.(*anypb.Any)
+						if ok && ({{ range $i, $name := .GrpcPayload }}{{ if $i }} || {{ end }}detailAny.MessageName() == "{{$name}}"{{ end }}) {
+							err = VisitPayloads(ctx, detailAny, *options.Inbound)
+							if err != nil {
+								return err
+							}
+						}
+						newDetails = append(newDetails, detailAny)
+					}
+					newStatus, err = newStatus.WithDetails(newDetails...)
+					if err != nil {
+						return err
+					}
+					return newStatus.Err()
+				}
+			}
 			return err
 		}
 
@@ -148,6 +177,9 @@ type FailureVisitorInterceptorOptions struct {
 }
 
 // NewFailureVisitorInterceptor creates a new gRPC interceptor for workflowservice messages.
+//
+// Note: Failure converters should come before payload codec converts, to allow the 
+// payloads generated by the failure convert to be intercepted by the payload codec converters.
 func NewFailureVisitorInterceptor(options FailureVisitorInterceptorOptions) (grpc.UnaryClientInterceptor, error) {
 	return func(ctx context.Context, method string, req, response interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
 		if reqMsg, ok := req.(proto.Message); ok && options.Outbound != nil {
@@ -159,6 +191,30 @@ func NewFailureVisitorInterceptor(options FailureVisitorInterceptorOptions) (grp
 
 		err := invoker(ctx, method, req, response, cc, opts...)
 		if err != nil {
+			if options.Inbound != nil {
+				stat, ok := status.FromError(err)
+				if ok {
+					// user provided payloads can sometimes end up in the status details of
+					// gRPC errors, make sure to visit those as well
+					newStatus := status.New(stat.Code(), stat.Message())
+					var newDetails []protoadapt.MessageV1
+					for _, detail := range stat.Details() {
+						detailAny, ok := detail.(*anypb.Any)
+						if ok && ({{ range $i, $name := .GrpcFailure }}{{ if $i }} || {{ end }}detailAny.MessageName() == "{{$name}}"{{ end }}) {
+							err = VisitFailures(ctx, detailAny, *options.Inbound)
+							if err != nil {	
+								return err
+							}
+						}
+						newDetails = append(newDetails, detailAny)
+					}
+					newStatus, err = newStatus.WithDetails(newDetails...)
+					if err != nil {
+						return err
+					}
+					return newStatus.Err()
+				}
+			}
 			return err
 		}
 
@@ -369,6 +425,13 @@ func visitFailures(ctx *VisitFailuresContext, options *VisitFailuresOptions, obj
 
 var interceptorTemplate = template.Must(template.New("interceptor").Parse(InterceptorTemplateText))
 
+type TemplateInput struct {
+	PayloadTypes map[string]*TypeRecord
+	FailureTypes map[string]*TypeRecord
+	GrpcPayload  []string
+	GrpcFailure  []string
+}
+
 // TypeRecord holds the state for a type referred to by the workflow service
 type TypeRecord struct {
 	Methods  []string // List of methods on this type that can eventually lead to Payload(s)
@@ -776,11 +839,31 @@ func generateInterceptor(cfg config) error {
 		return err
 	}
 
+	// gather a list of errordetails that can contain user payloads when included in
+	// gRPC error messages
+	var grpcPayload []string
+	for _, msg := range allPayloadContainingMessages {
+		if strings.Contains(string(msg.FullName()), "temporal.api.errordetails.v1.") && strings.Count(string(msg.FullName()), ".") == 4 {
+			grpcPayload = append(grpcPayload, string(msg.FullName()))
+		}
+	}
+
+	var grpcFailure []string
+	for _, msg := range allFailureContainingMessages {
+		if strings.Contains(string(msg.FullName()), "temporal.api.errordetails.v1.") && strings.Count(string(msg.FullName()), ".") == 4 {
+			grpcFailure = append(grpcFailure, string(msg.FullName()))
+		}
+	}
+
 	buf := &bytes.Buffer{}
 	fmt.Fprint(buf, cfg.license)
 
-	err = interceptorTemplate.Execute(buf, map[string]map[string]*TypeRecord{
-		"PayloadTypes": payloadRecords, "FailureTypes": failureRecords})
+	err = interceptorTemplate.Execute(buf, TemplateInput{
+		PayloadTypes: payloadRecords,
+		FailureTypes: failureRecords,
+		GrpcFailure:  grpcFailure,
+		GrpcPayload:  grpcPayload,
+	})
 	if err != nil {
 		return err
 	}