gRPC streaming client. #1054

htuch · 2017-06-06T15:09:47Z

This is a templatized gRPC client than can be used to implement unary/streaming (client, server,
bidi) for arbitrary protobuf request/response messages. This will be used by the v2 xDS API clients
for gRPC streams.

htuch · 2017-06-06T15:10:02Z

@lizan @fengli79 @mattklein123

mattklein123 · 2017-06-06T23:07:36Z

At a very high level (interface level), this approach LGTM.

fengli79 · 2017-06-07T01:21:37Z

Do you want to get the API closer to the gRPC async stream API?
https://github.com/grpc/grpc/blob/master/include/grpc%2B%2B/impl/codegen/async_stream.h

The void* tag can be replaced by a callback which indicates the finishing of corresponding operation.

PiotrSikora · 2017-06-07T22:28:30Z

@lizan @qiwzhang

htuch · 2017-06-08T21:02:18Z

@fengli79 I don't think it makes sense to use the exact gRPC async stream API, as what I'm doing is closer to the Envoy model for callbacks (declaring a callback object rather than supplying individual callbacks). With that said, I think there's useful stuff in the interface you point to wrt stream termination, status/error handling and metadata that we can use to make the Grpc::AsyncClient interface a bit richer and closer in terms of capabilities.

lizan · 2017-06-08T02:46:36Z

include/envoy/grpc/async_client.h

+   * TODO(htuch): Add support for metadata.
+   * @param service_full_name gRPC service name.
+   * @param method_name gRPC method name.
+   * @param end_stream supplies whether this is a headers-only request.


There is no headers-only request

In general, there's no headers-only request. However, the server side API allows to send back response as soon as request metadata is received. So, in theory, you can implement a grpc server like that.

lizan · 2017-06-08T21:04:25Z

include/envoy/grpc/async_client.h

+   * @param request protobuf serializable message.
+   * @param end_stream supplies whether this is the last message.
+   */
+  virtual void sendMessage(const RequestType& request, bool end_stream) PURE;


You probably need a method to just send end_stream (empty DATA frame with end_stream)

lizan · 2017-06-08T21:05:22Z

include/envoy/grpc/async_client.h

+   * Called when an async gRPC stream is remotely closed with successful status.
+   * @param metadata the trailing metadata.
+   */
+  virtual void onRemoteClose() PURE;


where is metadata? also there should be a callback to handle initial metadata too.

Need following semantic on the context, means they are available during a grpc call:
getServerInitialMetadata()
getServerTrailingMetadata()

Need following semantic as a sender:
AddMetadata. (Modify the request metadata)
Write and its callback. (Write and the callback of the write operation)
WriteDone and its callback. (A half close from client side and done callback of the half close operation)
Write with Options and with its callback. (Send a message with half close all together, important for unary request performance, grpc uses a write operation to implement this semantic, you may seek for a better method name. Other options may be supported to set compression algorithm, etc.)

Need following semantic as a receiver:
onInitialMetadata. (Callback when receiving response metadata)
onMessage.
onFinish. (Whenever the grpc call is done, success or failure. Prefer to get a single point to handle finish.)

lizan · 2017-06-08T21:08:39Z

include/envoy/grpc/async_client.h

