tracing: Add support for sending data in Zipkin v2 format

api/bazel/repositories.bzl

@@ -34,6 +34,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")
@@ -149,3 +154,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

@@ -18,6 +18,9 @@ PROMETHEUS_SHA = "783bdaf8ee0464b35ec0c8704871e1e72afa0005c3f3587f65d9d6694bf391
 
 KAFKA_SOURCE_SHA = "ae7a1696c0a0302b43c5b21e515c37e6ecd365941f68a510a7e442eebddf39a1"  # 2.2.0-rc2
 
+ZIPKINAPI_RELEASE = "0.2.2"  # Aug 23, 2019
+ZIPKINAPI_SHA256 = "688c4fe170821dd589f36ec45aaadc03a618a40283bc1f97da8fa11686fc816b"
+
 REPOSITORY_LOCATIONS = dict(
     bazel_skylib = dict(
         sha256 = BAZEL_SKYLIB_SHA256,
@@ -54,4 +57,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
@@ -83,6 +84,28 @@ message ZipkinConfig {
   // Determines whether client and server spans will shared the same span id.
   // The default value is true.
   google.protobuf.BoolValue shared_span_context = 4;
+
+  // Available Zipkin collector endpoint versions.
+  enum CollectorEndpointVersion {
+    // If not set, it fallbacks to ``HTTP_JSON_V1``.
+    NOT_SET = 0 [deprecated = true];
+
+    // Zipkin API v1, JSON over HTTP.
+    HTTP_JSON_V1 = 1 [deprecated = true];
+
+    // Zipkin API v2, JSON over HTTP.
+    HTTP_JSON = 2;
+
+    // Zipkin API v2, protobuf over HTTP.
+    HTTP_PROTO = 3;
+
+    // [#not-implemented-hide:]
+    GRPC = 4;
+  }
+
+  // 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/version_history.rst

@@ -37,6 +37,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,12 +1,29 @@
 #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"
+
 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) {
+bool SpanBuffer::addSpan(Span&& span) {
   if (span_buffer_.size() == span_buffer_.capacity()) {
     // Buffer full
     return false;
@@ -16,22 +33,202 @@ bool SpanBuffer::addSpan(const Span& span) {
   return true;
 }
 
-std::string SpanBuffer::toStringifiedJsonArray() {
+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) {
   std::string stringified_json_array = "[";
 
-  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();
+  if (!zipkin_spans.empty()) {
+    absl::StrAppend(&stringified_json_array, zipkin_spans[0].toJson());
+    for (uint64_t i = 1; i < zipkin_spans.size(); i++) {
+      absl::StrAppend(&stringified_json_array, ",", zipkin_spans[i].toJson());
     }
   }
-  stringified_json_array += "]";
+  absl::StrAppend(&stringified_json_array, "]");
 
   return stringified_json_array;
 }
 
+JsonV2Serializer::JsonV2Serializer(const bool shared_span_context)
+    : shared_span_context_{shared_span_context} {}
+
+std::string JsonV2Serializer::serialize(std::vector<Span>&& zipkin_spans) {
+  std::vector<zipkin::jsonv2::Span> spans;
+  spans.reserve(zipkin_spans.size() * 2);
+  for (const Span& zipkin_span : zipkin_spans) {
+    const std::vector<zipkin::jsonv2::Span>& converted = toListOfSpans(zipkin_span);
+    std::copy(converted.begin(), converted.end(), std::back_inserter(spans));
+  }
+
+  std::string stringified_json_array = "[";
+  const uint64_t spans_size = spans.size();
+  if (spans_size > 0) {
+    std::string entry;
+    Protobuf::util::MessageToJsonString(spans.at(0), &entry);
+    absl::StrAppend(&stringified_json_array, entry);
+    for (uint64_t i = 1; i < spans_size; i++) {
+      entry.clear();
+      Protobuf::util::MessageToJsonString(spans.at(i), &entry);
+      absl::StrAppend(&stringified_json_array, ",", entry);
+    }
+  }
+  absl::StrAppend(&stringified_json_array, "]");
+
+  return stringified_json_array;
+}
+
+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);
+  }
+
+  return endpoint;
+}
+
 } // namespace Zipkin
 } // namespace Tracers
 } // namespace Extensions