finagle-core: add StackParam that overrides the WeightedApertureToggle

Problem:
It's occasionally not feasible to update
flags for all clients when dealing with
large services.

Solution:
Let's create a Stack Param that will
override the WeightedApertureToggle. This
will be particularly useful for fat clients

JIRA Issues: CSL-11405

Differential Revision: https://phabricator.twitter.biz/D769335

finagle-core/src/main/scala/com/twitter/finagle/loadbalancer/LoadBalancerFactory.scala

@@ -34,6 +34,15 @@ object LoadBalancerFactory {
     implicit val param = Stack.Param(ManageWeights(false))
   }
 
+  /** A temporary stack param that allows you to override the [[WeightedApertureToggle]]
+   * for clients without modifying their flags directly.
+   */
+  private[twitter] case class UseWeightedBalancers(enabled: Boolean)
+
+  private[twitter] implicit object UseWeightedBalancers extends Stack.Param[UseWeightedBalancers] {
+    val default = UseWeightedBalancers(false)
+  }
+
   /**
    * A class eligible for configuring a client's load balancer probation setting.
    * When enabled, the balancer treats removals as advisory and flags them. If a
@@ -406,10 +415,13 @@ object LoadBalancerFactory {
         )
       }
 
+      def shouldUseWeighted: Boolean =
+        if (params.contains[UseWeightedBalancers]) params[UseWeightedBalancers].enabled
+        else WeightedApertureToggle(label)
+
       // If weight-aware aperture load balancers are enabled, we do not wrap the
       // newBalancer in a TrafficDistributor.
-      if (loadBalancerFactory.supportsWeighted && WeightedApertureToggle(label)) {
-
+      if (loadBalancerFactory.supportsWeighted && shouldUseWeighted) {
         // Add the newBalancer to the stack
         Stack.leaf(
           role,

finagle-core/src/test/scala/com/twitter/finagle/loadbalancer/LoadBalancerFactoryTest.scala

@@ -481,14 +481,6 @@ class LoadBalancerFactoryTest extends AnyFunSuite with Eventually with Integrati
   }
 
   test("WeightedAperture has weights available to it from address metadata") {
-    val serverInfo: ServerInfo = new ServerInfo {
-      def environment: Option[String] = Some("staging")
-      def id: String = "testing"
-      def instanceId: Option[Long] = None
-      def clusterId: String = id
-      def zone: Option[String] = Some("smf1")
-    }
-
     com.twitter.finagle.toggle.flag.overrides
       .let("com.twitter.finagle.loadbalancer.WeightedAperture", 1.0) {
 
@@ -534,4 +526,39 @@ class LoadBalancerFactoryTest extends AnyFunSuite with Eventually with Integrati
         assert(weights == Set(1.0, 2.0))
       }
   }
+
+  test("can override WeightedApertureToggle with UseWeightedBalancers Stack.Param") {
+    com.twitter.finagle.toggle.flag.overrides
+      .let("com.twitter.finagle.loadbalancer.WeightedAperture", 0.0) {
+
+        val endpoint: Stack[ServiceFactory[String, String]] =
+          Stack.leaf(
+            Stack.Role("endpoint"),
+            ServiceFactory.const[String, String](Service.mk[String, String](req => ???))
+          )
+
+        val mockBalancer = new LoadBalancerFactory {
+          override def supportsWeighted: Boolean = true
+
+          def newBalancer[Req, Rep](
+            endpoints: Activity[IndexedSeq[EndpointFactory[Req, Rep]]],
+            emptyException: NoBrokersAvailableException,
+            params: Stack.Params
+          ): ServiceFactory[Req, Rep] = {
+            ServiceFactory.const(Service.mk(_ => ???))
+          }
+        }
+
+        val stack = LoadBalancerFactory.module[String, String].toStack(endpoint)
+
+        val params = Stack.Params.empty +
+          LoadBalancerFactory.Param(mockBalancer) +
+          LoadBalancerFactory.ManageWeights(true) +
+          LoadBalancerFactory.UseWeightedBalancers(true)
+
+        val a: ServiceFactory[String, String] = stack.make(params)
+
+        assert(!a.isInstanceOf[TrafficDistributor[String, String]])
+      }
+  }
 }