Support gRPC traffic

include/proxy/http/HttpSM.h

@@ -446,6 +446,7 @@ class HttpSM : public Continuation, public PluginUserArgs<TS_USER_ARGS_TXN>
   void perform_cache_write_action();
   void perform_transform_cache_write_action();
   void setup_blind_tunnel(bool send_response_hdr, IOBufferReader *initial = nullptr);
+  void setup_tunnel_handler_trailer(HttpTunnelProducer *p);
   HttpTunnelProducer *setup_server_transfer_to_transform();
   HttpTunnelProducer *setup_transfer_from_transform();
   HttpTunnelProducer *setup_cache_transfer_to_transform();

include/proxy/http2/Http2Stream.h

@@ -82,7 +82,7 @@ class Http2Stream : public ProxyTransaction
   void set_expect_receive_trailer() override;
 
   Http2ErrorCode decode_header_blocks(HpackHandle &hpack_handle, uint32_t maximum_table_size);
-  void send_request(Http2ConnectionState &cstate);
+  void send_headers(Http2ConnectionState &cstate);
   void initiating_close();
   bool is_outbound_connection() const;
   bool is_tunneling() const;
@@ -217,6 +217,12 @@ class Http2Stream : public ProxyTransaction
   History<HISTORY_DEFAULT_SIZE> _history;
   Milestones<Http2StreamMilestone, static_cast<size_t>(Http2StreamMilestone::LAST_ENTRY)> _milestones;
 
+  /** Any headers received while this is true are trailing headers.
+   *
+   * This is set to true when processing DATA frames are done. Therefore any
+   * headers seen after that point are trailing headers. The qualification
+   * "possible" is added because the peer may or may not send trailing headers.
+   */
   bool _trailing_header_is_possible = false;
   bool _expect_send_trailer         = false;
   bool _expect_receive_trailer      = false;
@@ -373,7 +379,7 @@ Http2Stream::reset_send_headers()
   this->_send_header.create(HTTP_TYPE_RESPONSE);
 }
 
-// Check entire DATA payload length if content-length: header is exist
+// Check entire DATA payload length if content-length: header exists
 inline void
 Http2Stream::increment_data_length(uint64_t length)
 {

include/tscore/ink_string++.h

@@ -32,6 +32,7 @@
 
 #pragma once
 #include <cstdio>
+#include <cstring>
 #include <strings.h>
 
 /***********************************************************************

src/proxy/hdrs/HdrToken.cc

@@ -230,9 +230,6 @@ static HdrTokenFieldInfo _hdrtoken_strs_field_initializers[] = {
   {"Strict-Transport-Security", MIME_SLOTID_NONE,                MIME_PRESENCE_NONE,                (HTIF_MULTVALS)                              },
   {"Subject",                   MIME_SLOTID_NONE,                MIME_PRESENCE_SUBJECT,             HTIF_NONE                                    },
   {"Summary",                   MIME_SLOTID_NONE,                MIME_PRESENCE_SUMMARY,             HTIF_NONE                                    },
- // TODO: In the past we have observed issues with having hop-by-hop in here
-  // for gRPC. We plan to work on gRPC in a future. We should experiment with
-  // this and verify that it works as expected.
   {"TE",                        MIME_SLOTID_TE,                  MIME_PRESENCE_TE,                  (HTIF_COMMAS | HTIF_MULTVALS | HTIF_HOPBYHOP)},
   {"Transfer-Encoding",         MIME_SLOTID_TRANSFER_ENCODING,   MIME_PRESENCE_TRANSFER_ENCODING,
    (HTIF_COMMAS | HTIF_MULTVALS | HTIF_HOPBYHOP)                                                                                                 },

src/proxy/http/HttpSM.cc

@@ -2780,6 +2780,23 @@ HttpSM::tunnel_handler_post(int event, void *data)
   return 0;
 }
 
+void
+HttpSM::setup_tunnel_handler_trailer(HttpTunnelProducer *p)
+{
+  p->read_success               = true;
+  t_state.current.server->state = HttpTransact::TRANSACTION_COMPLETE;
+  t_state.current.server->abort = HttpTransact::DIDNOT_ABORT;
+
+  SMDbg(dbg_ctl_http, "Wait for the trailing header");
+
+  // Swap out the default hander to set up the new tunnel for the trailer exchange.
+  HTTP_SM_SET_DEFAULT_HANDLER(&HttpSM::tunnel_handler_trailer);
+  if (_ua.get_txn()) {
+    _ua.get_txn()->set_expect_send_trailer();
+  }
+  tunnel.local_finish_all(p);
+}
+
 int
 HttpSM::tunnel_handler_trailer(int event, void *data)
 {
@@ -3085,6 +3102,10 @@ HttpSM::tunnel_handler_server(int event, HttpTunnelProducer *p)
         tunnel.append_message_to_producer_buffer(p, reason, reason_len);
         }
       */
+      if (server_txn->expect_receive_trailer()) {
+        setup_tunnel_handler_trailer(p);
+        return 0;
+      }
       tunnel.local_finish_all(p);
     }
     break;
