Lightstep gPRC generation for tracing

ci/do_ci.sh

@@ -41,6 +41,7 @@ $EXTRA_CMAKE_FLAGS -DENVOY_DEBUG:BOOL=OFF \
 -DENVOY_TCLAP_INCLUDE_DIR:FILEPATH=/thirdparty/tclap-1.2.1/include \
 -DENVOY_JANSSON_INCLUDE_DIR:FILEPATH=/thirdparty_build/include \
 -DENVOY_OPENSSL_INCLUDE_DIR:FILEPATH=/thirdparty_build/include \
+-DENVOY_LIGHTSTEP_TRACER_INCLUDE_DIR:FILEPATH=/thirdparty_build/include \
 -DENVOY_PROTOBUF_INCLUDE_DIR:FILEPATH=/thirdparty_build/include \
 -DENVOY_PROTOBUF_PROTOC:FILEPATH=/thirdparty_build/bin/protoc \
 -DENVOY_GCOVR:FILEPATH=/thirdparty/gcovr-3.3/scripts/gcovr \

include/envoy/grpc/rpc_channel.h

@@ -2,6 +2,7 @@
 
 #include "envoy/common/optional.h"
 #include "envoy/common/pure.h"
+#include "envoy/http/async_client.h"
 #include "envoy/http/header_map.h"
 
 #include "google/protobuf/service.h"
@@ -69,4 +70,19 @@ class RpcChannelFactory {
                                const Optional<std::chrono::milliseconds>& timeout) PURE;
 };
 
+/**
+ * Interface for sending async gRPC requests.
+ */
+class RpcAsyncClient {
+public:
+  virtual ~RpcAsyncClient() {}
+
+  /**
+   * Send gRPC request to upstream cluster using method and callbacks.
+   */
+  virtual void send(const std::string& upstream_cluster, const std::string& service_full_name,
+                    const std::string& method_name, proto::Message&& grpc_request,
+                    Http::AsyncClient::Callbacks& callbacks) PURE;
+};
+
 } // Grpc

source/common/CMakeLists.txt

@@ -47,6 +47,7 @@ add_library(
   grpc/common.cc
   grpc/http1_bridge_filter.cc
   grpc/rpc_channel_impl.cc
+  grpc/utility.cc
   http/access_log/access_log_formatter.cc
   http/access_log/access_log_impl.cc
   http/async_client_impl.cc
@@ -113,6 +114,7 @@ endif()
 include_directories(${ENVOY_LIBEVENT_INCLUDE_DIR})
 include_directories(${ENVOY_NGHTTP2_INCLUDE_DIR})
 include_directories(SYSTEM ${ENVOY_OPENSSL_INCLUDE_DIR})
+include_directories(SYSTEM ${ENVOY_LIGHTSTEP_TRACER_INCLUDE_DIR})
 
 set_target_properties(envoy-common PROPERTIES COTIRE_CXX_PREFIX_HEADER_INIT
                       "../precompiled/precompiled.h")

source/common/common/utility.cc

@@ -102,3 +102,8 @@ bool StringUtil::startsWith(const std::string& source, const std::string& start,
     return strncasecmp(source.c_str(), start.c_str(), start.size()) == 0;
   }
 }
+
+const std::string& StringUtil::valueOrDefault(const std::string& input,
+                                              const std::string& default_value) {
+  return input.empty() ? default_value : input;
+}

source/common/common/utility.h

@@ -92,4 +92,10 @@ class StringUtil {
    */
   static bool startsWith(const std::string& source, const std::string& start,
                          bool case_sensitive = true);
+
+  /**
+   * @return original @param input string if it's not empty or @param default_value otherwise.
+   */
+  static const std::string& valueOrDefault(const std::string& input,
+                                           const std::string& default_value = "-");
 };

source/common/grpc/rpc_channel_impl.cc

@@ -1,5 +1,6 @@
 #include "common.h"
 #include "rpc_channel_impl.h"
+#include "utility.h"
 
 #include "common/common/enum_to_int.h"
 #include "common/common/utility.h"
