[Geneva Metrics Exporter] - Add support for abstract socket (#264)

Author: lalitb

exporters/geneva/CMakeLists.txt

@@ -95,8 +95,16 @@ if(BUILD_EXAMPLE)
   add_executable(example_metrics example/example_metrics.cc
                                  example/foo_library.cc)
   target_link_libraries(example_metrics opentelemetry_exporter_geneva_metrics)
+
+  add_executable(stress_test_linux example/stress_test_linux.cc)
+  find_package(Boost QUIET)
+  if(Boost_FOUND)
+    include_directories(${Boost_INCLUDE_DIRS})
+    target_compile_definitions(stress_test_linux PRIVATE HAVE_BOOST)
+  endif()
+  target_link_libraries(stress_test_linux opentelemetry_exporter_geneva_metrics)
 endif()
 if(MAIN_PROJECT AND BUILD_PACKAGE)
   include(cmake/package.cmake)
   include(CPack)
-endif()
+endif()

exporters/geneva/example/stress_test_linux.cc

@@ -5,7 +5,6 @@
 
 #include "opentelemetry/exporters/geneva/metrics/macros.h"
 #include "opentelemetry/exporters/geneva/metrics/socket_tools.h"
-#include "opentelemetry/ext/http/common/url_parser.h"
 #include "opentelemetry/version.h"
 #include <memory>
 #include <string>
@@ -26,8 +25,7 @@ class ConnectionStringParser {
 
 public:
   ConnectionStringParser(const std::string &connection_string)
-      : account_(""), namespace_(""),
-        url_(nullptr), transport_protocol_{TransportProtocol::kUnknown} {
+      : account_(""), namespace_(""), transport_protocol_{TransportProtocol::kUnknown} {
     std::string::size_type key_pos = 0;
     std::string::size_type key_end;
     std::string::size_type val_pos;
@@ -36,14 +34,16 @@ class ConnectionStringParser {
     while ((key_end = connection_string.find(kEqual, key_pos)) !=
            std::string::npos) {
       if ((val_pos = connection_string.find_first_not_of(kEqual, key_end)) ==
-          std::string::npos) {
+          std::string::npos) 
+      {
         break;
       }
       val_end = connection_string.find(kSemicolon, val_pos);
       auto key = connection_string.substr(key_pos, key_end - key_pos);
       auto value = connection_string.substr(val_pos, val_end - val_pos);
       key_pos = val_end;
-      if (key_pos != std::string::npos) {
+      if (key_pos != std::string::npos) 
+      {
         ++key_pos;
       }
       if (key == kNamespace) {
@@ -52,22 +52,24 @@ class ConnectionStringParser {
         account_ = value;
       } else if (key == kEndpoint) {
         is_endpoint_found = true;
-        url_ = std::unique_ptr<ext::http::common::UrlParser>(
-            new ext::http::common::UrlParser(value));
-        if (url_->success_) {
+        size_t pos = value.find("://", 0);
+        if (pos != std::string::npos)
+        {
+          auto scheme = std::string(value.begin(), value.begin() + pos);
+          connection_string_ = value.substr(pos + strlen("://"));
 #ifdef HAVE_UNIX_DOMAIN
-          if (url_->scheme_ == "unix") {
+          if (scheme == "unix") {
             transport_protocol_ = TransportProtocol::kUNIX;
           }
 #else
-          if (url_->scheme_ == "unix") {
+          if (scheme == "unix") {
             LOG_ERROR("Unix domain socket not supported on this platform")
           }
 #endif
-          if (url_->scheme_ == "tcp") {
+          if (scheme == "tcp") {
             transport_protocol_ = TransportProtocol::kTCP;
           }
-          if (url_->scheme_ == "udp") {
+          if (scheme == "udp") {
             transport_protocol_ = TransportProtocol::kUDP;
           }
         }
@@ -84,8 +86,8 @@ class ConnectionStringParser {
 
   std::string account_;
   std::string namespace_;
-  std::unique_ptr<ext::http::common::UrlParser> url_;
   TransportProtocol transport_protocol_;
+  std::string connection_string_;
 };
 } // namespace metrics
 } // namespace geneva

exporters/geneva/include/opentelemetry/exporters/geneva/metrics/connection_string_parser.h

@@ -214,13 +214,15 @@ struct SocketAddr {
 
   // Indicator that the sockaddr is sockaddr_un
   bool isUnixDomain;
+  size_t abstract_socket_size;
 
   /// <summary>
   /// SocketAddr constructor
   /// </summary>
   /// <returns>SocketAddr</returns>
   SocketAddr() {
     isUnixDomain = false;
+    abstract_socket_size = 0;
 #ifdef HAVE_UNIX_DOMAIN
     memset(&m_data_un, 0, sizeof(m_data_un));
 #else
@@ -252,6 +254,12 @@ struct SocketAddr {
       // Max length of Unix domain filename is up to 108 chars
       strncpy_s(m_data_un.sun_path, sizeof(m_data_un.sun_path),
                 unix_domain_path, sizeof(m_data_un.sun_path));
+      // special handling for abstract socket - they should be
+      // prefixed with '@' character.
+      if (unix_domain_path[0] == '@') {
+        m_data_un.sun_path[0] = '\0';
+        abstract_socket_size = ipAddress.size();
+      } 
       return;
     }
 #endif
@@ -305,8 +313,10 @@ struct SocketAddr {
 
   size_t size() const {
 #ifdef HAVE_UNIX_DOMAIN
-    // Unix domain struct m_data_un
-    if (isUnixDomain)
+    if (isUnixDomain && abstract_socket_size)
+      return sizeof(sa_family_t) + abstract_socket_size;
+    else if(isUnixDomain)
+      // Unix domain struct m_data_un
       return sizeof(m_data_un);
 #endif
     // IPv4 struct m_data_in

exporters/geneva/include/opentelemetry/exporters/geneva/metrics/socket_tools.h

@@ -26,7 +26,7 @@ Exporter::Exporter(const ExporterOptions &options)
         TransportProtocol::kUNIX) {
       data_transport_ =
           std::unique_ptr<DataTransport>(new UnixDomainSocketDataTransport(
-              connection_string_parser_.url_->path_));
+              connection_string_parser_.connection_string_));
     }
 #ifdef _WIN32
     else if (connection_string_parser_.transport_protocol_ ==

exporters/geneva/src/exporter.cc

@@ -10,7 +10,8 @@ namespace geneva {
 namespace metrics {
 
 UnixDomainSocketDataTransport::UnixDomainSocketDataTransport(
-    const std::string &connection_string) {
+    const std::string &connection_string) 
+{      
   addr_.reset(new SocketTools::SocketAddr(connection_string.c_str(), true));
 }
 

exporters/geneva/src/unix_domain_socket_data_transport.cc

@@ -17,7 +17,9 @@ using namespace kaitai;
 using namespace opentelemetry::sdk::metrics;
 using namespace opentelemetry::exporter::geneva::metrics;
 
-const std::string kUnixDomainPath = "/tmp/ifx_unix_socket";
+std::string kUnixDomainPathUDS = "/tmp/ifx_unix_socket";
+std::string kUnixDomainPathAbstractSocket = "@/tmp/ifx_unix_socket";
+
 const std::string kNamespaceName = "test_ns";
 const std::string kAccountName = "test_account";
 
@@ -202,7 +204,12 @@ struct TestServer {
   }
 };
 
-TEST(GenevaMetricsExporter, BasicTests) {
+class GenericMetricsExporterTextFixture: public ::testing::TestWithParam<std::string>
+{};
+
+TEST_P(GenericMetricsExporterTextFixture, BasicTests) {
+
+  std::string kUnixDomainPath = GetParam();
   bool isRunning = true;
 
   // Start test server
@@ -255,3 +262,8 @@ TEST(GenevaMetricsExporter, BasicTests) {
 
   testServer.Stop();
 }
+
+INSTANTIATE_TEST_SUITE_P(GenericMetricsExporterText, GenericMetricsExporterTextFixture,
+    ::testing::Values(kUnixDomainPathUDS, kUnixDomainPathAbstractSocket));
+
+