finagle-opencensus-tracing: upgraded opencensus and added TextFormat param

Problem

B3Format is currently the only propagation method we can use.
Invalid or missing propagation headers result in a 500
response.

Solution

raised opencensusVersion in build.sbt to "0.24.0"

added optional TextFormat parameter to the withOpenCensusTracing for both Server
and Client that defaults to previously used B3Format

in StackServerOps.scala previous version seemed to expect SpanContext.INVALID as
a failure result from TextFormat.extract, which was incorrect. This resulted in
missing or invalid propagation headers causing a 500 response.
TextFormat.extract throws an exception on bad or missing data, which is now
treated as SpanContext.INVALID

added a test dependency to opencensus-contrib-http-util which provides an
implementation of TextFormat for stackdriver style "x-cloud-trace-context"
headers.

added copy of existing test that uses the CloudTraceFormat from
opencensus-contrib-http-util

Result

You can now use other forms of tracing header propagation, i.e. the
CloudTraceFormat format from the opencensus-contrib-http-util package.

JIRA Issues: CSL-9378

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

Author: josephhajduk

CHANGELOG.rst

@@ -7,6 +7,12 @@ Note that ``PHAB_ID=#`` and ``RB_ID=#`` correspond to associated messages in com
 Unreleased
 ----------
 
+New Features
+~~~~~~~~~~~~
+
+* finagle-opencensus-tracing: Add support for providing a custom TextFormat for header
+  propagation. ``PHAB_ID=D432003``
+
 Runtime Behavior Changes
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
@@ -27,6 +33,9 @@ Bug Fixes
   from a HTTP/1.x pipeline, removing some traps such as unintended timeouts.
   ``PHAB_ID=D429416``
 
+* finagle-opencensus-tracing: Fixed internal server error when invalid or no propagation headers
+  are provided. ``PHAB_ID=D432003``
+
 20.1.0
 ------
 

build.sbt

