WIP TraceZ Processor

exporters/otlp/BUILD

@@ -28,12 +28,3 @@ cc_library(
         "@com_github_opentelemetry_proto//:trace_proto_cc",
     ],
 )
-
-cc_test(
-    name = "recordable_test",
-    srcs = ["recordable_test.cc"],
-    deps = [
-        ":recordable",
-        "@com_google_googletest//:gtest_main",
-    ],
-)

exporters/otlp/CMakeLists.txt

@@ -1,11 +1,3 @@
 add_library(opentelemetry_exporter_otprotocol recordable.cc)
 target_link_libraries(opentelemetry_exporter_otprotocol
                       $<TARGET_OBJECTS:opentelemetry_proto>)
-
-add_executable(recordable_test recordable_test.cc)
-target_link_libraries(recordable_test
-                      ${GTEST_BOTH_LIBRARIES}
-                      ${CMAKE_THREAD_LIBS_INIT}
-                      opentelemetry_exporter_otprotocol
-                      protobuf::libprotobuf)
-gtest_add_tests(TARGET recordable_test TEST_PREFIX exporter. TEST_LIST recordable_test)

exporters/otlp/recordable.cc

@@ -5,23 +5,6 @@ namespace exporter
 {
 namespace otlp
 {
-void Recordable::SetIds(trace::TraceId trace_id,
-                        trace::SpanId span_id,
-                        trace::SpanId parent_span_id) noexcept
-{
-  span_.set_trace_id(reinterpret_cast<const char *>(trace_id.Id().data()), trace::TraceId::kSize);
-  span_.set_span_id(reinterpret_cast<const char *>(span_id.Id().data()), trace::SpanId::kSize);
-  span_.set_parent_span_id(reinterpret_cast<const char *>(parent_span_id.Id().data()),
-                           trace::SpanId::kSize);
-}
-
-void Recordable::SetAttribute(nostd::string_view key,
-                              const opentelemetry::common::AttributeValue &&value) noexcept
-{
-  (void)key;
-  (void)value;
-}
-
 void Recordable::AddEvent(nostd::string_view name, core::SystemTimestamp timestamp) noexcept
 {
   (void)name;
@@ -37,18 +20,6 @@ void Recordable::SetName(nostd::string_view name) noexcept
 {
   span_.set_name(name.data(), name.size());
 }
-
-void Recordable::SetStartTime(opentelemetry::core::SystemTimestamp start_time) noexcept
-{
-  const uint64_t nano_unix_time = start_time.time_since_epoch().count();
-  span_.set_start_time_unix_nano(nano_unix_time);
-}
-
-void Recordable::SetDuration(std::chrono::nanoseconds duration) noexcept
-{
-  const uint64_t unix_end_time = span_.start_time_unix_nano() + duration.count();
-  span_.set_end_time_unix_nano(unix_end_time);
-}
 }  // namespace otlp
 }  // namespace exporter
 OPENTELEMETRY_END_NAMESPACE

exporters/otlp/recordable.h

@@ -14,23 +14,13 @@ class Recordable final : public sdk::trace::Recordable
 public:
   const proto::trace::v1::Span &span() const noexcept { return span_; }
 
-  void SetIds(trace::TraceId trace_id,
-              trace::SpanId span_id,
-              trace::SpanId parent_span_id) noexcept override;
-
-  void SetAttribute(nostd::string_view key,
-                    const opentelemetry::common::AttributeValue &&value) noexcept override;
-
+  // sdk::trace::Recordable
   void AddEvent(nostd::string_view name, core::SystemTimestamp timestamp) noexcept override;
 
   void SetStatus(trace::CanonicalCode code, nostd::string_view description) noexcept override;
 
   void SetName(nostd::string_view name) noexcept override;
 
-  void SetStartTime(opentelemetry::core::SystemTimestamp start_time) noexcept override;
-
-  void SetDuration(std::chrono::nanoseconds duration) noexcept override;
-
 private:
   proto::trace::v1::Span span_;
 };

ext/BUILD

@@ -0,0 +1,7 @@
+package(default_visibility = ["//visibility:public"])
+
+cc_library(
+    name = "headers",
+    hdrs = glob(["include/**/*.h"]),
+    strip_include_prefix = "include",
+)

ext/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_subdirectory(src)
+
+if(BUILD_TESTING)
+  add_subdirectory(test)
+endif()

ext/include/opentelemetry/ext/zpages/tracez_processor.h

