http: adding 100-Continue support to Envoy

RAW_RELEASE_NOTES.md

@@ -51,6 +51,7 @@ final version.
   :ref:`Squash <envoy_api_msg_filter.http.Squash>` allows debugging an incoming request to a microservice in the mesh.
 * lua: added headers replace() API.
 * Added support for direct responses -- i.e., sending a preconfigured HTTP response without proxying anywhere.
+* Added support for proxying 100-Continue responses.
 * Added DOWNSTREAM_LOCAL_ADDRESS, DOWNSTREAM_LOCAL_ADDRESS_WITHOUT_PORT header formatters, and
   DOWNSTREAM_LOCAL_ADDRESS access log formatter.
 * Added support for HTTPS redirects on specific routes.

include/envoy/http/codec.h

@@ -21,6 +21,12 @@ class StreamEncoder {
 public:
   virtual ~StreamEncoder() {}
 
+  /**
+   * Encode 100-Continue headers.
+   * @param headers supplies the 100-Continue header map to encode.
+   */
+  virtual void encode100ContinueHeaders(const HeaderMap& headers) PURE;
+
   /**
    * Encode headers, optionally indicating end of stream.
    * @param headers supplies the header map to encode.
@@ -54,6 +60,12 @@ class StreamDecoder {
 public:
   virtual ~StreamDecoder() {}
 
+  /**
+   * Called with decoded 100-Continue headers.
+   * @param headers supplies the decoded 100-Continue headers map that is moved into the callee.
+   */
+  virtual void decode100ContinueHeaders(HeaderMapPtr&& headers) PURE;
+
   /**
    * Called with decoded headers, optionally indicating end of stream.
    * @param headers supplies the decoded headers map that is moved into the callee.

include/envoy/http/filter.h

@@ -190,6 +190,17 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual void addDecodedData(Buffer::Instance& data, bool streaming_filter) PURE;
 
+  /**
+   * Called with 100-Continue headers to be encoded.
+   *
+   * This is not folded into encodeHeaders because most Envoy users and filters
+   * will not be proxying 100-continue and with it split out, can ignore the
+   * complexity of multiple encodeHeaders calls.
+   *
+   * @param headers supplies the headers to be encoded.
+   */
+  virtual void encode100ContinueHeaders(HeaderMapPtr&& headers) PURE;
+
   /**
    * Called with headers to be encoded, optionally indicating end of stream.
    *
@@ -401,6 +412,19 @@ class StreamEncoderFilterCallbacks : public virtual StreamFilterCallbacks {
  */
 class StreamEncoderFilter : public StreamFilterBase {
 public:
+  /*
+   * Called with 100-continue headers.
+   *
+   * This is not folded into encodeHeaders because most Envoy users and filters
+   * will not be proxying 100-continue and with it split out, can ignore the
+   * complexity of multiple encodeHeaders calls.
+   *
+   * @param headers supplies the 100-continue response headers to be encoded.
+   * @return FilterHeadersStatus determines how filter chain iteration proceeds.
+   *
+   */
+  virtual FilterHeadersStatus encode100ContinueHeaders(HeaderMap& headers) PURE;
+
   /**
    * Called with headers to be encoded, optionally indicating end of stream.
    * @param headers supplies the headers to be encoded.

source/common/dynamo/dynamo_filter.h

@@ -38,6 +38,9 @@ class DynamoFilter : public Http::StreamFilter {
   }
 
   // Http::StreamEncoderFilter
+  Http::FilterHeadersStatus encode100ContinueHeaders(Http::HeaderMap&) override {
+    return Http::FilterHeadersStatus::Continue;
+  }
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap&, bool) override;
   Http::FilterDataStatus encodeData(Buffer::Instance&, bool) override;
   Http::FilterTrailersStatus encodeTrailers(Http::HeaderMap&) override;

source/common/grpc/grpc_web_filter.h

@@ -34,6 +34,9 @@ class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
   }
 
   // Implements StreamEncoderFilter.
+  Http::FilterHeadersStatus encode100ContinueHeaders(Http::HeaderMap&) override {
+    return Http::FilterHeadersStatus::Continue;
+  }
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap&, bool) override;
   Http::FilterDataStatus encodeData(Buffer::Instance&, bool) override;
   Http::FilterTrailersStatus encodeTrailers(Http::HeaderMap& trailers) override;

source/common/grpc/http1_bridge_filter.h

@@ -31,6 +31,9 @@ class Http1BridgeFilter : public Http::StreamFilter {
   }
 
   // Http::StreamEncoderFilter
+  Http::FilterHeadersStatus encode100ContinueHeaders(Http::HeaderMap&) override {
+    return Http::FilterHeadersStatus::Continue;
+  }
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
   Http::FilterDataStatus encodeData(Buffer::Instance& data, bool end_stream) override;
   Http::FilterTrailersStatus encodeTrailers(Http::HeaderMap& trailers) override;

source/common/grpc/json_transcoder_filter.h

@@ -92,6 +92,9 @@ class JsonTranscoderFilter : public Http::StreamFilter, public Logger::Loggable<
   void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override;
 
   // Http::StreamEncoderFilter
+  Http::FilterHeadersStatus encode100ContinueHeaders(Http::HeaderMap&) override {
+    return Http::FilterHeadersStatus::Continue;
+  }
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
   Http::FilterDataStatus encodeData(Buffer::Instance& data, bool end_stream) override;
   Http::FilterTrailersStatus encodeTrailers(Http::HeaderMap& trailers) override;

source/common/http/async_client_impl.h

@@ -217,6 +217,9 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   void continueDecoding() override { NOT_IMPLEMENTED; }
   void addDecodedData(Buffer::Instance&, bool) override { NOT_IMPLEMENTED; }
   const Buffer::Instance* decodingBuffer() override { return buffered_body_.get(); }
+  // The async client won't pause if sending an Expect: 100-Continue so simply
+  // swallows any incoming encode100Continue.
+  void encode100ContinueHeaders(HeaderMapPtr&&) override {}
   void encodeHeaders(HeaderMapPtr&& headers, bool end_stream) override;
   void encodeData(Buffer::Instance& data, bool end_stream) override;
   void encodeTrailers(HeaderMapPtr&& trailers) override;

source/common/http/codec_wrappers.h

@@ -11,6 +11,10 @@ namespace Http {
 class StreamDecoderWrapper : public StreamDecoder {
 public:
   // StreamDecoder
+  void decode100ContinueHeaders(HeaderMapPtr&& headers) override {
+    inner_.decode100ContinueHeaders(std::move(headers));
+  }
+
   void decodeHeaders(HeaderMapPtr&& headers, bool end_stream) override {
     if (end_stream) {
       onPreDecodeComplete();
@@ -60,6 +64,10 @@ class StreamDecoderWrapper : public StreamDecoder {
 class StreamEncoderWrapper : public StreamEncoder {
 public:
   // StreamEncoder
+  void encode100ContinueHeaders(const HeaderMap& headers) override {
+    inner_.encode100ContinueHeaders(headers);
+  }
+
   void encodeHeaders(const HeaderMap& headers, bool end_stream) override {
     inner_.encodeHeaders(headers, end_stream);
     if (end_stream) {

source/common/http/conn_manager_impl.cc

@@ -64,6 +64,12 @@ ConnectionManagerImpl::ConnectionManagerImpl(ConnectionManagerConfig& config,
       runtime_(runtime), local_info_(local_info), cluster_manager_(cluster_manager),
       listener_stats_(config_.listenerStats()) {}
 
+const HeaderMapImpl& ConnectionManagerImpl::continueHeader() {
+  CONSTRUCT_ON_FIRST_USE(HeaderMapImpl,
+                         Http::HeaderMapImpl{{Http::Headers::get().Status,
+                                              std::to_string(enumToInt(Code::Continue))}});
+}
+
 void ConnectionManagerImpl::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {
   read_callbacks_ = &callbacks;
   stats_.named_.downstream_cx_total_.inc();
@@ -395,15 +401,18 @@ void ConnectionManagerImpl::ActiveStream::addAccessLogHandler(
   access_log_handlers_.push_back(handler);
 }
 
-void ConnectionManagerImpl::ActiveStream::chargeStats(HeaderMap& headers) {
+void ConnectionManagerImpl::ActiveStream::chargeStats(const HeaderMap& headers) {
   uint64_t response_code = Utility::getResponseStatus(headers);
   request_info_.response_code_.value(response_code);
 
   if (request_info_.hc_request_) {
     return;
   }
 
-  if (CodeUtility::is2xx(response_code)) {
+  if (CodeUtility::is1xx(response_code)) {
+    connection_manager_.stats_.named_.downstream_rq_1xx_.inc();
+    connection_manager_.listener_stats_.downstream_rq_1xx_.inc();
+  } else if (CodeUtility::is2xx(response_code)) {
     connection_manager_.stats_.named_.downstream_rq_2xx_.inc();
     connection_manager_.listener_stats_.downstream_rq_2xx_.inc();
   } else if (CodeUtility::is3xx(response_code)) {
@@ -446,6 +455,14 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
       this);
 #endif
 
+  if (!connection_manager_.config_.proxy100Continue() && request_headers_->Expect() &&
+      request_headers_->Expect()->value() == Headers::get().ExpectValues._100Continue.c_str()) {
+    // Note in the case Envoy is handling 100-Continue complexity, it skips the filter chain
+    // and sends the 100-Continue directly to the encoder.
+    chargeStats(continueHeader());
+    response_encoder_->encode100ContinueHeaders(continueHeader());
+  }
+
   connection_manager_.user_agent_.initializeFromHeaders(
       *request_headers_, connection_manager_.stats_.prefix_, connection_manager_.stats_.scope_);
 
@@ -773,6 +790,50 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
   cached_route_.value(std::move(route));
 }
 
+void ConnectionManagerImpl::ActiveStream::encode100ContinueHeaders(
+    ActiveStreamEncoderFilter* filter, HeaderMap& headers) {
+  // Make sure commonContinue continues encode100ContinueHeaders.
+  has_continue_headers_ = true;
+
+  // Similar to the block in encodeHeaders, run encode100ContinueHeaders on each
+  // filter. This is simpler than that case because 100 continue implies no
+  // end-stream, and because there are normal headers coming there's no need for
+  // complex continuation logic.
+  std::list<ActiveStreamEncoderFilterPtr>::iterator entry = commonEncodePrefix(filter, false);
+  for (; entry != encoder_filters_.end(); entry++) {
+    ASSERT(!(state_.filter_call_state_ & FilterCallState::Encode100ContinueHeaders));
+    state_.filter_call_state_ |= FilterCallState::Encode100ContinueHeaders;
+    FilterHeadersStatus status = (*entry)->handle_->encode100ContinueHeaders(headers);
+    state_.filter_call_state_ &= ~FilterCallState::Encode100ContinueHeaders;
+    ENVOY_STREAM_LOG(trace, "encode 100 continue headers called: filter={} status={}", *this,
+                     static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));
+    if (!(*entry)->commonHandleAfter100ContinueHeadersCallback(status)) {
+      return;
+    }
+  }
+
+  // Strip the T-E headers etc. Defer other header additions as well as drain-close logic to the
+  // continuation headers.
+  ConnectionManagerUtility::mutateResponseHeaders(headers, *request_headers_);
+
+  // Count both the 1xx and follow-up response code in stats.
+  chargeStats(headers);
+
+  ENVOY_STREAM_LOG(debug, "encoding 100 continue headers via codec", *this);
+#ifndef NVLOG
+  headers.iterate(
+      [](const HeaderEntry& header, void* context) -> HeaderMap::Iterate {
+        ENVOY_STREAM_LOG(debug, "  '{}':'{}'", *static_cast<ActiveStream*>(context),
+                         header.key().c_str(), header.value().c_str());
+        return HeaderMap::Iterate::Continue;
+      },
+      this);
+#endif
+
+  // Now actually encode via the codec.
+  response_encoder_->encode100ContinueHeaders(headers);
+}
+
 void ConnectionManagerImpl::ActiveStream::encodeHeaders(ActiveStreamEncoderFilter* filter,
                                                         HeaderMap& headers, bool end_stream) {
   std::list<ActiveStreamEncoderFilterPtr>::iterator entry = commonEncodePrefix(filter, end_stream);
@@ -1039,6 +1100,17 @@ void ConnectionManagerImpl::ActiveStreamFilterBase::commonContinue() {
   ASSERT(stopped_);
   stopped_ = false;
 
+  // Only resume with do100ContinueHeaders() if we've actually seen a 100-Continue.
+  if (parent_.has_continue_headers_ && !continue_headers_continued_) {
+    continue_headers_continued_ = true;
+    do100ContinueHeaders();
+    // If the response headers have not yet come in, don't continue on with
+    // headers and body. doHeaders expects request headers to exist.
+    if (!parent_.response_headers_.get()) {
+      return;
+    }
+  }
+
   // Make sure that we handle the zero byte data frame case. We make no effort to optimize this
   // case in terms of merging it into a header only request/response. This could be done in the
   // future.
@@ -1058,9 +1130,24 @@ void ConnectionManagerImpl::ActiveStreamFilterBase::commonContinue() {
   }
 }
 
-bool ConnectionManagerImpl::ActiveStreamFilterBase::commonHandleAfterHeadersCallback(
+bool ConnectionManagerImpl::ActiveStreamFilterBase::commonHandleAfter100ContinueHeadersCallback(
     FilterHeadersStatus status) {
+  ASSERT(parent_.has_continue_headers_);
+  ASSERT(!continue_headers_continued_);
+  ASSERT(!stopped_);
 
+  if (status == FilterHeadersStatus::StopIteration) {
+    stopped_ = true;
+    return false;
+  } else {
+    ASSERT(status == FilterHeadersStatus::Continue);
+    continue_headers_continued_ = true;
+    return true;
+  }
+}
+
+bool ConnectionManagerImpl::ActiveStreamFilterBase::commonHandleAfterHeadersCallback(
+    FilterHeadersStatus status) {
   ASSERT(!headers_continued_);
   ASSERT(!stopped_);
 
@@ -1182,6 +1269,12 @@ void ConnectionManagerImpl::ActiveStreamDecoderFilter::addDecodedData(Buffer::In
 
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::continueDecoding() { commonContinue(); }
 
+void ConnectionManagerImpl::ActiveStreamDecoderFilter::encode100ContinueHeaders(
+    HeaderMapPtr&& headers) {
+  parent_.continue_headers_ = std::move(headers);
+  parent_.encode100ContinueHeaders(nullptr, *parent_.continue_headers_);
+}
+
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::encodeHeaders(HeaderMapPtr&& headers,
                                                                      bool end_stream) {
   parent_.response_headers_ = std::move(headers);