@@ -65,7 +65,7 @@ val netty4LibsTest = Seq(
 val netty4Http = "io.netty" % "netty-codec-http" % netty4Version
 val netty4Http2 = "io.netty" % "netty-codec-http2" % netty4Version
 val netty4StaticSsl = "io.netty" % "netty-tcnative-boringssl-static" % netty4StaticSslVersion
-val opencensusVersion = "0.19.1"
+val opencensusVersion = "0.24.0"
 val jacksonVersion = "2.9.10"
 val jacksonDatabindVersion = "2.9.10.1"
 val jacksonLibs = Seq(
@@ -808,7 +808,8 @@ lazy val finagleOpenCensusTracing = Project(
   name := "finagle-opencensus-tracing",
   libraryDependencies ++= Seq(
     "io.opencensus" % "opencensus-api" % opencensusVersion,
-    "io.opencensus" % "opencensus-impl" % opencensusVersion
+    "io.opencensus" % "opencensus-impl" % opencensusVersion,
+    "io.opencensus" % "opencensus-contrib-http-util" % opencensusVersion % "test",
   ) ++ scroogeLibs
 ).dependsOn(
   finagleCore,

finagle-opencensus-tracing/src/main/scala/com/twitter/finagle/tracing/opencensus/StackClientOps.scala

@@ -45,20 +45,24 @@ object StackClientOps {
 
   implicit final class HttpOpenCensusTracing(private val client: Http.Client) extends AnyVal {
     def withOpenCensusTracing: Http.Client =
+      withOpenCensusTracing(Tracing.getPropagationComponent.getB3Format)
+
+    def withOpenCensusTracing(textFormat: TextFormat): Http.Client =
       client.withStack { stack =>
         // serialization must happen after we attach the OpenCensus span
         stack
           .insertBefore(
             TraceInitializerFilter.role,
             ClientTraceContextFilter.module[http.Request, http.Response]
           )
-          .replace(TraceInitializerFilter.role, httpSerializeModule)
+          .replace(TraceInitializerFilter.role, httpSerializeModule(textFormat))
       }
   }
 
-  private[this] val httpSerializeFilter: SimpleFilter[http.Request, http.Response] =
+  private[this] def httpSerializeFilter(
+    textFormat: TextFormat
+  ): SimpleFilter[http.Request, http.Response] =
     new SimpleFilter[http.Request, http.Response] {
-      private[this] val textFormat = Tracing.getPropagationComponent.getB3Format
       private[this] val setter = new TextFormat.Setter[http.Request] {
         def put(carrier: http.Request, key: String, value: String): Unit =
           carrier.headerMap.set(key, value)
@@ -80,12 +84,14 @@ object StackClientOps {
   private[opencensus] val HttpSerializationStackRole: Stack.Role =
     Stack.Role("OpenCensusHeaderSerialization")
 
-  private[this] val httpSerializeModule: Stackable[ServiceFactory[http.Request, http.Response]] =
+  private[this] def httpSerializeModule(
+    textFormat: TextFormat
+  ): Stackable[ServiceFactory[http.Request, http.Response]] =
     new Stack.Module0[ServiceFactory[http.Request, http.Response]] {
       def make(
         next: ServiceFactory[http.Request, http.Response]
       ): ServiceFactory[http.Request, http.Response] =
-        httpSerializeFilter.andThen(next)
+        httpSerializeFilter(textFormat).andThen(next)
 
       def role: Stack.Role = HttpSerializationStackRole
 

finagle-opencensus-tracing/src/main/scala/com/twitter/finagle/tracing/opencensus/StackServerOps.scala

@@ -2,7 +2,7 @@ package com.twitter.finagle.tracing.opencensus
 
 import com.twitter.finagle.context.Contexts
 import com.twitter.finagle._
-import com.twitter.util.Future
+import com.twitter.util.{Future, Try}
 import io.opencensus.trace.{SpanContext, Tracing}
 import io.opencensus.trace.propagation.TextFormat
 
@@ -44,18 +44,20 @@ object StackServerOps {
   }
 
   implicit final class HttpOpenCensusTracing(private val server: Http.Server) extends AnyVal {
-    def withOpenCensusTracing: Http.Server = {
+    def withOpenCensusTracing: Http.Server =
+      withOpenCensusTracing(Tracing.getPropagationComponent.getB3Format)
+
+    def withOpenCensusTracing(textFormat: TextFormat): Http.Server = {
       server.withStack { stack =>
         stack
           .prepend(ServerTraceContextFilter.module)
-          .prepend(httpDeserModule) // attach to broadcast ctx before setting OC
+          .prepend(httpDeserModule(textFormat)) // attach to broadcast ctx before setting OC
       }
     }
   }
 
-  private val httpDeserFilter: SimpleFilter[http.Request, http.Response] =
+  private def httpDeserFilter(textFormat: TextFormat): SimpleFilter[http.Request, http.Response] =
     new SimpleFilter[http.Request, http.Response] {
-      private[this] val textFormat = Tracing.getPropagationComponent.getB3Format
       private[this] val getter = new TextFormat.Getter[http.Request] {
         def get(carrier: http.Request, key: String): String =
           carrier.headerMap.getOrNull(key)
@@ -65,7 +67,7 @@ object StackServerOps {
         request: http.Request,
         service: Service[http.Request, http.Response]
       ): Future[http.Response] = {
-        val spanContext = textFormat.extract(request, getter)
+        val spanContext = Try(textFormat.extract(request, getter)).getOrElse(SpanContext.INVALID)
         if (spanContext != SpanContext.INVALID) {
           Contexts.broadcast.let(TraceContextFilter.SpanContextKey, spanContext) {
             service(request)
@@ -80,12 +82,14 @@ object StackServerOps {
   private[opencensus] val HttpDeserializationStackRole: Stack.Role =
     Stack.Role("OpenCensusHeaderDeserialization")
 
-  private val httpDeserModule: Stackable[ServiceFactory[http.Request, http.Response]] =
+  private def httpDeserModule(
+    textFormat: TextFormat
+  ): Stackable[ServiceFactory[http.Request, http.Response]] =
     new Stack.Module0[ServiceFactory[http.Request, http.Response]] {
       def make(
         next: ServiceFactory[http.Request, http.Response]
       ): ServiceFactory[http.Request, http.Response] =
-        httpDeserFilter.andThen(next)
+        httpDeserFilter(textFormat).andThen(next)
 
       def role: Stack.Role = HttpDeserializationStackRole
 

finagle-opencensus-tracing/src/test/scala/com/twitter/finagle/tracing/opencensus/BUILD

@@ -2,6 +2,7 @@ junit_tests(
     compiler_option_sets = {"fatal_warnings"},
     dependencies = [
         "3rdparty/jvm/io/opencensus:opencensus-api",
+        "3rdparty/jvm/io/opencensus:opencensus-contrib-http-util",
         "3rdparty/jvm/io/opencensus:opencensus-impl",
         "3rdparty/jvm/junit",
         "3rdparty/jvm/org/mockito:mockito-all",

finagle-opencensus-tracing/src/test/scala/com/twitter/finagle/tracing/opencensus/HttpEndToEndTest.scala

@@ -5,7 +5,9 @@ import com.twitter.finagle.http.{Request, Response, Status}
 import com.twitter.finagle.tracing.opencensus.TracingOps._
 import com.twitter.finagle.{Address, Http, Name, Service}
 import com.twitter.util.{Await, Duration, Future}
+import io.opencensus.contrib.http.util.HttpPropagationUtil
 import io.opencensus.trace.{SpanContext, Tracing}
+import io.opencensus.trace.propagation.TextFormat
 import java.net.{InetAddress, InetSocketAddress}
 import java.util.concurrent.atomic.AtomicReference
 import org.scalatest.FunSuite
@@ -15,14 +17,61 @@ class HttpEndToEndTest extends FunSuite {
   private def await[T](f: Future[T]): T =
     Await.result(f, Duration.fromSeconds(15))
 
-  private def httpServer: Http.Server = {
+  private def httpServer(
+    textFormat: TextFormat = Tracing.getPropagationComponent.getB3Format
+  ): Http.Server = {
     import com.twitter.finagle.tracing.opencensus.StackServerOps._
-    Http.server.withOpenCensusTracing
+    Http.server.withOpenCensusTracing(textFormat)
   }
 
-  private def httpClient: Http.Client = {
+  private def httpClient(
+    textFormat: TextFormat = Tracing.getPropagationComponent.getB3Format
+  ): Http.Client = {
     import com.twitter.finagle.tracing.opencensus.StackClientOps._
-    Http.client.withOpenCensusTracing
+    Http.client.withOpenCensusTracing(textFormat)
+  }
+
+  test("SpanContext is propagated when using CloudTraceFormat") {
+    val spanCtx = new AtomicReference[SpanContext](SpanContext.INVALID)
+
+    val svc: Service[Request, Response] = Service.mk { req =>
+      val rep = Contexts.broadcast.get(TraceContextFilter.SpanContextKey) match {
+        case None =>
+          val rep = Response(Status.BadRequest)
+          rep.contentString = "Broadcast context not set"
+          rep
+        case Some(ctx) if ctx.getTraceId != spanCtx.get.getTraceId =>
+          val rep = Response(Status.BadRequest)
+          rep.contentString =
+            s"TraceId mismatch, expected ${spanCtx.get.getTraceId}, but was ${ctx.getTraceId}"
+          rep
+        case Some(_) =>
+          Response(Status.Ok)
+      }
+
+      Future.value(rep)
+    }
+    val server = httpServer(HttpPropagationUtil.getCloudTraceFormat).serve(
+      new InetSocketAddress(InetAddress.getLoopbackAddress, 0),
+      svc
+    )
+    val client = httpClient(HttpPropagationUtil.getCloudTraceFormat).newService(
+      Name.bound(Address(server.boundAddress.asInstanceOf[InetSocketAddress])),
+      "cheese-processor"
+    )
+
+    val span = Tracing.getTracer
+      .spanBuilder("test")
+      .startSpan()
+    assert(span.getContext.isValid)
+    span.scopedAndEnd {
+      spanCtx.set(Tracing.getTracer.getCurrentSpan.getContext)
+
+      val rep = await(client(Request()))
+      assert(rep.status == Status.Ok, rep.contentString)
+    }
+    client.close()
+    server.close()
   }
 
   test("SpanContext is propagated") {
@@ -45,11 +94,11 @@ class HttpEndToEndTest extends FunSuite {
 
       Future.value(rep)
     }
-    val server = httpServer.serve(
+    val server = httpServer().serve(
       new InetSocketAddress(InetAddress.getLoopbackAddress, 0),
       svc
     )
-    val client = httpClient.newService(
+    val client = httpClient().newService(
       Name.bound(Address(server.boundAddress.asInstanceOf[InetSocketAddress])),
       "cheese-processor"
     )