@@ -0,0 +1,106 @@
+#pragma once
+
+#include <chrono>
+#include <memory>
+#include <unordered_set>
+#include <vector>
+#include <utility>
+#include <thread>
+#include <iostream>
+
+#include "opentelemetry/sdk/trace/recordable.h"
+#include "opentelemetry/sdk/trace/span_data.h"
+#include "opentelemetry/sdk/trace/exporter.h"
+#include "opentelemetry/sdk/trace/processor.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace ext
+{
+namespace zpages
+{
+/**
+ * The span processor passes finished recordables to be used by the
+ * Data Aggregator to store spans before aggregating for TraceZ.
+ *
+ */
+class TracezSpanProcessor : public opentelemetry::sdk::trace::SpanProcessor {
+ public:
+
+  struct CollectedSpans {
+    std::unordered_set<opentelemetry::sdk::trace::Recordable*> running;
+    std::vector<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> completed;
+  };
+
+  /**
+   * Initialize a span processor.
+   */
+  explicit TracezSpanProcessor() noexcept {}
+
+  /**
+   * Create a span recordable
+   * @return a newly initialized recordable
+   */
+  std::unique_ptr<opentelemetry::sdk::trace::Recordable> MakeRecordable() noexcept override
+  {
+    return std::unique_ptr<opentelemetry::sdk::trace::Recordable>(new opentelemetry::sdk::trace::SpanData);
+  }
+
+  /**
+   * OnStart is called when a span starts; that span is added to running_spans.
+   * @param span a recordable for a span that was just started
+   */
+  void OnStart(opentelemetry::sdk::trace::Recordable &span) noexcept override;
+
+  /**
+   * OnEnd is called when a span ends; that span is moved from running_spans to
+   * completed_spans
+   * @param span a recordable for a span that was ended
+   */
+  void OnEnd(std::unique_ptr<opentelemetry::sdk::trace::Recordable> &&span) noexcept override;
+
+  /**
+   * Returns a snapshot of all spans stored. This snapshot has a copy of the
+   * stored running_spans and gives ownership of completed spans. Stored
+   * completed_spans are cleared from the processor. Currently, copy-on-write
+   * is utilized where possible to minimize contention, but locks may be added
+   * in the future.
+   * @return snapshot of all currently running spans and newly completed spans
+   * at the time that the function is called
+   */
+  CollectedSpans GetSpanSnapshot() noexcept;
+
+  /**
+   * Send all ended spans that have not yet been sent.
+   * @param timeout an optional timeout, the default timeout of 0 means that no
+   * timeout is applied.
+   */
+  void ForceFlush(
+      std::chrono::microseconds timeout = std::chrono::microseconds(0)) noexcept override
+  {
+    std::this_thread::sleep_for(timeout);
+    if (shutdown_signal_received_) return;
+    // TODO: figure out how to send spans to data aggregator when this is called?
+    // should running spans be forced to end, and their status reflect this
+    // accordingly?
+  }
+  /**
+   * Shut down the processor and do any cleanup required. Ended spans are
+   * send before shutdown. After the call to Shutdown, subsequent calls to
+   * OnStart, OnEnd, ForceFlush or Shutdown will return immediately without
+   * doing anything.
+   * @param timeout an optional timeout, the default timeout of 0 means that no
+   * timeout is applied.
+   */
+  void Shutdown(std::chrono::microseconds timeout = std::chrono::microseconds(0)) noexcept override
+  {
+    ForceFlush(timeout);
+    shutdown_signal_received_ = true; // TODO: what cleanup do we need?
+  }
+
+ private:
+  CollectedSpans spans_;
+  bool shutdown_signal_received_ = false;
+};
+}  // namespace zpages
+}  // namespace ext
+OPENTELEMETRY_END_NAMESPACE

ext/src/CMakeLists.txt

@@ -0,0 +1 @@
+add_subdirectory(zpages)

ext/src/zpages/BUILD

@@ -0,0 +1,27 @@
+# Copyright 2020, OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+package(default_visibility = ["//visibility:public"])
+
+cc_library(
+    name = "zpages",
+    srcs = glob(["**/*.cc"]),
+    hdrs = glob(["**/*.h"]),
+    include_prefix = "ext/zpages",
+    deps = [
+        "//api",
+        "//sdk:headers",
+        "//ext:headers",
+    ],
+)

ext/src/zpages/CMakeLists.txt

@@ -0,0 +1 @@
+add_library(opentelemetry_trace tracez_processer.cc)

ext/src/zpages/tracez_processor.cc

@@ -0,0 +1,34 @@
+#include "opentelemetry/ext/zpages/tracez_processor.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace ext {
+namespace zpages {
+
+  void TracezSpanProcessor::OnStart(opentelemetry::sdk::trace::Recordable &span) noexcept {
+    if (shutdown_signal_received_) return;
+    spans_.running.insert(&span);
+  }
+
+  void TracezSpanProcessor::OnEnd(std::unique_ptr<opentelemetry::sdk::trace::Recordable> &&span) noexcept {
+     if (shutdown_signal_received_ || span == nullptr) return;
+     auto completed_span = spans_.running.find(span.get());
+     if (completed_span != spans_.running.end()) {
+       spans_.running.erase(completed_span);
+       spans_.completed.push_back(std::move(span));
+     }
+  }
+
+
+  TracezSpanProcessor::CollectedSpans TracezSpanProcessor::GetSpanSnapshot() noexcept {
+    if (shutdown_signal_received_) return TracezSpanProcessor::CollectedSpans();
+    CollectedSpans snapshot;
+    snapshot.running = spans_.running;
+    snapshot.completed = std::move(spans_.completed);
+    spans_.completed.clear();
+    return snapshot;
+  }
+
+
+}  // namespace zpages
+}  // namespace ext
+OPENTELEMETRY_END_NAMESPACE

ext/test/CMakeLists.txt

@@ -0,0 +1 @@
+add_subdirectory(zpages)

ext/test/zpages/BUILD

@@ -0,0 +1,11 @@
+cc_test(
+    name = "tracez_processor_tests",
+    srcs = [
+        "tracez_processor_test.cc",
+    ],
+    deps = [
+        "//sdk/src/trace",
+        "//ext/src/zpages",
+        "@com_google_googletest//:gtest_main",
+    ],
+)

ext/test/zpages/CMakeLists.txt

@@ -0,0 +1,9 @@
+foreach(testname tracez_processer_test)
+  add_executable(${testname} "${testname}.cc")
+  target_link_libraries(${testname} ${GTEST_BOTH_LIBRARIES}
+                        ${CMAKE_THREAD_LIBS_INIT} opentelemetry_trace)
+  gtest_add_tests(
+    TARGET ${testname}
+    TEST_PREFIX trace.
+    TEST_LIST ${testname})
+endforeach()