Support drop_overload API

api/envoy/config/endpoint/v3/endpoint.proto

@@ -40,7 +40,6 @@ message ClusterLoadAssignment {
     option (udpa.annotations.versioning).previous_message_type =
         "envoy.api.v2.ClusterLoadAssignment.Policy";
 
-    // [#not-implemented-hide:]
     message DropOverload {
       option (udpa.annotations.versioning).previous_message_type =
           "envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload";
@@ -75,7 +74,10 @@ message ClusterLoadAssignment {
     //    "throttle"_drop = 60%
     //    "lb"_drop = 20%  // 50% of the remaining 'actual' load, which is 40%.
     //    actual_outgoing_load = 20% // remaining after applying all categories.
-    // [#not-implemented-hide:]
+    //
+    // There are issues to support multiple categories at same time. It is decided
+    // to only support one category at this moment. If there are more than one category
+    // configured, the configuration will be rejected.
     repeated DropOverload drop_overloads = 2;
 
     // Priority levels and localities are considered overprovisioned with this

envoy/stream_info/stream_info.h

@@ -91,8 +91,10 @@ enum ResponseFlag {
   OverloadManager = 0x2000000,
   // DNS resolution failed.
   DnsResolutionFailed = 0x4000000,
+  // Drop certain percentage of overloaded traffic.
+  DropOverLoad = 0x8000000,
   // ATTENTION: MAKE SURE THIS REMAINS EQUAL TO THE LAST FLAG.
-  LastFlag = DnsResolutionFailed,
+  LastFlag = DropOverLoad,
 };
 
 /**
@@ -156,6 +158,8 @@ struct ResponseCodeDetailValues {
   const std::string ClusterNotFound = "cluster_not_found";
   // The request was rejected by the router filter because the cluster was in maintenance mode.
   const std::string MaintenanceMode = "maintenance_mode";
+  // The request was rejected by the router filter because the DROP_OVERLOAD configuration.
+  const std::string DropOverload = "drop_overload";
   // The request was rejected by the router filter because there was no healthy upstream found.
   const std::string NoHealthyUpstream = "no_healthy_upstream";
   // The request was forwarded upstream but the response timed out.

envoy/upstream/thread_local_cluster.h

@@ -151,6 +151,16 @@ class ThreadLocalCluster {
   virtual Tcp::AsyncTcpClientPtr
   tcpAsyncClient(LoadBalancerContext* context,
                  Tcp::AsyncTcpClientOptionsConstSharedPtr options) PURE;
+
+  /**
+   * @return the thread local cluster drop_overload configuration.
+   */
+  virtual UnitFloat dropOverload() const PURE;
+
+  /**
+   * Set up the drop_overload value for the thread local cluster.
+   */
+  virtual void setDropOverload(UnitFloat drop_overload) PURE;
 };
 
 using ThreadLocalClusterOptRef = absl::optional<std::reference_wrapper<ThreadLocalCluster>>;

envoy/upstream/upstream.h

@@ -743,6 +743,7 @@ class PrioritySet {
   COUNTER(upstream_rq_cancelled)                                                                   \
   COUNTER(upstream_rq_completed)                                                                   \
   COUNTER(upstream_rq_maintenance_mode)                                                            \
+  COUNTER(upstream_rq_drop_overload)                                                               \
   COUNTER(upstream_rq_max_duration_reached)                                                        \
   COUNTER(upstream_rq_pending_failure_eject)                                                       \
   COUNTER(upstream_rq_pending_overflow)                                                            \
@@ -1339,6 +1340,16 @@ class Cluster {
    * @return the const PrioritySet for the cluster.
    */
   virtual const PrioritySet& prioritySet() const PURE;
+
+  /**
+   * @return the cluster drop_overload configuration.
+   */
+  virtual UnitFloat dropOverload() const PURE;
+
+  /**
+   * Set up the drop_overload value for the cluster.
+   */
+  virtual void setDropOverload(UnitFloat drop_overload) PURE;
 };
 
 using ClusterSharedPtr = std::shared_ptr<Cluster>;

source/common/http/headers.h

@@ -176,6 +176,7 @@ class HeaderValues {
   const LowerCaseString EnvoyOriginalMethod{absl::StrCat(prefix(), "-original-method")};
   const LowerCaseString EnvoyOriginalPath{absl::StrCat(prefix(), "-original-path")};
   const LowerCaseString EnvoyOverloaded{absl::StrCat(prefix(), "-overloaded")};
+  const LowerCaseString EnvoyDropOverload{absl::StrCat(prefix(), "-drop-overload")};
   const LowerCaseString EnvoyRateLimited{absl::StrCat(prefix(), "-ratelimited")};
   const LowerCaseString EnvoyRetryOn{absl::StrCat(prefix(), "-retry-on")};
   const LowerCaseString EnvoyRetryGrpcOn{absl::StrCat(prefix(), "-retry-grpc-on")};
@@ -279,6 +280,10 @@ class HeaderValues {
     const std::string True{"true"};
   } EnvoyOverloadedValues;
 
+  struct {
+    const std::string True{"true"};
+  } EnvoyDropOverloadValues;
+
   struct {
     const std::string True{"true"};
   } EnvoyRateLimitedValues;

source/common/router/router.cc

@@ -562,6 +562,31 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
     return Http::FilterHeadersStatus::StopIteration;
   }
 
+  // Support DROP_OVERLOAD config from control plane to drop certain percentage of traffic.
+  if (cluster->dropOverload().value()) {
+    ENVOY_STREAM_LOG(debug, "Router filter: cluster has DROP_OVERLOAD configuration: {}\n",
+                     *callbacks_, cluster->dropOverload().value());
+    if (config_.random_.bernoulli(cluster->dropOverload())) {
+      ENVOY_STREAM_LOG(debug, "The request is dropped by DROP_OVERLOAD\n", *callbacks_);
+      callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::DropOverLoad);
+      chargeUpstreamCode(Http::Code::ServiceUnavailable, nullptr, true);
+      callbacks_->sendLocalReply(
+          Http::Code::ServiceUnavailable, "drop overload",
+          [modify_headers, this](Http::ResponseHeaderMap& headers) {
+            if (!config_.suppress_envoy_headers_) {
+              headers.addReference(Http::Headers::get().EnvoyDropOverload,
+                                   Http::Headers::get().EnvoyDropOverloadValues.True);
+            }
+            // Note: append_cluster_info does not respect suppress_envoy_headers.
+            modify_headers(headers);
+          },
+          absl::nullopt, StreamInfo::ResponseCodeDetails::get().DropOverload);
+
+      cluster_->trafficStats()->upstream_rq_drop_overload_.inc();
+      return Http::FilterHeadersStatus::StopIteration;
+    }
+  }
+
   // Fetch a connection pool for the upstream cluster.
   const auto& upstream_http_protocol_options = cluster_->upstreamHttpProtocolOptions();
 

source/common/stream_info/utility.cc

@@ -37,7 +37,7 @@ const std::string ResponseFlagUtils::toString(const StreamInfo& stream_info, boo
 }
 
 absl::flat_hash_map<std::string, ResponseFlag> ResponseFlagUtils::getFlagMap() {
-  static_assert(ResponseFlag::LastFlag == 0x4000000,
+  static_assert(ResponseFlag::LastFlag == 0x8000000,
                 "A flag has been added. Add the new flag to ALL_RESPONSE_STRINGS_FLAGS.");
   absl::flat_hash_map<std::string, ResponseFlag> res;
   for (auto [flag_strings, flag] : ResponseFlagUtils::ALL_RESPONSE_STRINGS_FLAGS) {

source/common/upstream/cluster_manager_impl.cc

@@ -1213,9 +1213,10 @@ void ClusterManagerImpl::postThreadLocalClusterUpdate(ClusterManagerCluster& cm_
   // Populate the cluster initialization object based on this update.
   ClusterInitializationObjectConstSharedPtr cluster_initialization_object =
       addOrUpdateClusterInitializationObjectIfSupported(params, cm_cluster.cluster().info(),
-                                                        load_balancer_factory, host_map);
+                                                        load_balancer_factory, host_map,
+                                                        cm_cluster.cluster().dropOverload());
 
-  tls_.runOnAllThreads([info = cm_cluster.cluster().info(), params = std::move(params),
+  tls_.runOnAllThreads([&cm_cluster, info = cm_cluster.cluster().info(), params = std::move(params),
                         add_or_update_cluster, load_balancer_factory, map = std::move(host_map),
                         cluster_initialization_object = std::move(cluster_initialization_object)](
                            OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
@@ -1270,6 +1271,8 @@ void ClusterManagerImpl::postThreadLocalClusterUpdate(ClusterManagerCluster& cm_
         new_cluster = new ThreadLocalClusterManagerImpl::ClusterEntry(*cluster_manager, info,
                                                                       load_balancer_factory);
         cluster_manager->thread_local_clusters_[info->name()].reset(new_cluster);
+        cluster_manager->thread_local_clusters_[info->name()]->setDropOverload(
+            cm_cluster.cluster().dropOverload());
         cluster_manager->local_stats_.clusters_inflated_.set(
             cluster_manager->thread_local_clusters_.size());
       }
@@ -1308,7 +1311,8 @@ void ClusterManagerImpl::postThreadLocalClusterUpdate(ClusterManagerCluster& cm_
 ClusterManagerImpl::ClusterInitializationObjectConstSharedPtr
 ClusterManagerImpl::addOrUpdateClusterInitializationObjectIfSupported(
     const ThreadLocalClusterUpdateParams& params, ClusterInfoConstSharedPtr cluster_info,
-    LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map) {
+    LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map,
+    UnitFloat drop_overload) {
   if (!deferralIsSupportedForCluster(cluster_info)) {
     return nullptr;
   }
@@ -1339,13 +1343,13 @@ ClusterManagerImpl::addOrUpdateClusterInitializationObjectIfSupported(
         entry->second->per_priority_state_, params, std::move(cluster_info),
         load_balancer_factory == nullptr ? entry->second->load_balancer_factory_
                                          : load_balancer_factory,
-        map);
+        map, drop_overload);
     cluster_initialization_map_[cluster_name] = new_initialization_object;
     return new_initialization_object;
   } else {
     // We need to create a fresh Cluster Initialization Object.
     auto new_initialization_object = std::make_shared<ClusterInitializationObject>(
-        params, std::move(cluster_info), load_balancer_factory, map);
+        params, std::move(cluster_info), load_balancer_factory, map, drop_overload);
     cluster_initialization_map_[cluster_name] = new_initialization_object;
     return new_initialization_object;
   }
@@ -1378,6 +1382,7 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::initializeClusterInlineIfExis
                             per_priority.overprovisioning_factor_,
                             initialization_object->cross_priority_host_map_);
   }
+  thread_local_clusters_[cluster]->setDropOverload(initialization_object->drop_overload_);
 
   // Remove the CIO as we've initialized the cluster.
   thread_local_deferred_clusters_.erase(entry);
@@ -1387,9 +1392,10 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::initializeClusterInlineIfExis
 
 ClusterManagerImpl::ClusterInitializationObject::ClusterInitializationObject(
     const ThreadLocalClusterUpdateParams& params, ClusterInfoConstSharedPtr cluster_info,
-    LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map)
+    LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map,
+    UnitFloat drop_overload)
     : cluster_info_(std::move(cluster_info)), load_balancer_factory_(load_balancer_factory),
-      cross_priority_host_map_(map) {
+      cross_priority_host_map_(map), drop_overload_(drop_overload) {
   // Copy the update since the map is empty.
   for (const auto& update : params.per_priority_update_params_) {
     per_priority_state_.emplace(update.priority_, update);
@@ -1399,9 +1405,11 @@ ClusterManagerImpl::ClusterInitializationObject::ClusterInitializationObject(
 ClusterManagerImpl::ClusterInitializationObject::ClusterInitializationObject(
     const absl::flat_hash_map<int, ThreadLocalClusterUpdateParams::PerPriority>& per_priority_state,
     const ThreadLocalClusterUpdateParams& update_params, ClusterInfoConstSharedPtr cluster_info,
-    LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map)
+    LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map,
+    UnitFloat drop_overload)
     : per_priority_state_(per_priority_state), cluster_info_(std::move(cluster_info)),
-      load_balancer_factory_(load_balancer_factory), cross_priority_host_map_(map) {
+      load_balancer_factory_(load_balancer_factory), cross_priority_host_map_(map),
+      drop_overload_(drop_overload) {
 
   // Because EDS Clusters receive the entire ClusterLoadAssignment but only
   // provides the delta we must process the hosts_added and hosts_removed and

source/common/upstream/cluster_manager_impl.h

@@ -434,18 +434,21 @@ class ClusterManagerImpl : public ClusterManager,
     ClusterInitializationObject(const ThreadLocalClusterUpdateParams& params,
                                 ClusterInfoConstSharedPtr cluster_info,
                                 LoadBalancerFactorySharedPtr load_balancer_factory,
-                                HostMapConstSharedPtr map);
+                                HostMapConstSharedPtr map,
+                                UnitFloat drop_overload);
 
     ClusterInitializationObject(
         const absl::flat_hash_map<int, ThreadLocalClusterUpdateParams::PerPriority>&
             per_priority_state,
         const ThreadLocalClusterUpdateParams& update_params, ClusterInfoConstSharedPtr cluster_info,
-        LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map);
+        LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map,
+        UnitFloat drop_overload);
 
     absl::flat_hash_map<int, ThreadLocalClusterUpdateParams::PerPriority> per_priority_state_;
     const ClusterInfoConstSharedPtr cluster_info_;
     const LoadBalancerFactorySharedPtr load_balancer_factory_;
     const HostMapConstSharedPtr cross_priority_host_map_;
+    UnitFloat drop_overload_{0};
   };
 
   using ClusterInitializationObjectConstSharedPtr =
@@ -614,6 +617,8 @@ class ClusterManagerImpl : public ClusterManager,
       // Drain any connection pools associated with the hosts filtered by the predicate.
       void drainConnPools(DrainConnectionsHostPredicate predicate,
                           ConnectionPool::DrainBehavior behavior);
+      UnitFloat dropOverload() const override { return drop_overload_;}
+      void setDropOverload(UnitFloat drop_overload) override { drop_overload_ = drop_overload; }
 
     private:
       Http::ConnectionPool::Instance*
@@ -629,6 +634,7 @@ class ClusterManagerImpl : public ClusterManager,
 
       ThreadLocalClusterManagerImpl& parent_;
       PrioritySetImpl priority_set_;
+      UnitFloat drop_overload_{0};
 
       // Don't change the order of cluster_info_ and lb_factory_/lb_ as the the lb_factory_/lb_
       // may keep a reference to the cluster_info_.
@@ -875,7 +881,8 @@ class ClusterManagerImpl : public ClusterManager,
    */
   ClusterInitializationObjectConstSharedPtr addOrUpdateClusterInitializationObjectIfSupported(
       const ThreadLocalClusterUpdateParams& params, ClusterInfoConstSharedPtr cluster_info,
-      LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map);
+      LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map,
+      UnitFloat drop_overload);
 
   bool deferralIsSupportedForCluster(const ClusterInfoConstSharedPtr& info) const;
 

source/common/upstream/health_discovery_service.h

@@ -72,7 +72,8 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
 
   std::vector<Upstream::HealthCheckerSharedPtr> healthCheckers() { return health_checkers_; };
   std::vector<HostSharedPtr> hosts() { return *hosts_; };
-
+  UnitFloat dropOverload() const override { return drop_overload_; }
+  void setDropOverload(UnitFloat drop_overload) override { drop_overload_ = drop_overload; }
 protected:
   PrioritySetImpl priority_set_;
   HealthCheckerSharedPtr health_checker_;
@@ -97,6 +98,7 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
   std::vector<Upstream::HealthCheckerSharedPtr> health_checkers_;
   HealthCheckerMap health_checkers_map_;
   TimeSource& time_source_;
+  UnitFloat drop_overload_{0};
 
   absl::Status updateHealthchecks(
       const Protobuf::RepeatedPtrField<envoy::config::core::v3::HealthCheck>& health_checks);

source/common/upstream/upstream_impl.cc

@@ -1521,6 +1521,37 @@ ClusterImplBase::ClusterImplBase(const envoy::config::cluster::v3::Cluster& clus
         info_->endpointStats().membership_degraded_.set(degraded_hosts);
         info_->endpointStats().membership_excluded_.set(excluded_hosts);
       });
+
+  if (cluster.load_assignment().has_policy()) {
+    auto policy = cluster.load_assignment().policy();
+    if (policy.drop_overloads().size() > 0) {
+      if (policy.drop_overloads().size() > 1) {
+        throwEnvoyExceptionOrPanic(
+            fmt::format("Only one drop_overload category configuration is supported. This cluster {} has {} configured",
+                      cluster.name(), policy.drop_overloads().size()));
+      } else {
+        auto drop_percentage = policy.drop_overloads(0).drop_percentage();
+        float denominator;
+        switch (drop_percentage.denominator()) {
+          case envoy::type::v3::FractionalPercent::HUNDRED:
+            denominator = 100;
+            break;
+          case envoy::type::v3::FractionalPercent::TEN_THOUSAND:
+            denominator = 10000;
+            break;
+          case envoy::type::v3::FractionalPercent::MILLION:
+            denominator = 1000000;
+            break;
+          default:
+            throwEnvoyExceptionOrPanic(
+                fmt::format("DROP_OVERLOAD denominator config is missing in the cluster {}.",
+                            cluster.name()));
+            break;
+        }
+        drop_overload_ = UnitFloat(float(drop_percentage.numerator()) / (denominator));
+      }
+    }
+  }
 }
 
 namespace {

source/common/upstream/upstream_impl.h

@@ -1219,6 +1219,8 @@ class ClusterImplBase : public Cluster, protected Logger::Loggable<Logger::Id::u
   Outlier::Detector* outlierDetector() override { return outlier_detector_.get(); }
   const Outlier::Detector* outlierDetector() const override { return outlier_detector_.get(); }
   void initialize(std::function<void()> callback) override;
+  UnitFloat dropOverload() const override { return drop_overload_; }
+  void setDropOverload(UnitFloat drop_overload) override { drop_overload_ = drop_overload; }
 
 protected:
   ClusterImplBase(const envoy::config::cluster::v3::Cluster& cluster,
@@ -1283,6 +1285,7 @@ class ClusterImplBase : public Cluster, protected Logger::Loggable<Logger::Id::u
   const bool local_cluster_;
   Config::ConstMetadataSharedPoolSharedPtr const_metadata_shared_pool_;
   Common::CallbackHandlePtr priority_update_cb_;
+  UnitFloat drop_overload_{0};
 };
 
 using ClusterImplBaseSharedPtr = std::shared_ptr<ClusterImplBase>;

source/extensions/access_loggers/grpc/grpc_access_log_utils.cc

@@ -40,7 +40,7 @@ void Utility::responseFlagsToAccessLogResponseFlags(
     envoy::data::accesslog::v3::AccessLogCommon& common_access_log,
     const StreamInfo::StreamInfo& stream_info) {
 
-  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x4000000,
+  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x8000000,
                 "A flag has been added. Fix this code.");
 
   if (stream_info.hasResponseFlag(StreamInfo::ResponseFlag::FailedLocalHealthCheck)) {