envoyproxy · mattklein123 · Oct 23, 2019 · Oct 7, 2019 · Oct 7, 2019 · Oct 7, 2019
diff --git a/api/envoy/config/filter/accesslog/v2/accesslog.proto b/api/envoy/config/filter/accesslog/v2/accesslog.proto
@@ -200,6 +200,7 @@ message ResponseFlagFilter {
         in: "URX"
         in: "SI"
         in: "IH"
+        in: "DPE"
       }
     }
   }];

diff --git a/api/envoy/config/filter/accesslog/v3alpha/accesslog.proto b/api/envoy/config/filter/accesslog/v3alpha/accesslog.proto
@@ -200,6 +200,7 @@ message ResponseFlagFilter {
         in: "URX"
         in: "SI"
         in: "IH"
+        in: "DPE"
       }
     }
   }];

diff --git a/api/envoy/data/accesslog/v2/accesslog.proto b/api/envoy/data/accesslog/v2/accesslog.proto
@@ -165,7 +165,7 @@ message AccessLogCommon {
 }
 
 // Flags indicating occurrences during request/response processing.
-// [#next-free-field: 19]
+// [#next-free-field: 20]
 message ResponseFlags {
   message Unauthorized {
     // Reasons why the request was unauthorized
@@ -233,6 +233,9 @@ message ResponseFlags {
   // Indicates that the request was rejected because an envoy request header failed strict
   // validation.
   bool invalid_envoy_request_headers = 18;
+
+  // Indicates there was an HTTP protocol error on the downstream request.
+  bool downstream_protocol_error = 19;
 }
 
 // Properties of a negotiated TLS connection.

diff --git a/api/envoy/data/accesslog/v3alpha/accesslog.proto b/api/envoy/data/accesslog/v3alpha/accesslog.proto
@@ -165,7 +165,7 @@ message AccessLogCommon {
 }
 
 // Flags indicating occurrences during request/response processing.
-// [#next-free-field: 19]
+// [#next-free-field: 20]
 message ResponseFlags {
   message Unauthorized {
     // Reasons why the request was unauthorized
@@ -233,6 +233,9 @@ message ResponseFlags {
   // Indicates that the request was rejected because an envoy request header failed strict
   // validation.
   bool invalid_envoy_request_headers = 18;
+
+  // Indicates there was an HTTP protocol error on the downstream request.
+  bool downstream_protocol_error = 19;
 }
 
 // Properties of a negotiated TLS connection.

diff --git a/docs/root/configuration/observability/access_log.rst b/docs/root/configuration/observability/access_log.rst
@@ -221,6 +221,7 @@ The following command operators are supported:
     * **IH**: The request was rejected because it set an invalid value for a
       :ref:`strictly-checked header <envoy_api_field_config.filter.http.router.v2.Router.strict_check_headers>` in addition to 400 response code.
     * **SI**: Stream idle timeout in addition to 408 response code.
+    * **DPE**: The downstream request had an HTTP protocol error.
 
 %RESPONSE_TX_DURATION%
   HTTP

diff --git a/docs/root/intro/version_history.rst b/docs/root/intro/version_history.rst
@@ -3,6 +3,7 @@ Version history
 
 1.12.0 (pending)
 ================
+* access log: added a new flag for :ref:`downstream protocol error <envoy_api_field_data.accesslog.v2.ResponseFlags.downstream_protocol_error>`.
 * access log: added :ref:`buffering <envoy_api_field_config.accesslog.v2.CommonGrpcAccessLogConfig.buffer_size_bytes>` and :ref:`periodical flushing <envoy_api_field_config.accesslog.v2.CommonGrpcAccessLogConfig.buffer_flush_interval>` support to gRPC access logger. Defaults to 16KB buffer and flushing every 1 second.
 * access log: added DOWNSTREAM_DIRECT_REMOTE_ADDRESS and DOWNSTREAM_DIRECT_REMOTE_ADDRESS_WITHOUT_PORT :ref:`access log formatters <config_access_log_format>` and gRPC access logger.
 * access log: gRPC Access Log Service (ALS) support added for :ref:`TCP access logs <envoy_api_msg_config.accesslog.v2.TcpGrpcAccessLogConfig>`.

diff --git a/include/envoy/stream_info/stream_info.h b/include/envoy/stream_info/stream_info.h
@@ -63,8 +63,10 @@ enum ResponseFlag {
   StreamIdleTimeout = 0x10000,
   // Request specified x-envoy-* header values that failed strict header checks.
   InvalidEnvoyRequestHeaders = 0x20000,
+  // Downstream request had an HTTP protocol error
+  DownstreamProtocolError = 0x40000,
   // ATTENTION: MAKE SURE THIS REMAINS EQUAL TO THE LAST FLAG.
-  LastFlag = InvalidEnvoyRequestHeaders
+  LastFlag = DownstreamProtocolError
 };
 
 /**

diff --git a/source/common/http/conn_manager_impl.cc b/source/common/http/conn_manager_impl.cc
@@ -270,7 +270,7 @@ void ConnectionManagerImpl::handleCodecException(const char* error) {
 
   // In the protocol error case, we need to reset all streams now. The connection might stick around
   // long enough for a pending stream to come back and try to encode.
-  resetAllStreams();
+  resetAllStreams(StreamInfo::ResponseFlag::DownstreamProtocolError);
 
   // HTTP/1.1 codec has already sent a 400 response if possible. HTTP/2 codec has already sent
   // GOAWAY.
@@ -359,7 +359,8 @@ Network::FilterStatus ConnectionManagerImpl::onNewConnection() {
   return Network::FilterStatus::StopIteration;
 }
 
-void ConnectionManagerImpl::resetAllStreams() {
+void ConnectionManagerImpl::resetAllStreams(
+    absl::optional<StreamInfo::ResponseFlag> response_flag) {
   while (!streams_.empty()) {
     // Mimic a downstream reset in this case. We must also remove callbacks here. Though we are
     // about to close the connection and will disable further reads, it is possible that flushing
@@ -372,6 +373,9 @@ void ConnectionManagerImpl::resetAllStreams() {
     auto& stream = *streams_.front();
     stream.response_encoder_->getStream().removeCallbacks(stream);
     stream.onResetStream(StreamResetReason::ConnectionTermination, absl::string_view());
+    if (response_flag.has_value()) {
+      stream.stream_info_.setResponseFlag(response_flag.value());
+    }
   }
 }
 
@@ -407,7 +411,7 @@ void ConnectionManagerImpl::onEvent(Network::ConnectionEvent event) {
 
     stats_.named_.downstream_cx_destroy_active_rq_.inc();
     user_agent_.onConnectionDestroy(event, true);
-    resetAllStreams();
+    resetAllStreams(absl::nullopt);
   }
 }
 

diff --git a/source/common/http/conn_manager_impl.h b/source/common/http/conn_manager_impl.h
@@ -653,7 +653,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
    */
   void doEndStream(ActiveStream& stream);
 
-  void resetAllStreams();
+  void resetAllStreams(absl::optional<StreamInfo::ResponseFlag> response_flag);
   void onIdleTimeout();
   void onDrainTimeout();
   void startDrainSequence();

diff --git a/source/common/stream_info/utility.cc b/source/common/stream_info/utility.cc
@@ -24,6 +24,7 @@ const std::string ResponseFlagUtils::UNAUTHORIZED_EXTERNAL_SERVICE = "UAEX";
 const std::string ResponseFlagUtils::RATELIMIT_SERVICE_ERROR = "RLSE";
 const std::string ResponseFlagUtils::STREAM_IDLE_TIMEOUT = "SI";
 const std::string ResponseFlagUtils::INVALID_ENVOY_REQUEST_HEADERS = "IH";
+const std::string ResponseFlagUtils::DOWNSTREAM_PROTOCOL_ERROR = "DPE";
 
 void ResponseFlagUtils::appendString(std::string& result, const std::string& append) {
   if (result.empty()) {
@@ -36,7 +37,7 @@ void ResponseFlagUtils::appendString(std::string& result, const std::string& app
 const std::string ResponseFlagUtils::toShortString(const StreamInfo& stream_info) {
   std::string result;
 
-  static_assert(ResponseFlag::LastFlag == 0x20000, "A flag has been added. Fix this code.");
+  static_assert(ResponseFlag::LastFlag == 0x40000, "A flag has been added. Fix this code.");
 
   if (stream_info.hasResponseFlag(ResponseFlag::FailedLocalHealthCheck)) {
     appendString(result, FAILED_LOCAL_HEALTH_CHECK);
@@ -109,6 +110,9 @@ const std::string ResponseFlagUtils::toShortString(const StreamInfo& stream_info
   if (stream_info.hasResponseFlag(ResponseFlag::InvalidEnvoyRequestHeaders)) {
     appendString(result, INVALID_ENVOY_REQUEST_HEADERS);
   }
+  if (stream_info.hasResponseFlag(ResponseFlag::DownstreamProtocolError)) {
+    appendString(result, DOWNSTREAM_PROTOCOL_ERROR);
+  }
 
   return result.empty() ? NONE : result;
 }
@@ -135,6 +139,7 @@ absl::optional<ResponseFlag> ResponseFlagUtils::toResponseFlag(const std::string
       {ResponseFlagUtils::UPSTREAM_RETRY_LIMIT_EXCEEDED, ResponseFlag::UpstreamRetryLimitExceeded},
       {ResponseFlagUtils::STREAM_IDLE_TIMEOUT, ResponseFlag::StreamIdleTimeout},
       {ResponseFlagUtils::INVALID_ENVOY_REQUEST_HEADERS, ResponseFlag::InvalidEnvoyRequestHeaders},
+      {ResponseFlagUtils::DOWNSTREAM_PROTOCOL_ERROR, ResponseFlag::DownstreamProtocolError},
   };
   const auto& it = map.find(flag);
   if (it != map.end()) {

diff --git a/source/common/stream_info/utility.h b/source/common/stream_info/utility.h
@@ -39,6 +39,7 @@ class ResponseFlagUtils {
   const static std::string RATELIMIT_SERVICE_ERROR;
   const static std::string STREAM_IDLE_TIMEOUT;
   const static std::string INVALID_ENVOY_REQUEST_HEADERS;
+  const static std::string DOWNSTREAM_PROTOCOL_ERROR;
 };
 
 /**

diff --git a/source/extensions/access_loggers/grpc/grpc_access_log_utils.cc b/source/extensions/access_loggers/grpc/grpc_access_log_utils.cc
@@ -35,7 +35,7 @@ void Utility::responseFlagsToAccessLogResponseFlags(
     envoy::data::accesslog::v2::AccessLogCommon& common_access_log,
     const StreamInfo::StreamInfo& stream_info) {
 
-  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x20000,
+  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x40000,
                 "A flag has been added. Fix this code.");
 
   if (stream_info.hasResponseFlag(StreamInfo::ResponseFlag::FailedLocalHealthCheck)) {
@@ -111,6 +111,10 @@ void Utility::responseFlagsToAccessLogResponseFlags(
   if (stream_info.hasResponseFlag(StreamInfo::ResponseFlag::InvalidEnvoyRequestHeaders)) {
     common_access_log.mutable_response_flags()->set_invalid_envoy_request_headers(true);
   }
+
+  if (stream_info.hasResponseFlag(StreamInfo::ResponseFlag::DownstreamProtocolError)) {
+    common_access_log.mutable_response_flags()->set_downstream_protocol_error(true);
+  }
 }
 
 void Utility::extractCommonAccessLogProperties(

diff --git a/test/common/access_log/access_log_impl_test.cc b/test/common/access_log/access_log_impl_test.cc
@@ -916,11 +916,12 @@ name: envoy.file_access_log
       - URX
       - SI
       - IH
+      - DPE
 config:
   path: /dev/null
   )EOF";
 
-  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x20000,
+  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x40000,
                 "A flag has been added. Fix this code.");
 
   const std::vector<StreamInfo::ResponseFlag> all_response_flags = {
@@ -942,6 +943,7 @@ name: envoy.file_access_log
       StreamInfo::ResponseFlag::UpstreamRetryLimitExceeded,
       StreamInfo::ResponseFlag::StreamIdleTimeout,
       StreamInfo::ResponseFlag::InvalidEnvoyRequestHeaders,
+      StreamInfo::ResponseFlag::DownstreamProtocolError,
   };
 
   InstanceSharedPtr log = AccessLogFactory::fromProto(parseAccessLogFromV2Yaml(yaml), context_);
@@ -973,7 +975,7 @@ name: envoy.file_access_log
       "[\"embedded message failed validation\"] | caused by "
       "ResponseFlagFilterValidationError.Flags[i]: [\"value must be in list \" [\"LH\" \"UH\" "
       "\"UT\" \"LR\" \"UR\" \"UF\" \"UC\" \"UO\" \"NR\" \"DI\" \"FI\" \"RL\" \"UAEX\" \"RLSE\" "
-      "\"DC\" \"URX\" \"SI\" \"IH\"]]): name: \"envoy.file_access_log\"\nfilter {\n  "
+      "\"DC\" \"URX\" \"SI\" \"IH\" \"DPE\"]]): name: \"envoy.file_access_log\"\nfilter {\n  "
       "response_flag_filter {\n    flags: \"UnsupportedFlag\"\n  }\n}\nconfig {\n  fields {\n    "
       "key: \"path\"\n    value {\n      string_value: \"/dev/null\"\n    }\n  }\n}\n");
 }
@@ -998,7 +1000,7 @@ name: envoy.file_access_log
       "[\"embedded message failed validation\"] | caused by "
       "ResponseFlagFilterValidationError.Flags[i]: [\"value must be in list \" [\"LH\" \"UH\" "
       "\"UT\" \"LR\" \"UR\" \"UF\" \"UC\" \"UO\" \"NR\" \"DI\" \"FI\" \"RL\" \"UAEX\" \"RLSE\" "
-      "\"DC\" \"URX\" \"SI\" \"IH\"]]): name: \"envoy.file_access_log\"\nfilter {\n  "
+      "\"DC\" \"URX\" \"SI\" \"IH\" \"DPE\"]]): name: \"envoy.file_access_log\"\nfilter {\n  "
       "response_flag_filter {\n    flags: \"UnsupportedFlag\"\n  }\n}\ntyped_config {\n  "
       "[type.googleapis.com/envoy.config.accesslog.v2.FileAccessLog] {\n    path: \"/dev/null\"\n  "
       "}\n}\n");

diff --git a/test/common/http/conn_manager_impl_test.cc b/test/common/http/conn_manager_impl_test.cc
@@ -2415,6 +2415,66 @@ TEST_F(HttpConnectionManagerImplTest, DownstreamProtocolError) {
   conn_manager_->onData(fake_input, false);
 }
 
+TEST_F(HttpConnectionManagerImplTest, TestDownstreamProtocolErrorAccessLog) {
+  std::shared_ptr<AccessLog::MockInstance> handler(new NiceMock<AccessLog::MockInstance>());
+  access_logs_ = {handler};
+  setup(false, "");
+
+  EXPECT_CALL(*handler, log(_, _, _, _))
+      .WillOnce(Invoke([](const HeaderMap*, const HeaderMap*, const HeaderMap*,
+                          const StreamInfo::StreamInfo& stream_info) {
+        EXPECT_FALSE(stream_info.responseCode());
+        EXPECT_TRUE(stream_info.hasAnyResponseFlag());
+        EXPECT_TRUE(stream_info.hasResponseFlag(StreamInfo::ResponseFlag::DownstreamProtocolError));
+      }));
+
+  StreamDecoder* decoder = nullptr;
+  NiceMock<MockStreamEncoder> encoder;
+  EXPECT_CALL(*codec_, dispatch(_)).WillRepeatedly(Invoke([&](Buffer::Instance&) -> void {
+    decoder = &conn_manager_->newStream(encoder);
+    throw CodecProtocolException("protocol error");
+  }));
+
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+}
+
+TEST_F(HttpConnectionManagerImplTest, TestDownstreamProtocolErrorAfterHeadersAccessLog) {
+  setup(false, "");
+
+  std::shared_ptr<MockStreamDecoderFilter> filter(new NiceMock<MockStreamDecoderFilter>());
+  std::shared_ptr<AccessLog::MockInstance> handler(new NiceMock<AccessLog::MockInstance>());
+
+  EXPECT_CALL(filter_factory_, createFilterChain(_))
+      .WillOnce(Invoke([&](FilterChainFactoryCallbacks& callbacks) -> void {
+        callbacks.addStreamDecoderFilter(filter);
+        callbacks.addAccessLogHandler(handler);
+      }));
+
+  EXPECT_CALL(*handler, log(_, _, _, _))
+      .WillOnce(Invoke([](const HeaderMap*, const HeaderMap*, const HeaderMap*,
+                          const StreamInfo::StreamInfo& stream_info) {
+        EXPECT_FALSE(stream_info.responseCode());
+        EXPECT_TRUE(stream_info.hasAnyResponseFlag());
+        EXPECT_TRUE(stream_info.hasResponseFlag(StreamInfo::ResponseFlag::DownstreamProtocolError));
+      }));
+
+  StreamDecoder* decoder = nullptr;
+  NiceMock<MockStreamEncoder> encoder;
+  EXPECT_CALL(*codec_, dispatch(_)).WillRepeatedly(Invoke([&](Buffer::Instance&) -> void {
+    decoder = &conn_manager_->newStream(encoder);
+
+    HeaderMapPtr headers{
+        new TestHeaderMapImpl{{":method", "GET"}, {":authority", "host"}, {":path", "/"}}};
+    decoder->decodeHeaders(std::move(headers), true);
+
+    throw CodecProtocolException("protocol error");
+  }));
+
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+}
+
 // Verify that FrameFloodException causes connection to be closed abortively.
 TEST_F(HttpConnectionManagerImplTest, FrameFloodError) {
   InSequence s;

diff --git a/test/common/stream_info/utility_test.cc b/test/common/stream_info/utility_test.cc
@@ -15,7 +15,7 @@ namespace StreamInfo {
 namespace {
 
 TEST(ResponseFlagUtilsTest, toShortStringConversion) {
-  static_assert(ResponseFlag::LastFlag == 0x20000, "A flag has been added. Fix this code.");
+  static_assert(ResponseFlag::LastFlag == 0x40000, "A flag has been added. Fix this code.");
 
   std::vector<std::pair<ResponseFlag, std::string>> expected = {
       std::make_pair(ResponseFlag::FailedLocalHealthCheck, "LH"),
@@ -36,6 +36,7 @@ TEST(ResponseFlagUtilsTest, toShortStringConversion) {
       std::make_pair(ResponseFlag::UpstreamRetryLimitExceeded, "URX"),
       std::make_pair(ResponseFlag::StreamIdleTimeout, "SI"),
       std::make_pair(ResponseFlag::InvalidEnvoyRequestHeaders, "IH"),
+      std::make_pair(ResponseFlag::DownstreamProtocolError, "DPE"),
   };
 
   for (const auto& test_case : expected) {
@@ -64,7 +65,7 @@ TEST(ResponseFlagUtilsTest, toShortStringConversion) {
 }
 
 TEST(ResponseFlagsUtilsTest, toResponseFlagConversion) {
-  static_assert(ResponseFlag::LastFlag == 0x20000, "A flag has been added. Fix this code.");
+  static_assert(ResponseFlag::LastFlag == 0x40000, "A flag has been added. Fix this code.");
 
   std::vector<std::pair<std::string, ResponseFlag>> expected = {
       std::make_pair("LH", ResponseFlag::FailedLocalHealthCheck),
@@ -85,6 +86,7 @@ TEST(ResponseFlagsUtilsTest, toResponseFlagConversion) {
       std::make_pair("URX", ResponseFlag::UpstreamRetryLimitExceeded),
       std::make_pair("SI", ResponseFlag::StreamIdleTimeout),
       std::make_pair("IH", ResponseFlag::InvalidEnvoyRequestHeaders),
+      std::make_pair("DPE", ResponseFlag::DownstreamProtocolError),
   };
 
   EXPECT_FALSE(ResponseFlagUtils::toResponseFlag("NonExistentFlag").has_value());

diff --git a/test/extensions/access_loggers/grpc/grpc_access_log_utils_test.cc b/test/extensions/access_loggers/grpc/grpc_access_log_utils_test.cc
@@ -40,6 +40,7 @@ TEST(UtilityResponseFlagsToAccessLogResponseFlagsTest, All) {
   common_access_log_expected.mutable_response_flags()->set_upstream_retry_limit_exceeded(true);
   common_access_log_expected.mutable_response_flags()->set_stream_idle_timeout(true);
   common_access_log_expected.mutable_response_flags()->set_invalid_envoy_request_headers(true);
+  common_access_log_expected.mutable_response_flags()->set_downstream_protocol_error(true);
 
   EXPECT_EQ(common_access_log_expected.DebugString(), common_access_log.DebugString());
 }