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

Signed-off-by: Balazs Scheidler <[email protected]> Signed-off-by: László Várady <[email protected]>
bazsi · May 8, 2024 · 90731f9 · 90731f9
1 parent dbf43c7
commit 90731f9
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 2 deletions.
diff --git a/modules/grpc/otel/otel-protobuf-parser.cpp b/modules/grpc/otel/otel-protobuf-parser.cpp
@@ -230,6 +230,37 @@ _extract_hostname(const grpc::string &peer)
   return "";
 }
 
+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)
+    {
+      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 NULL;
+}
+
 static bool
 _parse_metadata(LogMessage *msg)
 {
@@ -1087,6 +1118,8 @@ syslogng::grpc::otel::ProtobufParser::store_raw_metadata(LogMessage *msg, const
   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);
   _set_value(msg, logmsg_handle::RAW_RESOURCE, serialized, LM_VT_PROTOBUF);

diff --git a/modules/grpc/otel/tests/test-otel-protobuf-parser.cpp b/modules/grpc/otel/tests/test-otel-protobuf-parser.cpp
@@ -46,7 +46,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;
@@ -150,7 +150,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);
@@ -207,6 +208,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);