grpc-gateway don't work well when using github.com/golang/protobuf/ptypes/struct with streaming

[yfei1]

Steps you follow to reproduce the error:

My code has the following API definition

message EvaluateRequest {
  api.Match match = 1;
}

message EvaluateResponse {
  api.Match match = 1;
}

message Match {
  string match_id = 1;

  repeated Ticket tickets = 4;

  google.protobuf.Struct properties = 6;
}

service Evaluator {
  rpc Evaluate(stream EvaluateRequest) returns (stream EvaluateResponse) {
    option (google.api.http) = {
      post: "/v1/evaluator/matches:evaluate"
      body: "*"
    };
  }
}

With the code logic following the example in the integration test. https://github.com/grpc-ecosystem/grpc-gateway/blob/master/examples/integration/integration_test.go#956

func (ec *evaluatorClient) httpEvaluate(ctx context.Context, proposals []*pb.Match) (results []*pb.Match, err error) {
	reqr, reqw := io.Pipe()
	proposalIDs := getMatchIds(proposals)
	waitc := make(chan error)

	go func() {
                var m jsonpb.Marshaler
		defer close(waitc)
		defer func() {
			sendErr := reqw.Close()
			if sendErr != nil {
				logger.WithError(sendErr).Warning("failed to close response body read closer")
			}
		}()

		for _, proposal := range proposals {
			buf, sendErr := m.MarshalToString(&pb.EvaluateRequest{Match: proposal})
			if sendErr != nil {
				waitc <- status.Errorf(codes.FailedPrecondition, "failed to marshal proposal to string: %s", sendErr.Error())
				return
			}
			_, sendErr = io.WriteString(reqw, buf)
			if sendErr != nil {
				waitc <- status.Errorf(codes.FailedPrecondition, "failed to write proto string to io writer: %s", sendErr.Error())
				return
			}
		}
	}()

	req, err := http.NewRequest("POST", ec.baseURL+"/v1/evaluator/matches:evaluate", reqr)
	if err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "failed to create evaluator http request for proposals %s: %s", proposalIDs, err.Error())
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Transfer-Encoding", "chunked")

	resp, err := http.DefaultClient.Do(req.WithContext(ctx))
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get response from evaluator for proposals %s: %s", proposalIDs, err.Error())
	}
	defer func() {
		err = resp.Body.Close()
		if err != nil {
			logger.WithError(err).Warning("failed to close response body read closer")
		}
	}()

	var got = []*pb.Match{}
	dec := json.NewDecoder(resp.Body)
	for {
		var item struct {
			Result json.RawMessage        `json:"result"`
			Error  map[string]interface{} `json:"error"`
		}
		err := dec.Decode(&item)
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, status.Errorf(codes.Unavailable, "failed to read response from HTTP JSON stream: %s", err.Error())
		}
		if len(item.Error) != 0 {
			return nil, status.Errorf(codes.Unavailable, "failed to execute evaluator.Evaluate: %v", item.Error)
		}
		resp := &pb.EvaluateResponse{}
		if err := json.Unmarshal(item.Result, resp); err != nil {
			return nil, status.Errorf(codes.Unavailable, "failed to execute jsonpb.UnmarshalString(%s, &proposal): %v.", item.Result, err)
		}
		got = append(got, resp.GetMatch())
	}

	if err, ok := <-waitc; ok && err != nil {
		return nil, err
	}
	return got, nil
}

When running the tests against the http endpoint I got the following error, yet the same test works for the grpc endpoint.

panic: reflect.Set: value of type map[string]*types.Value is not assignable to type map[string]*structpb.Value [recovered]
	panic: reflect.Set: value of type map[string]*types.Value is not assignable to type map[string]*structpb.Value

goroutine 12 [running]:
testing.tRunner.func1(0xc000108800)
	/usr/lib/google-golang/src/testing/testing.go:830 +0x392
panic(0xf40200, 0xc00060e260)
	/usr/lib/google-golang/src/runtime/panic.go:522 +0x1b5
reflect.Value.assignTo(0xf7b100, 0xc0001493e0, 0x15, 0x10e3c4b, 0xb, 0xf7b1c0, 0x0, 0x52458d, 0x1093660, 0xc0001493e0)
	/usr/lib/google-golang/src/reflect/value.go:2339 +0x437
reflect.Value.Set(0xf7b1c0, 0xc000149260, 0x195, 0xf7b100, 0xc0001493e0, 0x15)
	/usr/lib/google-golang/src/reflect/value.go:1473 +0xa8
