Support for multiple processors

examples/CMakeLists.txt

@@ -6,4 +6,5 @@ add_subdirectory(simple)
 add_subdirectory(batch)
 add_subdirectory(metrics_simple)
 add_subdirectory(multithreaded)
+add_subdirectory(multi_processor)
 add_subdirectory(http)

examples/http/tracer_common.hpp

@@ -4,7 +4,7 @@
 #include "opentelemetry/sdk/trace/tracer_provider.h"
 #include "opentelemetry/trace/provider.h"
 
-#include <iostream>
+#include<vector>
 
 namespace {
 
@@ -13,8 +13,10 @@ void initTracer() {
       new opentelemetry::exporter::trace::OStreamSpanExporter);
   auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::SimpleSpanProcessor(std::move(exporter)));
+  std::vector<std::unique_ptr<sdktrace::SpanProcessor>> processors;
+  processors.push_back(std::move(processor));
   // Default is an always-on sampler.
-  auto context = std::make_shared<sdktrace::TracerContext>(std::move(processor));
+  auto context = std::make_shared<sdktrace::TracerContext>(std::move(processors));
   auto provider =  nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
       new sdktrace::TracerProvider(context));
   // Set the global trace provider

examples/multi_processor/BUILD

@@ -0,0 +1,26 @@
+cc_library(
+    name = "foo_multi_library",
+    srcs = [
+        "foo_library/foo_library.cc",
+    ],
+    hdrs = [
+        "foo_library/foo_library.h",
+    ],
+    deps = [
+        "//api",
+    ],
+)
+
+cc_binary(
+    name = "example_multi_processor",
+    srcs = [
+        "main.cc",
+    ],
+    deps = [
+        ":foo_multi_library",
+        "//api",
+        "//exporters/memory:in_memory_span_exporter",
+        "//exporters/ostream:ostream_span_exporter",
+        "//sdk/src/trace",
+    ],
+)

examples/multi_processor/CMakeLists.txt

@@ -0,0 +1,10 @@
+include_directories(${CMAKE_SOURCE_DIR}/exporters/ostream/include
+                    ${CMAKE_SOURCE_DIR}/exporters/memory/include)
+
+add_library(foo_multi_library foo_library/foo_library.cc)
+target_link_libraries(foo_multi_library opentelemetry_exporter_ostream_span
+                      ${CMAKE_THREAD_LIBS_INIT} opentelemetry_api)
+
+add_executable(example_multi_processor main.cc)
+target_link_libraries(example_multi_processor ${CMAKE_THREAD_LIBS_INIT}
+                      foo_multi_library opentelemetry_trace)

examples/multi_processor/README.md

