Feature/tracing poc

.github/workflows/build-ubuntu.yml

@@ -169,6 +169,12 @@ jobs:
       # Always save cache if configure succeeded (even if the build failed)
       if: ${{ always() && steps.cmake-configure.outcome == 'success' }}
 
+    - name: Create tracing output directory
+      run: |
+        ECAL_TRACING_DATA_DIR="${{ runner.temp }}/ecal_tracing"
+        mkdir -p "$ECAL_TRACING_DATA_DIR"
+        echo "ECAL_TRACING_DATA_DIR=$ECAL_TRACING_DATA_DIR" >> "$GITHUB_ENV"
+
     - name: Run Tests
       run: ctest -V --test-dir _build
 

.github/workflows/build-windows.yml

@@ -200,6 +200,13 @@ jobs:
       run: cmake --build "${{ runner.workspace }}/_build/csharp" --config Release
       shell: cmd
 
+    - name: Create tracing output directory
+      run: |
+        $TRACING_DIR = "${{ runner.temp }}\ecal_tracing"
+        mkdir $TRACING_DIR
+        echo "ECAL_TRACING_DATA_DIR=$TRACING_DIR" >> $Env:GITHUB_ENV
+      shell: powershell
+
     - name: Run C# Tests
       run: ctest -C Release -V --test-dir "${{ runner.workspace }}/_build/csharp"
 

cmake/submodule_dependencies.cmake

@@ -15,6 +15,7 @@ set(ecal_submodule_dependencies
   HDF5
   #libssh2
   nanobind
+  nlohmann_json
   Protobuf
   protozero
   qwt

conanfile.py

@@ -17,6 +17,7 @@ def build_requirements(self):
     def requirements(self):
         self.requires("hdf5/1.10.6")
         self.requires("protobuf/3.17.1")
+        self.requires("nlohmann_json/3.11.2")
         self.requires("libcurl/7.78.0")
         self.requires("qt/5.15.2")
         self.requires("spdlog/1.9.2")
@@ -51,6 +52,6 @@ def generate(self):
         else:
             tc.variables["Protobuf_PROTOC_EXECUTABLE"] = os.path.join(self.deps_cpp_info["protobuf"].rootpath, "bin", "protoc")
         tc.generate()
-        
-        
-       
+
+
+

ecal/core/CMakeLists.txt

@@ -43,6 +43,7 @@ find_package(asio             REQUIRED)
 find_package(Threads          REQUIRED)
 find_package(ecaludp          REQUIRED)
 find_package(protozero        REQUIRED)
+find_package(nlohmann_json    REQUIRED)
 
 if (ECAL_CORE_CONFIGURATION)
   find_package(yaml-cpp       REQUIRED)
@@ -459,6 +460,20 @@ if(ECAL_CORE_TIMEPLUGIN)
   )
 endif()
 
+######################################
+# tracing
+######################################
+set(ecal_tracing_src
+    src/tracing/tracing.h
+    src/tracing/span.cpp
+    src/tracing/span.h
+    src/tracing/trace_provider.cpp
+    src/tracing/trace_provider.h
+    src/tracing/tracing_writer.h
+    src/tracing/tracing_writer_jsonl.cpp
+    src/tracing/tracing_writer_jsonl.h
+)
+
 ######################################
 # util
 ######################################
