Fix consul service observation release (linkerd#1871)

Consul namer releases the observation of a service
by raising a specific Exception - ServiceRelease.
Decision whether to release an observation is
taken upon inspection of an observation result
(SvcAddr:67) - if it's a Failure and
ServiceRelease is its imeediate cause, then the
observation should be released. However the cause
may be nested deeply. So the observation is not
always released.

This commit unwraps the root cause of a Thrown
received during observation so we can detect
ServiceRelease even if it was wrapped arbitrary
number of times by underlying code.

Fixed linkerd#1820

Signed-off-by: Dmytro Kostiuchenko <[email protected]>

namer/consul/src/main/scala/io/buoyant/namer/consul/SvcAddr.scala

@@ -64,7 +64,7 @@ private[consul] object SvcAddr {
 
         if (stopped) Future.Unit
         else getAddresses(blockingIndex).transform {
-          case Throw(Failure(Some(cause))) if cause == ServiceRelease =>
+          case Throw(RootCause(cause)) if cause == ServiceRelease =>
             // this exception is raised when we close a watch - thus, it needs
             // to be special-cased so that we don't continue observing that
             // service.
@@ -117,7 +117,7 @@ private[consul] object SvcAddr {
         stopped = true
         stats.closes.incr()
         pending.raise(Failure("service observation released", ServiceRelease, Failure.Interrupted))
-        Future.Unit
+        pending
       }
     }
   }
@@ -176,4 +176,8 @@ private[consul] object SvcAddr {
 
   private[this] val NoIndexException =
     Failure(new IllegalArgumentException("consul did not return an index") with NoStackTrace)
+
+  object RootCause {
+    def unapply(e: Throwable): Option[Throwable] = Option(e.getCause).flatMap(unapply).orElse(Some(e))
+  }
 }

namer/consul/src/test/scala/io/buoyant/namer/consul/SvcAddrTest.scala

@@ -3,7 +3,7 @@ package io.buoyant.namer.consul
 import com.twitter.conversions.time._
 import com.twitter.finagle.stats.NullStatsReceiver
 import com.twitter.finagle.util.DefaultTimer
-import com.twitter.finagle.{Addr, Address}
+import com.twitter.finagle.{Addr, Address, Failure}
 import com.twitter.util.{Await, Duration, Future, Promise, Timer, Var}
 import io.buoyant.consul.v1.{CatalogApi, ConsistencyMode, HealthStatus, Indexed, ServiceNode}
 import io.buoyant.namer.consul.SvcAddr.Stats
@@ -113,4 +113,27 @@ class SvcAddrTest extends FunSuite with Matchers {
     Await.ready(retried, 1.second)
     numOfRequests.intValue() should equal(numOfAttempts)
   }
+
+  test("should extract nested root cause correctly") {
+    // given
+    val cause = Failure("cause")
+    val failure = Failure("one", Failure("two", Failure("three", cause)))
+
+    // when
+    val extracted = SvcAddr.RootCause.unapply(failure)
+
+    // then
+    extracted should be(Some(cause))
+  }
+
+  test("should extract root cause if there are no nested causes") {
+    // given
+    val cause = Failure("cause")
+
+    // when
+    val extracted = SvcAddr.RootCause.unapply(cause)
+
+    // then
+    extracted should be(Some(cause))
+  }
 }