finagle-core: Do not load balance over 0-weighted empty NameTrees

Problem:
--------
NameFactoryTree equally load balances over both empty and non-empty
resolutions if the total weight of the Union is zero. This results in
NoBrokersAvailable for a portion of requests through this client, even though
viable brokers are available.

Solution:
--------
Filter zero-weighted empty resolutions (NameTree.{Neg, Fail, Empty}) out
of Unions.

JIRA Issues: CSL-10943

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

CHANGELOG.rst

@@ -12,7 +12,7 @@ New Features
 
 * finagle-http2: Added `c.t.f.http2.param.EnforceMaxConcurrentStreams` which allows users to
   configure http2 clients to buffer streams once a connection has hit the max concurrent stream
-  limit rather than rejecting them.  A `buffered_streams` gauge has been added to track the 
+  limit rather than rejecting them.  A `buffered_streams` gauge has been added to track the
   current number of buffered streams.  ``PHAB_ID=D643138``
 
 Breaking API Changes
@@ -31,6 +31,9 @@ Bug Fixes
 * finagle-http2: The `streams` gauge is now correctly added for http2 connections over TLS.
   ``PHAB_ID=D643138``
 
+* finagle-core: `c.t.f.n.NameTreeFactory` will now discard empty elements in
+  `c.t.f.NameTree.Union`s with zero weight. ``PHAB_ID=D666635``
+
 21.4.0
 ------
 
@@ -173,7 +176,7 @@ Runtime Behavior Changes
 New Features
 ~~~~~~~~~~~~
 
-* finagle-core: Add `clnt/<FilterName>_rejected` annotation to filters that may throttle requests, 
+* finagle-core: Add `clnt/<FilterName>_rejected` annotation to filters that may throttle requests,
   including `c.t.finagle.filter.NackAdmissionFilter` and `c.t.finagle.filter.RequestSemaphoreFilter`.
   ``PHAB_ID=D597875``
 
@@ -208,7 +211,7 @@ New Features
 * finagle-benchmark: Add a benchmark for LocalContext. ``PHAB_ID=D588632``
 
 * finagle-core: Add a new filter, `ClientExceptionTracingFilter`, that records error annotations for
-  completed spans. Annotations include `error`, `exception.type`, and `exception.message`. 
+  completed spans. Annotations include `error`, `exception.type`, and `exception.message`.
   See https://github.com/open-telemetry/opentelemetry-specification for naming details.
   ``PHAB_ID=D583001``
 

finagle-core/src/main/scala/com/twitter/finagle/naming/NameTreeFactory.scala

@@ -44,6 +44,17 @@ private object NameTreeFactory {
       def close(deadline: Time) = Future.Done
     }
 
+    // Filter 0-weighted empty factories out of Unions. In the case where an
+    // entire Union is zero-weighted, we want to load balance over
+    // non-empty factories.
+    def shouldKeepInUnion(t: NameTree.Weighted[Key]): Boolean = {
+      t.tree match {
+        case NameTree.Neg | NameTree.Fail | NameTree.Empty =>
+          t.weight != 0.0
+        case _ => true
+      }
+    }
+
     def factoryOfTree(tree: NameTree[Key]): ServiceFactory[Req, Rep] =
       tree match {
         case NameTree.Neg | NameTree.Fail | NameTree.Empty => noBrokersAvailableFactory
@@ -53,8 +64,15 @@ private object NameTreeFactory {
         case NameTree.Alt(_*) => Failed(new IllegalArgumentException("NameTreeFactory"))
 
         case NameTree.Union(weightedTrees @ _*) =>
-          val (weights, trees) = weightedTrees.unzip { case NameTree.Weighted(w, t) => (w, t) }
-          Weighted(Drv.fromWeights(weights), trees.map(factoryOfTree))
+          val (weights, trees) = weightedTrees
+            .filter(shouldKeepInUnion)
+            .unzip { case NameTree.Weighted(w, t) => (w, t) }
+
+          if (weights.isEmpty) {
+            noBrokersAvailableFactory
+          } else {
+            Weighted(Drv.fromWeights(weights), trees.map(factoryOfTree))
+          }
       }
 
     factoryOfTree(tree)

finagle-core/src/test/scala/com/twitter/finagle/naming/NameTreeFactoryTest.scala

@@ -129,4 +129,93 @@ class NameTreeFactoryTest extends FunSuite {
       )
     )
   }