@@ -637,6 +652,7 @@ set(ecal_sources
     ${ecal_serialization_src}
     ${ecal_service_src}
     ${ecal_time_src}
+    ${ecal_tracing_src}
     ${ecal_util_src}
     ${ecal_cmn_src}
     ${ecal_builder_src}
@@ -717,6 +733,7 @@ target_link_libraries(ecal_core_private
     $<$<BOOL:${ECAL_CORE_TRANSPORT_TCP}>:tcp_pubsub::tcp_pubsub>
     $<$<BOOL:${ECAL_CORE_TIMEPLUGIN}>:ecaltime>
     ecaludp::ecaludp
+    nlohmann_json::nlohmann_json
     $<$<AND:$<BOOL:${UNIX}>,$<NOT:$<BOOL:${QNXNTO}>>>:dl>
     $<$<AND:$<BOOL:${UNIX}>,$<NOT:$<BOOL:${APPLE}>>,$<NOT:$<BOOL:${QNXNTO}>>>:rt>
     $<$<AND:$<BOOL:${UNIX}>,$<NOT:$<BOOL:${APPLE}>>,$<NOT:$<BOOL:${QNXNTO}>>>:atomic>

ecal/core/src/ecal.cpp

@@ -118,6 +118,7 @@ namespace eCAL
       SetGlobalUnitName(unit_name_.c_str());
 
       if ((components_ & Init::Logging) != 0u) InitializeLogging(config_);
+      InitializeTracing();
 
       auto globals_instance = CreateGlobalsInstance();
       if (!globals_instance) return false;
@@ -169,6 +170,7 @@ namespace eCAL
     ResetGlobalEcalConfiguration();
 
     ResetLogging();
+    ResetTracing();
 
     return finalized;
   }

ecal/core/src/ecal_global_accessors.cpp

@@ -32,6 +32,7 @@
 #include "config/builder/logging_attribute_builder.h"
 #include "logging/ecal_log_provider.h"
 #include "logging/ecal_log_receiver.h"
+#include "tracing/trace_provider.h"
 
 #include <atomic>
 #include <string>
@@ -57,6 +58,8 @@ namespace eCAL
   std::shared_ptr<Logging::CLogProvider> g_log_provider_instance;
   std::shared_ptr<Logging::CLogReceiver> g_log_receiver_instance;
 
+  std::shared_ptr<tracing::CTraceProvider> g_trace_provider_instance;
+
   void SetGlobalUnitName(const char *unit_name_)
   {
     if(unit_name_ != nullptr) g_unit_name = unit_name_;
@@ -118,6 +121,22 @@ namespace eCAL
     return nullptr;
   }
 
+  void InitializeTracing()
+  {
+    g_trace_provider_instance = tracing::CTraceProvider::Create();
+  }
+
+  void ResetTracing()
+  {
+    g_trace_provider_instance.reset();
+  }
+
+    std::shared_ptr<tracing::CTraceProvider> g_trace_provider()
+  {
+    if (auto provider = g_trace_provider_instance; provider) return provider;
+    return nullptr;
+  }
+
   std::shared_ptr<CGlobals> g_globals()
   {
     if (auto globals_instance = g_globals_instance; globals_instance) 

ecal/core/src/ecal_global_accessors.h

@@ -43,6 +43,11 @@ namespace eCAL
     class CLogReceiver;
   }
 
+  namespace tracing
+  {
+    class CTraceProvider;
+  }
+
 #if ECAL_CORE_MONITORING
   class  CMonitoring;
 #endif
@@ -80,6 +85,9 @@ namespace eCAL
   void                      InitializeLogging(const eCAL::Configuration& config_);
   void                      ResetLogging();
 
+  void                      InitializeTracing();
+  void                      ResetTracing();
+
   // Declaration of getter functions for globally accessible variable instances
   std::shared_ptr<CGlobals>               g_globals();
 #if ECAL_CORE_MONITORING
@@ -110,6 +118,8 @@ namespace eCAL
   std::shared_ptr<Logging::CLogProvider>  g_logging_provider();
   std::shared_ptr<Logging::CLogReceiver>  g_logging_receiver();
 
+  std::shared_ptr<tracing::CTraceProvider> g_trace_provider();
+
   // declaration of globally accessible variables
   extern std::string                   g_default_ini_file;
   extern Configuration                 g_ecal_configuration;

ecal/core/src/pubsub/ecal_publisher.cpp

@@ -98,9 +98,10 @@ namespace eCAL
   }
 
   bool CPublisher::Send(CPayloadWriter& payload_, long long time_)
