Better ACLK debug communication log (#13281)

6 files changed, 128 insertions, 13 deletions

diff --git a/aclk/aclk.c b/aclk/aclk.c
index 2489d02c3c..4477f1bb5f 100644
--- a/aclk/aclk.c
+++ b/aclk/aclk.c
@@ -254,7 +254,7 @@ static void msg_callback(const char *topic, const void *msg, size_t msglen, int
     close(logfd);
 #endif
 
-    aclk_handle_new_cloud_msg(msgtype, msg, msglen);
+    aclk_handle_new_cloud_msg(msgtype, msg, msglen, topic);
 }
 
 static void puback_callback(uint16_t packet_id)
diff --git a/aclk/aclk_rx_msgs.c b/aclk/aclk_rx_msgs.c
index a47918706b..4afe6335a8 100644
--- a/aclk/aclk_rx_msgs.c
+++ b/aclk/aclk_rx_msgs.c
@@ -6,6 +6,8 @@
 #include "aclk_query_queue.h"
 #include "aclk.h"
 
+#include "schema-wrappers/proto_2_json.h"
+
 #define ACLK_V2_PAYLOAD_SEPARATOR "\x0D\x0A\x0D\x0A"
 #define ACLK_CLOUD_REQ_V2_PREFIX "GET /"
 
@@ -478,7 +480,7 @@ unsigned int aclk_init_rx_msg_handlers(void)
     return i;
 }
 
-void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len)
+void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len, const char *topic)
 {
     if (aclk_stats_enabled) {
         ACLK_STATS_LOCK;
@@ -496,6 +498,17 @@ void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t
         }
         return;
     }
+
+#ifdef NETDATA_INTERNAL_CHECKS
+    if (!strncmp(message_type, "cmd", strlen("cmd"))) {
+        log_aclk_message_bin(msg, msg_len, 0, topic, msg_descriptor->name);
+    } else {
+        char *json = protomsg_to_json(msg, msg_len, msg_descriptor->name);
+        log_aclk_message_bin(json, strlen(json), 0, topic, msg_descriptor->name);
+        freez(json);
+    }
+#endif
+
     if (aclk_stats_enabled) {
         ACLK_STATS_LOCK;
         aclk_proto_rx_msgs_sample[msg_descriptor-rx_msgs]++;
diff --git a/aclk/aclk_rx_msgs.h b/aclk/aclk_rx_msgs.h
index 0df612786e..61921faec4 100644
--- a/aclk/aclk_rx_msgs.h
+++ b/aclk/aclk_rx_msgs.h
@@ -12,6 +12,6 @@ int aclk_handle_cloud_cmd_message(char *payload);
 
 const char *rx_handler_get_name(size_t i);
 unsigned int aclk_init_rx_msg_handlers(void);
-void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len);
+void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len, const char *topic);
 
 #endif /* ACLK_RX_MSGS_H */
diff --git a/aclk/aclk_tx_msgs.c b/aclk/aclk_tx_msgs.c
index 4f094b94ee..6b87bba759 100644
--- a/aclk/aclk_tx_msgs.c
+++ b/aclk/aclk_tx_msgs.c
@@ -6,6 +6,8 @@
 #include "aclk_stats.h"
 #include "aclk.h"
 
+#include "schema-wrappers/proto_2_json.h"
+
 #ifndef __GNUC__
 #pragma region aclk_tx_msgs helper functions
 #endif
@@ -33,16 +35,9 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
 
 #ifdef NETDATA_INTERNAL_CHECKS
     aclk_stats_msg_published(packet_id);
-#endif
-#ifdef ACLK_LOG_CONVERSATION_DIR
-#define FN_MAX_LEN 1024
-    char filename[FN_MAX_LEN];
-    snprintf(filename, FN_MAX_LEN, ACLK_LOG_CONVERSATION_DIR "/%010d-tx-%s.bin", ACLK_GET_CONV_LOG_NEXT(), msgname);
-    FILE *fptr;
-    if (fptr = fopen(filename,"w")) {
-        fwrite(msg, msg_len, 1, fptr);
-        fclose(fptr);
-    }
+    char *json = protomsg_to_json(msg, msg_len, msgname);
+    log_aclk_message_bin(json, strlen(json), 1, topic, msgname);
+    freez(json);
 #endif
 
     return packet_id;