+
+  def canBuildServiceFromTree(tree: NameTree[String]): Boolean = {
+    val factoryCache = new ServiceFactoryCache[String, Unit, Unit](
+      _ =>
+        new ServiceFactory[Unit, Unit] {
+          def apply(conn: ClientConnection): Future[Service[Unit, Unit]] =
+            Future.value(Service.const(Future.Done))
+          def close(deadline: Time) = Future.Done
+          override def status = Status.Open
+        },
+      Timer.Nil
+    )
+
+    NameTreeFactory(Path.empty, tree, factoryCache).isAvailable
+  }
+
+  test("filters Zero-weighted Neg/Empty/Fail out of unions") {
+    val oneNonEmpty = NameTree.Union(
+      NameTree.Weighted(0.0, NameTree.Neg),
+      NameTree.Weighted(0.0, NameTree.Fail),
+      NameTree.Weighted(0.0, NameTree.Empty),
+      NameTree.Weighted(0.0, NameTree.Leaf("foo"))
+    )
+
+    assert(canBuildServiceFromTree(oneNonEmpty))
+  }
+
+  test("if only zero-weighted Neg/Empty/Fail in union, then NoBrokersAvailable") {
+    val allEmpty = NameTree.Union(
+      NameTree.Weighted(0.0, NameTree.Neg),
+      NameTree.Weighted(0.0, NameTree.Fail),
+      NameTree.Weighted(0.0, NameTree.Empty)
+    )
+
+    assert(!canBuildServiceFromTree(allEmpty))
+  }
+
+  test("nested all-zero weighted unions filters correctly") {
+    val embeddedNonEmpty = NameTree.Union(
+      NameTree.Weighted(0.0, NameTree.Neg),
+      NameTree.Weighted(0.0, NameTree.Fail),
+      NameTree.Weighted(0.0, NameTree.Empty),
+      NameTree.Weighted(
+        0.0,
+        NameTree.Union(
+          NameTree.Weighted(0.0, NameTree.Neg),
+          NameTree.Weighted(0.0, NameTree.Leaf("foo"))
+        ))
+    )
+
+    assert(canBuildServiceFromTree(embeddedNonEmpty))
+  }
+
+  test("nested not all-zero weighted unions filters correctly") {
+    val embeddedNotAllZerosNonEmpty = NameTree.Union(
+      NameTree.Weighted(0.0, NameTree.Empty),
+      NameTree.Weighted(
+        1.0,
+        NameTree.Union(
+          NameTree.Weighted(0.0, NameTree.Neg),
+          NameTree.Weighted(0.0, NameTree.Union(NameTree.Weighted(0.0, NameTree.Leaf("foo"))))
+        ))
+    )
+
+    assert(canBuildServiceFromTree(embeddedNotAllZerosNonEmpty))
+  }
+
+  test("deeply nested not all-zero weighted unions filters correctly") {
+    val embeddedNotAllZerosNonEmpty = NameTree.Union(
+      NameTree.Weighted(0.0, NameTree.Empty),
+      NameTree.Weighted(
+        0.0,
+        NameTree.Union(
+          NameTree.Weighted(0.0, NameTree.Neg),
+          NameTree.Weighted(0.0, NameTree.Union(NameTree.Weighted(1.0, NameTree.Leaf("foo"))))
+        ))
+    )
+
+    assert(canBuildServiceFromTree(embeddedNotAllZerosNonEmpty))
+  }
+
+  test("non-zero weighted empty are not filtered") {
+    val emptyWeightedNonZero = NameTree.Union(
+      NameTree.Weighted(100.0, NameTree.Empty),
+      NameTree.Weighted(0.0, NameTree.Leaf("foo"))
+    )
+
+    assert(!canBuildServiceFromTree(emptyWeightedNonZero))
+  }
 }