-  {
+  {    
     auto publisher_impl = m_publisher_impl.lock();
     if (!publisher_impl) return false;
+
     // in an optimization case the
      // publisher can send an empty package
      // or we do not have any subscription at all

ecal/core/src/pubsub/ecal_publisher_impl.cpp

@@ -46,6 +46,10 @@
 
 #include "registration/ecal_registration_provider.h"
 
+#include "tracing/tracing.h"
+#include "tracing/span.h"
+#include "tracing/trace_provider.h"
+
 #include <algorithm>
 #include <chrono>
 #include <functional>
@@ -122,6 +126,19 @@ namespace eCAL
     m_topic_id.topic_id.host_name = m_attributes.host_name;
     m_topic_id.topic_id.process_id = m_attributes.process_id;
 
+    // record topic metadata for tracing
+    {
+      eCAL::tracing::STopicMetadata meta;
+      meta.entity_id  = m_publisher_id;
+      meta.process_id = m_attributes.process_id;
+      meta.host_name  = m_attributes.host_name;
+      meta.topic_name = m_attributes.topic_name;
+      meta.encoding   = m_topic_info.encoding;
+      meta.type_name  = m_topic_info.name;
+      meta.direction  = eCAL::tracing::topic_direction::publisher;
+      if (auto provider = g_trace_provider(); provider) provider->addTopicMetadata(meta);
+    }
+
     // mark as created
     m_created = true;
   }
@@ -179,6 +196,29 @@ namespace eCAL
     // prepare counter and internal states
     const size_t snd_hash = PrepareWrite(filter_id_, payload_buf_size);
 
+    // determine active transport layer for tracing
+    eCAL::tracing::eTracingLayerType active_layer = eCAL::tracing::tl_trace_none;
+    {
+#if ECAL_CORE_TRANSPORT_SHM
+      if (m_writer_shm) { active_layer = static_cast<eCAL::tracing::eTracingLayerType>(active_layer | eCAL::tracing::tl_trace_shm); }
+#endif
+#if ECAL_CORE_TRANSPORT_UDP
+      if (m_writer_udp) { active_layer = static_cast<eCAL::tracing::eTracingLayerType>(active_layer | eCAL::tracing::tl_trace_udp); }
+#endif
+#if ECAL_CORE_TRANSPORT_TCP
+      if (m_writer_tcp) { active_layer = static_cast<eCAL::tracing::eTracingLayerType>(active_layer | eCAL::tracing::tl_trace_tcp); }
+#endif
+    }
+
+    // create tracing span for the send operation
+    eCAL::tracing::CSpan send_span(
+      m_topic_id,
+      m_clock,
+      active_layer,
+      payload_buf_size,
+      eCAL::tracing::operation_type::send
+    );
+
     // did we write anything
     bool written(false);
 

ecal/core/src/pubsub/ecal_publisher_impl.h

@@ -118,6 +118,7 @@ namespace eCAL
 
     void GetRegistrationSample(Registration::Sample& sample);
     void GetUnregistrationSample(Registration::Sample& sample);
+    long long GetClock() const { return m_clock; }
 
     bool StartUdpLayer();
     bool StartShmLayer();

ecal/core/src/pubsub/ecal_subscriber_impl.cpp

@@ -23,6 +23,9 @@
 **/
 
 #include "ecal_subscriber_impl.h"
+#include "tracing/tracing.h"
+#include "tracing/span.h"
+#include "tracing/trace_provider.h"
 #include <ecal/config.h>
 #include <ecal/log.h>
 #include <ecal/process.h>
@@ -93,6 +96,19 @@ namespace eCAL
     m_topic_id.topic_id.host_name = m_attributes.host_name;
     m_topic_id.topic_id.process_id = m_attributes.process_id;
 
+    // record topic metadata for tracing
+    {
+      eCAL::tracing::STopicMetadata meta;
+      meta.entity_id  = m_subscriber_id;
+      meta.process_id = m_attributes.process_id;
+      meta.host_name  = m_attributes.host_name;
+      meta.topic_name = m_attributes.topic_name;
+      meta.encoding   = m_topic_info.encoding;
+      meta.type_name  = m_topic_info.name;
+      meta.direction  = eCAL::tracing::topic_direction::subscriber;
+      if (auto provider = g_trace_provider(); provider) provider->addTopicMetadata(meta);
+    }
+
     // start transport layers
     InitializeLayers();
     StartTransportLayer();
@@ -373,6 +389,16 @@ namespace eCAL
 
   size_t CSubscriberImpl::ApplySample(const Payload::TopicInfo& topic_info_, const char* payload_, size_t size_, long long id_, long long clock_, long long time_, size_t /*hash_*/, eTLayerType layer_)
   {
+
+    eCAL::tracing::CSpan receive_span(
+      m_subscriber_id,
+      topic_info_,
+      clock_,
+      eCAL::tracing::toTracingLayerType(layer_),
+      size_,
+      eCAL::tracing::operation_type::receive
+    );
+
     // ensure thread safety
     const std::lock_guard<std::mutex> lock(m_receive_callback_mutex);
     if (!m_created) return(0);
@@ -450,7 +476,18 @@ namespace eCAL
 
         // execute it
         const std::lock_guard<std::mutex> exec_lock(m_connection_map_mtx);
-        (m_receive_callback)(topic_id, m_connection_map[pub_info].data_type_info, cb_data);
+        {
+          eCAL::tracing::CSpan receive_span(
+            m_subscriber_id,
+            topic_info_,
+            clock_,
+            eCAL::tracing::toTracingLayerType(layer_),
+            size_,
+            eCAL::tracing::operation_type::callback_execution
+          );
+
+
+        (m_receive_callback)(topic_id, m_connection_map[pub_info].data_type_info, cb_data); }
         processed = true;
       }
     }