diff --git a/aclk/schema-wrappers/proto_2_json.cc b/aclk/schema-wrappers/proto_2_json.cc
new file mode 100644
index 0000000000..2fd2bb3179
--- /dev/null
+++ b/aclk/schema-wrappers/proto_2_json.cc
@@ -0,0 +1,89 @@
+#include <google/protobuf/message.h>
+#include <google/protobuf/util/json_util.h>
+
+#include "proto/alarm/v1/config.pb.h"
+#include "proto/alarm/v1/stream.pb.h"
+#include "proto/aclk/v1/lib.pb.h"
+#include "proto/chart/v1/config.pb.h"
+#include "proto/chart/v1/stream.pb.h"
+#include "proto/agent/v1/connection.pb.h"
+#include "proto/agent/v1/disconnect.pb.h"
+#include "proto/nodeinstance/connection/v1/connection.pb.h"
+#include "proto/nodeinstance/create/v1/creation.pb.h"
+#include "proto/nodeinstance/info/v1/info.pb.h"
+
+#include "libnetdata/libnetdata.h"
+
+#include "proto_2_json.h"
+
+using namespace google::protobuf::util;
+
+static google::protobuf::Message *msg_name_to_protomsg(const char *msgname)
+{
+//tx side
+    if (!strcmp(msgname, "UpdateAgentConnection"))
+        return new agent::v1::UpdateAgentConnection;
+    if (!strcmp(msgname, "UpdateNodeInstanceConnection"))
+        return new nodeinstance::v1::UpdateNodeInstanceConnection;
+    if (!strcmp(msgname, "CreateNodeInstance"))
+        return new nodeinstance::create::v1::CreateNodeInstance;
+    if (!strcmp(msgname, "ChartsAndDimensionsUpdated"))
+        return new chart::v1::ChartsAndDimensionsUpdated;
+    if (!strcmp(msgname, "ChartConfigsUpdated"))
+        return new chart::v1::ChartConfigsUpdated;
+    if (!strcmp(msgname, "ResetChartMessages"))
+        return new chart::v1::ResetChartMessages;
+    if (!strcmp(msgname, "RetentionUpdated"))
+        return new chart::v1::RetentionUpdated;
+    if (!strcmp(msgname, "UpdateNodeInfo"))
+        return new nodeinstance::info::v1::UpdateNodeInfo;
+    if (!strcmp(msgname, "AlarmLogHealth"))
+        return new alarms::v1::AlarmLogHealth;
+    if (!strcmp(msgname, "ProvideAlarmConfiguration"))
+        return new alarms::v1::ProvideAlarmConfiguration;
+    if (!strcmp(msgname, "AlarmSnapshot"))
+        return new alarms::v1::AlarmSnapshot;
+    if (!strcmp(msgname, "AlarmLogEntry"))
+        return new alarms::v1::AlarmLogEntry;
+
+//rx side
+    if (!strcmp(msgname, "CreateNodeInstanceResult"))
+        return new nodeinstance::create::v1::CreateNodeInstanceResult;
+    if (!strcmp(msgname, "SendNodeInstances"))
+        return new agent::v1::SendNodeInstances;
+    if (!strcmp(msgname, "StreamChartsAndDimensions"))
+        return new chart::v1::StreamChartsAndDimensions;
+    if (!strcmp(msgname, "ChartsAndDimensionsAck"))
+        return new chart::v1::ChartsAndDimensionsAck;
+    if (!strcmp(msgname, "UpdateChartConfigs"))
+        return new chart::v1::UpdateChartConfigs;
+    if (!strcmp(msgname, "StartAlarmStreaming"))
+        return new alarms::v1::StartAlarmStreaming;
+    if (!strcmp(msgname, "SendAlarmLogHealth"))
+        return new alarms::v1::SendAlarmLogHealth;
+    if (!strcmp(msgname, "SendAlarmConfiguration"))
+        return new alarms::v1::SendAlarmConfiguration;
+    if (!strcmp(msgname, "SendAlarmSnapshot"))
+        return new alarms::v1::SendAlarmSnapshot;
+    if (!strcmp(msgname, "DisconnectReq"))
+        return new agent::v1::DisconnectReq;
+
+    return NULL;
+}
+
+char *protomsg_to_json(const void *protobin, size_t len, const char *msgname)
+{
+    google::protobuf::Message *msg = msg_name_to_protomsg(msgname);
+    if (msg == NULL)
+        return strdupz("Don't know this message type by name.");
+
+    if (!msg->ParseFromArray(protobin, len))
+        return strdupz("Can't parse this message. Malformed or wrong parser used.");
+
+    JsonPrintOptions options;
+
+    std::string output;
+    google::protobuf::util::MessageToJsonString(*msg, &output, options);
+    delete msg;
+    return strdupz(output.c_str());
+}
diff --git a/aclk/schema-wrappers/proto_2_json.h b/aclk/schema-wrappers/proto_2_json.h
new file mode 100644
index 0000000000..3bd98478c9
--- /dev/null
+++ b/aclk/schema-wrappers/proto_2_json.h
@@ -0,0 +1,18 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#ifndef PROTO_2_JSON_H
+#define PROTO_2_JSON_H
+
+#include <sys/types.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+char *protomsg_to_json(const void *protobin, size_t len, const char *msgname);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PROTO_2_JSON_H */