@@ -3111,10 +3132,7 @@ HttpSM::tunnel_handler_server(int event, HttpTunnelProducer *p)
       }
     }
     if (server_txn->expect_receive_trailer()) {
-      SMDbg(dbg_ctl_http, "wait for that trailing header");
-      // Swap out the default hander to set up the new tunnel for the trailer exchange.
-      HTTP_SM_SET_DEFAULT_HANDLER(&HttpSM::tunnel_handler_trailer);
-      tunnel.local_finish_all(p);
+      setup_tunnel_handler_trailer(p);
       return 0;
     }
     break;

src/proxy/http/HttpTransactHeaders.cc

@@ -33,6 +33,7 @@
 #include "proxy/hdrs/HTTP.h"
 #include "proxy/hdrs/HdrUtils.h"
 #include "proxy/hdrs/HttpCompat.h"
+#include "proxy/hdrs/MIME.h"
 #include "proxy/http/HttpSM.h"
 #include "proxy/PoolableSession.h"
 
@@ -222,7 +223,9 @@ HttpTransactHeaders::copy_header_fields(HTTPHdr *src_hdr, HTTPHdr *new_hdr, bool
   //         my opinion error prone and if the client doesn't follow the spec
   //         we'll have problems with the TE being forwarded to the server
   //         and us caching the transfer encoded documents and then
-  //         serving it to a client that can not handle it
+  //         serving it to a client that can not handle it. The exception
+  //         to this is that we will allow "TE: trailers" to be forwarded
+  //         because that is required for gRPC traffic.
   //      2) Transfer encoding is copied.  If the transfer encoding
   //         is changed for example by dechunking, the transfer encoding
   //         should be modified when the decision is made to dechunk it
@@ -235,10 +238,19 @@ HttpTransactHeaders::copy_header_fields(HTTPHdr *src_hdr, HTTPHdr *new_hdr, bool
     int field_flags = hdrtoken_index_to_flags(field.m_wks_idx);
 
     if (field_flags & HTIF_HOPBYHOP) {
+      std::string_view name(field.name_get());
+      std::string_view value(field.value_get());
+      bool const is_te_trailers = name == MIME_FIELD_TE && value == "trailers";
+      if (is_te_trailers) {
+        // te: trailers is used by gRPC, do not delete it.
+        continue;
+      }
+
       // Delete header if not in special proxy_auth retention mode
-      if ((!retain_proxy_auth_hdrs) || (!(field_flags & HTIF_PROXYAUTH))) {
-        new_hdr->field_delete(&field);
+      if (retain_proxy_auth_hdrs && (field_flags & HTIF_PROXYAUTH)) {
+        continue;
       }
+      new_hdr->field_delete(&field);
     } else if (field.m_wks_idx == MIME_WKSIDX_DATE) {
       date_hdr = true;
     }

src/proxy/http2/Http2ConnectionState.cc

@@ -475,16 +475,16 @@ Http2ConnectionState::rcv_headers_frame(const Http2Frame &frame)
       stream->mark_milestone(Http2StreamMilestone::START_TXN);
       stream->new_transaction(frame.is_from_early_data());
       // Send request header to SM
-      stream->send_request(*this);
+      stream->send_headers(*this);
     } else {
       // If this is a trailer, first signal to the SM that the body is done
       if (stream->trailing_header_is_possible()) {
         stream->set_expect_receive_trailer();
         // Propagate the  trailer header
-        stream->send_request(*this);
+        stream->send_headers(*this);
       } else {
         // Propagate the response
-        stream->send_request(*this);
+        stream->send_headers(*this);
       }
     }
     // Give a chance to send response before reading next frame.
@@ -1061,7 +1061,7 @@ Http2ConnectionState::rcv_continuation_frame(const Http2Frame &frame)
     // "from_early_data" flag from the associated HEADERS frame.
     stream->new_transaction(frame.is_from_early_data());
     // Send request header to SM
-    stream->send_request(*this);
+    stream->send_headers(*this);
     // Give a chance to send response before reading next frame.
     this->session->interrupt_reading_frames();
   } else {
@@ -2132,15 +2132,18 @@ Http2ConnectionState::send_a_data_frame(Http2Stream *stream, size_t &payload_len
   stream->update_sent_count(payload_length);
 
   // Are we at the end?
+  // We have no payload to send but might expect data from either trailer or body
+  // TODO(KS): does the expect send trailer and empty payload need a flush, or does it
+  //           warrant a separate flow with NO_ERROR?
   // If we return here, we never send the END_STREAM in the case of a early terminating OS.
   // OK if there is no body yet. Otherwise continue on to send a DATA frame and delete the stream
-  if (!stream->is_write_vio_done() && payload_length == 0) {
+  if ((!stream->is_write_vio_done() || stream->expect_send_trailer()) && payload_length == 0) {
     Http2StreamDebug(this->session, stream->get_id(), "No payload");
     this->session->flush();
     return Http2SendDataFrameResult::NO_PAYLOAD;
   }
 
-  if (stream->is_write_vio_done()) {
+  if (stream->is_write_vio_done() && !resp_reader->is_read_avail_more_than(payload_length) && !stream->expect_send_trailer()) {
     Http2StreamDebug(this->session, stream->get_id(), "End of Data Frame");
     flags |= HTTP2_FLAGS_DATA_END_STREAM;
   }
@@ -2430,7 +2433,7 @@ Http2ConnectionState::send_push_promise_frame(Http2Stream *stream, URL &url, con
   stream->set_receive_headers(hdr);
   stream->new_transaction();
   stream->receive_end_stream = true; // No more data with the request
-  stream->send_request(*this);
+  stream->send_headers(*this);
 
   return true;
 }

src/proxy/http2/Http2Stream.cc

@@ -29,6 +29,7 @@
 #include "proxy/http/HttpDebugNames.h"
 #include "proxy/http/HttpSM.h"
 #include "tscore/HTTPVersion.h"
+#include "tscore/ink_assert.h"
 
 #include <numeric>
 
@@ -259,45 +260,47 @@ Http2Stream::decode_header_blocks(HpackHandle &hpack_handle, uint32_t maximum_ta
 }
 
 void
-Http2Stream::send_request(Http2ConnectionState &cstate)
+Http2Stream::send_headers(Http2ConnectionState &cstate)
 {
   if (closed) {
     return;
   }
   REMEMBER(NO_EVENT, this->reentrancy_count);
 
-  // Convert header to HTTP/1.1 format
-  if (http2_convert_header_from_2_to_1_1(&_receive_header) == PARSE_RESULT_ERROR) {
-    Http2StreamDebug("Error converting HTTP/2 headers to HTTP/1.1.");
-    if (_receive_header.type_get() == HTTP_TYPE_REQUEST) {
-      // There's no way to cause Bad Request directly at this time.
-      // Set an invalid method so it causes an error later.
-      _receive_header.method_set("\xffVOID", 1);
+  // Convert header to HTTP/1.1 format. Trailing headers need no conversion
+  // because they, by definition, do not contain pseudo headers.
+  if (this->trailing_header_is_possible()) {
+    Http2StreamDebug("trailing header: Skipping send_headers initialization.");
+  } else {
+    if (http2_convert_header_from_2_to_1_1(&_receive_header) == PARSE_RESULT_ERROR) {
+      Http2StreamDebug("Error converting HTTP/2 headers to HTTP/1.1.");
+      if (_receive_header.type_get() == HTTP_TYPE_REQUEST) {
+        // There's no way to cause Bad Request directly at this time.
+        // Set an invalid method so it causes an error later.
+        _receive_header.method_set("\xffVOID", 1);
+      }
     }
-  }
 
-  if (_receive_header.type_get() == HTTP_TYPE_REQUEST) {
-    // Check whether the request uses CONNECT method
-    int method_len;
-    const char *method = _receive_header.method_get(&method_len);
-    if (method_len == HTTP_LEN_CONNECT && strncmp(method, HTTP_METHOD_CONNECT, HTTP_LEN_CONNECT) == 0) {
-      this->_is_tunneling = true;
+    if (_receive_header.type_get() == HTTP_TYPE_REQUEST) {
+      // Check whether the request uses CONNECT method
+      int method_len;
+      const char *method = _receive_header.method_get(&method_len);
+      if (method_len == HTTP_LEN_CONNECT && strncmp(method, HTTP_METHOD_CONNECT, HTTP_LEN_CONNECT) == 0) {
+        this->_is_tunneling = true;
+      }
     }
+    ink_release_assert(this->_sm != nullptr);
+    this->_http_sm_id = this->_sm->sm_id;
   }
 
-  if (this->expect_send_trailer()) {
-    // Send read complete to terminate previous data tunnel
-    this->read_vio.nbytes = this->read_vio.ndone;
-    this->signal_read_event(VC_EVENT_READ_COMPLETE);
-  }
-
-  ink_release_assert(this->_sm != nullptr);
-  this->_http_sm_id = this->_sm->sm_id;
-
   // Write header to a buffer.  Borrowing logic from HttpSM::write_header_into_buffer.
   // Seems like a function like this ought to be in HTTPHdr directly
   int bufindex;
   int dumpoffset = 0;
+  // The name dumpoffset is used here for parity with
+  // HttpSM::write_header_into_buffer, but create an alias for clarity in the
+  // use of this variable below this loop.
+  int &num_header_bytes = dumpoffset;
   int done, tmp;
   do {
     bufindex             = 0;
@@ -315,19 +318,43 @@ Http2Stream::send_request(Http2ConnectionState &cstate)
     }
   } while (!done);
 
-  if (bufindex == 0) {
+  if (num_header_bytes == 0) {
     // No data to signal read event
     return;
   }
 
   // Is the _sm ready to process the header?
   if (this->read_vio.nbytes > 0) {
     if (this->receive_end_stream) {
-      this->read_vio.nbytes = bufindex;
-      this->read_vio.ndone  = bufindex;
+      // These headers may be standard or trailer headers:
+      //
+      // * If they are standard, then there is no body (note again that the
+      // END_STREAM flag was sent with them), data_length will be 0, and
+      // num_header_bytes will simply be the length of the headers.
+      //
+      // * If they are trailers, then the tunnel behind the SM was set up after
+      // the original headers were sent, and thus nbytes should not include the
+      // size of the original standard headers. Rather, for trailers, nbytes
+      // only needs to include the body length (i.e., DATA frame payload
+      // length), and the length of these current trailer headers calculated in
+      // num_header_bytes.
+      this->read_vio.nbytes = this->data_length + num_header_bytes;
+      Http2StreamDebug("nbytes: %" PRId64 ", ndone: %" PRId64 ", num_header_bytes: %d, data_length: %" PRId64,
+                       this->read_vio.nbytes, this->read_vio.ndone, num_header_bytes, this->data_length);
       if (this->is_outbound_connection()) {
+        // This is a response header.
+        // We don't set ndone because the VC_EVENT_EOS will
+        // first flush the remaining content to consumers,
+        // after which the TUNNEL_EVENT_DONE will be fired
+        // and the header handler will be set up.
+        // The header handler will read the buffer, and not
+        // get its content from the VIO
+        // This can break if the implementation
+        // changes.
         this->signal_read_event(VC_EVENT_EOS);
       } else {
+        // Request headers.
+        this->read_vio.ndone = this->read_vio.nbytes;
         this->signal_read_event(VC_EVENT_READ_COMPLETE);
       }
     } else {

tests/Pipfile

@@ -49,5 +49,9 @@ jsonschema = "*"
 python-jose = "*"
 pyyaml ="*"
 
+# For the grpc tests.
+grpcio = "*"
+grpcio-tools = "*"
+
 [requires]
 python_version = "3"