Merge pull request #4954 from MrAnno/otel-source-hostname

otel: hostname handling

modules/grpc/otel/otel-grammar.ym

@@ -74,6 +74,7 @@ GrpcClientCredentialsBuilderW *last_grpc_client_credentials_builder;
 %token KW_BATCH_BYTES
 %token KW_CONCURRENT_REQUESTS
 %token KW_CHANNEL_ARGS
+%token KW_SET_HOSTNAME
 
 %type <ptr> source_otel
 %type <ptr> parser_otel
@@ -143,6 +144,7 @@ parser_otel_options
 
 parser_otel_option
   : parser_opt
+  | KW_SET_HOSTNAME '(' yesno ')' { otel_protobuf_parser_set_hostname(last_parser, $3); }
   ;
 
 destination_otel
@@ -294,4 +296,3 @@ grpc_client_credentials_builder_alts_target_service_accounts
 /* INCLUDE_RULES */
 
 %%
-

modules/grpc/otel/otel-parser.c

@@ -52,6 +52,7 @@ static CfgLexerKeyword otel_keywords[] =
   { "batch_bytes",               KW_BATCH_BYTES },
   { "concurrent_requests",       KW_CONCURRENT_REQUESTS },
   { "channel_args",              KW_CHANNEL_ARGS },
+  { "set_hostname",              KW_SET_HOSTNAME },
   { NULL }
 };
 

modules/grpc/otel/otel-protobuf-parser.cpp

@@ -218,16 +218,23 @@ _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 void
+_set_hostname_from_attributes(LogMessage *msg, const RepeatedPtrField<KeyValue> &key_values)
 {
-  size_t first = peer.find_first_of(':');
-  size_t last = peer.find_last_of(':');
+  for (const KeyValue &kv : key_values)
+    {
+      if (kv.key() == "host.name")
+        {
+          if (kv.value().value_case() != AnyValue::kStringValue)
+            return;
 
-  if (first != grpc::string::npos && last != grpc::string::npos)
-    return peer.substr(first + 1, last - first - 1);
+          std::string hostname = kv.value().string_value();
+          if (!hostname.empty())
+            log_msg_set_value(msg, LM_V_HOST, hostname.c_str(), hostname.length());
 
-  return "";
+          return;
+        }
+    }
 }
 
 static GSockAddr *
@@ -262,7 +269,7 @@ _extract_saddr(const grpc::string &peer)
 }
 
 static bool
-_parse_metadata(LogMessage *msg)
+_parse_metadata(LogMessage *msg, bool set_hostname)
 {
   char number_buf[G_ASCII_DTOSTR_BUF_SIZE];
   gssize len;
@@ -282,6 +289,8 @@ _parse_metadata(LogMessage *msg)
 
   /* .otel.resource.attributes */
   _add_repeated_KeyValue_fields(msg, ".otel.resource.attributes", resource.attributes());
+  if (set_hostname)
+    _set_hostname_from_attributes(msg, resource.attributes());
 
   /* .otel.resource.dropped_attributes_count */
   std::snprintf(number_buf, G_N_ELEMENTS(number_buf), "%" PRIu32, resource.dropped_attributes_count());
@@ -1113,11 +1122,6 @@ 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 */
@@ -1393,7 +1397,9 @@ syslogng::grpc::otel::ProtobufParser::process(LogMessage *msg)
 
   gssize len;
   LogMessageValueType log_msg_type;
-  const gchar *type = log_msg_get_value_with_type(msg, logmsg_handle::RAW_TYPE, &len, &log_msg_type);
+
+  /* _parse_metadata() may invalidate the returned char pointer, so a copy is made with std::string */
+  std::string type = log_msg_get_value_with_type(msg, logmsg_handle::RAW_TYPE, &len, &log_msg_type);
 
   if (log_msg_type == LM_VT_NULL)
     {
@@ -1409,17 +1415,20 @@ syslogng::grpc::otel::ProtobufParser::process(LogMessage *msg)
       return false;
     }
 
-  if (strncmp(type, "log", len) == 0)
+  if (!_parse_metadata(msg, this->set_host))
+    return false;
+
+  if (type == "log")
     {
       if (!_parse_log_record(msg))
         return false;
     }
-  else if (strncmp(type, "metric", len) == 0)
+  else if (type == "metric")
     {
       if (!_parse_metric(msg))
         return false;
     }
-  else if (strncmp(type, "span", len) == 0)
+  else if (type == "span")
     {
       if (!_parse_span(msg))
         return false;
@@ -1428,13 +1437,10 @@ syslogng::grpc::otel::ProtobufParser::process(LogMessage *msg)
     {
       msg_error("OpenTelemetry: unexpected .otel_raw.type",
                 evt_tag_msg_reference(msg),
-                evt_tag_str("type", type));
+                evt_tag_str("type", type.c_str()));
       return false;
     }
 
-  if (!_parse_metadata(msg))
-    return false;
-
   _unset_raw_fields(msg);
 
   return true;
@@ -1458,6 +1464,12 @@ _clone(LogPipe *s)
   return &cloned->super.super;
 }
 
