thrift proxy: missing parts from the previous PR

docs/root/configuration/other_protocols/thrift_filters/router_filter.rst

@@ -23,6 +23,7 @@ The filter outputs generic routing error statistics in the *thrift.<stat_prefix>
   unknown_cluster, Counter, Total requests with a route that has an unknown cluster.
   upstream_rq_maintenance_mode, Counter, Total requests with a destination cluster in maintenance mode.
   no_healthy_upstream, Counter, Total requests with no healthy upstream endpoints available.
+  shadow_request_submit_failure, Counter, Total shadow requests that failed to be submitted.
 
 
 The filter is also responsible for cluster-level statistics derived from routed upstream clusters.

source/extensions/filters/network/thrift_proxy/router/router_impl.cc

@@ -452,7 +452,7 @@ FilterStatus Router::setEnd() {
 
 void Router::onUpstreamData(Buffer::Instance& data, bool end_stream) {
   const bool done =
-      upstream_request_->handleUpstreamData(data, end_stream, *this, *upstream_response_callbacks_);
+      upstream_request_->handleUpstreamData(data, end_stream, *upstream_response_callbacks_);
   if (done) {
     cleanup();
   }

source/extensions/filters/network/thrift_proxy/router/shadow_writer_impl.cc

@@ -21,6 +21,7 @@ ShadowWriterImpl::submit(const std::string& cluster_name, MessageMetadataSharedP
                                                           original_transport, original_protocol);
   const bool created = shadow_router->createUpstreamRequest();
   if (!created) {
+    stats_.shadow_request_submit_failure_.inc();
     return absl::nullopt;
   }
 
@@ -339,7 +340,7 @@ void ShadowRouterImpl::maybeCleanup() {
 
 void ShadowRouterImpl::onUpstreamData(Buffer::Instance& data, bool end_stream) {
   const bool done =
-      upstream_request_->handleUpstreamData(data, end_stream, *this, *upstream_response_callbacks_);
+      upstream_request_->handleUpstreamData(data, end_stream, *upstream_response_callbacks_);
   if (done) {
     maybeCleanup();
   }

source/extensions/filters/network/thrift_proxy/router/shadow_writer_impl.h

@@ -227,11 +227,18 @@ class ShadowRouterImpl : public ShadowRouterHandle,
   std::list<ConverterCallback> pending_callbacks_;
 };
 
+#define ALL_SHADOW_WRITER_STATS(COUNTER, GAUGE, HISTOGRAM) COUNTER(shadow_request_submit_failure)
+
+struct ShadowWriterStats {
+  ALL_SHADOW_WRITER_STATS(GENERATE_COUNTER_STRUCT, GENERATE_GAUGE_STRUCT, GENERATE_HISTOGRAM_STRUCT)
+};
+
 class ShadowWriterImpl : public ShadowWriter, Logger::Loggable<Logger::Id::thrift> {
 public:
   ShadowWriterImpl(Upstream::ClusterManager& cm, const std::string& stat_prefix,
                    Stats::Scope& scope, Event::Dispatcher& dispatcher)
-      : cm_(cm), stat_prefix_(stat_prefix), scope_(scope), dispatcher_(dispatcher) {}
+      : cm_(cm), stat_prefix_(stat_prefix), scope_(scope), dispatcher_(dispatcher),
+        stats_(generateStats(stat_prefix, scope)) {}
 
   ~ShadowWriterImpl() override {
     while (!active_routers_.empty()) {
@@ -253,11 +260,18 @@ class ShadowWriterImpl : public ShadowWriter, Logger::Loggable<Logger::Id::thrif
 private:
   friend class ShadowRouterImpl;
 
+  ShadowWriterStats generateStats(const std::string& prefix, Stats::Scope& scope) {
+    return ShadowWriterStats{ALL_SHADOW_WRITER_STATS(POOL_COUNTER_PREFIX(scope, prefix),
+                                                     POOL_GAUGE_PREFIX(scope, prefix),
+                                                     POOL_HISTOGRAM_PREFIX(scope, prefix))};
+  }
+
   Upstream::ClusterManager& cm_;
   const std::string stat_prefix_;
   Stats::Scope& scope_;
   Event::Dispatcher& dispatcher_;
   std::list<std::unique_ptr<ShadowRouterImpl>> active_routers_;
+  ShadowWriterStats stats_;
 };
 
 } // namespace Router

source/extensions/filters/network/thrift_proxy/router/upstream_request.cc

@@ -114,7 +114,7 @@ void UpstreamRequest::handleUpgradeResponse(Buffer::Instance& data) {
 }
 
 ThriftFilters::ResponseStatus
-UpstreamRequest::handleRegularResponse(Buffer::Instance& data, RequestOwner& owner,
+UpstreamRequest::handleRegularResponse(Buffer::Instance& data,
                                        UpstreamResponseCallbacks& callbacks) {
   ENVOY_LOG(trace, "reading response: {} bytes", data.length());
 
@@ -123,36 +123,36 @@ UpstreamRequest::handleRegularResponse(Buffer::Instance& data, RequestOwner& own
     response_started_ = true;
   }
 
-  const auto& cluster = owner.cluster();
+  const auto& cluster = parent_.cluster();
 
   const auto status = callbacks.upstreamData(data);
   if (status == ThriftFilters::ResponseStatus::Complete) {
     ENVOY_LOG(debug, "response complete");
 
-    owner.recordUpstreamResponseSize(cluster, response_size_);
+    parent_.recordUpstreamResponseSize(cluster, response_size_);
 
     switch (callbacks.responseMetadata()->messageType()) {
     case MessageType::Reply:
-      owner.incResponseReply(cluster);
+      parent_.incResponseReply(cluster);
       if (callbacks.responseSuccess()) {
         upstream_host_->outlierDetector().putResult(
             Upstream::Outlier::Result::ExtOriginRequestSuccess);
-        owner.incResponseReplySuccess(cluster);
+        parent_.incResponseReplySuccess(cluster);
       } else {
         upstream_host_->outlierDetector().putResult(
             Upstream::Outlier::Result::ExtOriginRequestFailed);
-        owner.incResponseReplyError(cluster);
+        parent_.incResponseReplyError(cluster);
       }
       break;
 
     case MessageType::Exception:
       upstream_host_->outlierDetector().putResult(
           Upstream::Outlier::Result::ExtOriginRequestFailed);
-      owner.incResponseException(cluster);
+      parent_.incResponseException(cluster);
       break;
 
     default:
-      owner.incResponseInvalidType(cluster);
+      parent_.incResponseInvalidType(cluster);
       break;
     }
     onResponseComplete();
@@ -167,7 +167,6 @@ UpstreamRequest::handleRegularResponse(Buffer::Instance& data, RequestOwner& own
 }
 
 bool UpstreamRequest::handleUpstreamData(Buffer::Instance& data, bool end_stream,
-                                         RequestOwner& owner,
                                          UpstreamResponseCallbacks& callbacks) {
   ASSERT(!response_complete_);
 
@@ -176,7 +175,7 @@ bool UpstreamRequest::handleUpstreamData(Buffer::Instance& data, bool end_stream
   if (upgrade_response_ != nullptr) {
     handleUpgradeResponse(data);
   } else {
-    const auto status = handleRegularResponse(data, owner, callbacks);
+    const auto status = handleRegularResponse(data, callbacks);
     if (status != ThriftFilters::ResponseStatus::MoreData) {
       return true;
     }

source/extensions/filters/network/thrift_proxy/router/upstream_request.h

@@ -44,10 +44,10 @@ struct UpstreamRequest : public Tcp::ConnectionPool::Callbacks,
   void onPoolReady(Tcp::ConnectionPool::ConnectionDataPtr&& conn,
                    Upstream::HostDescriptionConstSharedPtr host) override;
 
-  bool handleUpstreamData(Buffer::Instance& data, bool end_stream, RequestOwner& owner,
+  bool handleUpstreamData(Buffer::Instance& data, bool end_stream,
                           UpstreamResponseCallbacks& callbacks);
   void handleUpgradeResponse(Buffer::Instance& data);
-  ThriftFilters::ResponseStatus handleRegularResponse(Buffer::Instance& data, RequestOwner& owner,
+  ThriftFilters::ResponseStatus handleRegularResponse(Buffer::Instance& data,
                                                       UpstreamResponseCallbacks& callbacks);
   uint64_t encodeAndWrite(Buffer::OwnedImpl& request_buffer);
   void onEvent(Network::ConnectionEvent event);

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

@@ -254,6 +254,7 @@ TEST_F(ShadowWriterTest, SubmitClusterNotFound) {
   auto router_handle = shadow_writer_->submit("shadow_cluster", metadata_, TransportType::Framed,
                                               ProtocolType::Binary);
   EXPECT_EQ(absl::nullopt, router_handle);
+  EXPECT_EQ(1U, context_.scope().counterFromString("test.shadow_request_submit_failure").value());
 }
 
 TEST_F(ShadowWriterTest, SubmitClusterInMaintenance) {
@@ -264,6 +265,7 @@ TEST_F(ShadowWriterTest, SubmitClusterInMaintenance) {
   auto router_handle = shadow_writer_->submit("shadow_cluster", metadata_, TransportType::Framed,
                                               ProtocolType::Binary);
   EXPECT_EQ(absl::nullopt, router_handle);
+  EXPECT_EQ(1U, context_.scope().counterFromString("test.shadow_request_submit_failure").value());
 }
 
 TEST_F(ShadowWriterTest, SubmitNoHealthyUpstream) {
@@ -277,6 +279,7 @@ TEST_F(ShadowWriterTest, SubmitNoHealthyUpstream) {
   auto router_handle = shadow_writer_->submit("shadow_cluster", metadata_, TransportType::Framed,
                                               ProtocolType::Binary);
   EXPECT_EQ(absl::nullopt, router_handle);
+  EXPECT_EQ(1U, context_.scope().counterFromString("test.shadow_request_submit_failure").value());
 
   // We still count the request, even if it didn't go through.
   EXPECT_EQ(