Create shared context for updating span pipeline from TracerProvider …

…and affecting Tracer. (#650)

examples/batch/main.cc

@@ -28,11 +28,11 @@ void initTracer()
   // We export `kNumSpans` after every `schedule_delay_millis` milliseconds.
   options.max_export_batch_size = kNumSpans;
 
-  auto processor = std::shared_ptr<sdktrace::SpanProcessor>(
+  auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::BatchSpanProcessor(std::move(exporter), options));
 
   auto provider = nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
-      new sdktrace::TracerProvider(processor));
+      new sdktrace::TracerProvider(std::move(processor)));
   // Set the global trace provider.
   opentelemetry::trace::Provider::SetTracerProvider(provider);
 }

examples/http/BUILD

@@ -0,0 +1,42 @@
+cc_binary(
+    name = "example_http_client",
+    srcs = [
+        "client.cc",
+        "tracer_common.hpp",
+    ],
+    # TODO: Move copts/linkopts for static CURL usage into shared bzl file.
+    copts = [
+        "-DCURL_STATICLIB",
+        "-DWITH_CURL",
+    ],
+    linkopts = select({
+        "//bazel:windows": [
+            "-DEFAULTLIB:advapi32.lib",
+            "-DEFAULTLIB:crypt32.lib",
+            "-DEFAULTLIB:Normaliz.lib",
+        ],
+        "//conditions:default": [],
+    }),
+    deps = [
+        "//api",
+        "//exporters/ostream:ostream_span_exporter",
+        "//ext:headers",
+        "//ext/src/http/client/curl:http_client_curl",
+        "//sdk/src/trace",
+    ],
+)
+
+cc_binary(
+    name = "example_http_server",
+    srcs = [
+        "server.cc",
+        "server.hpp",
+        "tracer_common.hpp",
+    ],
+    deps = [
+        "//api",
+        "//exporters/ostream:ostream_span_exporter",
+        "//ext:headers",
+        "//sdk/src/trace",
+    ],
+)

examples/http/tracer_common.hpp

@@ -11,11 +11,12 @@ namespace {
 void initTracer() {
   auto exporter = std::unique_ptr<sdktrace::SpanExporter>(
       new opentelemetry::exporter::trace::OStreamSpanExporter);
-  auto processor = std::shared_ptr<sdktrace::SpanProcessor>(
+  auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::SimpleSpanProcessor(std::move(exporter)));
-  auto provider = nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
-      new sdktrace::TracerProvider(processor, opentelemetry::sdk::resource::Resource::Create({}),
-                                   std::make_shared<opentelemetry::sdk::trace::AlwaysOnSampler>()));
+  // Default is an always-on sampler.
+  auto context = std::make_shared<sdktrace::TracerContext>(std::move(processor));
+  auto provider =  nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
+      new sdktrace::TracerProvider(context));
   // Set the global trace provider
   opentelemetry::trace::Provider::SetTracerProvider(provider);
 }

examples/multithreaded/main.cc

@@ -16,10 +16,11 @@ void initTracer()
 {
   auto exporter = std::unique_ptr<sdktrace::SpanExporter>(
       new opentelemetry::exporter::trace::OStreamSpanExporter);
-  auto processor = std::shared_ptr<sdktrace::SpanProcessor>(
+  auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::SimpleSpanProcessor(std::move(exporter)));
-  auto provider = nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
-      new sdktrace::TracerProvider(processor, opentelemetry::sdk::resource::Resource::Create({})));
+  auto provider =
+      nostd::shared_ptr<opentelemetry::trace::TracerProvider>(new sdktrace::TracerProvider(
+          std::move(processor), opentelemetry::sdk::resource::Resource::Create({})));
   // Set the global trace provider
   opentelemetry::trace::Provider::SetTracerProvider(provider);
 }

examples/otlp/main.cc

@@ -17,9 +17,10 @@ void InitTracer()
 {
   // Create OTLP exporter instance
   auto exporter  = std::unique_ptr<sdktrace::SpanExporter>(new otlp::OtlpExporter(opts));
-  auto processor = std::shared_ptr<sdktrace::SpanProcessor>(
+  auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::SimpleSpanProcessor(std::move(exporter)));
-  auto provider = nostd::shared_ptr<trace::TracerProvider>(new sdktrace::TracerProvider(processor));
+  auto provider =
+      nostd::shared_ptr<trace::TracerProvider>(new sdktrace::TracerProvider(std::move(processor)));
   // Set the global trace provider
   trace::Provider::SetTracerProvider(provider);
 }