+void
+otel_protobuf_parser_set_hostname(LogParser *s, gboolean set_hostname)
+{
+  get_ProtobufParser(s)->set_hostname(set_hostname);
+}
+
 static void
 _free(LogPipe *s)
 {

modules/grpc/otel/otel-protobuf-parser.h

@@ -32,6 +32,7 @@
 typedef struct OtelProtobufParser_ OtelProtobufParser;
 
 LogParser *otel_protobuf_parser_new(GlobalConfig *cfg);
+void otel_protobuf_parser_set_hostname(LogParser *s, gboolean set_hostname);
 
 #include "compat/cpp-end.h"
 

modules/grpc/otel/otel-protobuf-parser.hpp

@@ -53,6 +53,11 @@ class ProtobufParser
 public:
   bool process(LogMessage *msg);
 
+  void set_hostname(bool s)
+  {
+    this->set_host = s;
+  }
+
   static void store_raw_metadata(LogMessage *msg, const ::grpc::string &peer,
                                  const Resource &resource, const std::string &resource_schema_url,
                                  const InstrumentationScope &scope, const std::string &scope_schema_url);
@@ -69,6 +74,9 @@ class ProtobufParser
   static void set_syslog_ng_macros(LogMessage *msg, const KeyValueList &macros);
   static void set_syslog_ng_address(LogMessage *msg, GSockAddr **sa, const KeyValueList &addr);
   static void parse_syslog_ng_tags(LogMessage *msg, const std::string &tags_as_str);
+
+private:
+  bool set_host = true;
 };
 
 }

modules/grpc/otel/otel-source.cpp

@@ -121,6 +121,12 @@ syslogng::grpc::otel::SourceDriver::init()
   if (fetch_limit == -1)
     fetch_limit = super->super.worker_options.super.init_window_size;
 
+  /*
+   * syslog-ng-otlp(): the original HOST is always kept
+   * opentelemetry(): there is no parsing in this source, HOST always falls back to saddr (LogSource)
+   */
+  super->super.worker_options.super.keep_hostname = TRUE;
+
   return log_threaded_source_driver_init_method(&super->super.super.super.super);
 }
 

modules/grpc/otel/tests/test-otel-protobuf-parser.cpp

@@ -143,6 +143,10 @@ Test(otel_protobuf_parser, metadata)
   bytes_attr->set_key("bytes_key");
   bytes_attr->mutable_value()->set_bytes_value({0, 1, 2, 3, 4, 5, 6, 7});
 
+  KeyValue *hostname = resource.add_attributes();
+  hostname->set_key("host.name");
+  hostname->mutable_value()->set_string_value("myhost");
+
   std::string scope_schema_url = "my_scope_schema_url";
 
   LogMessage *msg = log_msg_new_empty();
@@ -152,6 +156,7 @@ Test(otel_protobuf_parser, metadata)
 
   cr_assert(msg->saddr != NULL);
   _assert_log_msg_value(msg, "SOURCEIP", "::1", -1, LM_VT_STRING);
+  _assert_log_msg_value(msg, "HOST", "myhost", -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);