thrift_proxy: introduce header transport

api/envoy/config/filter/network/thrift_proxy/v2alpha1/thrift_proxy.proto

@@ -23,6 +23,9 @@ message ThriftProxy {
 
     // The Thrift proxy will assume the client is using the Thrift unframed transport.
     UNFRAMED = 2;
+
+    // The Thrift proxy will assume the client is using the Thrift header transport.
+    HEADER = 3;
   }
 
   // Supplies the type of transport that the Thrift proxy should use. Defaults to `AUTO_TRANSPORT`.

source/extensions/filters/network/thrift_proxy/BUILD

@@ -187,11 +187,13 @@ envoy_cc_library(
     name = "transport_lib",
     srcs = [
         "framed_transport_impl.cc",
+        "header_transport_impl.cc",
         "transport_impl.cc",
         "unframed_transport_impl.cc",
     ],
     hdrs = [
         "framed_transport_impl.h",
+        "header_transport_impl.h",
         "transport_impl.h",
         "unframed_transport_impl.h",
     ],

source/extensions/filters/network/thrift_proxy/config.cc

@@ -41,6 +41,8 @@ static const TransportTypeMap& transportTypeMap() {
           {envoy::config::filter::network::thrift_proxy::v2alpha1::
                ThriftProxy_TransportType_UNFRAMED,
            TransportType::Unframed},
+          {envoy::config::filter::network::thrift_proxy::v2alpha1::ThriftProxy_TransportType_HEADER,
+           TransportType::Header},
       });
 }
 
@@ -92,6 +94,20 @@ ConfigImpl::ConfigImpl(
       transport_(config.transport()), proto_(config.protocol()),
       route_matcher_(new Router::RouteMatcher(config.route_config())) {
 
+  if (transportTypeMap().find(transport_) == transportTypeMap().end()) {
+    throw EnvoyException(fmt::format(
+        "unknown transport {}",
+        envoy::config::filter::network::thrift_proxy::v2alpha1::ThriftProxy_TransportType_Name(
+            transport_)));
+  }
+
+  if (protocolTypeMap().find(proto_) == protocolTypeMap().end()) {
+    throw EnvoyException(fmt::format(
+        "unknown protocol {}",
+        envoy::config::filter::network::thrift_proxy::v2alpha1::ThriftProxy_ProtocolType_Name(
+            proto_)));
+  }
+
   // Construct the only Thrift DecoderFilter: the Router
   auto& factory =
       Envoy::Config::Utility::getAndCheckFactory<ThriftFilters::NamedThriftFilterConfigFactory>(
@@ -115,14 +131,14 @@ DecoderPtr ConfigImpl::createDecoder(DecoderCallbacks& callbacks) {
 
 TransportPtr ConfigImpl::createTransport() {
   TransportTypeMap::const_iterator i = transportTypeMap().find(transport_);
-  RELEASE_ASSERT(i != transportTypeMap().end(), "invalid transport type");
+  ASSERT(i != transportTypeMap().end());
 
   return NamedTransportConfigFactory::getFactory(i->second).createTransport();
 }
 
 ProtocolPtr ConfigImpl::createProtocol() {
   ProtocolTypeMap::const_iterator i = protocolTypeMap().find(proto_);
-  RELEASE_ASSERT(i != protocolTypeMap().end(), "invalid protocol type");
+  ASSERT(i != protocolTypeMap().end());
   return NamedProtocolConfigFactory::getFactory(i->second).createProtocol();
 }
 

source/extensions/filters/network/thrift_proxy/decoder.cc

@@ -385,6 +385,25 @@ ThriftFilters::FilterStatus Decoder::onData(Buffer::Instance& data, bool& buffer
     }
     ENVOY_LOG(debug, "thrift: {} transport started", transport_->name());
 
+    if (metadata_->hasProtocol()) {
+      if (protocol_->type() == ProtocolType::Auto) {
+        protocol_->setType(metadata_->protocol());
+        ENVOY_LOG(debug, "thrift: {} transport forced {} protocol", transport_->name(),
+                  protocol_->name());
+      } else if (metadata_->protocol() != protocol_->type()) {
+        throw EnvoyException(fmt::format("transport reports protocol {}, but configured for {}",
+                                         ProtocolNames::get().fromType(metadata_->protocol()),
+                                         ProtocolNames::get().fromType(protocol_->type())));
+      }
+    }
+    if (metadata_->hasAppException()) {
+      AppExceptionType ex_type = metadata_->appExceptionType();
+      std::string ex_msg = metadata_->appExceptionMessage();
+      // Force new metadata if we get called again.
+      metadata_.reset();
+      throw AppException(ex_type, ex_msg);
+    }
+
     request_ = std::make_unique<ActiveRequest>(callbacks_.newDecoderFilter());
     frame_started_ = true;
     state_machine_ =