tracing: Add support for sending data in Zipkin v2 format

api/bazel/repositories.bzl

@@ -38,6 +38,11 @@ def api_dependencies():
         locations = REPOSITORY_LOCATIONS,
         build_file_content = KAFKASOURCE_BUILD_CONTENT,
     )
+    envoy_http_archive(
+        name = "com_github_openzipkin_zipkinapi",
+        locations = REPOSITORY_LOCATIONS,
+        build_file_content = ZIPKINAPI_BUILD_CONTENT,
+    )
 
 GOGOPROTO_BUILD_CONTENT = """
 load("@com_google_protobuf//:protobuf.bzl", "cc_proto_library", "py_proto_library")
@@ -153,3 +158,23 @@ filegroup(
 )
 
 """
+
+ZIPKINAPI_BUILD_CONTENT = """
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_library", "api_go_proto_library")
+load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
+
+api_proto_library(
+    name = "zipkin",
+    srcs = [
+        "zipkin-jsonv2.proto",
+        "zipkin.proto",
+    ],
+    visibility = ["//visibility:public"],
+)
+
+api_go_proto_library(
+    name = "zipkin",
+    proto = ":zipkin",
+)
+"""

api/bazel/repository_locations.bzl

@@ -21,6 +21,9 @@ KAFKA_SOURCE_SHA = "ae7a1696c0a0302b43c5b21e515c37e6ecd365941f68a510a7e442eebddf
 UDPA_GIT_SHA = "4cbdcb9931ca743a915a7c5fda51b2ee793ed157"  # Aug 22, 2019
 UDPA_SHA256 = "6291d0c0e3a4d5f08057ea7a00ed0b0ec3dd4e5a3b1cf20f803774680b5a806f"
 
