otel: set the $SOURCEIP to the IP of the gRPC peer

Signed-off-by: Balazs Scheidler <[email protected]>
bazsi · Feb 4, 2024 · fa3ff2e · fa3ff2e
1 parent a9c807f
commit fa3ff2e
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 11 deletions.
diff --git a/lib/ivykis b/lib/ivykis
diff --git a/modules/grpc/otel/otel-protobuf-parser.cpp b/modules/grpc/otel/otel-protobuf-parser.cpp
@@ -223,16 +223,35 @@ _add_repeated_KeyValue_fields(LogMessage *msg, const char *key, const RepeatedPt
   _add_repeated_KeyValue_fields_with_prefix(msg, key_buffer, 0, key, key_values);
 }
 
-static std::string
-_extract_hostname(const grpc::string &peer)
+static GSockAddr *
+_extract_saddr(const grpc::string &peer)
 {
   size_t first = peer.find_first_of(':');
   size_t last = peer.find_last_of(':');
 
+  /* expected format:  ipv6:[::1]:32768 or ipv4:1.2.3.4:32768 */
   if (first != grpc::string::npos && last != grpc::string::npos)
-    return peer.substr(first + 1, last - first - 1);
+    {
+      const std::string ip_version = peer.substr(0, first);
+      std::string host;
+      int port = std::stoi(peer.substr(last + 1, grpc::string::npos), nullptr, 10);
+
+      if (peer.at(first + 1) == '[')
+        host = peer.substr(first + 2, last - first - 3);
+      else
+        host = peer.substr(first + 1, last - first - 1);
+
+      if (ip_version.compare("ipv6") == 0)
+        {
+          return g_sockaddr_inet6_new(host.c_str(), port);
+        }
+      else if (ip_version.compare("ipv4") == 0)
+        {
+          return g_sockaddr_inet_new(host.c_str(), port);
+        }
+    }
 
-  return "";
+  return NULL;
 }
 
 static bool
@@ -1091,10 +1110,7 @@ syslogng::grpc::otel::ProtobufParser::store_raw_metadata(LogMessage *msg, const
 {
   std::string serialized;
 
-  /* HOST */
-  std::string hostname = _extract_hostname(peer);
-  if (hostname.length())
-    log_msg_set_value(msg, LM_V_HOST, hostname.c_str(), hostname.length());
+  msg->saddr = _extract_saddr(peer);
 
   /* .otel_raw.resource */
   resource.SerializePartialToString(&serialized);

diff --git a/modules/grpc/otel/tests/test-otel-protobuf-parser.cpp b/modules/grpc/otel/tests/test-otel-protobuf-parser.cpp
@@ -45,7 +45,7 @@ _create_dummy_log_msg()
 {
   LogMessage *msg = log_msg_new_empty();
 
-  grpc::string peer = "ipv6:[::1]:36372";
+  grpc::string peer = "ipv4:127.0.0.5:36372";
   Resource resource;
   std::string resource_schema_url = "dummy_resource_schema_url";
   InstrumentationScope scope;
@@ -149,7 +149,8 @@ Test(otel_protobuf_parser, metadata)
   ProtobufParser::store_raw(msg, LogRecord());
   ProtobufParser().process(msg);
 
-  _assert_log_msg_value(msg, "HOST", "[::1]", -1, LM_VT_STRING);
+  cr_assert(msg->saddr != NULL);
+  _assert_log_msg_value(msg, "SOURCEIP", "::1", -1, LM_VT_STRING);
 
   _assert_log_msg_value(msg, ".otel.resource.attributes.null_key", "", -1, LM_VT_NULL);
   _assert_log_msg_value(msg, ".otel.resource.attributes.string_key", "string_attribute", -1, LM_VT_STRING);
@@ -206,6 +207,10 @@ Test(otel_protobuf_parser, log_record)
   ProtobufParser::store_raw(msg, log_record);
   cr_assert(ProtobufParser().process(msg));
 
+  cr_assert(msg->saddr != NULL);
+
+  _assert_log_msg_value(msg, "SOURCEIP", "127.0.0.5", -1, LM_VT_STRING);
+
   _assert_log_msg_value(msg, ".otel.type", "log", -1, LM_VT_STRING);
   _assert_log_msg_value(msg, ".otel.log.time_unix_nano", "111000222000", -1, LM_VT_INTEGER);
   _assert_log_msg_value(msg, ".otel.log.observed_time_unix_nano", "333000444000", -1, LM_VT_INTEGER);
+3 −6		configure.ac
+3 −3		man3/iv_work.3
+2 −3		src/Makefile.am
+20 −17		src/iv_fd.c
+7 −31		src/iv_fd_epoll.c
+2 −9		src/iv_fd_poll.c
+4 −3		src/iv_private_posix.h
+8 −9		src/iv_work.c
+1 −5		src/spinlock.h
+3 −3		test.mt/configure.ac
+2 −2		test.mt/iv_event_bench.c
+2 −2		test/iv_event_raw_bench.c
+2 −2		test/iv_signal_bench.c