Add option to not publish topic names to Cloudwatch Logs (#58)

* add option to not publish topic names to cloudwatch logs

* fix unit test failure

* add unit tests, improve code coverage

* update documentation

* address PR comments

* make the deprecated constructor use constructor delegation

Author: mm318

README.md

@@ -119,7 +119,8 @@ An example configuration file called `sample_configuration.yaml` is provided. Wh
 | log_group_name | AWS CloudWatch log group name | *std::string* | 'string'<br/>*note*: Log group names must be unique within a region foran AWS account | ros_log_group |
 | log_stream_name | AWS CloudWatch log stream name | *std::string* | 'string'<br/>*note*: The : (colon) and * (asterisk) characters are not allowed | ros_log_stream |
 | topics | A list of topics to get logs from (excluding `rosout_agg`) | *std::vector<std::string>* | ['string', 'string', 'string'] | `[]` |
-| min_log_verbosity| The minimum log severity for sending logs selectively to AWS CloudWatch Logs, log messages with a severity lower than `min_log_verbosity` will be ignored | *std::string* | DEBUG/INFO/WARN/ERROR/FATAL | DEBUG |
+| min_log_verbosity | The minimum log severity for sending logs selectively to AWS CloudWatch Logs, log messages with a severity lower than `min_log_verbosity` will be ignored | *std::string* | DEBUG/INFO/WARN/ERROR/FATAL | DEBUG |
+| publish_topic_names | Whether or not to include topic name information in the log messsages that are uploaded to AWS CloudWatch Logs | *bool* | true/false | true |
 | storage_directory | The location where all offline metrics will be stored | *string* | string | ~/.ros/cwlogs/ |
 | storage_limit | The maximum size of all offline storage files in KB. Once this limit is reached offline logs will start to be deleted oldest first. | *int* | number | 1048576 |
 | aws_client_configuration | AWS region configuration | *std::string* | *region*: "us-west-2"/"us-east-1"/"us-east-2"/etc. | region: us-west-2 |

cloudwatch_logger/config/sample_configuration.yaml

@@ -34,6 +34,10 @@ ignore_nodes: ["/cloudwatch_logger", "/cloudwatch_metrics_collector"]
 # default value is: DEBUG == logs of all log verbosity levels get sent to CloudWatch Logs
 min_log_verbosity: DEBUG
 
+# whether or not to include topic name information in the log messsages that are uploaded to cloudwatch
+# default value is: true
+publish_topic_names: true
+
 # The absolute path to a folder that all offline logs will be stored in
 storage_directory: "~/.ros/cwlogs/"
 

cloudwatch_logger/include/cloudwatch_logger/log_node.h

@@ -33,15 +33,32 @@ namespace Utils {
 class LogNode : public Service
 {
 public:
+  struct Options {
+    int8_t min_log_severity;
+    bool publish_topic_names;
+    std::unordered_set<std::string> ignore_nodes;
+  };
+
+  /**
+   * Creates a new CloudWatchLogNode
+   *
+   * @param options an options struct that specifies some behaviors of this CloudWatchLogNode
+   */
+  explicit LogNode(const Options & options);
+
   /**
-   * @brief Creates a new CloudWatchLogNode
+   * @deprecated Creates a new CloudWatchLogNode
    *
    * @param min_log_severity the minimum log severity level defined in the configuration file
    *                         logs with severity level equal or above get sent to CloudWatch Logs
    * @param ignore_nodes The set of node names to ignore logs from
    */
   explicit LogNode(int8_t min_log_severity, std::unordered_set<std::string> ignore_nodes);
 
+  LogNode(const LogNode & other) = delete;
+
+  LogNode & operator=(const LogNode & other) = delete;
+
   /**
    *  @brief Tears down a AWSCloudWatchLogNode object
    */
@@ -91,9 +108,11 @@ class LogNode : public Service
 private:
   bool ShouldSendToCloudWatchLogs(const int8_t log_severity_level);
   const std::string FormatLogs(const rosgraph_msgs::Log::ConstPtr & log_msg);
+
   std::shared_ptr<Aws::CloudWatchLogs::LogService> log_service_;
   int8_t min_log_severity_;
   std::unordered_set<std::string> ignore_nodes_;
+  bool publish_topic_names_;
 };
 
 }  // namespace Utils

cloudwatch_logger/include/cloudwatch_logger/log_node_param_helper.h

@@ -35,6 +35,7 @@ constexpr char kNodeParamSubscribeToRosoutKey[] = "sub_to_rosout";
 constexpr char kNodeParamLogGroupNameKey[] = "log_group_name";
 constexpr char kNodeParamLogTopicsListKey[] = "topics";
 constexpr char kNodeParamMinLogVerbosityKey[] = "min_log_verbosity";
+constexpr char kNodeParamPublishTopicNamesKey[] = "publish_topic_names";
 constexpr char kNodeParamIgnoreNodesKey[] = "ignore_nodes";
 
 /** Configuration params for Aws::DataFlow::UploaderOptions **/
@@ -51,11 +52,13 @@ constexpr char kNodeParamFileExtension[] = "file_extension";
 constexpr char kNodeParamMaximumFileSize[] = "maximum_file_size";
 constexpr char kNodeParamStorageLimit[] = "storage_limit";
 
+/** Default values for parameters **/
 constexpr char kNodeLogGroupNameDefaultValue[] = "ros_log_group";
 constexpr char kNodeLogStreamNameDefaultValue[] = "ros_log_stream";
 constexpr int8_t kNodeMinLogVerbosityDefaultValue = rosgraph_msgs::Log::DEBUG;
 constexpr double kNodePublishFrequencyDefaultValue = 5.0;
 constexpr bool kNodeSubscribeToRosoutDefaultValue = true;
+constexpr bool kNodePublishTopicNamesDefaultValue = true;
 
 /**
  * Fetch the parameter for the log publishing frequency.
@@ -116,6 +119,19 @@ Aws::AwsError ReadMinLogVerbosity(
   std::shared_ptr<Aws::Client::ParameterReaderInterface> parameter_reader,
   int8_t & min_log_verbosity);
 
+/**
+ * Fetch the parameter for whether or not to include topic name information in the log messsages
+ * that are uploaded to AWS CloudWatch Logs.
+ *
+ * @param parameter_reader to retrieve the parameters from.
+ * @param publish_topic_names the parameter is stored here when it is read successfully.
+ * @return an error code that indicates whether the parameter was read successfully or not, 
+ * as returned by \p parameter_reader
+ */
+Aws::AwsError ReadPublishTopicNames(
+  const std::shared_ptr<Aws::Client::ParameterReaderInterface>& parameter_reader,
+  bool & publish_topic_names);
+
 /**
  * Fetch the parameter for the list of topics to get logs from, and subscribe \p nh
  * to them. Also subscribe to rout if \p subscribe_to_rosout is true. 

cloudwatch_logger/package.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0"?>
 <package format="2">
   <name>cloudwatch_logger</name>
-  <version>2.2.1</version>
+  <version>2.3.1</version>
   <description>CloudWatch Logger node for publishing logs to AWS CloudWatch Logs</description>
 
   <url>http://wiki.ros.org/cloudwatch_logger</url>

cloudwatch_logger/src/log_node.cpp

@@ -30,12 +30,13 @@
 
 using namespace Aws::CloudWatchLogs::Utils;
 
-LogNode::LogNode(int8_t min_log_severity, std::unordered_set<std::string> ignore_nodes) 
-    : ignore_nodes_(std::move(ignore_nodes))
-{
-  this->log_service_ = nullptr;
-  this->min_log_severity_ = min_log_severity;
-}
+LogNode::LogNode(const Options & options)
+  : min_log_severity_(options.min_log_severity),
+    ignore_nodes_(options.ignore_nodes),
+    publish_topic_names_(options.publish_topic_names) {}
+
+LogNode::LogNode(int8_t min_log_severity, std::unordered_set<std::string> ignore_nodes)
+  : LogNode(Options{min_log_severity, true, std::move(ignore_nodes)}) {}
 
 LogNode::~LogNode() { this->log_service_ = nullptr; }
 
@@ -130,19 +131,21 @@ const std::string LogNode::FormatLogs(const rosgraph_msgs::Log::ConstPtr & log_m
   }
   ss << "[node name: " << log_msg->name << "] ";
 
-  ss << "[topics: ";
-  std::vector<std::string>::const_iterator it = log_msg->topics.begin();
-  std::vector<std::string>::const_iterator end = log_msg->topics.end();
-  for (; it != end; ++it) {
-    const std::string & topic = *it;
-
-    if (it != log_msg->topics.begin()) {
-      ss << ", ";
+  if (publish_topic_names_) {
+    ss << "[topics: ";
+    auto it = log_msg->topics.begin();
+    auto end = log_msg->topics.end();
+    for (; it != end; ++it) {
+      const std::string & topic = *it;
+      if (it != log_msg->topics.begin()) {
+        ss << ", ";
+      }
+      ss << topic;
     }
-
-    ss << topic;
+    ss  << "] ";
   }
-  ss << "] " << log_msg->msg << "\n";
+
+  ss << log_msg->msg << "\n";
 
   return ss.str();
-}
+}

cloudwatch_logger/src/log_node_param_helper.cpp

@@ -171,6 +171,37 @@ Aws::AwsError ReadMinLogVerbosity(
   return ret;
 }
 
+Aws::AwsError ReadPublishTopicNames(
+  const std::shared_ptr<Aws::Client::ParameterReaderInterface>& parameter_reader,
+  bool & publish_topic_names)
+{
+  Aws::AwsError ret =
+    parameter_reader->ReadParam(ParameterPath(kNodeParamPublishTopicNamesKey), publish_topic_names);
+
+  switch (ret) {
+    case Aws::AwsError::AWS_ERR_NOT_FOUND:
+      publish_topic_names = kNodePublishTopicNamesDefaultValue;
+      AWS_LOGSTREAM_INFO(
+      __func__,
+      "Whether to publish topic names to Cloudwatch Logs configuration not found, setting to default value: "
+        << kNodePublishTopicNamesDefaultValue);
+      break;
+    case Aws::AwsError::AWS_ERR_OK:
+      AWS_LOGSTREAM_INFO(
+      __func__, "Whether to publish topic names to Cloudwatch Logs is set to: " << publish_topic_names);
+      break;
+    default:
+      publish_topic_names = kNodePublishTopicNamesDefaultValue;
+      AWS_LOGSTREAM_ERROR(
+        __func__,
+        "Error " << ret 
+        << "retrieving parameter for whether to publish topic names to Cloudwatch Logs" 
+        << ", setting to default value: " << kNodePublishTopicNamesDefaultValue);
+  }
+
+  return ret;
+}
+
 Aws::AwsError ReadSubscriberList(
   const bool subscribe_to_rosout,
   std::shared_ptr<Aws::Client::ParameterReaderInterface> parameter_reader,

cloudwatch_logger/src/main.cpp

@@ -25,8 +25,6 @@
 
 #include <cloudwatch_logs_common/cloudwatch_options.h>
 
-using namespace Aws::CloudWatchLogs::Utils;
-
 constexpr char kNodeName[] = "cloudwatch_logger";
 
 int main(int argc, char ** argv)
@@ -41,9 +39,8 @@ int main(int argc, char ** argv)
   std::string log_group;
   std::string log_stream;
   bool subscribe_to_rosout;
-  int8_t min_log_verbosity;
-  std::vector<ros::Subscriber> subscriptions;
-  std::unordered_set<std::string> ignore_nodes;
+  Aws::SDKOptions sdk_options;
+  Aws::CloudWatchLogs::Utils::LogNode::Options logger_options;
   Aws::CloudWatchLogs::CloudWatchOptions cloudwatch_options;
 
   ros::NodeHandle nh;
@@ -52,22 +49,20 @@ int main(int argc, char ** argv)
     std::make_shared<Aws::Client::Ros1NodeParameterReader>();
 
   // checking configurations to set values or set to default values;
-  ReadPublishFrequency(parameter_reader, publish_frequency);
-  ReadLogGroup(parameter_reader, log_group);
-  ReadLogStream(parameter_reader, log_stream);
-  ReadSubscribeToRosout(parameter_reader, subscribe_to_rosout);
-  ReadMinLogVerbosity(parameter_reader, min_log_verbosity);
-  ReadIgnoreNodesSet(parameter_reader, ignore_nodes);
-
-  ReadCloudWatchOptions(parameter_reader, cloudwatch_options);
+  Aws::CloudWatchLogs::Utils::ReadPublishFrequency(parameter_reader, publish_frequency);
+  Aws::CloudWatchLogs::Utils::ReadLogGroup(parameter_reader, log_group);
+  Aws::CloudWatchLogs::Utils::ReadLogStream(parameter_reader, log_stream);
+  Aws::CloudWatchLogs::Utils::ReadSubscribeToRosout(parameter_reader, subscribe_to_rosout);
+  Aws::CloudWatchLogs::Utils::ReadMinLogVerbosity(parameter_reader, logger_options.min_log_severity);
+  Aws::CloudWatchLogs::Utils::ReadPublishTopicNames(parameter_reader, logger_options.publish_topic_names);
+  Aws::CloudWatchLogs::Utils::ReadIgnoreNodesSet(parameter_reader, logger_options.ignore_nodes);
+  Aws::CloudWatchLogs::Utils::ReadCloudWatchOptions(parameter_reader, cloudwatch_options);
 
   // configure aws settings
   Aws::Client::ClientConfigurationProvider client_config_provider(parameter_reader);
   Aws::Client::ClientConfiguration config = client_config_provider.GetClientConfiguration();
 
-  Aws::SDKOptions sdk_options;
-
-  Aws::CloudWatchLogs::Utils::LogNode cloudwatch_logger(min_log_verbosity, ignore_nodes);
+  Aws::CloudWatchLogs::Utils::LogNode cloudwatch_logger(logger_options);
   cloudwatch_logger.Initialize(log_group, log_stream, config, sdk_options, cloudwatch_options);
 
   ros::ServiceServer service = nh.advertiseService(kNodeName,
@@ -83,7 +78,8 @@ int main(int argc, char ** argv)
   };
 
   // subscribe to additional topics, if any
-  ReadSubscriberList(subscribe_to_rosout, parameter_reader, callback, nh, subscriptions);
+  std::vector<ros::Subscriber> subscriptions;
+  Aws::CloudWatchLogs::Utils::ReadSubscriberList(subscribe_to_rosout, parameter_reader, callback, nh, subscriptions);
   AWS_LOGSTREAM_INFO(__func__, "Initialized " << kNodeName << ".");
 
   bool publish_when_size_reached = cloudwatch_options.uploader_options.batch_trigger_publish_size

cloudwatch_logger/test/log_node_param_helper_test.cpp

@@ -252,6 +252,41 @@ TEST_F(LogNodeParamHelperFixture, TestReadReadMinLogVerbosity)
     EXPECT_EQ(kNodeMinLogVerbosityDefaultValue, param);
 }
 
+TEST_F(LogNodeParamHelperFixture, TestPublishTopicNames)
+{
+    {
+      InSequence read_param_seq;
+
+      EXPECT_CALL(*param_reader_, ReadParam(Eq(ParameterPath(kNodeParamPublishTopicNamesKey)), A<bool &>()))
+        .WillOnce(Return(AwsError::AWS_ERR_FAILURE));
+
+      EXPECT_CALL(*param_reader_, ReadParam(Eq(ParameterPath(kNodeParamPublishTopicNamesKey)), A<bool &>()))
+        .WillOnce(Return(AwsError::AWS_ERR_NOT_FOUND)); 
+
+      EXPECT_CALL(*param_reader_, ReadParam(Eq(ParameterPath(kNodeParamPublishTopicNamesKey)), A<bool &>()))
+        .WillOnce(DoAll(SetArgReferee<1>(true), Return(AwsError::AWS_ERR_OK)));
+
+      EXPECT_CALL(*param_reader_, ReadParam(Eq(ParameterPath(kNodeParamPublishTopicNamesKey)), A<bool &>()))
+        .WillOnce(DoAll(SetArgReferee<1>(false), Return(AwsError::AWS_ERR_OK)));
+    }
+
+    bool param = false;
+    EXPECT_EQ(AwsError::AWS_ERR_FAILURE, ReadPublishTopicNames(param_reader_, param));
+    EXPECT_EQ(kNodePublishTopicNamesDefaultValue, param);
+
+    param = false;
+    EXPECT_EQ(AwsError::AWS_ERR_NOT_FOUND, ReadPublishTopicNames(param_reader_, param));
+    EXPECT_EQ(kNodePublishTopicNamesDefaultValue, param);
+
+    param = false;
+    EXPECT_EQ(AwsError::AWS_ERR_OK, ReadPublishTopicNames(param_reader_, param));
+    EXPECT_EQ(true, param);
+
+    param = true;
+    EXPECT_EQ(AwsError::AWS_ERR_OK, ReadPublishTopicNames(param_reader_, param));
+    EXPECT_EQ(false, param);
+}
+
 AwsError MockReadParamAddStringToList(const ParameterPath & param_path, std::vector<std::string> & out)
 {
   (void)param_path;

cloudwatch_logger/test/log_node_test.cpp

@@ -103,10 +103,17 @@ class LogNodeFixture : public ::testing::Test
     log_service_factory = std::make_shared<LogServiceFactoryMock>();
   }
 
-  std::shared_ptr<LogNode> build_test_subject(int8_t severity_level = rosgraph_msgs::Log::DEBUG,
-      std::unordered_set<std::string> ignore_nodes = std::unordered_set<std::string>())
+  std::shared_ptr<LogNode> build_test_subject(
+    int8_t severity_level = rosgraph_msgs::Log::DEBUG,
+    bool publish_topic_names = true,
+    const std::unordered_set<std::string> & ignore_nodes = std::unordered_set<std::string>())
   {
-    return std::make_shared<LogNode>(severity_level, ignore_nodes);
+    Aws::CloudWatchLogs::Utils::LogNode::Options logger_options = {
+      severity_level,
+      publish_topic_names,
+      ignore_nodes
+    };
+    return std::make_shared<LogNode>(logger_options);
   }
 
   rosgraph_msgs::Log::ConstPtr message_to_constptr(rosgraph_msgs::Log log_message)
@@ -190,6 +197,26 @@ TEST_F(LogNodeFixture, TestRecordLogSevBelowMinSeverity)
   test_subject->RecordLogs(message_to_constptr(log_message_));
 }
 
+TEST_F(LogNodeFixture, TestDontPublishTopicNames)
+{
+  std::shared_ptr<LogNode> test_subject = build_test_subject(rosgraph_msgs::Log::DEBUG, false);
+  initialize_log_node(test_subject);
+
+  const std::string log_name_reference_str = std::string("[node name: ") + log_message_.name + "]";
+  const std::string log_topics_reference_str = "[topics: ]";
+
+  EXPECT_CALL(*log_service,
+    batchData(AllOf(
+      HasSubstr("DEBUG"),
+      HasSubstr(log_message_.msg),
+      HasSubstr(log_name_reference_str),
+      Not(HasSubstr(log_topics_reference_str))
+      )))
+    .WillOnce(Return(true));
+
+  test_subject->RecordLogs(message_to_constptr(log_message_));
+}
+
 TEST_F(LogNodeFixture, TestRecordLogSevEqGtMinSeverity)
 {
   std::shared_ptr<LogNode> test_subject = build_test_subject(rosgraph_msgs::Log::ERROR);
@@ -218,7 +245,7 @@ TEST_F(LogNodeFixture, TestRecordLogSevEqGtMinSeverity)
 
 TEST_F(LogNodeFixture, TestRecordLogTopicsOk)
 {
-  std::shared_ptr<LogNode> test_subject = build_test_subject(rosgraph_msgs::Log::DEBUG);
+  std::shared_ptr<LogNode> test_subject = build_test_subject();
 
   initialize_log_node(test_subject);
 
@@ -280,7 +307,7 @@ TEST_F(LogNodeFixture, TestRecordLogIgnoreList)
 {
   std::unordered_set<std::string> ignore_nodes;
   ignore_nodes.emplace(log_message_.name);
-  std::shared_ptr<LogNode> test_subject = build_test_subject(rosgraph_msgs::Log::DEBUG, ignore_nodes);
+  std::shared_ptr<LogNode> test_subject = build_test_subject(rosgraph_msgs::Log::DEBUG, true, ignore_nodes);
 
   initialize_log_node(test_subject);