+ZIPKINAPI_RELEASE = "0.2.2"  # Aug 23, 2019
+ZIPKINAPI_SHA256 = "688c4fe170821dd589f36ec45aaadc03a618a40283bc1f97da8fa11686fc816b"
+
 REPOSITORY_LOCATIONS = dict(
     bazel_skylib = dict(
         sha256 = BAZEL_SKYLIB_SHA256,
@@ -62,4 +65,9 @@ REPOSITORY_LOCATIONS = dict(
         strip_prefix = "kafka-2.2.0-rc2/clients/src/main/resources/common/message",
         urls = ["https://github.com/apache/kafka/archive/2.2.0-rc2.zip"],
     ),
+    com_github_openzipkin_zipkinapi = dict(
+        sha256 = ZIPKINAPI_SHA256,
+        strip_prefix = "zipkin-api-" + ZIPKINAPI_RELEASE,
+        urls = ["https://github.com/openzipkin/zipkin-api/archive/" + ZIPKINAPI_RELEASE + ".tar.gz"],
+    ),
 )

api/envoy/config/trace/v2/trace.proto

@@ -65,6 +65,7 @@ message LightstepConfig {
   string access_token_file = 2 [(validate.rules).string.min_bytes = 1];
 }
 
+// Configuration for the Zipkin tracer.
 message ZipkinConfig {
   // The cluster manager cluster that hosts the Zipkin collectors. Note that the
   // Zipkin cluster must be defined in the :ref:`Bootstrap static cluster
@@ -80,9 +81,34 @@ message ZipkinConfig {
   // trace instance. The default value is false, which will result in a 64 bit trace id being used.
   bool trace_id_128bit = 3;
 
-  // Determines whether client and server spans will shared the same span id.
+  // Determines whether client and server spans will share the same span context.
   // The default value is true.
   google.protobuf.BoolValue shared_span_context = 4;
+
+  // Available Zipkin collector endpoint versions.
+  enum CollectorEndpointVersion {
+    // Zipkin API v1, JSON over HTTP.
+    // [#comment: The default implementation of Zipkin client before this field is added was only v1
+    // and the way user configure this was by not explicitly specifying the version. Consequently,
+    // before this is added, the corresponding Zipkin collector expected to receive v1 payload.
+    // Hence the motivation of adding HTTP_JSON_V1 as the default is to avoid a breaking change when
+    // user upgrading Envoy with this change. Furthermore, we also immediately deprecate this field,
+    // since in Zipkin realm this v1 version is considered to be not preferable anymore.]
+    HTTP_JSON_V1 = 0 [deprecated = true];
+
+    // Zipkin API v2, JSON over HTTP.
+    HTTP_JSON = 1;
+
+    // Zipkin API v2, protobuf over HTTP.
+    HTTP_PROTO = 2;
+
+    // [#not-implemented-hide:]
+    GRPC = 3;
+  }
+
+  // Determines the selected collector endpoint version. By default, the ``HTTP_JSON_V1`` will be
+  // used.
+  CollectorEndpointVersion collector_endpoint_version = 5;
 }
 
 // DynamicOtConfig is used to dynamically load a tracer from a shared library

docs/root/intro/deprecated.rst

@@ -31,6 +31,9 @@ Version 1.12.0 (pending)
   and `present_match` fields.
 * The :option:`--allow-unknown-fields` command-line option,
   use :option:`--allow-unknown-static-fields` instead.
+* The use of HTTP_JSON_V1 :ref:`Zipkin collector endpoint version
+  <envoy_api_field_config.trace.v2.ZipkinConfig.collector_endpoint_version>` or not explicitly
+  specifying it is deprecated, use HTTP_JSON or HTTP_PROTO instead.
 
 Version 1.11.0 (July 11, 2019)
 ==============================

docs/root/intro/version_history.rst

@@ -42,6 +42,7 @@ Version history
 * router: added :ref:`rq_retry_skipped_request_not_complete <config_http_filters_router_stats>` counter stat to router stats.
 * router check tool: add coverage reporting & enforcement.
 * router check tool: add comprehensive coverage reporting.
+* tracing: added support to the Zipkin reporter for sending list of spans as Zipkin JSON v2 and protobuf message over HTTP.
 * tls: added verification of IP address SAN fields in certificates against configured SANs in the
   certificate validation context.
 * upstream: added network filter chains to upstream connections, see :ref:`filters<envoy_api_field_Cluster.filters>`.

source/common/http/headers.h

@@ -178,6 +178,7 @@ class HeaderValues {
     const std::string GrpcWebText{"application/grpc-web-text"};
     const std::string GrpcWebTextProto{"application/grpc-web-text+proto"};
     const std::string Json{"application/json"};
+    const std::string Protobuf{"application/x-protobuf"};
     const std::string FormUrlEncoded{"application/x-www-form-urlencoded"};
   } ContentTypeValues;
 

source/extensions/tracers/zipkin/BUILD

@@ -57,6 +57,7 @@ envoy_cc_library(
         "//source/common/singleton:const_singleton",
         "//source/common/tracing:http_tracer_lib",
         "//source/extensions/tracers:well_known_names",
+        "@com_github_openzipkin_zipkinapi//:zipkin_cc",
     ],
 )
 

source/extensions/tracers/zipkin/span_buffer.cc

@@ -1,35 +1,225 @@
 #include "extensions/tracers/zipkin/span_buffer.h"
 
+#include "common/protobuf/protobuf.h"
+
+#include "extensions/tracers/zipkin/util.h"
+#include "extensions/tracers/zipkin/zipkin_core_constants.h"
+
+#include "absl/strings/str_join.h"
+
 namespace Envoy {
 namespace Extensions {
 namespace Tracers {
 namespace Zipkin {
 
+SpanBuffer::SpanBuffer(
+    const envoy::config::trace::v2::ZipkinConfig::CollectorEndpointVersion& version,
+    const bool shared_span_context)
+    : serializer_{makeSerializer(version, shared_span_context)} {}
+
+SpanBuffer::SpanBuffer(
+    const envoy::config::trace::v2::ZipkinConfig::CollectorEndpointVersion& version,
+    const bool shared_span_context, uint64_t size)
+    : serializer_{makeSerializer(version, shared_span_context)} {
+  allocateBuffer(size);
+}
+
 // TODO(fabolive): Need to avoid the copy to improve performance.
-bool SpanBuffer::addSpan(const Span& span) {
-  if (span_buffer_.size() == span_buffer_.capacity()) {
-    // Buffer full
+bool SpanBuffer::addSpan(Span&& span) {
+  const auto& annotations = span.annotations();
+  if (span_buffer_.size() == span_buffer_.capacity() || annotations.empty() ||
+      annotations.end() ==
+          std::find_if(annotations.begin(), annotations.end(), [](const auto& annotation) {
+            return annotation.value() == ZipkinCoreConstants::get().CLIENT_SEND ||
+                   annotation.value() == ZipkinCoreConstants::get().SERVER_RECV;
+          })) {
+
+    // Buffer full or invalid span.
     return false;
   }
+
   span_buffer_.push_back(std::move(span));
 
   return true;
 }
 
-std::string SpanBuffer::toStringifiedJsonArray() {
-  std::string stringified_json_array = "[";
+SerializerPtr SpanBuffer::makeSerializer(
+    const envoy::config::trace::v2::ZipkinConfig::CollectorEndpointVersion& version,
+    const bool shared_span_context) {
+  switch (version) {
+  case envoy::config::trace::v2::ZipkinConfig::HTTP_JSON_V1:
+    return std::make_unique<JsonV1Serializer>();
+  case envoy::config::trace::v2::ZipkinConfig::HTTP_JSON:
+    return std::make_unique<JsonV2Serializer>(shared_span_context);
+  case envoy::config::trace::v2::ZipkinConfig::HTTP_PROTO:
+    return std::make_unique<ProtobufSerializer>(shared_span_context);
+  default:
+    NOT_REACHED_GCOVR_EXCL_LINE;
+  }
+}
+
+std::string JsonV1Serializer::serialize(std::vector<Span>&& zipkin_spans) {
+  const std::string serialized_elements =
+      absl::StrJoin(zipkin_spans, ",", [](std::string* element, Span zipkin_span) {
+        absl::StrAppend(element, zipkin_span.toJson());
+      });
+  return absl::StrCat("[", serialized_elements, "]");
+}
+
+JsonV2Serializer::JsonV2Serializer(const bool shared_span_context)
+    : shared_span_context_{shared_span_context} {}
+
+std::string JsonV2Serializer::serialize(std::vector<Span>&& zipkin_spans) {
+  const std::string serialized_elements =
+      absl::StrJoin(zipkin_spans, ",", [this](std::string* out, const Span& zipkin_span) {
+        absl::StrAppend(out,
+                        absl::StrJoin(toListOfSpans(zipkin_span), ",",
+                                      [](std::string* element, const zipkin::jsonv2::Span& span) {
+                                        std::string entry;
+                                        Protobuf::util::MessageToJsonString(span, &entry);
+                                        absl::StrAppend(element, entry);
+                                      }));
+      });
+  return absl::StrCat("[", serialized_elements, "]");
+}
 
-  if (pendingSpans()) {
-    stringified_json_array += span_buffer_[0].toJson();
-    const uint64_t size = span_buffer_.size();
-    for (uint64_t i = 1; i < size; i++) {
-      stringified_json_array += ",";
-      stringified_json_array += span_buffer_[i].toJson();
+const std::vector<zipkin::jsonv2::Span>
+JsonV2Serializer::toListOfSpans(const Span& zipkin_span) const {
+  std::vector<zipkin::jsonv2::Span> spans;
+  spans.reserve(zipkin_span.annotations().size());
+  for (const auto& annotation : zipkin_span.annotations()) {
+    zipkin::jsonv2::Span span;
+
+    if (annotation.value() == ZipkinCoreConstants::get().CLIENT_SEND) {
+      span.set_kind(ZipkinCoreConstants::get().KIND_CLIENT);
+    } else if (annotation.value() == ZipkinCoreConstants::get().SERVER_RECV) {
+      span.set_shared(shared_span_context_ && zipkin_span.annotations().size() > 1);
+      span.set_kind(ZipkinCoreConstants::get().KIND_SERVER);
+    } else {
+      continue;
+    }
+
+    if (annotation.isSetEndpoint()) {
+      span.set_timestamp(annotation.timestamp());
+      span.mutable_local_endpoint()->MergeFrom(toProtoEndpoint(annotation.endpoint()));
+    }
+
+    span.set_trace_id(zipkin_span.traceIdAsHexString());
+    if (zipkin_span.isSetParentId()) {
+      span.set_parent_id(zipkin_span.parentIdAsHexString());
+    }
+
+    span.set_id(zipkin_span.idAsHexString());
+    span.set_name(zipkin_span.name());
+
+    if (zipkin_span.isSetDuration()) {
+      span.set_duration(zipkin_span.duration());
+    }
+
+    auto& tags = *span.mutable_tags();
+    for (const auto& binary_annotation : zipkin_span.binaryAnnotations()) {
+      tags[binary_annotation.key()] = binary_annotation.value();
     }
+
+    spans.push_back(span);
+  }
+  return spans;
+}
+
+const zipkin::jsonv2::Endpoint
+JsonV2Serializer::toProtoEndpoint(const Endpoint& zipkin_endpoint) const {
+  zipkin::jsonv2::Endpoint endpoint;
+  Network::Address::InstanceConstSharedPtr address = zipkin_endpoint.address();
+  if (address) {
+    if (address->ip()->version() == Network::Address::IpVersion::v4) {
+      endpoint.set_ipv4(address->ip()->addressAsString());
+    } else {
+      endpoint.set_ipv6(address->ip()->addressAsString());
+    }
+    endpoint.set_port(address->ip()->port());
+  }
+
+  const std::string& service_name = zipkin_endpoint.serviceName();
+  if (!service_name.empty()) {
+    endpoint.set_service_name(service_name);
+  }
+
+  return endpoint;
+}
+
+ProtobufSerializer::ProtobufSerializer(const bool shared_span_context)
+    : shared_span_context_{shared_span_context} {}
+
+std::string ProtobufSerializer::serialize(std::vector<Span>&& zipkin_spans) {
+  zipkin::proto3::ListOfSpans spans;
+  for (const Span& zipkin_span : zipkin_spans) {
+    spans.MergeFrom(toListOfSpans(zipkin_span));
+  }
+  std::string serialized;
+  spans.SerializeToString(&serialized);
+  return serialized;
+}
+
+const zipkin::proto3::ListOfSpans ProtobufSerializer::toListOfSpans(const Span& zipkin_span) const {
+  zipkin::proto3::ListOfSpans spans;
+  for (const auto& annotation : zipkin_span.annotations()) {
+    zipkin::proto3::Span span;
+    if (annotation.value() == ZipkinCoreConstants::get().CLIENT_SEND) {
+      span.set_kind(zipkin::proto3::Span::CLIENT);
+    } else if (annotation.value() == ZipkinCoreConstants::get().SERVER_RECV) {
+      span.set_shared(shared_span_context_ && zipkin_span.annotations().size() > 1);
+      span.set_kind(zipkin::proto3::Span::SERVER);
+    } else {
+      continue;
+    }
+
+    if (annotation.isSetEndpoint()) {
+      span.set_timestamp(annotation.timestamp());
+      span.mutable_local_endpoint()->MergeFrom(toProtoEndpoint(annotation.endpoint()));
+    }
+
+    span.set_trace_id(zipkin_span.traceIdAsByteString());
+    if (zipkin_span.isSetParentId()) {
+      span.set_parent_id(zipkin_span.parentIdAsByteString());
+    }
+
+    span.set_id(zipkin_span.idAsByteString());
+    span.set_name(zipkin_span.name());
+
+    if (zipkin_span.isSetDuration()) {
+      span.set_duration(zipkin_span.duration());
+    }
+
+    auto& tags = *span.mutable_tags();
+    for (const auto& binary_annotation : zipkin_span.binaryAnnotations()) {
+      tags[binary_annotation.key()] = binary_annotation.value();
+    }
+
+    auto* mutable_span = spans.add_spans();
+    mutable_span->MergeFrom(span);
+  }
+  return spans;
+}
+
+const zipkin::proto3::Endpoint
+ProtobufSerializer::toProtoEndpoint(const Endpoint& zipkin_endpoint) const {
+  zipkin::proto3::Endpoint endpoint;
+  Network::Address::InstanceConstSharedPtr address = zipkin_endpoint.address();
+  if (address) {
+    if (address->ip()->version() == Network::Address::IpVersion::v4) {
+      endpoint.set_ipv4(Util::toByteString(address->ip()->ipv4()->address()));
+    } else {
+      endpoint.set_ipv6(Util::toByteString(address->ip()->ipv6()->address()));
+    }
+    endpoint.set_port(address->ip()->port());
+  }
+
+  const std::string& service_name = zipkin_endpoint.serviceName();
+  if (!service_name.empty()) {
+    endpoint.set_service_name(service_name);
   }
-  stringified_json_array += "]";
 
-  return stringified_json_array;
+  return endpoint;
 }
 
 } // namespace Zipkin