@@ -0,0 +1,12 @@
+
+# Simple Trace Example
+
+In this example, the application in `main.cc` initializes and registers a tracer
+provider from the [OpenTelemetry
+SDK](https://github.com/open-telemetry/opentelemetry-cpp). The application then
+calls a `foo_library` which has been instrumented using the [OpenTelemetry
+API](https://github.com/open-telemetry/opentelemetry-cpp/tree/main/api).
+Resulting telemetry is directed to stdout through the StdoutSpanExporter.
+
+See [CONTRIBUTING.md](../../CONTRIBUTING.md) for instructions on building and
+running the example.

examples/multi_processor/foo_library/foo_library.cc

@@ -0,0 +1,42 @@
+#include "opentelemetry/trace/provider.h"
+
+namespace trace = opentelemetry::trace;
+namespace nostd = opentelemetry::nostd;
+
+namespace
+{
+nostd::shared_ptr<trace::Tracer> get_tracer()
+{
+  auto provider = trace::Provider::GetTracerProvider();
+  return provider->GetTracer("foo_library");
+}
+
+void f1()
+{
+  auto span  = get_tracer()->StartSpan("f1");
+  auto scope = get_tracer()->WithActiveSpan(span);
+
+  span->End();
+}
+
+void f2()
+{
+  auto span  = get_tracer()->StartSpan("f2");
+  auto scope = get_tracer()->WithActiveSpan(span);
+
+  f1();
+  f1();
+
+  span->End();
+}
+}  // namespace
+
+void foo_library()
+{
+  auto span  = get_tracer()->StartSpan("library");
+  auto scope = get_tracer()->WithActiveSpan(span);
+
+  f2();
+
+  span->End();
+}

examples/multi_processor/foo_library/foo_library.h

@@ -0,0 +1,3 @@
+#pragma once
+
+void foo_library();

examples/multi_processor/main.cc

@@ -0,0 +1,85 @@
+#include "opentelemetry/sdk/trace/simple_processor.h"
+#include "opentelemetry/sdk/trace/tracer_context.h"
+#include "opentelemetry/sdk/trace/tracer_provider.h"
+#include "opentelemetry/trace/provider.h"
+
+// Using an exporter that simply dumps span data to stdout.
+#include "foo_library/foo_library.h"
+#include "opentelemetry/exporters/memory/in_memory_span_exporter.h"
+#include "opentelemetry/exporters/ostream/span_exporter.h"
+
+using opentelemetry::exporter::memory::InMemorySpanExporter;
+
+InMemorySpanExporter *memory_span_exporter;
+
+namespace
+{
+void initTracer()
+{
+  auto exporter1 = std::unique_ptr<sdktrace::SpanExporter>(
+      new opentelemetry::exporter::trace::OStreamSpanExporter);
+  auto processor1 = std::unique_ptr<sdktrace::SpanProcessor>(
+      new sdktrace::SimpleSpanProcessor(std::move(exporter1)));
+
+  auto exporter2 = std::unique_ptr<sdktrace::SpanExporter>(new InMemorySpanExporter());
+
+  // fetch the exporter for dumping data later
+  memory_span_exporter = dynamic_cast<InMemorySpanExporter *>(exporter2.get());
+
+  auto processor2 = std::unique_ptr<sdktrace::SpanProcessor>(
+      new sdktrace::SimpleSpanProcessor(std::move(exporter2)));
+
+  std::vector<std::unique_ptr<sdktrace::SpanProcessor>> processors;
+  processors.push_back(std::move(processor1));
+  processors.push_back(std::move(processor2));
+  // Default is an always-on sampler.
+  auto context = std::make_shared<sdktrace::TracerContext>(std::move(processors));
+
+  auto provider = nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
+      new sdktrace::TracerProvider(context));
+
+  // Set the global trace provider
+  opentelemetry::trace::Provider::SetTracerProvider(provider);
+}
+
+void dumpSpans(std::vector<std::unique_ptr<opentelemetry::sdk::trace::SpanData>> &spans)
+{
+  char span_buf[opentelemetry::trace::SpanId::kSize * 2];
+  char trace_buf[opentelemetry::trace::TraceId::kSize * 2];
+  char parent_span_buf[opentelemetry::trace::SpanId::kSize * 2];
+  std::cout << "\nSpans from memory :" << std::endl;
+
+  for (auto &span : spans)
+  {
+    std::cout << "\n\tSpan: " << std::endl;
+    std::cout << "\t\tName: " << span->GetName() << std::endl;
+    span->GetSpanId().ToLowerBase16(span_buf);
+    span->GetTraceId().ToLowerBase16(trace_buf);
+    span->GetParentSpanId().ToLowerBase16(parent_span_buf);
+    std::cout << "\t\tTraceId: " << std::string(trace_buf, sizeof(trace_buf)) << std::endl;
+    std::cout << "\t\tSpanId: " << std::string(span_buf, sizeof(span_buf)) << std::endl;
+    std::cout << "\t\tParentSpanId: " << std::string(parent_span_buf, sizeof(parent_span_buf))
+              << std::endl;
+
+    std::cout << "\t\tDescription: " << span->GetDescription() << std::endl;
+    std::cout << "\t\tSpan kind:"
+              << static_cast<typename std::underlying_type<opentelemetry::trace::SpanKind>::type>(
+                     span->GetSpanKind())
+              << std::endl;
+    std::cout << "\t\tSpan Status: "
+              << static_cast<typename std::underlying_type<opentelemetry::trace::StatusCode>::type>(
+                     span->GetStatus())
+              << std::endl;
+  }
+}
+}  // namespace
+
+int main()
+{
+  // Removing this line will leave the default noop TracerProvider in place.
+  initTracer();
+
+  foo_library();
+  auto memory_spans = memory_span_exporter->GetData()->GetSpans();
+  dumpSpans(memory_spans);
+}

ext/test/w3c_tracecontext_test/main.cc

@@ -42,7 +42,9 @@ void initTracer()
       new opentelemetry::exporter::trace::OStreamSpanExporter);
   auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::SimpleSpanProcessor(std::move(exporter)));
-  auto context  = std::make_shared<sdktrace::TracerContext>(std::move(processor));
+  std::vector<std::unique_ptr<sdktrace::SpanProcessor>> processors;
+  processors.push_back(std::move(processor));
+  auto context  = std::make_shared<sdktrace::TracerContext>(std::move(processors));
   auto provider = nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
       new sdktrace::TracerProvider(context));
   // Set the global trace provider

ext/test/zpages/tracez_data_aggregator_test.cc