examples/simple/main.cc

@@ -12,11 +12,10 @@ void initTracer()
 {
   auto exporter = std::unique_ptr<sdktrace::SpanExporter>(
       new opentelemetry::exporter::trace::OStreamSpanExporter);
-  auto processor = std::shared_ptr<sdktrace::SpanProcessor>(
+  auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::SimpleSpanProcessor(std::move(exporter)));
   auto provider = nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
-      new sdktrace::TracerProvider(processor, opentelemetry::sdk::resource::Resource::Create({}),
-                                   std::make_shared<opentelemetry::sdk::trace::AlwaysOnSampler>()));
+      new sdktrace::TracerProvider(std::move(processor)));
 
   // Set the global trace provider
   opentelemetry::trace::Provider::SetTracerProvider(provider);

exporters/otlp/test/otlp_exporter_test.cc

@@ -73,10 +73,10 @@ TEST_F(OtlpExporterTestPeer, ExportIntegrationTest)
 
   auto exporter = GetExporter(stub_interface);
 
-  auto processor = std::shared_ptr<sdk::trace::SpanProcessor>(
+  auto processor = std::unique_ptr<sdk::trace::SpanProcessor>(
       new sdk::trace::SimpleSpanProcessor(std::move(exporter)));
-  auto provider =
-      nostd::shared_ptr<trace::TracerProvider>(new sdk::trace::TracerProvider(processor));
+  auto provider = nostd::shared_ptr<trace::TracerProvider>(
+      new sdk::trace::TracerProvider(std::move(processor)));
   auto tracer = provider->GetTracer("test");
 
   EXPECT_CALL(*mock_stub, Export(_, _, _))

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

@@ -51,10 +51,11 @@ class ZPages
   /** Replaces the global tracer provider with an instance that exports to tracez. */
   void ReplaceGlobalProvider()
   {
-    std::shared_ptr<opentelemetry::sdk::trace::SpanProcessor> tracez_processor(
+    // GCC 4.8 can't infer the type coercion.
+    std::unique_ptr<opentelemetry::sdk::trace::SpanProcessor> processor(
         MakeSpanProcessor().release());
     auto tracez_provider_ = opentelemetry::nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
-        new opentelemetry::sdk::trace::TracerProvider(tracez_processor));
+        new opentelemetry::sdk::trace::TracerProvider(std::move(processor)));
     opentelemetry::trace::Provider::SetTracerProvider(tracez_provider_);
   }
 

ext/src/http/client/curl/BUILD

@@ -6,7 +6,21 @@ cc_library(
         "http_client_curl.cc",
         "http_client_factory_curl.cc",
     ],
+    # TODO: Move copts/linkopts for static CURL usage into shared bzl file.
+    copts = [
+        "-DCURL_STATICLIB",
+        "-DWITH_CURL",
+    ],
     include_prefix = "src/http/client/curl",
