request shadowing (#32)

This change implements fire and forget request shadow logic in the router.

docs/configuration/http_conn_man/route_config/route.rst

@@ -22,6 +22,7 @@ next (e.g., redirect, forward, rewrite, etc.).
     "runtime": "{...}",
     "retry_policy": "{...}",
     "rate_limit": "{...}",
+    "shadow": "{...}"
   }
 
 prefix
@@ -85,6 +86,9 @@ content_type
 :ref:`rate_limit <config_http_conn_man_route_table_route_rate_limit>`
   *(optional, object)* Indicates that the route has a rate limit policy.
 
+:ref:`shadow <config_http_conn_man_route_table_route_shadow>`
+  *(optional, object)* Indicates that the route has a shadow policy.
+
 .. _config_http_conn_man_route_table_route_runtime:
 
 Runtime
@@ -151,3 +155,26 @@ global
   *(optional, boolean)* Specifies whether the global rate limit service should be called for a
   request that matches this route. This information is used by the :ref:`rate limit filter
   <config_http_filters_rate_limit>` if it is installed. Defaults to false if not specified.
+
+.. _config_http_conn_man_route_table_route_shadow:
+
+Shadow
+------
+
+.. code-block:: json
+
+  {
+    "cluster": "...",
+    "runtime_key": "..."
+  }
+
+cluster
+  *(required, string)* Specifies the cluster that requests will be shadowed to. The cluster must
+  exist in the :ref:`cluster manager configuration <config_cluster_manager>`.
+
+runtime_key
+  *(optional, string)* If not specified, **all** requests to the target cluster will be shadowed.
+  If specified, Envoy will lookup the runtime key to get the % of requests to shadow. Valid values are
+  from 0 to 10000, allowing for increments of 0.01% of requests to be shadowed. If the runtime key
+  is specified in the configuration but not present in runtime, 0 is the default and thus 0% of
+  requests will be shadowed.

include/envoy/http/async_client.h

