Remote name splitting (#525)

* Instead of splitting the hardware_id we now split the name of the diagnostic message into its parts. User still needing the hardware_id can find is as a field in influxdb. Next to this the setting of "none" for unknown data has been removed as it is not part of the influxdb spec, now the field is not set.

* Small update to diagnostic_updater README, explaining the parameters that can be set

* changed response_code to long instead of CURLcode as it is a HTTP return and not CURL error code.

* Removed the use of send.agg, now aggregate all /diagnostics and send them on a timer

* Removed the use of send.agg, now aggregate all /diagnostics and send them on a timer

* Removed the use of send.agg, now aggregate all /diagnostics and send them on a timer

* Updated README

* Forgot a  merge conflict marker

* Forgot a  merge conflict marker

---------

Co-authored-by: Daan Wijffels <dwijffels@lely.com>
Co-authored-by: Christian Henkel <christian.henkel2@de.bosch.com>

Author: 3 people

diagnostic_remote_logging/README.md

@@ -30,11 +30,13 @@ The `influx` node supports several parameters. Below is an example configuration
       bucket:
       organization:
     send:
-      agg: true
+      diagnostics: true
+      period: 1.0
       top_level_state: true
 ```
 
-- `send.agg`: Enables or disables subscription to the `/diagnostics_agg` topic.
+- `send.diagnostics`: Enables or disables subscription to the `/diagnostics` topic.
+- `send.period`: Specifies the interval in seconds for sending diagnostic data to InfluxDB. During each period, all incoming `/diagnostics` messages are collected and transmitted as a batch to InfluxDB.
 - `send.top_level_state`: Enables or disables subscription to the `/diagnostics_toplevel_state` topic.
 
 #### InfluxDB Configuration

diagnostic_remote_logging/include/diagnostic_remote_logging/influx_line_protocol.hpp

@@ -40,11 +40,12 @@
 #define DIAGNOSTIC_REMOTE_LOGGING__INFLUX_LINE_PROTOCOL_HPP_
 
 #include <string>
+#include <tuple>
 #include <utility>
 #include <vector>
 
-#include "rclcpp/rclcpp.hpp"
 #include "diagnostic_msgs/msg/diagnostic_array.hpp"
+#include "rclcpp/rclcpp.hpp"
 
 std::string toInfluxTimestamp(const rclcpp::Time & time)
 {
@@ -105,68 +106,75 @@ std::string formatValues(const std::vector<diagnostic_msgs::msg::KeyValue> & val
   return formatted;
 }
 
-std::pair<std::string, std::string> splitHardwareID(const std::string & input)
+std::tuple<std::string, std::string, std::string> splitName(const std::string & input)
 {
-  size_t first_slash_pos = input.find('/');
+  size_t colon_pos = input.find(':');
+  std::string ns_node = (colon_pos != std::string::npos) ? input.substr(0, colon_pos) : input;
+  std::string name = (colon_pos != std::string::npos) ? input.substr(colon_pos + 1) : "";
 
-  // If no slash is found, treat the entire input as the node_name
-  if (first_slash_pos == std::string::npos) {
-    return {"none", input};
+  // Trim leading whitespace from diagnostic name
+  if (!name.empty() && name[0] == ' ') {
+    name = name.substr(1);
   }
 
-  size_t second_slash_pos = input.find('/', first_slash_pos + 1);
-
-  // If the second slash is found, extract the "ns" and "node" parts
-  if (second_slash_pos != std::string::npos) {
-    std::string ns = input.substr(first_slash_pos + 1, second_slash_pos - first_slash_pos - 1);
-    std::string node = input.substr(second_slash_pos + 1);
-    return {ns, node};
+  // Handle use_fqn is false, only node name
+  if (ns_node.empty() || ns_node[0] != '/') {
+    // No leading slash - treat entire thing as node name
+    return {"", ns_node, name};
   }
 
-  // If no second slash is found, everything after the first slash is the node
-  std::string node = input.substr(first_slash_pos + 1);
-  return {"none", node};    // ns is empty, node is the remaining string
+  // Remove leading slash when we get a fully qualified name
+  ns_node = ns_node.substr(1);
+  size_t last_slash = ns_node.rfind('/');
+
+  std::string ns = ns_node.substr(0, last_slash);
+  std::string node = ns_node.substr(last_slash + 1);
+
+  return {ns, node, name};
 }
 
 void statusToInfluxLineProtocol(
-  std::string & output,
-  const diagnostic_msgs::msg::DiagnosticStatus & status,
+  std::string & output, const diagnostic_msgs::msg::DiagnosticStatus & status,
   const std::string & timestamp_str)
 {
-  // hardware_id is empty for analyzer groups, so skip them
-  if (status.hardware_id.empty()) {
-    return;
+  auto [ns, node, name] = splitName(status.name);
+  output += escapeSpace(node);
+  if (!ns.empty()) {
+    output += ",ns=" + escapeSpace(ns);
+  }
+  if (!name.empty()) {
+    output += ",name=" + escapeSpace(name);
+  }
+  if (!status.hardware_id.empty()) {
+    output += ",hardware_id=" + escapeSpace(status.hardware_id);
+  }
+  output += " level=" + std::to_string(status.level);
+
+  if (!status.message.empty()) {
+    output += ",message=\"" + status.message + "\"";
   }
 
-  auto [ns, identifier] = splitHardwareID(status.hardware_id);
-  output += escapeSpace(identifier) + ",ns=" + escapeSpace(ns) +
-    " level=" + std::to_string(status.level) + ",message=\"" + status.message + "\"";
   auto formatted_key_values = formatValues(status.values);
   if (!formatted_key_values.empty()) {
     output += "," + formatted_key_values;
   }
   output += " " + timestamp_str + "\n";
 }
 
-std::string diagnosticStatusToInfluxLineProtocol(
-  const diagnostic_msgs::msg::DiagnosticStatus::SharedPtr & msg, const rclcpp::Time & time)
+void statusToInfluxLineProtocol(
+  std::string & output, const diagnostic_msgs::msg::DiagnosticStatus & status,
+  const rclcpp::Time & time)
 {
-  std::string output =
-    msg->name + " level=" + std::to_string(msg->level) + " " + toInfluxTimestamp(time) + "\n";
-  return output;
+  statusToInfluxLineProtocol(output, status, toInfluxTimestamp(time));
 }
 
-std::string diagnosticArrayToInfluxLineProtocol(
-  const diagnostic_msgs::msg::DiagnosticArray::SharedPtr & diag_msg)
+void diagnosticArrayToInfluxLineProtocol(
+  std::string & output, const diagnostic_msgs::msg::DiagnosticArray::SharedPtr & diag_msg)
 {
-  std::string output;
   std::string timestamp = toInfluxTimestamp(diag_msg->header.stamp);
-
   for (auto & status : diag_msg->status) {
     statusToInfluxLineProtocol(output, status, timestamp);
   }
-
-  return output;
 }
 
 #endif  // DIAGNOSTIC_REMOTE_LOGGING__INFLUX_LINE_PROTOCOL_HPP_

diagnostic_remote_logging/include/diagnostic_remote_logging/influxdb.hpp

@@ -44,12 +44,12 @@
 #endif
 
 #include <curl/curl.h>
+
 #include <string>
 
+#include "diagnostic_msgs/msg/diagnostic_array.hpp"
 #include "diagnostic_remote_logging/influx_line_protocol.hpp"
-
 #include "rclcpp/rclcpp.hpp"
-#include "diagnostic_msgs/msg/diagnostic_array.hpp"
 
 class InfluxDB : public rclcpp::Node
 {
@@ -64,10 +64,15 @@ class InfluxDB : public rclcpp::Node
   std::string post_url_, influx_token_;
   CURL * curl_;
 
+  std::string output_string_;
+
+  rclcpp::TimerBase::SharedPtr diagnostics_send_timer_;
+
   void setupConnection(const std::string & telegraf_url);
 
   void diagnosticsCallback(const diagnostic_msgs::msg::DiagnosticArray::SharedPtr msg);
   void topLevelCallback(const diagnostic_msgs::msg::DiagnosticStatus::SharedPtr msg);
+  void sendTimerCallback();
 
   bool sendToInfluxDB(const std::string & data);
 };

diagnostic_remote_logging/src/influxdb.cpp

@@ -57,8 +57,8 @@ InfluxDB::InfluxDB(const rclcpp::NodeOptions & opt)
     // Ensure all parameters are set
     if (organization.empty() || bucket.empty() || influx_token_.empty()) {
       throw std::runtime_error(
-          "All parameters (connection.organization, connection.bucket, connection.token) "
-          "must be set, or when using a proxy like Telegraf none have to be set.");
+        "All parameters (connection.organization, connection.bucket, connection.token) "
+        "must be set, or when using a proxy like Telegraf none have to be set.");
     }
 
     // Construct the Telegraf URL
@@ -68,33 +68,56 @@ InfluxDB::InfluxDB(const rclcpp::NodeOptions & opt)
 
   setupConnection(post_url_);
 
-  if (declare_parameter("send.agg", true)) {
+  double send_period = declare_parameter<double>("send.period", 1.0);
+  bool send_diagnostics = declare_parameter<bool>("send.diagnostics", true);
+
+  if (send_period <= 0.0 && send_diagnostics) {
+    throw std::runtime_error(
+      "Parameter send.period must be greater than 0.0 if send.diagnostics is set to true");
+  }
+
+  if (send_diagnostics) {
+    diagnostics_send_timer_ = this->create_wall_timer(
+      std::chrono::duration<double>(send_period), std::bind(&InfluxDB::sendTimerCallback, this));
+
     diag_sub_ = this->create_subscription<diagnostic_msgs::msg::DiagnosticArray>(
-        "/diagnostics_agg", rclcpp::SensorDataQoS(),
-        std::bind(&InfluxDB::diagnosticsCallback, this, std::placeholders::_1));
+      "/diagnostics", rclcpp::SensorDataQoS(),
+      std::bind(&InfluxDB::diagnosticsCallback, this, std::placeholders::_1));
+  }
+
+  if (declare_parameter("send.agg", false)) {
+    throw std::runtime_error(
+      "The option send.agg is deprecated and will be removed in a future version. Use "
+      "send.diagnostics and send.period instead.");
   }
 
   if (declare_parameter<bool>("send.top_level_state", true)) {
     top_level_sub_ = this->create_subscription<diagnostic_msgs::msg::DiagnosticStatus>(
-        "/diagnostics_toplevel_state", rclcpp::SensorDataQoS(),
-        std::bind(&InfluxDB::topLevelCallback, this, std::placeholders::_1));
+      "/diagnostics_toplevel_state", rclcpp::SensorDataQoS(),
+      std::bind(&InfluxDB::topLevelCallback, this, std::placeholders::_1));
   }
 }
 
 void InfluxDB::diagnosticsCallback(const diagnostic_msgs::msg::DiagnosticArray::SharedPtr msg)
 {
-  std::string output = diagnosticArrayToInfluxLineProtocol(msg);
+  diagnosticArrayToInfluxLineProtocol(output_string_, msg);
+}
 
-  if (!sendToInfluxDB(output)) {
-    RCLCPP_ERROR(this->get_logger(), "Failed to send /diagnostics_agg to telegraf");
+void InfluxDB::sendTimerCallback()
+{
+  if (!sendToInfluxDB(output_string_)) {
+    RCLCPP_ERROR(this->get_logger(), "Failed to send /diagnostics to telegraf");
   }
 
-  RCLCPP_DEBUG(this->get_logger(), "%s", output.c_str());
+  RCLCPP_DEBUG(this->get_logger(), "%s", output_string_.c_str());
+
+  output_string_.clear();
 }
 
 void InfluxDB::topLevelCallback(const diagnostic_msgs::msg::DiagnosticStatus::SharedPtr msg)
 {
-  std::string output = diagnosticStatusToInfluxLineProtocol(msg, this->get_clock()->now());
+  std::string output;
+  statusToInfluxLineProtocol(output, *msg, this->get_clock()->now());
 
   if (!sendToInfluxDB(output)) {
     RCLCPP_ERROR(this->get_logger(), "Failed to send /diagnostics_toplevel_state to telegraf");
@@ -145,12 +168,12 @@ bool InfluxDB::sendToInfluxDB(const std::string & data)
     RCLCPP_ERROR(this->get_logger(), "cURL error: %s", curl_easy_strerror(res));
     return false;
   }
-  CURLcode response_code;
+  int32_t response_code;
   curl_easy_getinfo(curl_, CURLINFO_RESPONSE_CODE, &response_code);
 
   if (response_code != 204) {
-    RCLCPP_ERROR(this->get_logger(), "Error (%d) when sending to telegraf:\n%s", response_code,
-                 data.c_str());
+    RCLCPP_ERROR(
+      this->get_logger(), "Error (%d) when sending to telegraf:\n%s", response_code, data.c_str());
     return false;
   }
 

diagnostic_remote_logging/test/influx_line_protocol.cpp

@@ -40,10 +40,11 @@
 
 #include <gtest/gtest.h>
 
+#include <rclcpp/rclcpp.hpp>
+
 #include "diagnostic_msgs/msg/diagnostic_array.hpp"
 #include "diagnostic_msgs/msg/diagnostic_status.hpp"
 #include "diagnostic_msgs/msg/key_value.hpp"
-#include <rclcpp/rclcpp.hpp>
 
 diagnostic_msgs::msg::KeyValue createKeyValue(const std::string & key, const std::string & value)
 {
@@ -95,31 +96,40 @@ TEST(FormatValuesTests, FormatsKeyValuePairs)
   EXPECT_EQ(formatValues(values), expected);
 }
 
-// Test splitHardwareID
-TEST(SplitHardwareIDTests, SplitsCorrectly)
+// Test splitName
+TEST(SplitNameTests, SplitsCorrectly)
 {
-  EXPECT_EQ(splitHardwareID("node_name"), std::make_pair(std::string("none"), std::string("node_"
-                                                                                          "name")));
-  EXPECT_EQ(splitHardwareID("/ns/node_name"), std::make_pair(std::string("ns"), std::string("node_"
-                                                                                            "nam"
-                                                                                            "e")));
-  EXPECT_EQ(splitHardwareID("/ns/prefix/node_name"),
-            std::make_pair(std::string("ns"), std::string("prefix/"
-                                                          "node_name")));
+  EXPECT_EQ(splitName("diagnostic_name"), std::make_tuple("", "diagnostic_name", ""));
+
+  EXPECT_EQ(splitName("/ns/node_name"), std::make_tuple("ns", "node_name", ""));
+
+  EXPECT_EQ(splitName("/ns/prefix/node_name"), std::make_tuple("ns/prefix", "node_name", ""));
+
+  EXPECT_EQ(
+    splitName("/ns/ns2/ns3/node_name: diagnostic description"),
+    std::make_tuple("ns/ns2/ns3", "node_name", "diagnostic description"));
+
+  EXPECT_EQ(
+    splitName("/simple_ns/node: some long diagnostic name here"),
+    std::make_tuple("simple_ns", "node", "some long diagnostic name here"));
+
+  EXPECT_EQ(splitName("just_node: diag"), std::make_tuple("", "just_node", "diag"));
 }
 
 // Test statusToInfluxLineProtocol
 TEST(StatusToInfluxLineProtocolTests, FormatsCorrectly)
 {
   diagnostic_msgs::msg::DiagnosticStatus status;
-  status.hardware_id = "/ns/node_name";
+  status.name = "/test/test2/diagnostic_updater_example: topic1 topic status";
+  status.hardware_id = "Device-27-46";
   status.level = 2;
   status.message = "Test message";
   status.values.push_back(createKeyValue("key1", "value1"));
   status.values.push_back(createKeyValue("key2", "42"));
 
   std::string expected =
-    "node_name,ns=ns level=2,message=\"Test message\",key1=\"value1\",key2=42"
+    "diagnostic_updater_example,ns=test/test2,name=topic1\\ topic\\ "
+    "status,hardware_id=Device-27-46 level=2,message=\"Test message\",key1=\"value1\",key2=42"
     " 1672531200123456789\n";
   std::string output;
   statusToInfluxLineProtocol(output, status, "1672531200123456789");
@@ -134,36 +144,48 @@ TEST(DiagnosticArrayToInfluxLineProtocolTests, HandlesMultipleStatuses)
   diag_msg->header.stamp = rclcpp::Time(1672531200, 123456789);
 
   diagnostic_msgs::msg::DiagnosticStatus status1;
-  status1.hardware_id = "/ns1/node1";
+  status1.name = "/ns1/node1: diagnostic description";
+  status1.hardware_id = "Device-27-46";
   status1.level = 1;
   status1.message = "First status";
   status1.values.push_back(createKeyValue("keyA", "valueA"));
 
   diagnostic_msgs::msg::DiagnosticStatus status2;
-  status2.hardware_id = "node2";
+  status2.name = "node2";
+  status2.hardware_id = "12345";
   status2.level = 2;
   status2.message = "Second status";
   status2.values.push_back(createKeyValue("keyB", "42"));
 
-  diag_msg->status = {status1, status2};
+  diagnostic_msgs::msg::DiagnosticStatus status3;
+  status3.name = "node3: status";
+  status3.level = 3;
+
+  diag_msg->status = {status1, status2, status3};
+
+  std::string output;
+  diagnosticArrayToInfluxLineProtocol(output, diag_msg);
 
   std::string expected =
-    "node1,ns=ns1 level=1,message=\"First "
+    "node1,ns=ns1,name=diagnostic\\ description,hardware_id=Device-27-46 level=1,message=\"First "
     "status\",keyA=\"valueA\" 1672531200123456789\n"
-    "node2,ns=none level=2,message=\"Second "
-    "status\",keyB=42 1672531200123456789\n";
+    "node2,hardware_id=12345 level=2,message=\"Second status\",keyB=42 1672531200123456789\n"
+    "node3,name=status level=3 1672531200123456789\n";
 
-  EXPECT_EQ(diagnosticArrayToInfluxLineProtocol(diag_msg), expected);
+  EXPECT_EQ(output, expected);
 }
 
-// Test diagnosticStatusToInfluxLineProtocol
-TEST(DiagnosticStatusToInfluxLineProtocol, HandlesSingleStatus)
+// Test topLevelStatus
+TEST(TopLevelStatusTests, HandlesTopLevelStatus)
 {
   auto status = std::make_shared<diagnostic_msgs::msg::DiagnosticStatus>();
   status->level = 1;
   status->name = "toplevel_state";
   auto time = rclcpp::Time(1672531200, 123456789);
 
+  std::string output;
+  statusToInfluxLineProtocol(output, *status, time);
+
   std::string expected = "toplevel_state level=1 1672531200123456789\n";
-  EXPECT_EQ(diagnosticStatusToInfluxLineProtocol(status, time), expected);
+  EXPECT_EQ(output, expected);
 }