+   *         the handle can be used to send more messages or close the stream.
+   */
+  virtual AsyncClientStream<RequestType>*
+  start(const std::string& service_full_name, const std::string& method_name, bool end_stream,


Consider take a protobuf descriptor pointer const google::protobuf::MethodDescriptor* here instead of service name and method name, it should be easy to get from generated pb.h and it contains more information you might want to use.

Can you provide an example of how to extract this for https://github.com/lyft/envoy/blob/master/test/proto/helloworld.proto?

Don't worry, figured it out. helloworld::Greeter::descriptor()->FindMethodByName("SayHello");

lizan · 2017-06-09T18:48:37Z

include/envoy/grpc/status.h

+  Internal = 13,
+  Unavailable = 14,
+  DataLoss = 15,
+  InvalidCode = 16,


If you decided copy this but not reusing protobuf (which is fine), this shouldn't be 16. maybe -1?

https://github.com/grpc/grpc/blob/master/include/grpc%2B%2B/impl/codegen/status_code_enum.h

This is copied from #1067 - I will rebase on that when it merges and use whatever is there. Please direct feedback there.

Both Unauthenticated and InvalidCode get same value (16)?
I don't think you need InvalidCode, as if grpc expand the enum in the future, it introduces conflict for sure.
Also, please copy all the comments. And add one additional comment to indicate that this enum should be kept sync with grpc.

As mentioned in the other comment, please see #1067 for the status code bits that will be part of this final PR once it is non-WIP.

lizan · 2017-06-09T18:50:42Z

source/common/grpc/async_client_impl.h

+    Http::AsyncClient::Stream* http_stream =
+        cm_.httpAsyncClientForCluster(remote_cluster_name_)
+            .start(*grpc_stream, Optional<std::chrono::milliseconds>(timeout));
+    grpc_stream->set_stream(http_stream);


nit: move this after the null check?

lizan · 2017-06-09T18:53:30Z

source/common/grpc/common.cc

-void Common::checkForHeaderOnlyError(Http::Message& http_response) {
+void Common::validateHeaders(const Http::HeaderMap& headers) {
+  if (Http::Utility::getResponseStatus(headers) != enumToInt(Http::Code::OK)) {
+    throw Exception(Optional<uint64_t>(), "non-200 response code");


This should fail gracefully, not an exception.

http://www.grpc.io/docs/guides/wire.html

Implementations should expect broken deployments to send non-200 HTTP status codes in responses as well as a variety of non-GRPC content-types and to omit Status & Status-Message. Implementations must synthesize a Status & Status-Message to propagate to the application layer when this occurs.

This was here before. I'm actually going to undo my changes to this file, since I've inlined the relevant bits into async_client_impl.h.

lizan · 2017-06-09T18:55:32Z

test/common/grpc/async_client_impl_test.cc

+using testing::Invoke;
+using testing::Eq;
+using testing::NiceMock;
+// using testing::Ref;


nit: remove

lizan · 2017-06-09T18:59:41Z

source/common/grpc/async_client_impl.h

+    for (const auto& frame : frames) {
+      std::unique_ptr<ResponseType> response(new ResponseType());
+      if (frame.flags_ != GRPC_FH_DEFAULT ||
+          !response->ParseFromArray(frame.data_->linearize(frame.data_->length()),


Add a TODO here for not do linearize. I will have a PR to convert Envoy Buffer to protobuf ZeroCopyInputStream

fengli79 · 2017-06-09T17:35:47Z

include/envoy/grpc/async_client.h

+   * @param request protobuf serializable message.
+   * @param end_stream supplies whether this is the last message.
+   */
+  virtual void sendMessage(const RequestType& request, bool end_stream) PURE;


fengli79 · 2017-06-09T17:39:06Z

include/envoy/grpc/async_client.h

+   * Called when an async gRPC stream is remotely closed with successful status.
+   * @param metadata the trailing metadata.
+   */
+  virtual void onRemoteClose() PURE;


Need following semantic on the context, means they are available during a grpc call:
getServerInitialMetadata()
getServerTrailingMetadata()

Need following semantic as a sender:
AddMetadata. (Modify the request metadata)
Write and its callback. (Write and the callback of the write operation)
WriteDone and its callback. (A half close from client side and done callback of the half close operation)
Write with Options and with its callback. (Send a message with half close all together, important for unary request performance, grpc uses a write operation to implement this semantic, you may seek for a better method name. Other options may be supported to set compression algorithm, etc.)

Need following semantic as a receiver:
onInitialMetadata. (Callback when receiving response metadata)
onMessage.
onFinish. (Whenever the grpc call is done, success or failure. Prefer to get a single point to handle finish.)

fengli79 · 2017-06-09T18:03:39Z

include/envoy/grpc/async_client.h

+   * TODO(htuch): Add support for metadata.
+   * @param service_full_name gRPC service name.
+   * @param method_name gRPC method name.
+   * @param end_stream supplies whether this is a headers-only request.


In general, there's no headers-only request. However, the server side API allows to send back response as soon as request metadata is received. So, in theory, you can implement a grpc server like that.

fengli79 · 2017-06-09T21:04:21Z

include/envoy/grpc/async_client.h

+   * Called when populating the headers to send with initial metadata.
+   * @param metadata initial metadata reference.
+   */
+  virtual void onCreateInitialMetadata(Http::HeaderMap& metadata) PURE;


"onCreate" is confusing here. The method name should be able to represent when this method is going to be invoked.
I guess it should be invoked right after the request initial metadata gets sent?
If it's intended to be invoked before the request initial metadata gets sent, seems we don't need a callback.

This is before the initial metadata is sent, and provides an opportunity to fill in the metadata. This is similar to how RpcChannel works today, see https://github.com/lyft/envoy/blob/master/include/envoy/grpc/rpc_channel.h#L30.

fengli79 · 2017-06-09T21:10:53Z

include/envoy/grpc/async_client.h

+  /**
+   * Finish the stream. No further operations are permitted on the stream.
+   */
+  virtual void finish() PURE;


The comment looks not right. The client is still be able to receive data on the stream. This is more like a half close on the request, the response processing should not be impacted.

I agree, what was there before was confusing, in particular if we compare with the Finish semantics in the gRPC C++ library, which is a full close. In the model we're adopting in AsyncClient here, there's independent local/remote closing of streams. I will rename this to close to make the semantics clearer.

fengli79 · 2017-06-09T21:25:28Z

source/common/grpc/async_client_impl.h

+        remote_cluster_name_, service_method.service()->full_name(), service_method.name());
+    callbacks.onCreateInitialMetadata(message->headers());
+
+    http_stream->sendHeaders(message->headers(), false);


Expose sendHeaders via grpc_stream. Sometimes, the initial metadata need to be prepared in an async routine.

The onCreateInitialMetadata method in callbacks provides this facility.

fengli79 · 2017-06-09T21:27:33Z

source/common/grpc/async_client_impl.h

+
+  // Http::AsyncClient::StreamCallbacks
+  void onHeaders(Http::HeaderMapPtr&& headers, bool end_stream) override {
+    if (remote_closed_) {


I think the headers should be still be delivered.

I seems that once we've invoked onRemoteClose(..) on callbacks (which is implied by remote_close_), we shouldn't be delivering any further response callbacks as the logical stream is now broken. We're basically waiting for the client owner to invoke finish() so we can mop up all the stream resources. I'll try and add some more comments to make this clearer in the interface docs.

fengli79 · 2017-06-09T21:32:00Z

source/common/grpc/async_client_impl.h

+      return;
+    }
+    if (end_stream) {
+      onTrailers(std::move(headers));


I don't think we should fake trailers when headers are received actually.

http://www.grpc.io/docs/guides/wire.html refers to this as a Trailers-Only response.

fengli79 · 2017-06-09T21:34:13Z

source/common/grpc/async_client_impl.h

+      std::unique_ptr<ResponseType> response(new ResponseType());
+      // TODO(htuch): We can avoid linearizing the buffer here when Buffer::Instance implements
+      // protobuf ZeroCopyInputStream.
+      if (frame.flags_ != GRPC_FH_DEFAULT ||


The response can be compressed. So, it can be GRPC_FH_DEFAULT | compressed flag.
Please add a TODO here.

fengli79 · 2017-06-09T21:35:17Z

source/common/grpc/async_client_impl.h

+    }
+
+    const GrpcStatus grpc_status = Common::getGrpcStatus(*trailers);
+    if (grpc_status == GrpcStatus::InvalidCode) {


Remove InvalidCode.

I'm not sure what the ask is here, can you elaborate some more? The status codes are not part of this PR, I temporarily patched in an earlier revision of #1067, which I plan to merge back in here after it hits master. In the most recent #1067, if grpc_status < 0 in this scenario, it indicates a missing/invalid code, so I will update this logic to reflect that.

wattli · 2017-06-12T17:10:22Z

test/test_common/utility.cc

@@ -64,6 +65,11 @@ std::string TestUtility::bufferToString(const Buffer::Instance& buffer) {
  return output;
 }

+std::string TestUtility::bufferToHexString(const Buffer::Instance& buffer) {


Is this func used?

No, but it was useful in the test development. I might split this out into a separate PR.

wattli · 2017-06-12T17:53:23Z

test/common/grpc/async_client_impl_test.cc

+
+const std::string HELLO_REQUEST = "ABC";
+// We expect the 5 byte header to only have a length of 5 indicating the size of the protobuf. The
+// protobuf beings with 0xa, indicating this is the first field of type string. This is followed


begins with x0a

please fix.

s/beings with 0xa/begins with x0a

wattli · 2017-06-12T17:56:27Z

test/common/grpc/async_client_impl_test.cc

+const char HELLO_REQUEST_DATA[] = "\x00\x00\x00\x00\x05\x0a\x03\x41\x42\x43";
+const size_t HELLO_REQUEST_SIZE = sizeof(HELLO_REQUEST_DATA) - 1;
+
+const std::string HELLO_REPLY = "CBA";


I would pick HELLO_REPLY as something like 'DEFG', so HELLO_REPLY_SIZE will be different with HELLO_REQUEST_SIZE. And the content will be different as well.

This is a templatized gRPC client than can be used to implement unary/streaming (client, server, bidi) for arbitrary protobuf request/response messages. This will be used by the v2 xDS API clients for gRPC streams.

htuch · 2017-06-12T20:41:57Z

@fengli79 @wattli @lizan @mattklein123 This is now ready for review.

lizan · 2017-06-13T00:30:32Z

source/common/grpc/async_client_impl.h

+  void onHeaders(Http::HeaderMapPtr&& headers, bool end_stream) override {
+    ASSERT(!remote_closed_);
+    if (Http::Utility::getResponseStatus(*headers) != enumToInt(Http::Code::OK)) {
+      streamError(Status::GrpcStatus::Internal);


The HTTP code to gRPC status code mapping is defined here
https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md

We should follow this for non-2xx HTTP codes.

lizan

LGTM

mattklein123

overall looks great. few things.

mattklein123 · 2017-06-13T18:24:00Z

include/envoy/grpc/async_client.h

@@ -0,0 +1,113 @@
+#pragma once


general comment. Can we open issue to track deprecating RpcChannel? This is superior and we should just get rid of the other code.

Done, see #1102.

mattklein123 · 2017-06-13T18:27:45Z

include/envoy/grpc/async_client.h

+   * Start a gRPC stream asynchronously.
+   * @param service_method protobuf descriptor of gRPC service method.
+   * @param callbacks the callbacks to be notified of stream status.
+   * @param timeout supplies the stream timeout, measured since when the frame with end_stream


not sure off the top of my head how to fix this yet, but calling out that I'm not sure this definition of timeout makes sense for a streaming RPC. Not sure if we want to leave this here, or split interface somehow for streaming and unary?

I was cribbing from the Http::AsyncClient comments here for stream start(). @lizan Do you know what the intended behavior is for Http::AsyncClient here, I can't figure it out, seems to be just setting a whole stream timeout.

We had previous discussion around this. The intended behavior for Http::AsyncClient is what you have comment here, since it is also used for unary. I think this only make sense for unary calls, so if you want use the AsyncClient for unary call too, probably just leave this here and comment it so.

OK, I think it's fine for now, we can revisit later, we will only be using it with infinite timeout for the APIs.

mattklein123 · 2017-06-13T20:06:57Z

source/common/grpc/async_client_impl.h

+        remote_cluster_name_, service_method.service()->full_name(), service_method.name());
+    callbacks.onCreateInitialMetadata(message->headers());
+
+    http_stream->sendHeaders(message->headers(), false);


I haven't looked at this code in a while, but fairly certain that sendHeaders() can result in an inline reset. Will the code work in that case or does the logic below need to be modified to potentially detect and return nullptr?

I'm not sure if I see that in the current implementation (it can do a closeLocal() if end_stream is true, but it isn't here). I'll add handling to be defensive though, since the interface allows this.

When you send headers, it's possible to get an immediate reset for example due to http/2 stream exhaustion or circuit breaking. I think that case, you want to return nullptr and raise a callback immediately?

mattklein123 · 2017-06-13T20:14:35Z

source/common/grpc/async_client_impl.h

+
+  void onData(Buffer::Instance& data, bool end_stream) override {
+    ASSERT(!remote_closed_);
+    std::vector<Frame> frames;


as a slight perf improvement might consider making this a class variable. In the streaming case, if we clear the frames after dispatch, I think we can retain the capacity and then use it again later (I think). Not a huge deal but though I would point it out.

mattklein123 · 2017-06-13T20:15:31Z

source/common/grpc/async_client_impl.h

+      }
+      callbacks_.onReceiveMessage(std::move(response));
+    }
+    if (end_stream) {


If this is a protocol failure, should we check this before even trying to do the decode up above?

mattklein123 · 2017-06-13T20:17:09Z

test/common/grpc/async_client_impl_test.cc

+#include "gtest/gtest.h"
+
+namespace Envoy {
+using testing::_;


nit: in the rest of the code we have been putting the using statements outside of the Envoy namespace.

mattklein123 · 2017-06-13T20:18:24Z

test/mocks/http/mocks.h

@@ -388,6 +388,14 @@ class MockAsyncClientRequest : public AsyncClient::Request {
  MockAsyncClient* client_;
 };

+class MockAsyncClientStream : public AsyncClient::Stream {
+public:


per gmock compile perf guidelines can we declare constructor/deconstructor here and then define in .cc file. For other mocks in this PR, I'm not sure if you can do that with template mocks, but if you can we should do that (probably not).

Yeah, I don't think this will work with the templates unless we instantiate for every type that is used.

htuch · 2017-06-13T22:59:05Z

@lizan Can you comment on the HTTP/gRPC async client timeout question above? If I don't get any further feedback, I'll merge and followup in a separate issue (in the interest of moving the v2 API implementation along).

lizan reviewed Jun 8, 2017

View reviewed changes

htuch force-pushed the async-grpc branch from 2200bd2 to 1afa238 Compare June 9, 2017 18:05

lizan reviewed Jun 9, 2017

View reviewed changes

fengli79 reviewed Jun 9, 2017

View reviewed changes

wattli reviewed Jun 12, 2017

View reviewed changes

htuch force-pushed the async-grpc branch from cb46d2a to a32018f Compare June 12, 2017 20:38

gRPC streaming client.

8e0a11d

This is a templatized gRPC client than can be used to implement unary/streaming (client, server, bidi) for arbitrary protobuf request/response messages. This will be used by the v2 xDS API clients for gRPC streams.

htuch force-pushed the async-grpc branch from a32018f to 8e0a11d Compare June 12, 2017 20:40

Typo from wattli@.

d715cae

lizan reviewed Jun 13, 2017

View reviewed changes

htuch changed the title ~~gRPC streaming client WIP.~~ gRPC streaming client. Jun 13, 2017

HTTP to gRPC response status code translation.

482b3bb

lizan approved these changes Jun 13, 2017

View reviewed changes

mattklein123 reviewed Jun 13, 2017

View reviewed changes

ASAN fixes.

2a608de

htuch mentioned this pull request Jun 13, 2017

Deprecate RpcChannel for unary gRPC #1102

Closed

htuch added 5 commits June 13, 2017 16:55

Deprecation for RpcChannel[Impl].

36b0c68

Handle sendHeaders() failure in AsyncClient start().

52c0985

Avoid reallocating onData frames vector.

5ac1896

Handle end_stream early in onData().

d21b76b

Move using statements outside Envoy namespace.

211836a

Constructor/destructor in .cc for better gmock compile performance.

a43d5c6

mattklein123 approved these changes Jun 13, 2017

View reviewed changes

htuch merged commit e82b3bd into envoyproxy:master Jun 14, 2017

htuch deleted the async-grpc branch June 14, 2017 13:29

gRPC streaming client. #1054

gRPC streaming client. #1054

Conversation

htuch commented Jun 6, 2017 • edited Loading

htuch commented Jun 6, 2017

mattklein123 commented Jun 6, 2017

fengli79 commented Jun 7, 2017 • edited Loading

PiotrSikora commented Jun 7, 2017

htuch commented Jun 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

htuch commented Jun 12, 2017

Choose a reason for hiding this comment

lizan left a comment

Choose a reason for hiding this comment

mattklein123 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

htuch commented Jun 13, 2017

htuch commented Jun 6, 2017 •

edited

Loading

fengli79 commented Jun 7, 2017 •

edited

Loading