+    linkopts = select({
+        "//bazel:windows": [
+            "-DEFAULTLIB:advapi32.lib",
+            "-DEFAULTLIB:crypt32.lib",
+            "-DEFAULTLIB:Normaliz.lib",
+            "-DEFAULTLIB:Ws2_32.lib",
+        ],
+        "//conditions:default": [],
+    }),
     deps = [
         "//api",
         "//ext:headers",

ext/test/http/BUILD

@@ -3,19 +3,6 @@ cc_test(
     srcs = [
         "curl_http_test.cc",
     ],
-    # TODO: Move copts/linkopts for static CURL usage into shared bzl file.
-    copts = [
-        "-DCURL_STATICLIB",
-        "-DWITH_CURL",
-    ],
-    linkopts = select({
-        "//bazel:windows": [
-            "-DEFAULTLIB:advapi32.lib",
-            "-DEFAULTLIB:crypt32.lib",
-            "-DEFAULTLIB:Normaliz.lib",
-        ],
-        "//conditions:default": [],
-    }),
     deps = [
         "//ext:headers",
         "//ext/src/http/client/curl:http_client_curl",

ext/test/w3c_tracecontext_test/main.cc

@@ -38,10 +38,11 @@ void initTracer()
 {
   auto exporter = std::unique_ptr<sdktrace::SpanExporter>(
       new opentelemetry::exporter::trace::OStreamSpanExporter);
-  auto processor = std::shared_ptr<sdktrace::SpanProcessor>(
+  auto processor = std::unique_ptr<sdktrace::SpanProcessor>(
       new sdktrace::SimpleSpanProcessor(std::move(exporter)));
+  auto context  = std::make_shared<sdktrace::TracerContext>(std::move(processor));
   auto provider = nostd::shared_ptr<opentelemetry::trace::TracerProvider>(
-      new sdktrace::TracerProvider(processor));
+      new sdktrace::TracerProvider(context));
   // Set the global trace provider
   opentelemetry::trace::Provider::SetTracerProvider(provider);
 }

ext/test/zpages/tracez_data_aggregator_test.cc

@@ -35,9 +35,10 @@ class TracezDataAggregatorTest : public ::testing::Test
   void SetUp() override
   {
     std::shared_ptr<TracezSharedData> shared_data(new TracezSharedData());
-    std::shared_ptr<TracezSpanProcessor> processor(new TracezSpanProcessor(shared_data));
     auto resource = opentelemetry::sdk::resource::Resource::Create({});
-    tracer        = std::shared_ptr<opentelemetry::trace::Tracer>(new Tracer(processor, resource));
+    auto context  = std::make_shared<TracerContext>(
+        std::unique_ptr<SpanProcessor>(new TracezSpanProcessor(shared_data)), 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

@@ -178,8 +178,12 @@ class TracezProcessor : public ::testing::Test
     shared_data   = std::shared_ptr<TracezSharedData>(new TracezSharedData());
     processor     = std::shared_ptr<TracezSpanProcessor>(new TracezSpanProcessor(shared_data));
     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);
 
-    tracer     = std::shared_ptr<opentelemetry::trace::Tracer>(new Tracer(processor, resource));
+    tracer     = std::shared_ptr<opentelemetry::trace::Tracer>(new Tracer(context));
     auto spans = shared_data->GetSpanSnapshot();
     running    = spans.running;
     completed  = std::move(spans.completed);

sdk/include/opentelemetry/sdk/common/atomic_unique_ptr.h

@@ -20,6 +20,8 @@ class AtomicUniquePtr
 public:
   AtomicUniquePtr() noexcept {}
 
+  explicit AtomicUniquePtr(std::unique_ptr<T> &&other) noexcept : ptr_(other.release()) {}
+
   ~AtomicUniquePtr() noexcept { Reset(); }
 
   T &operator*() const noexcept { return *Get(); }

sdk/include/opentelemetry/sdk/trace/tracer.h

@@ -4,6 +4,7 @@
 #include "opentelemetry/sdk/resource/resource.h"
 #include "opentelemetry/sdk/trace/processor.h"
 #include "opentelemetry/sdk/trace/samplers/always_on.h"
+#include "opentelemetry/sdk/trace/tracer_context.h"
 #include "opentelemetry/trace/noop.h"
 #include "opentelemetry/trace/tracer.h"
 #include "opentelemetry/version.h"
@@ -18,33 +19,8 @@ namespace trace
 class Tracer final : public trace_api::Tracer, public std::enable_shared_from_this<Tracer>
 {
 public:
-  /**
-   * Initialize a new tracer.
-   * @param processor The span processor for this tracer. This must not be a
-   * nullptr.
-   */
-  explicit Tracer(std::shared_ptr<SpanProcessor> processor,
-                  const opentelemetry::sdk::resource::Resource &resource,
-                  std::shared_ptr<Sampler> sampler = std::make_shared<AlwaysOnSampler>()) noexcept;
-
-  /**
-   * Set the span processor associated with this tracer.
-   * @param processor The new span processor for this tracer. This must not be
-   * a nullptr.
-   */
-  void SetProcessor(std::shared_ptr<SpanProcessor> processor) noexcept;
-
-  /**
-   * Obtain the span processor associated with this tracer.
-   * @return The span processor for this tracer.
-   */
-  std::shared_ptr<SpanProcessor> GetProcessor() const noexcept;
-
-  /**
-   * Obtain the sampler associated with this tracer.
-   * @return The sampler for this tracer.
-   */
-  std::shared_ptr<Sampler> GetSampler() const noexcept;
+  /** Construct a new Tracer with the given context pipeline. */
+  explicit Tracer(std::shared_ptr<sdk::trace::TracerContext> context) noexcept;
 
   nostd::shared_ptr<trace_api::Span> StartSpan(
       nostd::string_view name,
@@ -56,10 +32,14 @@ class Tracer final : public trace_api::Tracer, public std::enable_shared_from_th
 
   void CloseWithMicroseconds(uint64_t timeout) noexcept override;
 
+  /** Returns the currently active span processor. */
+  SpanProcessor &GetActiveProcessor() noexcept { return context_->GetActiveProcessor(); }
+
+  // Note: Test only
+  Sampler &GetSampler() { return context_->GetSampler(); }
+
 private:
-  opentelemetry::sdk::common::AtomicSharedPtr<SpanProcessor> processor_;
-  const std::shared_ptr<Sampler> sampler_;
-  const opentelemetry::sdk::resource::Resource &resource_;
+  std::shared_ptr<sdk::trace::TracerContext> context_;
 };
 }  // namespace trace
 }  // namespace sdk

sdk/include/opentelemetry/sdk/trace/tracer_context.h

@@ -0,0 +1,84 @@
+#pragma once
+
+#include "opentelemetry/sdk/common/atomic_unique_ptr.h"
+#include "opentelemetry/sdk/resource/resource.h"
+#include "opentelemetry/sdk/trace/processor.h"
+#include "opentelemetry/sdk/trace/samplers/always_on.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace trace
+{
+
+/**
+ * A class which stores the TracerProvider context.
+ *
+ * This class meets the following design criteria:
+ * - A shared reference between TracerProvider and Tracers instantiated.
+ * - A thread-safe class that allows updating/altering processor/exporter pipelines
+ *   and sampling config.
+ * - The owner/destroyer of Processors/Exporters.  These will remain active until
+ *   this class is destroyed.  I.e. Sampling, Exporting, flushing etc. are all ok if this
+ *   object is alive, and they will work together. If this object is destroyed, then
+ *   no shared references to Processor, Exporter, Recordable etc. should exist, and all
+ *   associated pipelines will have been flushed.
+ */
+class TracerContext
+{
+public:
+  explicit TracerContext(std::unique_ptr<SpanProcessor> processor,
+                         opentelemetry::sdk::resource::Resource resource =
+                             opentelemetry::sdk::resource::Resource::Create({}),
+                         std::unique_ptr<Sampler> sampler =
+                             std::unique_ptr<AlwaysOnSampler>(new AlwaysOnSampler)) noexcept;
+  /**
+   * Attaches a span processor to this tracer context.
+   *
+   * @param processor The new span processor for this tracer. This must not be
+   * a nullptr. Ownership is given to the `TracerContext`.
+   */
+  void RegisterPipeline(std::unique_ptr<SpanProcessor> processor) noexcept;
+
+  /**
+   * Obtain the sampler associated with this tracer.
+   * @return The sampler for this tracer.
+   */
+  Sampler &GetSampler() const noexcept;
+
+  /**
+   * Obtain the (conceptual) active processor.
+   *
+   * Note: When more than one processor is active, this will
+   * return an "aggregate" processor
+   */
+  SpanProcessor &GetActiveProcessor() const noexcept;
+
+  /**
+   * Obtain the resource associated with this tracer context.
+   * @return The resource for this tracer context.
+   */
+  const opentelemetry::sdk::resource::Resource &GetResource() const noexcept;
+
+  /**
+   * Force all active SpanProcessors to flush any buffered spans
+   * within the given timeout.
+   */
+  bool ForceFlush(std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept;
+
+  /**
+   * Shutdown the span processor associated with this tracer provider.
+   */
+  bool Shutdown() noexcept;
+
+private:
+  // This is an atomic pointer so we can adapt the processor pipeline dynamically.
+  opentelemetry::sdk::common::AtomicUniquePtr<SpanProcessor> processor_;
+  opentelemetry::sdk::resource::Resource resource_;
+  std::unique_ptr<Sampler> sampler_;
+};
+
+}  // namespace trace
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE