[
https://issues.apache.org/jira/browse/PROTON-2487?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17509089#comment-17509089
]
ASF GitHub Bot commented on PROTON-2487:
----------------------------------------
astitcher commented on a change in pull request #355:
URL: https://github.com/apache/qpid-proton/pull/355#discussion_r830361622
##########
File path: cpp/src/tracing_private.hpp
##########
@@ -0,0 +1,54 @@
+#ifndef PROTON_INTERNAL_TRACING_HPP
+#define PROTON_INTERNAL_TRACING_HPP
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+namespace proton {
+
+class message;
+class delivery;
+class tracker;
+class binary;
+
+/// Base class for OpentelemetryTracing and DummyTracing. An interface for
Tracing.
+class Tracing {
+
+ static Tracing* the;
+
+ public:
+ /// Returns an object for the Tracing. If Tracing is enable, returns the
object of OpentelemetryTracing.
+ /// By default returns the object of DummyTracing.
+ static Tracing& getTracing();
+
+ /// Initialize the Tracing object with the OpentelemetryTracing object.
+ static void activate(Tracing& r);
+
+ virtual void message_encode(const message &m, std::vector<char> &buf,
const binary &tag, const tracker &track) = 0;
+ virtual void on_message_handler(messaging_handler& h, delivery& d,
message& message) = 0;
+ virtual void on_settled_span(delivery& d, tracker& track) = 0;
+ virtual void send_span(const message& message, const binary& tag, const
tracker& track) = 0;
Review comment:
This abstract method is now not needed.
What it does is now part of the OpentelemetryTracing implementation only and
so should just be part of Opentelemetry::message_encode.
##########
File path: cpp/src/tracing.cpp
##########
@@ -0,0 +1,255 @@
+/*
Review comment:
This file should be renemaed tracing_opentelemetry.cpp as it is the
Opentelemetry implementation of tracing - in theory there could be others.
##########
File path: cpp/src/tracing_dummy.cpp
##########
@@ -0,0 +1,54 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#include <proton/tracing.hpp>
+#include <proton/annotation_key.hpp>
+#include <proton/message.hpp>
+#include <proton/messaging_handler.hpp>
+
+#include "tracing_private.hpp"
+
+namespace proton
+{
+
+class DummyTracing : public Tracing {
Review comment:
Probably should rename this to StubTracing in line with new proposed
filename.
##########
File path: cpp/examples/tracing_demo.cpp
##########
@@ -0,0 +1,112 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#include <proton/connection.hpp>
+#include <proton/container.hpp>
+#include <proton/delivery.hpp>
+#include <proton/message.hpp>
+#include <proton/messaging_handler.hpp>
+#include <proton/tracker.hpp>
+#include <proton/tracing.hpp>
+
+#include <iostream>
+#include <sstream>
+#include <thread>
+#include <map>
+#include <string>
+#include <cctype>
+
+class tracing_demo : public proton::messaging_handler {
+ std::string conn_url_;
+ std::string addr_;
+ proton::connection conn_;
+ std::map<std::string, proton::sender> senders_;
+
+ public:
+ tracing_demo(const std::string& u, const std::string& a) :
+ conn_url_(u), addr_(a) {}
+
+ void on_container_start(proton::container& c) override {
+ conn_ = c.connect(conn_url_);
+ }
+
+ void on_connection_open(proton::connection& c) override {
+ c.open_receiver(addr_);
+ c.open_sender(addr_);
+ }
+
+ void on_sendable(proton::sender &s) override {
+ proton::message m("Hello Tracing World!");
+ s.send(m);
+ std::cout << "Message sent: " << m.body() << std::endl;
+ s.close();
+ }
+
+ void on_tracker_accept(proton::tracker &t) override {
+ std::cout << "all messages confirmed" << std::endl;
+ }
+
+ std::string to_upper(const std::string& s) {
+ std::string uc(s);
+ size_t l = uc.size();
+
+ for (size_t i=0; i<l; i++) {
+ uc[i] = static_cast<char>(std::toupper(uc[i]));
+ }
+
+ return uc;
+ }
+
+ void on_message(proton::delivery &d, proton::message &m) override {
+ std::cout << "Message received: " << m.body() << std::endl;
+
+ std::string reply_to = m.reply_to();
+ proton::message reply;
+
+ reply.to(reply_to);
+ reply.body(to_upper(proton::get<std::string>(m.body())));
+
+ if (!senders_[reply_to]) {
+ senders_[reply_to] = conn_.open_sender(reply_to);
+ }
+
+ senders_[reply_to].send(reply);
+ d.connection().close();
+ }
+
+};
+
+int main(int argc, char **argv) {
+ try {
+ std::string conn_url = argc > 1 ? argv[1] : "//127.0.0.1:5672";
+ std::string addr = argc > 2 ? argv[2] : "examples";
+
+ proton::initTracer(jaeger);
Review comment:
I think it would be worthwhile calling out that this is the only thing
that is necessary to make this example a tracing example.
##########
File path: cpp/examples/CMakeLists.txt
##########
@@ -89,3 +89,10 @@ foreach(example
add_executable(${example} ${example}.cpp)
target_link_libraries(${example} Proton::cpp Threads::Threads)
endforeach()
+
+if (ENABLE_OPENTELEMETRYCPP)
+ set(example tracing_demo)
+ add_executable(${example} ${example}.cpp)
+ include_directories(${CMAKE_SOURCE_DIR}/exporters/jaeger/include)
+ target_link_libraries(${example} Proton::cpp opentelemetry_trace
opentelemetry_exporter_jaeger_trace)
+endif()
Review comment:
As far as I can tell the tracing example should compile without any
special attention to whether Opentelemetry is compiled into Proton or not.
Obviously if there is no real tracing library then this example will do no
tracing but I don't think any of the stuff in the conditional should be
necessary - if it is necessary then we need to debug why and fix
that.(excluding the `add_executable(..)` which I'm suggesting could just go
directly into the single threaded examples foreach above.
##########
File path: cpp/src/tracing_private.hpp
##########
@@ -0,0 +1,54 @@
+#ifndef PROTON_INTERNAL_TRACING_HPP
+#define PROTON_INTERNAL_TRACING_HPP
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+namespace proton {
+
+class message;
+class delivery;
+class tracker;
+class binary;
+
+/// Base class for OpentelemetryTracing and DummyTracing. An interface for
Tracing.
+class Tracing {
+
+ static Tracing* the;
+
+ public:
+ /// Returns an object for the Tracing. If Tracing is enable, returns the
object of OpentelemetryTracing.
+ /// By default returns the object of DummyTracing.
+ static Tracing& getTracing();
+
+ /// Initialize the Tracing object with the OpentelemetryTracing object.
+ static void activate(Tracing& r);
+
+ virtual void message_encode(const message &m, std::vector<char> &buf,
const binary &tag, const tracker &track) = 0;
+ virtual void on_message_handler(messaging_handler& h, delivery& d,
message& message) = 0;
+ virtual void on_settled_span(delivery& d, tracker& track) = 0;
Review comment:
I think this may have the wrong signature you shouldn't need a delivery
here at all in fact it's even really wrong as the send side only has trackers
not deliveries. You should be able to get the tag from the tracker - if there
is no C++ API then use the C API directly.
##########
File path: cpp/src/tracing_dummy.cpp
##########
@@ -0,0 +1,54 @@
+/*
Review comment:
This file should be renamed as tracing_stub.cpp to be in line with
naming in other places.
##########
File path: cpp/src/tracing.cpp
##########
@@ -0,0 +1,255 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#include <opentelemetry/sdk/trace/simple_processor.h>
+#include <opentelemetry/sdk/trace/tracer_provider.h>
+#include <opentelemetry/trace/provider.h>
+#include <opentelemetry/trace/span.h>
+#include <opentelemetry/trace/tracer.h>
+#include <opentelemetry/trace/context.h>
+#include <opentelemetry/nostd/unique_ptr.h>
+#include <opentelemetry/context/propagation/global_propagator.h>
+#include <opentelemetry/trace/propagation/http_trace_context.h>
+#include <opentelemetry/exporters/jaeger/jaeger_exporter.h>
+#include <opentelemetry/exporters/ostream/span_exporter.h>
+
+#include <proton/messaging_handler.hpp>
+
+#include <proton/annotation_key.hpp>
+#include <proton/delivery.hpp>
+#include <proton/message.hpp>
+#include <proton/receiver.hpp>
+#include <proton/sender.hpp>
+#include <proton/source.hpp>
+#include <proton/target.hpp>
+#include <proton/tracing.hpp>
+#include <proton/tracker.hpp>
+#include <proton/transfer.hpp>
+
+#include "proton/link.hpp"
+#include <proton/link.h>
+
+#include "tracing_private.hpp"
+
+#include <iostream>
+#include <sstream>
+#include <memory>
+
+namespace proton
+{
+namespace nostd = opentelemetry::nostd;
+namespace sdktrace = opentelemetry::sdk::trace;
+
+const std::string kContextKey = "x-opt-qpid-tracestate";
+
+// TODO: Delete map entries to avoid memory leakage in future
+std::map<binary, nostd::shared_ptr<opentelemetry::trace::Span>> tag_span;
+
+class AMQPMapCarrier : public
opentelemetry::context::propagation::TextMapCarrier {
+ public:
+ AMQPMapCarrier(const proton::map<annotation_key, value>*
message_annotations) : message_annotations_(message_annotations) {}
+ virtual nostd::string_view Get(nostd::string_view key) const noexcept
override {
+ std::string key_to_compare = key.data();
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ if (tracing_map.exists(annotation_key(key_to_compare))) {
+ value extracted_value =
tracing_map.get(annotation_key(key_to_compare));
+ std::string extracted_string = to_string(extracted_value);
+ extracted_strings.push_back(extracted_string);
+ nostd::string_view final_extracted_string =
nostd::string_view(extracted_strings.back());
+
+ return final_extracted_string;
+ }
+ return "";
+ }
+ virtual void Set(nostd::string_view key,
+ nostd::string_view val) noexcept override {
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ tracing_map.put(annotation_key(std::string(key)),
value(std::string(val)));
+ ((proton::map<proton::annotation_key,
proton::value>*)message_annotations_)->put(annotation_key(kContextKey),
tracing_map);
+ }
+
+ const proton::map<annotation_key, value>* message_annotations_;
+ mutable std::vector<std::string> extracted_strings;
+};
+
+nostd::shared_ptr<opentelemetry::trace::Tracer> get_tracer() {
+ auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+ nostd::shared_ptr<opentelemetry::trace::Tracer> tracer =
provider->GetTracer("qpid-tracer");
+ return tracer;
+}
+
+class OpentelemetryTracing : public Tracing {
+ public:
+ void message_encode(const message& m, std::vector<char>& buf, const
binary& tag, const tracker& track) override {
+ proton::message message_cp = m;
+ Tracing& ot = Tracing::getTracing();
+ ot.send_span(message_cp, tag, track);
+ message_cp.encode(buf);
+ }
+
+ void on_message_handler(messaging_handler& h, delivery& d, message&
message) override {
+ opentelemetry::trace::StartSpanOptions options;
+ options.kind = opentelemetry::trace::SpanKind::kConsumer;
+
+ // Extract context from AMQP message annotations
+ const AMQPMapCarrier carrier(&message.message_annotations());
+
+
nostd::shared_ptr<opentelemetry::context::propagation::TextMapPropagator> prop =
+
opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
+ opentelemetry::context::Context current_ctx =
opentelemetry::context::RuntimeContext::GetCurrent();
+
+ opentelemetry::context::Context new_context = prop->Extract(carrier,
current_ctx);
+
+ options.parent =
opentelemetry::trace::GetSpan(new_context)->GetContext();
+
+ binary tag_in_binary = d.tag();
+ std::string tag_in_string = std::string(d.tag());
+ std::stringstream ss;
+ for (int i = 0; i < (int)tag_in_string.length(); ++i)
+ ss << std::hex << (int)tag_in_binary[i];
+ std::string delivery_tag = ss.str();
+
+ receiver r = d.receiver();
+ source s = r.source();
+ std::string s_addr = s.address();
+
+ transfer tt(d);
+ std::string delivery_state = to_string(tt.state());
+
+ nostd::shared_ptr<opentelemetry::trace::Span> span =
get_tracer()->StartSpan(
+ "amqp-message-received",
+ {{"Delivery_tag", delivery_tag}, {"Source_address", s_addr}},
+ options);
+
+ opentelemetry::trace::Scope scope = get_tracer()->WithActiveSpan(span);
+
+ h.on_message(d, message);
+
+ span->End();
+ }
+
+ void on_settled_span(delivery& d, tracker& track) override {
+
+ binary tag = d.tag();
+ nostd::shared_ptr<opentelemetry::trace::Span> span = tag_span[tag];
+ transfer tt(track);
+ std::string delivery_state = to_string(tt.state());
+ span->AddEvent("delivery state: " + delivery_state);
+
+ span->End();
+ }
+
+ void send_span(const message& message, const binary& tag, const tracker&
track) override {
Review comment:
This should be folded into message_encode and removed.
##########
File path: cpp/src/tracing.cpp
##########
@@ -0,0 +1,255 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#include <opentelemetry/sdk/trace/simple_processor.h>
+#include <opentelemetry/sdk/trace/tracer_provider.h>
+#include <opentelemetry/trace/provider.h>
+#include <opentelemetry/trace/span.h>
+#include <opentelemetry/trace/tracer.h>
+#include <opentelemetry/trace/context.h>
+#include <opentelemetry/nostd/unique_ptr.h>
+#include <opentelemetry/context/propagation/global_propagator.h>
+#include <opentelemetry/trace/propagation/http_trace_context.h>
+#include <opentelemetry/exporters/jaeger/jaeger_exporter.h>
+#include <opentelemetry/exporters/ostream/span_exporter.h>
+
+#include <proton/messaging_handler.hpp>
+
+#include <proton/annotation_key.hpp>
+#include <proton/delivery.hpp>
+#include <proton/message.hpp>
+#include <proton/receiver.hpp>
+#include <proton/sender.hpp>
+#include <proton/source.hpp>
+#include <proton/target.hpp>
+#include <proton/tracing.hpp>
+#include <proton/tracker.hpp>
+#include <proton/transfer.hpp>
+
+#include "proton/link.hpp"
+#include <proton/link.h>
+
+#include "tracing_private.hpp"
+
+#include <iostream>
+#include <sstream>
+#include <memory>
+
+namespace proton
+{
+namespace nostd = opentelemetry::nostd;
+namespace sdktrace = opentelemetry::sdk::trace;
+
+const std::string kContextKey = "x-opt-qpid-tracestate";
+
+// TODO: Delete map entries to avoid memory leakage in future
+std::map<binary, nostd::shared_ptr<opentelemetry::trace::Span>> tag_span;
+
+class AMQPMapCarrier : public
opentelemetry::context::propagation::TextMapCarrier {
+ public:
+ AMQPMapCarrier(const proton::map<annotation_key, value>*
message_annotations) : message_annotations_(message_annotations) {}
+ virtual nostd::string_view Get(nostd::string_view key) const noexcept
override {
+ std::string key_to_compare = key.data();
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ if (tracing_map.exists(annotation_key(key_to_compare))) {
+ value extracted_value =
tracing_map.get(annotation_key(key_to_compare));
+ std::string extracted_string = to_string(extracted_value);
+ extracted_strings.push_back(extracted_string);
+ nostd::string_view final_extracted_string =
nostd::string_view(extracted_strings.back());
+
+ return final_extracted_string;
+ }
+ return "";
+ }
+ virtual void Set(nostd::string_view key,
+ nostd::string_view val) noexcept override {
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ tracing_map.put(annotation_key(std::string(key)),
value(std::string(val)));
+ ((proton::map<proton::annotation_key,
proton::value>*)message_annotations_)->put(annotation_key(kContextKey),
tracing_map);
+ }
+
+ const proton::map<annotation_key, value>* message_annotations_;
+ mutable std::vector<std::string> extracted_strings;
+};
+
+nostd::shared_ptr<opentelemetry::trace::Tracer> get_tracer() {
+ auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+ nostd::shared_ptr<opentelemetry::trace::Tracer> tracer =
provider->GetTracer("qpid-tracer");
+ return tracer;
+}
+
+class OpentelemetryTracing : public Tracing {
+ public:
+ void message_encode(const message& m, std::vector<char>& buf, const
binary& tag, const tracker& track) override {
+ proton::message message_cp = m;
+ Tracing& ot = Tracing::getTracing();
Review comment:
You never need to use getTracing from within a tracing member function
as you use this there or in this case just access the member function directly.
As in `send_span(...)`
However I suggest removing send span completely.
Or if you really want to leave it then it (which would be fine) it doesn't
need to be virtual or to override the parent class as it is only used here in
this class and no one else needs to call it (so probably should be private as
well).
##########
File path: cpp/src/tracing.cpp
##########
@@ -0,0 +1,255 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#include <opentelemetry/sdk/trace/simple_processor.h>
+#include <opentelemetry/sdk/trace/tracer_provider.h>
+#include <opentelemetry/trace/provider.h>
+#include <opentelemetry/trace/span.h>
+#include <opentelemetry/trace/tracer.h>
+#include <opentelemetry/trace/context.h>
+#include <opentelemetry/nostd/unique_ptr.h>
+#include <opentelemetry/context/propagation/global_propagator.h>
+#include <opentelemetry/trace/propagation/http_trace_context.h>
+#include <opentelemetry/exporters/jaeger/jaeger_exporter.h>
+#include <opentelemetry/exporters/ostream/span_exporter.h>
+
+#include <proton/messaging_handler.hpp>
+
+#include <proton/annotation_key.hpp>
+#include <proton/delivery.hpp>
+#include <proton/message.hpp>
+#include <proton/receiver.hpp>
+#include <proton/sender.hpp>
+#include <proton/source.hpp>
+#include <proton/target.hpp>
+#include <proton/tracing.hpp>
+#include <proton/tracker.hpp>
+#include <proton/transfer.hpp>
+
+#include "proton/link.hpp"
+#include <proton/link.h>
+
+#include "tracing_private.hpp"
+
+#include <iostream>
+#include <sstream>
+#include <memory>
+
+namespace proton
+{
+namespace nostd = opentelemetry::nostd;
+namespace sdktrace = opentelemetry::sdk::trace;
+
+const std::string kContextKey = "x-opt-qpid-tracestate";
+
+// TODO: Delete map entries to avoid memory leakage in future
+std::map<binary, nostd::shared_ptr<opentelemetry::trace::Span>> tag_span;
+
+class AMQPMapCarrier : public
opentelemetry::context::propagation::TextMapCarrier {
+ public:
+ AMQPMapCarrier(const proton::map<annotation_key, value>*
message_annotations) : message_annotations_(message_annotations) {}
+ virtual nostd::string_view Get(nostd::string_view key) const noexcept
override {
+ std::string key_to_compare = key.data();
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ if (tracing_map.exists(annotation_key(key_to_compare))) {
+ value extracted_value =
tracing_map.get(annotation_key(key_to_compare));
+ std::string extracted_string = to_string(extracted_value);
+ extracted_strings.push_back(extracted_string);
+ nostd::string_view final_extracted_string =
nostd::string_view(extracted_strings.back());
+
+ return final_extracted_string;
+ }
+ return "";
+ }
+ virtual void Set(nostd::string_view key,
+ nostd::string_view val) noexcept override {
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ tracing_map.put(annotation_key(std::string(key)),
value(std::string(val)));
+ ((proton::map<proton::annotation_key,
proton::value>*)message_annotations_)->put(annotation_key(kContextKey),
tracing_map);
+ }
+
+ const proton::map<annotation_key, value>* message_annotations_;
+ mutable std::vector<std::string> extracted_strings;
+};
+
+nostd::shared_ptr<opentelemetry::trace::Tracer> get_tracer() {
+ auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+ nostd::shared_ptr<opentelemetry::trace::Tracer> tracer =
provider->GetTracer("qpid-tracer");
+ return tracer;
+}
+
+class OpentelemetryTracing : public Tracing {
+ public:
+ void message_encode(const message& m, std::vector<char>& buf, const
binary& tag, const tracker& track) override {
+ proton::message message_cp = m;
+ Tracing& ot = Tracing::getTracing();
+ ot.send_span(message_cp, tag, track);
+ message_cp.encode(buf);
+ }
+
+ void on_message_handler(messaging_handler& h, delivery& d, message&
message) override {
+ opentelemetry::trace::StartSpanOptions options;
+ options.kind = opentelemetry::trace::SpanKind::kConsumer;
+
+ // Extract context from AMQP message annotations
+ const AMQPMapCarrier carrier(&message.message_annotations());
+
+
nostd::shared_ptr<opentelemetry::context::propagation::TextMapPropagator> prop =
+
opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
+ opentelemetry::context::Context current_ctx =
opentelemetry::context::RuntimeContext::GetCurrent();
+
+ opentelemetry::context::Context new_context = prop->Extract(carrier,
current_ctx);
+
+ options.parent =
opentelemetry::trace::GetSpan(new_context)->GetContext();
+
+ binary tag_in_binary = d.tag();
+ std::string tag_in_string = std::string(d.tag());
+ std::stringstream ss;
+ for (int i = 0; i < (int)tag_in_string.length(); ++i)
+ ss << std::hex << (int)tag_in_binary[i];
+ std::string delivery_tag = ss.str();
+
+ receiver r = d.receiver();
+ source s = r.source();
+ std::string s_addr = s.address();
+
+ transfer tt(d);
+ std::string delivery_state = to_string(tt.state());
+
+ nostd::shared_ptr<opentelemetry::trace::Span> span =
get_tracer()->StartSpan(
+ "amqp-message-received",
+ {{"Delivery_tag", delivery_tag}, {"Source_address", s_addr}},
+ options);
+
+ opentelemetry::trace::Scope scope = get_tracer()->WithActiveSpan(span);
+
+ h.on_message(d, message);
+
+ span->End();
+ }
+
+ void on_settled_span(delivery& d, tracker& track) override {
+
+ binary tag = d.tag();
+ nostd::shared_ptr<opentelemetry::trace::Span> span = tag_span[tag];
+ transfer tt(track);
+ std::string delivery_state = to_string(tt.state());
Review comment:
I don't understand why we need a new copy of the transfer (`tt`) just to
get a string version of the delivery state, we should be able to get that
directly from the original `track` without copying.
##########
File path: cpp/src/tracing.cpp
##########
@@ -0,0 +1,255 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#include <opentelemetry/sdk/trace/simple_processor.h>
+#include <opentelemetry/sdk/trace/tracer_provider.h>
+#include <opentelemetry/trace/provider.h>
+#include <opentelemetry/trace/span.h>
+#include <opentelemetry/trace/tracer.h>
+#include <opentelemetry/trace/context.h>
+#include <opentelemetry/nostd/unique_ptr.h>
+#include <opentelemetry/context/propagation/global_propagator.h>
+#include <opentelemetry/trace/propagation/http_trace_context.h>
+#include <opentelemetry/exporters/jaeger/jaeger_exporter.h>
+#include <opentelemetry/exporters/ostream/span_exporter.h>
+
+#include <proton/messaging_handler.hpp>
+
+#include <proton/annotation_key.hpp>
+#include <proton/delivery.hpp>
+#include <proton/message.hpp>
+#include <proton/receiver.hpp>
+#include <proton/sender.hpp>
+#include <proton/source.hpp>
+#include <proton/target.hpp>
+#include <proton/tracing.hpp>
+#include <proton/tracker.hpp>
+#include <proton/transfer.hpp>
+
+#include "proton/link.hpp"
+#include <proton/link.h>
+
+#include "tracing_private.hpp"
+
+#include <iostream>
+#include <sstream>
+#include <memory>
+
+namespace proton
+{
+namespace nostd = opentelemetry::nostd;
+namespace sdktrace = opentelemetry::sdk::trace;
+
+const std::string kContextKey = "x-opt-qpid-tracestate";
+
+// TODO: Delete map entries to avoid memory leakage in future
+std::map<binary, nostd::shared_ptr<opentelemetry::trace::Span>> tag_span;
+
+class AMQPMapCarrier : public
opentelemetry::context::propagation::TextMapCarrier {
+ public:
+ AMQPMapCarrier(const proton::map<annotation_key, value>*
message_annotations) : message_annotations_(message_annotations) {}
+ virtual nostd::string_view Get(nostd::string_view key) const noexcept
override {
+ std::string key_to_compare = key.data();
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ if (tracing_map.exists(annotation_key(key_to_compare))) {
+ value extracted_value =
tracing_map.get(annotation_key(key_to_compare));
+ std::string extracted_string = to_string(extracted_value);
+ extracted_strings.push_back(extracted_string);
+ nostd::string_view final_extracted_string =
nostd::string_view(extracted_strings.back());
+
+ return final_extracted_string;
+ }
+ return "";
+ }
+ virtual void Set(nostd::string_view key,
+ nostd::string_view val) noexcept override {
+
+ proton::value v_tracing_map =
message_annotations_->get(annotation_key(kContextKey));
+ proton::map<proton::annotation_key, proton::value> tracing_map;
+
+ if (!v_tracing_map.empty())
+ get(v_tracing_map, tracing_map);
+
+ tracing_map.put(annotation_key(std::string(key)),
value(std::string(val)));
+ ((proton::map<proton::annotation_key,
proton::value>*)message_annotations_)->put(annotation_key(kContextKey),
tracing_map);
+ }
+
+ const proton::map<annotation_key, value>* message_annotations_;
+ mutable std::vector<std::string> extracted_strings;
+};
+
+nostd::shared_ptr<opentelemetry::trace::Tracer> get_tracer() {
+ auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+ nostd::shared_ptr<opentelemetry::trace::Tracer> tracer =
provider->GetTracer("qpid-tracer");
+ return tracer;
+}
+
+class OpentelemetryTracing : public Tracing {
+ public:
+ void message_encode(const message& m, std::vector<char>& buf, const
binary& tag, const tracker& track) override {
+ proton::message message_cp = m;
+ Tracing& ot = Tracing::getTracing();
+ ot.send_span(message_cp, tag, track);
+ message_cp.encode(buf);
+ }
+
+ void on_message_handler(messaging_handler& h, delivery& d, message&
message) override {
+ opentelemetry::trace::StartSpanOptions options;
+ options.kind = opentelemetry::trace::SpanKind::kConsumer;
+
+ // Extract context from AMQP message annotations
+ const AMQPMapCarrier carrier(&message.message_annotations());
+
+
nostd::shared_ptr<opentelemetry::context::propagation::TextMapPropagator> prop =
+
opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
+ opentelemetry::context::Context current_ctx =
opentelemetry::context::RuntimeContext::GetCurrent();
+
+ opentelemetry::context::Context new_context = prop->Extract(carrier,
current_ctx);
+
+ options.parent =
opentelemetry::trace::GetSpan(new_context)->GetContext();
+
+ binary tag_in_binary = d.tag();
+ std::string tag_in_string = std::string(d.tag());
+ std::stringstream ss;
+ for (int i = 0; i < (int)tag_in_string.length(); ++i)
+ ss << std::hex << (int)tag_in_binary[i];
+ std::string delivery_tag = ss.str();
+
+ receiver r = d.receiver();
+ source s = r.source();
+ std::string s_addr = s.address();
+
+ transfer tt(d);
+ std::string delivery_state = to_string(tt.state());
+
+ nostd::shared_ptr<opentelemetry::trace::Span> span =
get_tracer()->StartSpan(
+ "amqp-message-received",
+ {{"Delivery_tag", delivery_tag}, {"Source_address", s_addr}},
+ options);
+
+ opentelemetry::trace::Scope scope = get_tracer()->WithActiveSpan(span);
+
+ h.on_message(d, message);
+
+ span->End();
+ }
+
+ void on_settled_span(delivery& d, tracker& track) override {
+
Review comment:
Shouldn't need to pass in a delivery only a tracker as this is sender
side.
##########
File path: cpp/src/tracing_dummy.cpp
##########
@@ -0,0 +1,54 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#include <proton/tracing.hpp>
+#include <proton/annotation_key.hpp>
+#include <proton/message.hpp>
+#include <proton/messaging_handler.hpp>
+
+#include "tracing_private.hpp"
+
+namespace proton
+{
+
+class DummyTracing : public Tracing {
+ public:
+ void message_encode(const message& m, std::vector<char>& buf, const
binary& tag, const tracker& track) override {
+ m.encode(buf);
+ }
+
+ void on_message_handler(messaging_handler& h, delivery& d, message&
message) override {
+ h.on_message(d, message);
+ }
+
+ void on_settled_span(delivery& d, tracker& track) override {}
+
+ void send_span(const message& message, const binary& tag, const tracker&
track) override {}
Review comment:
shouldn't be needed - remove from parent class.
##########
File path: cpp/examples/README.dox
##########
@@ -137,3 +137,55 @@ A multithreaded sender and receiver enhanced for flow
control.
__Requires C++11__
*/
+
+/** @example tracing_demo.cpp
+
+A working example of distributed tracing.
+
+Steps to run this example:
+
+# Start Jaeger, for example:
+
+docker run -d --name jaeger \
+ -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
+ -p 5775:5775/udp \
+ -p 6831:6831/udp \
+ -p 6832:6832/udp \
+ -p 5778:5778 \
+ -p 16686:16686 \
+ -p 14268:14268 \
+ -p 14250:14250 \
+ -p 9411:9411 \
+ jaegertracing/all-in-one:1.25
+
+# Opentelemetry-cpp:
+
+1. Clone opentelemetry-cpp
+
+WORKDIR opentelemetry-cpp
+2. RUN mkdir bld
+
+WORKDIR bld
+3. RUN cmake .. -DBUILD_TESTING=OFF -DBUILD_SHARED_LIBS=ON
-DCMAKE_POSITION_INDEPENDENT_CODE=ON -DWITH_JAEGER=ON
+4. RUN make
+5. RUN sudo make install
+
+# Start broker, for example:
+
+cd examples
+./broker
+
+# Build and run the tracing_demo example:
+
+WORKDIR examples
+RUN ./tracing_demo
+
+# Look in the Jaeger UI:
+
+Browse to http://localhost:16686
+Select the Service dropdown at the top of the Search options (if not already
selected).
+Hit Find Traces.
+
+__Requires C++11 opentelemetry-cpp__
+
+*/
Review comment:
I think we should consider putting this whole section of the examples
readme into its own file - the instructions for running the tracing
infrastructure are intricate and might perhaps confuse people who aren't
interested in tracing?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
> [cpp] Implement distributed tracing
> -----------------------------------
>
> Key: PROTON-2487
> URL: https://issues.apache.org/jira/browse/PROTON-2487
> Project: Qpid Proton
> Issue Type: New Feature
> Components: cpp-binding
> Affects Versions: proton-c-0.37.0
> Reporter: Rakhi Kumari
> Assignee: Rakhi Kumari
> Priority: Major
> Fix For: proton-c-0.38.0
>
>
> Implement distributed tracing using opentelemetry.
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