@@ -30,15 +31,9 @@ void RpcChannelImpl::CallMethod(const proto::MethodDescriptor* method, proto::Rp
   // here for clarity.
   ASSERT(cm_.get(cluster_)->features() & Upstream::Cluster::Features::HTTP2);
 
-  Http::MessagePtr message(new Http::RequestMessageImpl());
-  message->headers().addViaMoveValue(Http::Headers::get().Scheme, "http");
-  message->headers().addViaMoveValue(Http::Headers::get().Method, "POST");
-  message->headers().addViaMoveValue(
-      Http::Headers::get().Path,
-      fmt::format("/{}/{}", method->service()->full_name(), method->name()));
-  message->headers().addViaCopy(Http::Headers::get().Host, cluster_);
-  message->headers().addViaCopy(Http::Headers::get().ContentType, Common::GRPC_CONTENT_TYPE);
-  message->body(serializeBody(*grpc_request));
+  Http::MessagePtr message =
+      Utility::prepareHeaders(cluster_, method->service()->full_name(), method->name());
+  message->body(Utility::serializeBody(*grpc_request));
 
   callbacks_.onPreRequestCustomizeHeaders(message->headers());
   http_request_ = cm_.httpAsyncClientForCluster(cluster_).send(std::move(message), *this, timeout_);
@@ -133,15 +128,23 @@ void RpcChannelImpl::onComplete() {
   grpc_response_ = nullptr;
 }
 
-Buffer::InstancePtr RpcChannelImpl::serializeBody(const proto::Message& message) {
-  // http://www.grpc.io/docs/guides/wire.html
-  Buffer::InstancePtr body(new Buffer::OwnedImpl());
-  uint8_t compressed = 0;
-  body->add(&compressed, sizeof(compressed));
-  uint32_t size = htonl(message.ByteSize());
-  body->add(&size, sizeof(size));
-  body->add(message.SerializeAsString());
-  return body;
+void RpcAsyncClientImpl::send(const std::string& upstream_cluster,
+                              const std::string& service_full_name, const std::string& method_name,
+                              proto::Message&& grpc_request,
+                              Http::AsyncClient::Callbacks& callbacks) {
+  // For proto3 messages this should always return true.
+  ASSERT(grpc_request->IsInitialized());
+  // This should be caught in configuration, and a request will fail normally anyway, but assert
+  // here for clarity.
+  ASSERT(cm_.get(cluster_)->features() & Upstream::Cluster::Features::HTTP2);
+
+  Http::MessagePtr message =
+      Utility::prepareHeaders(upstream_cluster, service_full_name, method_name);
+  message->body(Utility::serializeBody(grpc_request));
+
+  // TODO: pass timeout to send method.
+  cm_.httpAsyncClientForCluster(upstream_cluster)
+      .send(std::move(message), callbacks, std::chrono::milliseconds(5000));
 }
 
 } // Grpc

source/common/grpc/rpc_channel_impl.h

@@ -70,4 +70,16 @@ class RpcChannelImpl : public RpcChannel, public Http::AsyncClient::Callbacks {
   Optional<std::chrono::milliseconds> timeout_;
 };
 
+class RpcAsyncClientImpl : public RpcAsyncClient {
+public:
+  RpcAsyncClientImpl(Upstream::ClusterManager& cm) : cm_(cm) {}
+
+  void send(const std::string& upstream_cluster, const std::string& service_full_name,
+            const std::string& method_name, proto::Message&& grpc_request,
+            Http::AsyncClient::Callbacks& callbacks) override;
+
+private:
+  Upstream::ClusterManager& cm_;
+};
+
 } // Grpc

source/common/grpc/utility.cc

@@ -0,0 +1,34 @@
+#include "common.h"
+#include "utility.h"
+
+#include "common/http/headers.h"
+
+namespace Grpc {
+
+Buffer::InstancePtr Utility::serializeBody(const google::protobuf::Message& message) {
+  // http://www.grpc.io/docs/guides/wire.html
+  Buffer::InstancePtr body(new Buffer::OwnedImpl());
+  uint8_t compressed = 0;
+  body->add(&compressed, sizeof(compressed));
+  uint32_t size = htonl(message.ByteSize());
+  body->add(&size, sizeof(size));
+  body->add(message.SerializeAsString());
+
+  return body;
+}
+
+Http::MessagePtr Utility::prepareHeaders(const std::string& upstream_cluster,
+                                         const std::string& service_full_name,
+                                         const std::string& method_name) {
+  Http::MessagePtr message(new Http::RequestMessageImpl());
+  message->headers().addViaMoveValue(Http::Headers::get().Scheme, "http");
+  message->headers().addViaMoveValue(Http::Headers::get().Method, "POST");
+  message->headers().addViaMoveValue(Http::Headers::get().Path,
+                                     fmt::format("/{}/{}", service_full_name, method_name));
+  message->headers().addViaCopy(Http::Headers::get().Host, upstream_cluster);
+  message->headers().addViaCopy(Http::Headers::get().ContentType, Common::GRPC_CONTENT_TYPE);
+
+  return message;
+}
+
+} // Grpc

source/common/grpc/utility.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include "common/http/message_impl.h"
+
+#include "google/protobuf/message.h"
+
+namespace Grpc {
+
+class Utility {
+public:
+  static Buffer::InstancePtr serializeBody(const google::protobuf::Message& message);
+  static Http::MessagePtr prepareHeaders(const std::string& upstream_cluster,
+                                         const std::string& service_full_name,
+                                         const std::string& method_name);
+};
+
+} // Grpc