github.com/gogo/protobuf/jsonpb.(*Unmarshaler).unmarshalValue(0xc0005d8400, 0x102b900, 0xc000149260, 0x199, 0xc0002c2540, 0x11, 0x20, 0xc000502b00, 0x0, 0x0)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:916 +0x2f0f
github.com/gogo/protobuf/jsonpb.(*Unmarshaler).unmarshalValue(0xc0005d8400, 0x1062000, 0xc00049e4e0, 0x196, 0xc0002c2540, 0x11, 0x20, 0xc000502b00, 0xf8c5df, 0x49)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:808 +0x2a1
github.com/gogo/protobuf/jsonpb.(*Unmarshaler).unmarshalValue(0xc0005d8400, 0x10932a0, 0xc00049e480, 0x199, 0xc00041a180, 0x52, 0x60, 0xc000108100, 0x0, 0xfe180)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:1061 +0xaaf
github.com/gogo/protobuf/jsonpb.(*Unmarshaler).unmarshalValue(0xc0005d8400, 0x108f2e0, 0xc000148a80, 0x196, 0xc00041a180, 0x52, 0x60, 0xc000108100, 0xf69179, 0x3f)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:808 +0x2a1
github.com/gogo/protobuf/jsonpb.(*Unmarshaler).unmarshalValue(0xc0005d8400, 0x1033a00, 0xc000148a80, 0x199, 0xc00041a120, 0x5c, 0x60, 0x0, 0x523978, 0x160)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:1061 +0xaaf
github.com/gogo/protobuf/jsonpb.(*Unmarshaler).UnmarshalNext(0xc0005d8400, 0xc0000ee420, 0x12df020, 0xc000148a80, 0xc00041a0c0, 0xc00041a060)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:758 +0x18f
github.com/gogo/protobuf/jsonpb.(*Unmarshaler).Unmarshal(...)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:769
github.com/gogo/protobuf/jsonpb.UnmarshalString(0xc00041a0c0, 0x5c, 0x12df020, 0xc000148a80, 0xc00041a0c0, 0x5c)
	/usr/local/google/home/yufanfei/go/src/github.com/gogo/protobuf/jsonpb/jsonpb.go:790 +0xde
open-match.dev/open-match/internal/app/synchronizer.(*evaluatorClient).httpEvaluate(0xc00059c050, 0x12e2960, 0xc00003a0b0, 0xc000574020, 0x1, 0x1, 0x0, 0x0, 0x0, 0x0, ...)
	/usr/local/google/home/yufanfei/go/src/open-match.dev/open-match/internal/app/synchronizer/evaluator_client.go:226 +0x7e2
open-match.dev/open-match/internal/app/synchronizer.TestEvaluator.func2(0xc000108800)
	/usr/local/google/home/yufanfei/go/src/open-match.dev/open-match/internal/app/synchronizer/evaluator_client_test.go:119 +0x78
testing.tRunner(0xc000108800, 0xc0000cf6d0)
	/usr/lib/google-golang/src/testing/testing.go:865 +0xc0
created by testing.(*T).Run
	/usr/lib/google-golang/src/testing/testing.go:916 +0x35a
FAIL	open-match.dev/open-match/internal/app/synchronizer	0.119s
Error: Tests failed.

What did you expect to happen instead:

Test passed

What's your theory on why it isn't working:

I think the grpc-gateway plugin might be incompatible with messages having the google.protobuf.Struct field. I have another service with a stream response and it produced the same error when calling the API service via its HTTP endpoint. I was able to have a workaround by replacing https://github.com/grpc-ecosystem/grpc-gateway/blob/master/examples/integration/integration_test.go#1021 with

...
if err := json.Unmarshal(item.Result, resp); err != nil {
   ...
}
...

yet using the same trick in a bidirectional streaming service gives a partial response without the google.protobuf.Struct field.

--- FAIL: TestEvaluator (0.00s)
    --- FAIL: TestEvaluator/test_input_matches_output_when_receiving_one_match (0.00s)
        /usr/local/google/home/yufanfei/go/src/open-match.dev/open-match/internal/app/synchronizer/evaluator_client_test.go:125: 
matches: [match_id:"12" tickets:<id:"1" > tickets:<id:"2" > properties:<fields:<key:"match_score" value:<number_value:1 > > > ] does not contains match match_id:"12" tickets:<id:"1" > tickets:<id:"2" > properties:<> 
FAIL
FAIL	open-match.dev/open-match/internal/app/synchronizer	0.120s
Error: Tests failed.

[johanbrandhorst]

Thanks for the bug report! This looks pretty bad, but it also sounds like you have a good test to reproduce the panic, with it's a great way to start creating a fix! Would you be interested in contributing a fix for this?

[yfei1]

Thanks for the bug report! This looks pretty bad, but it also sounds like you have a good test to reproduce the panic, with it's a great way to start creating a fix! Would you be interested in contributing a fix for this?

I don't have enough context about how grpc-gateway works. Could you locate some files that might relate to this bug? I would love to help if it is a trivial bug.

[johanbrandhorst]

Wait, I've just realised you're using gogo/protobuf (as is evident by that backtrace). This could be another one of those subtle gogo/protobuf bugs. Can you try using golang/protobuf?

[yfei1]

Wait, I've just realised you're using gogo/protobuf (as is evident by that backtrace). This could be another one of those subtle gogo/protobuf bugs. Can you try using golang/protobuf?

Thanks for pointing it out. It seems like a bug in gogo/protobuf. Everything works as expected after using golang/protobuf instead!

[johanbrandhorst]

Glad that fixed it, but it's not so much a bug in gogo/protobuf as a simple incompatibility with our codebase... It's an unfortunate consequence of the way go protobuf was designed from the start. Anyway, glad it works now!