@@ -36,8 +36,11 @@ class TracezDataAggregatorTest : public ::testing::Test
   {
     std::shared_ptr<TracezSharedData> shared_data(new TracezSharedData());
     auto resource = opentelemetry::sdk::resource::Resource::Create({});
-    auto context  = std::make_shared<TracerContext>(
-        std::unique_ptr<SpanProcessor>(new TracezSpanProcessor(shared_data)), resource);
+    std::unique_ptr<SpanProcessor> processor(new TracezSpanProcessor(shared_data));
+    std::vector<std::unique_ptr<SpanProcessor>> processors;
+    processors.push_back(std::move(processor));
+
+    auto context           = std::make_shared<TracerContext>(std::move(processors), resource);
     tracer                 = std::shared_ptr<opentelemetry::trace::Tracer>(new Tracer(context));
     tracez_data_aggregator = std::unique_ptr<TracezDataAggregator>(
         new TracezDataAggregator(shared_data, milliseconds(10)));

ext/test/zpages/tracez_processor_test.cc

@@ -175,13 +175,16 @@ class TracezProcessor : public ::testing::Test
 protected:
   void SetUp() override
   {
-    shared_data   = std::shared_ptr<TracezSharedData>(new TracezSharedData());
-    processor     = std::shared_ptr<TracezSpanProcessor>(new TracezSpanProcessor(shared_data));
+    shared_data = std::shared_ptr<TracezSharedData>(new TracezSharedData());
+    processor   = std::shared_ptr<TracezSpanProcessor>(new TracezSpanProcessor(shared_data));
+    std::unique_ptr<SpanProcessor> processor2(new TracezSpanProcessor(shared_data));
+    std::vector<std::unique_ptr<SpanProcessor>> processors;
+    processors.push_back(std::move(processor2));
     auto resource = opentelemetry::sdk::resource::Resource::Create({});
+
     // Note: we make a *different* processor for the tracercontext. THis is because
     // all the tests use shared data, and we want to make sure this works correctly.
-    auto context = std::make_shared<TracerContext>(
-        std::unique_ptr<SpanProcessor>(new TracezSpanProcessor(shared_data)), resource);
+    auto context = std::make_shared<TracerContext>(std::move(processors), resource);
 
     tracer     = std::shared_ptr<opentelemetry::trace::Tracer>(new Tracer(context));
     auto spans = shared_data->GetSpanSnapshot();

sdk/include/opentelemetry/sdk/trace/multi_recordable.h

@@ -0,0 +1,141 @@
+#pragma once
+
+#include "opentelemetry/common/timestamp.h"
+#include "opentelemetry/sdk/trace/processor.h"
+#include "opentelemetry/sdk/trace/recordable.h"
+#include "opentelemetry/version.h"
+
+#include <map>
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace trace
+{
+namespace
+{
+std::size_t MakeKey(const SpanProcessor &processor)
+{
+  return reinterpret_cast<std::size_t>(&processor);
+}
+
+}  // namespace
+
+class MultiRecordable : public Recordable
+{
+public:
+  void AddRecordable(const SpanProcessor &processor,
+                     std::unique_ptr<Recordable> recordable) noexcept
+  {
+    recordables_[MakeKey(processor)] = std::move(recordable);
+  }
+
+  const std::unique_ptr<Recordable> &GetRecordable(const SpanProcessor &processor) const noexcept
+  {
+    // TODO - return nullptr ref on failed lookup?
+    static std::unique_ptr<Recordable> empty(nullptr);
+    auto i = recordables_.find(MakeKey(processor));
+    if (i != recordables_.end())
+    {
+      return i->second;
+    }
+    return empty;
+  }
+
+  std::unique_ptr<Recordable> ReleaseRecordable(const SpanProcessor &processor) noexcept
+  {
+    auto i = recordables_.find(MakeKey(processor));
+    if (i != recordables_.end())
+    {
+      std::unique_ptr<Recordable> result(i->second.release());
+      recordables_.erase(MakeKey(processor));
+      return result;
+    }
+    return std::unique_ptr<Recordable>(nullptr);
+  }
+
+  void SetIdentity(const opentelemetry::trace::SpanContext &span_context,
+                   opentelemetry::trace::SpanId parent_span_id) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->SetIdentity(span_context, parent_span_id);
+    }
+  }
+
+  virtual void SetAttribute(nostd::string_view key,
+                            const opentelemetry::common::AttributeValue &value) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->SetAttribute(key, value);
+    }
+  }
+
+  virtual void AddEvent(nostd::string_view name,
+                        opentelemetry::common::SystemTimestamp timestamp,
+                        const opentelemetry::common::KeyValueIterable &attributes) noexcept override
+  {
+
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->AddEvent(name, timestamp, attributes);
+    }
+  }
+
+  virtual void AddLink(const opentelemetry::trace::SpanContext &span_context,
+                       const opentelemetry::common::KeyValueIterable &attributes) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->AddLink(span_context, attributes);
+    }
+  }
+
+  virtual void SetStatus(opentelemetry::trace::StatusCode code,
+                         nostd::string_view description) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->SetStatus(code, description);
+    }
+  }
+
+  virtual void SetName(nostd::string_view name) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->SetName(name);
+    }
+  }
+
+  virtual void SetSpanKind(opentelemetry::trace::SpanKind span_kind) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->SetSpanKind(span_kind);
+    }
+  }
+
+  virtual void SetStartTime(opentelemetry::common::SystemTimestamp start_time) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->SetStartTime(start_time);
+    }
+  }
+
+  virtual void SetDuration(std::chrono::nanoseconds duration) noexcept override
+  {
+    for (auto &recordable : recordables_)
+    {
+      recordable.second->SetDuration(duration);
+    }
+  }
+
+private:
+  std::map<std::size_t, std::unique_ptr<Recordable>> recordables_;
+};
+}  // namespace trace
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE