thrift proxy: add request shadowing support

api/envoy/extensions/filters/network/thrift_proxy/v3/route.proto

@@ -81,11 +81,33 @@ message RouteMatch {
   repeated config.route.v3.HeaderMatcher headers = 4;
 }
 
-// [#next-free-field: 7]
+// [#next-free-field: 8]
 message RouteAction {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.network.thrift_proxy.v2alpha1.RouteAction";
 
+  // The router is capable of shadowing traffic from one cluster to another. The current
+  // implementation is "fire and forget," meaning Envoy will not wait for the shadow cluster to
+  // respond before returning the response from the primary cluster. All normal statistics are
+  // collected for the shadow cluster making this feature useful for testing.
+  //
+  // .. note::
+  //
+  //   Shadowing will not be triggered if the primary cluster does not exist.
+  message RequestMirrorPolicy {
+    // Specifies the cluster that requests will be mirrored to. The cluster must
+    // exist in the cluster manager configuration when the route configuration is loaded.
+    // If it disappears at runtime, the shadow request will silently be ignored.
+    string cluster = 1 [(validate.rules).string = {min_len: 1}];
+
+    // If not specified, all requests to the target cluster will be mirrored.
+    //
+    // For some fraction N/D, a random number in the range [0,D) is selected. If the
+    // number is <= the value of the numerator N, or if the key is not present, the default
+    // value, the request will be mirrored.
+    config.core.v3.RuntimeFractionalPercent runtime_fraction = 2;
+  }
+
   oneof cluster_specifier {
     option (validate.required) = true;
 
@@ -123,6 +145,9 @@ message RouteAction {
   // Strip the service prefix from the method name, if there's a prefix. For
   // example, the method call Service:method would end up being just method.
   bool strip_service_name = 5;
+
+  // Indicates that the route has request mirroring policies.
+  repeated RequestMirrorPolicy request_mirror_policies = 7;
 }
 
 // Allows for specification of multiple upstream clusters along with weights that indicate the

docs/root/version_history/current.rst

@@ -64,6 +64,7 @@ New Features
 * jwt_authn: added support for :ref:`Jwt Cache <envoy_v3_api_field_extensions.filters.http.jwt_authn.v3.JwtProvider.jwt_cache_config>` and its size can be specified by :ref:`jwt_cache_size <envoy_v3_api_field_extensions.filters.http.jwt_authn.v3.JwtCacheConfig.jwt_cache_size>`.
 * listener: new listener metric ``downstream_cx_transport_socket_connect_timeout`` to track transport socket timeouts.
 * rbac: added :ref:`destination_port_range <envoy_v3_api_field_config.rbac.v3.Permission.destination_port_range>` for matching range of destination ports.
+* thrift_proxy: added support for :ref:`mirroring requests <envoy_v3_api_field_extensions.filters.network.thrift_proxy.v3.RouteAction.request_mirror_policies>`.
 
 Deprecated
 ----------

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

@@ -30,6 +30,71 @@ class MessageMetadata {
 public:
   MessageMetadata() = default;
 
+  std::shared_ptr<MessageMetadata> clone() const {
+    auto copy = std::make_shared<MessageMetadata>();
+
+    if (hasFrameSize()) {
+      copy->setFrameSize(frameSize());
+    }
+
+    if (hasProtocol()) {
+      copy->setProtocol(protocol());
+    }
+
+    if (hasMethodName()) {
+      copy->setMethodName(methodName());
+    }
+
+    if (hasSequenceId()) {
+      copy->setSequenceId(sequenceId());
+    }
+
+    if (hasMessageType()) {
+      copy->setMessageType(messageType());
+    }
+
+    Http::HeaderMapImpl::copyFrom(copy->headers(), headers());
+    copy->mutableSpans().assign(spans().begin(), spans().end());
+
+    if (hasAppException()) {
+      copy->setAppException(appExceptionType(), appExceptionMessage());
+    }
+
+    copy->setProtocolUpgradeMessage(isProtocolUpgradeMessage());
+
+    auto trace_id = traceId();
+    if (trace_id.has_value()) {
+      copy->setTraceId(trace_id.value());
+    }
+
+    auto trace_id_high = traceIdHigh();
+    if (trace_id_high.has_value()) {
+      copy->setTraceIdHigh(trace_id_high.value());
+    }
+
+    auto span_id = spanId();
+    if (span_id.has_value()) {
+      copy->setSpanId(span_id.value());
+    }
+
+    auto parent_span_id = parentSpanId();
+    if (parent_span_id.has_value()) {
+      copy->setParentSpanId(parent_span_id.value());
+    }
+
+    auto flags_opt = flags();
+    if (flags_opt.has_value()) {
+      copy->setFlags(flags_opt.value());
+    }
+
+    auto sampled_opt = sampled();
+    if (sampled_opt.has_value()) {
+      copy->setSampled(sampled_opt.value());
+    }
+
+    return copy;
+  }
+
   bool hasFrameSize() const { return frame_size_.has_value(); }
   uint32_t frameSize() const { return frame_size_.value(); }
   void setFrameSize(uint32_t size) { frame_size_ = size; }

source/extensions/filters/network/thrift_proxy/router/BUILD

@@ -65,13 +65,37 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "shadow_writer_lib",
+    srcs = ["shadow_writer_impl.cc"],
+    hdrs = ["shadow_writer_impl.h"],
+    deps = [
+        ":router_interface",
+        ":upstream_request_lib",
+        "//envoy/tcp:conn_pool_interface",
+        "//envoy/upstream:cluster_manager_interface",
+        "//envoy/upstream:load_balancer_interface",
+        "//envoy/upstream:thread_local_cluster_interface",
+        "//source/common/common:linked_object",
+        "//source/common/upstream:load_balancer_lib",
+        "//source/extensions/filters/network:well_known_names",
+        "//source/extensions/filters/network/thrift_proxy:app_exception_lib",
+        "//source/extensions/filters/network/thrift_proxy:conn_manager_lib",
+        "//source/extensions/filters/network/thrift_proxy:protocol_converter_lib",
+        "//source/extensions/filters/network/thrift_proxy:protocol_interface",
+        "//source/extensions/filters/network/thrift_proxy:thrift_object_interface",
+        "//source/extensions/filters/network/thrift_proxy:transport_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "router_lib",
     srcs = ["router_impl.cc"],
     hdrs = ["router_impl.h"],
     deps = [
         ":router_interface",
         ":router_ratelimit_lib",
+        ":shadow_writer_lib",
         ":upstream_request_lib",
         "//envoy/tcp:conn_pool_interface",
         "//envoy/upstream:cluster_manager_interface",

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

@@ -5,6 +5,7 @@
 #include "envoy/registry/registry.h"
 
 #include "source/extensions/filters/network/thrift_proxy/router/router_impl.h"
+#include "source/extensions/filters/network/thrift_proxy/router/shadow_writer_impl.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -17,9 +18,13 @@ ThriftFilters::FilterFactoryCb RouterFilterConfig::createFilterFactoryFromProtoT
     const std::string& stat_prefix, Server::Configuration::FactoryContext& context) {
   UNREFERENCED_PARAMETER(proto_config);
 
-  return [&context, stat_prefix](ThriftFilters::FilterChainFactoryCallbacks& callbacks) -> void {
-    callbacks.addDecoderFilter(
-        std::make_shared<Router>(context.clusterManager(), stat_prefix, context.scope()));
+  auto shadow_writer = std::make_shared<ShadowWriterImpl>(context.clusterManager(), stat_prefix,
+                                                          context.scope(), context.dispatcher());
+
+  return [&context, stat_prefix,
+          shadow_writer](ThriftFilters::FilterChainFactoryCallbacks& callbacks) -> void {
+    callbacks.addDecoderFilter(std::make_shared<Router>(
+        context.clusterManager(), stat_prefix, context.scope(), context.runtime(), *shadow_writer));
   };
 }