@@ -15,7 +15,7 @@ class AsyncClient {
    */
   enum class FailureReason {
     // Request timeout has been reached.
-    RequestTimemout,
+    RequestTimeout,
     // The stream has been reset.
     Reset
   };

include/envoy/router/router.h

@@ -90,6 +90,28 @@ class RateLimitPolicy {
   virtual bool doGlobalLimiting() const PURE;
 };
 
+/**
+ * Per route policy for request shadowing.
+ */
+class ShadowPolicy {
+public:
+  virtual ~ShadowPolicy() {}
+
+  /**
+   * @return the name of the cluster that a matching request should be shadowed to. Returns empty
+   *         string if no shadowing should take place.
+   */
+  virtual const std::string& cluster() const PURE;
+
+  /**
+   * @return the runtime key that will be used to determine whether an individual request should
+   *         be shadowed. The lack of a key means that all requests will be shadowed. If a key is
+   *         present it will be used to drive random selection in the range 0-10000 for 0.01%
+   *         increments.
+   */
+  virtual const std::string& runtimeKey() const PURE;
+};
+
 /**
  * An individual resolved route entry.
  */
@@ -121,6 +143,12 @@ class RouteEntry {
    */
   virtual const RetryPolicy& retryPolicy() const PURE;
 
+  /**
+   * @return const ShadowPolicy& the shadow policy for the route. All routes have a shadow policy
+   *         even if no shadowing takes place.
+   */
+  virtual const ShadowPolicy& shadowPolicy() const PURE;
+
   /**
    * @return std::chrono::milliseconds the route's timeout.
    */

include/envoy/router/shadow_writer.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include "envoy/http/message.h"
+
+namespace Router {
+
+/**
+ * Interface used to shadow complete requests to an alternate upstream cluster in a "fire and
+ * forget" fashion. Right now this interface takes a fully buffered request and cannot be used for
+ * streaming. This is sufficient for current use cases.
+ */
+class ShadowWriter {
+public:
+  virtual ~ShadowWriter() {}
+
+  /**
+   * Shadow a request.
+   * @param cluster supplies the cluster name to shadow to.
+   * @param message supplies the complete request to shadow.
+   * @param timeout supplies the shadowed request timeout.
+   */
+  virtual void shadow(const std::string& cluster, Http::MessagePtr&& request,
+                      std::chrono::milliseconds timeout) PURE;
+};
+
+typedef std::unique_ptr<ShadowWriter> ShadowWriterPtr;
+
+} // Router

source/common/CMakeLists.txt

@@ -83,6 +83,7 @@ add_library(
   router/config_impl.cc
   router/retry_state_impl.cc
   router/router.cc
+  router/shadow_writer_impl.cc
   runtime/runtime_impl.cc
   runtime/uuid_util.cc
   ssl/connection_impl.cc

source/common/grpc/rpc_channel_impl.cc

@@ -123,7 +123,7 @@ void RpcChannelImpl::onFailure(Http::AsyncClient::FailureReason reason) {
   case Http::AsyncClient::FailureReason::Reset:
     onFailureWorker(Optional<uint64_t>(), "stream reset");
     break;
-  case Http::AsyncClient::FailureReason::RequestTimemout:
+  case Http::AsyncClient::FailureReason::RequestTimeout:
     onFailureWorker(Optional<uint64_t>(), "request timeout");
     break;
   }

source/common/http/async_client_impl.cc

@@ -136,7 +136,7 @@ void AsyncRequestImpl::onRequestTimeout() {
   CodeUtility::chargeResponseStat(info);
   parent_.cluster_.stats().upstream_rq_timeout_.inc();
   stream_encoder_->resetStream();
-  callbacks_.onFailure(AsyncClient::FailureReason::RequestTimemout);
+  callbacks_.onFailure(AsyncClient::FailureReason::RequestTimeout);
   cleanup();
 }
 

source/common/http/message_impl.h

@@ -34,6 +34,7 @@ class MessageImpl : public Http::Message {
 class RequestMessageImpl : public MessageImpl {
 public:
   RequestMessageImpl() : MessageImpl(HeaderMapPtr{new HeaderMapImpl()}) {}
+  RequestMessageImpl(HeaderMapPtr&& headers) : MessageImpl(std::move(headers)) {}
 };
 
 class ResponseMessageImpl : public MessageImpl {

source/common/router/config_impl.cc

@@ -36,6 +36,15 @@ RateLimitPolicyImpl::RateLimitPolicyImpl(const Json::Object& config) {
   do_global_limiting_ = config.getObject("rate_limit").getBoolean("global", false);
 }
 
+ShadowPolicyImpl::ShadowPolicyImpl(const Json::Object& config) {
+  if (!config.hasObject("shadow")) {
+    return;
+  }
+
+  cluster_ = config.getObject("shadow").getString("cluster");
+  runtime_key_ = config.getObject("shadow").getString("runtime_key", "");
+}
+
 RouteEntryImplBase::RouteEntryImplBase(const VirtualHost& vhost, const Json::Object& route,
                                        Runtime::Loader& loader)
     : case_sensitive_(route.getBoolean("case_sensitive", true)),
@@ -46,7 +55,8 @@ RouteEntryImplBase::RouteEntryImplBase(const VirtualHost& vhost, const Json::Obj
       runtime_(loadRuntimeData(route)), loader_(loader),
       host_redirect_(route.getString("host_redirect", "")),
       path_redirect_(route.getString("path_redirect", "")), retry_policy_(route),
-      content_type_(route.getString("content_type", "")), rate_limit_policy_(route) {
+      content_type_(route.getString("content_type", "")), rate_limit_policy_(route),
+      shadow_policy_(route) {
 
   // Check to make sure that we are either a redirect route or we have a cluster.
   if (!(isRedirect() ^ !cluster_name_.empty())) {
@@ -199,6 +209,13 @@ VirtualHost::VirtualHost(const Json::Object& virtual_host, Runtime::Loader& runt
             fmt::format("route: unknown cluster '{}'", routes_.back()->clusterName()));
       }
     }
+
+    if (!routes_.back()->shadowPolicy().cluster().empty()) {
+      if (!cm.get(routes_.back()->shadowPolicy().cluster())) {
+        throw EnvoyException(fmt::format("route: unknown shadow cluster '{}'",
+                                         routes_.back()->shadowPolicy().cluster()));
+      }
+    }
   }
 
   if (virtual_host.hasObject("virtual_clusters")) {

source/common/router/config_impl.h

@@ -103,6 +103,22 @@ class RateLimitPolicyImpl : public RateLimitPolicy {
   bool do_global_limiting_{};
 };
 
+/**
+ * Implementation of ShadowPolicy that reads from the JSON route config.
+ */
+class ShadowPolicyImpl : public ShadowPolicy {
+public:
+  ShadowPolicyImpl(const Json::Object& config);
+
+  // Router::ShadowPolicy
+  const std::string& cluster() const override { return cluster_; }
+  const std::string& runtimeKey() const override { return runtime_key_; }
+
+private:
+  std::string cluster_;
+  std::string runtime_key_;
+};
+
 /**
  * Base implementation for all route entries.
  */
@@ -117,6 +133,7 @@ class RouteEntryImplBase : public RouteEntry, public Matchable, public RedirectE
   void finalizeRequestHeaders(Http::HeaderMap& headers) const override;
   const RateLimitPolicy& rateLimitPolicy() const override { return rate_limit_policy_; }
   const RetryPolicy& retryPolicy() const override { return retry_policy_; }
+  const ShadowPolicy& shadowPolicy() const override { return shadow_policy_; }
   const std::string& virtualClusterName(const Http::HeaderMap& headers) const override {
     return vhost_.virtualClusterFromEntries(headers);
   }
@@ -157,6 +174,7 @@ class RouteEntryImplBase : public RouteEntry, public Matchable, public RedirectE
   const RetryPolicyImpl retry_policy_;
   const std::string content_type_;
   const RateLimitPolicyImpl rate_limit_policy_;
+  const ShadowPolicyImpl shadow_policy_;
 };
 
 /**