thrift_proxy: support header flags

docs/root/version_history/current.rst

@@ -61,6 +61,7 @@ New Features
 * thrift_proxy: add upstream response zone metrics in the form ``cluster.cluster_name.zone.local_zone.upstream_zone.thrift.upstream_resp_success``.
 * thrift_proxy: add upstream metrics to show decoding errors and whether exception is from local or remote, e.g. ``cluster.cluster_name.thrift.upstream_resp_exception_remote``.
 * thrift_proxy: add host level success/error metrics where success is a reply of type success and error is any other response to a call.
+* thrift_proxy: support header flags.
 * thrift_proxy: support subset lb when using request or route metadata.
 * transport_socket: added :ref:`envoy.transport_sockets.tcp_stats <envoy_v3_api_msg_extensions.transport_sockets.tcp_stats.v3.Config>` which generates additional statistics gathered from the OS TCP stack.
 * udp: add support for multiple listener filters.

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

@@ -63,7 +63,9 @@ bool HeaderTransportImpl::decodeFrameStart(Buffer::Instance& buffer, MessageMeta
     throw EnvoyException(fmt::format("invalid thrift header transport magic {:04x}", magic));
   }
 
-  // offset 6: 16 bit flags field, unused
+  // offset 6: 16 bit flags field
+  int16_t header_flags = buffer.peekBEInt<int16_t>(6);
+
   // offset 8: 32 bit sequence number field
   int32_t seq_id = buffer.peekBEInt<int32_t>(8);
 
