Create shared context for updating span pipeline from TracerProvider and affecting Tracer. #650
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| cc_binary( | ||
| name = "example_http_client", | ||
| srcs = [ | ||
| "tracer_common.hpp", | ||
| "client.cc", | ||
| ], | ||
| deps = [ | ||
| "//api", | ||
| "//ext:headers", | ||
| "//ext/src/http/client/curl:http_client_curl", | ||
| "//exporters/ostream:ostream_span_exporter", | ||
| "//sdk/src/trace", | ||
| ], | ||
| # 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": [], | ||
| }), | ||
| ) | ||
|
|
||
| cc_binary( | ||
| name = "example_http_server", | ||
| srcs = [ | ||
| "server.cc", | ||
| "tracer_common.hpp", | ||
| "server.hpp", | ||
| ], | ||
| deps = [ | ||
| "//api", | ||
| "//ext:headers", | ||
| "//exporters/ostream:ostream_span_exporter", | ||
| "//sdk/src/trace", | ||
| ], | ||
| ) | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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(); } | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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: | ||
| /** 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: | ||
| std::shared_ptr<sdk::trace::TracerContext> context_; | ||
| }; | ||
| } // namespace trace | ||
| } // namespace sdk | ||
|
|
||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,85 @@ | ||||||
| #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`. | ||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||
| */ | ||||||
| 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; | ||||||
|
Comment on lines
+51
to
+56
There was a problem hiding this comment. What would be an inactive processor? There was a problem hiding this comment. I don't think there is any There was a problem hiding this comment. 'inactive' would be any constructed but not registered processor. I'm not tied to this name, just took it from Java ;) There was a problem hiding this comment. Perhaps just name it There was a problem hiding this comment. We might theoretically run into issue here when some tracer does some work with a processor obtained by There was a problem hiding this comment. Shouldn't be the case. The interface is "RegisterPipeline", and it's a shared pointer. The TracerContext can guarantee that ALL registered processors remain alive for its own lifecycle. I.e. we're not going to delete a pipeline, we're only going to allow registration of new ones specifically to prevent this issue. We have the inverse problem around Recordable though, specifically if we have an "active" span that did NOT get a recordable registered when a new pipeline is registered, what happens when that span ends? If you look at my design around "ExportableSpan" where a pipeline can ask for its instance of Recordable (and handle a missing recordable), that's how I planned to solve that issue. There was a problem hiding this comment. I think I understand. So if a new pipeline is registered, the old one isn't just overwritten, but kept in store and just made inactive? There was a problem hiding this comment. My understanding was that all the pipelines(i.e, processors) remain active. And we can only add a new pipeline, but can't remove the existing one? And once we add multi-processor support, There was a problem hiding this comment. Yeah. i'm adopting the design I've seen elsewhere where you can "append a pipeline", but can't remove an existing pipeline. I.e. once a pipeline is active, it's forever active. We don't have a register/unregister mechanism. |
||||||
|
|
||||||
| /** | ||||||
| * 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 | ||||||
|
There was a problem hiding this comment. I'm a bit confused by all active span processor. Wouldn't we just have a There was a problem hiding this comment. Personally, I wouldn't mind leaving There was a problem hiding this comment. This is for users of TracerContext to understand what it does. Yes the implementation of this behavior may be implemented by "CompositeSpanProcessor" but it's not something that would be directly exposed, instead users just see "RegisterPipeline" and "Shutdown" and "ActiveProcessor". I.e. we still need to document the behavior of this class and how to use its methods. Note: I'm not tied to "active" processor, I borrowed the name from the Java SDK. |
||||||
| * 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 | ||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for adding the bazel build for this.