@@ -92,6 +94,7 @@ bool HeaderTransportImpl::decodeFrameStart(Buffer::Instance& buffer, MessageMeta
   // (header_size).
   metadata.setFrameSize(
       static_cast<uint32_t>(frame_size - header_size - MinFrameStartSizeNoHeaders));
+  metadata.setHeaderFlags(header_flags);
   metadata.setSequenceId(seq_id);
 
   ProtocolType proto = ProtocolType::Auto;
@@ -237,10 +240,14 @@ void HeaderTransportImpl::encodeFrame(Buffer::Instance& buffer, const MessageMet
   if (metadata.hasSequenceId()) {
     seq_id = metadata.sequenceId();
   }
+  int16_t header_flags = 0;
+  if (metadata.hasHeaderFlags()) {
+    header_flags = metadata.headerFlags();
+  }
 
   buffer.writeBEInt<uint32_t>(static_cast<uint32_t>(size));
   buffer.writeBEInt<uint16_t>(Magic);
-  buffer.writeBEInt<uint16_t>(0); // flags
+  buffer.writeBEInt<uint16_t>(header_flags); // flags
   buffer.writeBEInt<int32_t>(seq_id);
   buffer.writeBEInt<uint16_t>(static_cast<uint16_t>(header_size / 4));
 

source/extensions/filters/network/thrift_proxy/metadata.h

@@ -45,6 +45,10 @@ class MessageMetadata {
       copy->setMethodName(methodName());
     }
 
+    if (hasHeaderFlags()) {
+      copy->setHeaderFlags(headerFlags());
+    }
+
     if (hasSequenceId()) {
       copy->setSequenceId(sequenceId());
     }
@@ -111,6 +115,10 @@ class MessageMetadata {
   const std::string& methodName() const { return method_name_.value(); }
   void setMethodName(const std::string& method_name) { method_name_ = method_name; }
 
+  bool hasHeaderFlags() const { return header_flags_.has_value(); }
+  int16_t headerFlags() const { return header_flags_.value(); }
+  void setHeaderFlags(int16_t header_flags) { header_flags_ = header_flags; }
+
   bool hasSequenceId() const { return seq_id_.has_value(); }
   int32_t sequenceId() const { return seq_id_.value(); }
   void setSequenceId(int32_t seq_id) { seq_id_ = seq_id; }
@@ -174,6 +182,7 @@ class MessageMetadata {
   absl::optional<uint32_t> frame_size_{};
   absl::optional<ProtocolType> proto_{};
   absl::optional<std::string> method_name_{};
+  absl::optional<int16_t> header_flags_{};
   absl::optional<int32_t> seq_id_{};
   absl::optional<MessageType> msg_type_{};
   absl::optional<ReplyType> reply_type_{};

test/extensions/filters/network/thrift_proxy/header_transport_impl_test.cc

@@ -243,13 +243,14 @@ TEST(HeaderTransportTest, NoTransformsOrInfo) {
 
     buffer.writeBEInt<int32_t>(100);
     buffer.writeBEInt<int16_t>(0x0FFF);
-    buffer.writeBEInt<int16_t>(0);
+    buffer.writeBEInt<int16_t>(1); // header flags
     buffer.writeBEInt<int32_t>(1); // sequence number
     buffer.writeBEInt<int16_t>(1); // size 4
     addSeq(buffer, {0, 0, 0, 0});  // 0 = binary proto, 0 = num transforms, pad, pad
     EXPECT_TRUE(transport.decodeFrameStart(buffer, metadata));
     EXPECT_THAT(metadata, HasFrameSize(86U));
     EXPECT_THAT(metadata, HasProtocol(ProtocolType::Binary));
+    EXPECT_THAT(metadata, HasHeaderFlags(1));
     EXPECT_THAT(metadata, HasSequenceId(1));
     EXPECT_THAT(metadata, HasNoHeaders());
     EXPECT_EQ(buffer.length(), 0);
@@ -261,13 +262,14 @@ TEST(HeaderTransportTest, NoTransformsOrInfo) {
 
     buffer.writeBEInt<int32_t>(101);
     buffer.writeBEInt<int16_t>(0x0FFF);
-    buffer.writeBEInt<int16_t>(0);
+    buffer.writeBEInt<int16_t>(2); // header flags
     buffer.writeBEInt<int32_t>(2); // sequence number
     buffer.writeBEInt<int16_t>(1); // size 4
     addSeq(buffer, {2, 0, 0, 0});  // 2 = compact proto, 0 = num transforms, pad, pad
     EXPECT_TRUE(transport.decodeFrameStart(buffer, metadata));
     EXPECT_THAT(metadata, HasFrameSize(87U));
     EXPECT_THAT(metadata, HasProtocol(ProtocolType::Compact));
+    EXPECT_THAT(metadata, HasHeaderFlags(2));
     EXPECT_THAT(metadata, HasSequenceId(2));
     EXPECT_THAT(metadata, HasNoHeaders());
   }
@@ -341,7 +343,7 @@ TEST(HeaderTransportTest, InvalidInfoBlock) {
 
     buffer.writeBEInt<int32_t>(100);
     buffer.writeBEInt<int16_t>(0x0FFF);
-    buffer.writeBEInt<int16_t>(0);
+    buffer.writeBEInt<int16_t>(1); // header flags
     buffer.writeBEInt<int32_t>(1); // sequence number
     buffer.writeBEInt<int16_t>(1); // size 4
     addSeq(buffer, {0, 0, 2, 0});  // 0 = binary proto, 0 = num transforms, 2 = unknown info id, pad
@@ -350,6 +352,7 @@ TEST(HeaderTransportTest, InvalidInfoBlock) {
     EXPECT_TRUE(transport.decodeFrameStart(buffer, metadata));
     EXPECT_THAT(metadata, HasFrameSize(86U));
     EXPECT_THAT(metadata, HasProtocol(ProtocolType::Binary));
+    EXPECT_THAT(metadata, HasHeaderFlags(1));
     EXPECT_THAT(metadata, HasSequenceId(1));
     EXPECT_THAT(metadata, HasNoHeaders());
     EXPECT_EQ(buffer.length(), 0);

test/extensions/filters/network/thrift_proxy/metadata_test.cc

@@ -37,6 +37,12 @@ TEST(MessageMetadataTest, Fields) {
   EXPECT_TRUE(metadata.hasMessageType());
   EXPECT_EQ(MessageType::Call, metadata.messageType());
 
+  EXPECT_FALSE(metadata.hasHeaderFlags());
+  EXPECT_THROW(metadata.headerFlags(), absl::bad_optional_access);
+  metadata.setHeaderFlags(11);
+  EXPECT_TRUE(metadata.hasHeaderFlags());
+  EXPECT_EQ(11, metadata.headerFlags());
+
   EXPECT_FALSE(metadata.hasSequenceId());
   EXPECT_THROW(metadata.sequenceId(), absl::bad_optional_access);
   metadata.setSequenceId(101);

test/extensions/filters/network/thrift_proxy/utility.h

@@ -149,6 +149,7 @@ MATCHER_P(HasFrameSize, n, "") {
 
 MATCHER_P(HasProtocol, p, "") { return arg.hasProtocol() && arg.protocol() == p; }
 MATCHER_P(HasSequenceId, id, "") { return arg.hasSequenceId() && arg.sequenceId() == id; }
+MATCHER_P(HasHeaderFlags, flags, "") { return arg.hasHeaderFlags() && arg.headerFlags() == flags; }
 MATCHER(HasNoHeaders, "") { return arg.headers().size() == 0; }
 
 MATCHER_P2(HasAppException, t, m, "") {