Better ACLK debug communication log (#13281)
Timotej S committed
Jul 8, 2022 at 10:33 UTC
f8b7a9c63ba1ad01900623db563fff51dc9b5fa2
11 files changed
+189
-13
CMakeLists.txt
+2
@@ -846,6 +846,8 @@ set(ACLK_FILES
846
aclk/schema-wrappers/node_info.h
847
aclk/schema-wrappers/capability.cc
848
aclk/schema-wrappers/capability.h
849
+ aclk/schema-wrappers/proto_2_json.cc
850
+ aclk/schema-wrappers/proto_2_json.h
851
aclk/schema-wrappers/schema_wrappers.h
852
aclk/schema-wrappers/schema_wrapper_utils.cc
853
aclk/schema-wrappers/schema_wrapper_utils.h
Makefile.am
+2
@@ -669,6 +669,8 @@ ACLK_FILES = \
669
aclk/schema-wrappers/node_info.h \
670
aclk/schema-wrappers/capability.cc \
671
aclk/schema-wrappers/capability.h \
672
+ aclk/schema-wrappers/proto_2_json.cc \
673
+ aclk/schema-wrappers/proto_2_json.h \
674
aclk/schema-wrappers/schema_wrappers.h \
675
aclk/schema-wrappers/schema_wrapper_utils.cc \
676
aclk/schema-wrappers/schema_wrapper_utils.h \
aclk/aclk.c
+1
-1
@@ -254,7 +254,7 @@ static void msg_callback(const char *topic, const void *msg, size_t msglen, int
254
close(logfd);
255
#endif
256
257
- aclk_handle_new_cloud_msg(msgtype, msg, msglen);
257
+ aclk_handle_new_cloud_msg(msgtype, msg, msglen, topic);
258
}
259
260
static void puback_callback(uint16_t packet_id)
aclk/aclk_rx_msgs.c
+14
-1
@@ -6,6 +6,8 @@
6
#include "aclk_query_queue.h"
7
#include "aclk.h"
8
9
+#include "schema-wrappers/proto_2_json.h"
10
+
11
#define ACLK_V2_PAYLOAD_SEPARATOR "\x0D\x0A\x0D\x0A"
12
#define ACLK_CLOUD_REQ_V2_PREFIX "GET /"
13
@@ -478,7 +480,7 @@ unsigned int aclk_init_rx_msg_handlers(void)
480
return i;
481
}
482
481
-void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len)
483
+void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len, const char *topic)
484
{
485
if (aclk_stats_enabled) {
486
ACLK_STATS_LOCK;
@@ -496,6 +498,17 @@ void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t
498
}
499
return;
500
}
501
+
502
+#ifdef NETDATA_INTERNAL_CHECKS
503
+ if (!strncmp(message_type, "cmd", strlen("cmd"))) {
504
+ log_aclk_message_bin(msg, msg_len, 0, topic, msg_descriptor->name);
505
+ } else {
506
+ char *json = protomsg_to_json(msg, msg_len, msg_descriptor->name);
507
+ log_aclk_message_bin(json, strlen(json), 0, topic, msg_descriptor->name);
508
+ freez(json);
509
+ }
510
+#endif
511
+
512
if (aclk_stats_enabled) {
513
ACLK_STATS_LOCK;
514
aclk_proto_rx_msgs_sample[msg_descriptor-rx_msgs]++;
aclk/aclk_rx_msgs.h
+1
-1
@@ -12,6 +12,6 @@ int aclk_handle_cloud_cmd_message(char *payload);
12
13
const char *rx_handler_get_name(size_t i);
14
unsigned int aclk_init_rx_msg_handlers(void);
15
-void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len);
15
+void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len, const char *topic);
16
17
#endif /* ACLK_RX_MSGS_H */
aclk/aclk_tx_msgs.c
+5
-10
@@ -6,6 +6,8 @@
6
#include "aclk_stats.h"
7
#include "aclk.h"
8
9
+#include "schema-wrappers/proto_2_json.h"
10
+
11
#ifndef __GNUC__
12
#pragma region aclk_tx_msgs helper functions
13
#endif
@@ -33,16 +35,9 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
35
36
#ifdef NETDATA_INTERNAL_CHECKS
37
aclk_stats_msg_published(packet_id);
36
-#endif
37
-#ifdef ACLK_LOG_CONVERSATION_DIR
38
-#define FN_MAX_LEN 1024
39
- char filename[FN_MAX_LEN];
40
- snprintf(filename, FN_MAX_LEN, ACLK_LOG_CONVERSATION_DIR "/%010d-tx-%s.bin", ACLK_GET_CONV_LOG_NEXT(), msgname);
41
- FILE *fptr;
42
- if (fptr = fopen(filename,"w")) {
43
- fwrite(msg, msg_len, 1, fptr);
44
- fclose(fptr);
45
- }
38
+ char *json = protomsg_to_json(msg, msg_len, msgname);
39
+ log_aclk_message_bin(json, strlen(json), 1, topic, msgname);
40
+ freez(json);
41
#endif
42
43
return packet_id;
aclk/schema-wrappers/proto_2_json.cc
new
+89
@@ -0,0 +1,89 @@
1
+#include <google/protobuf/message.h>
2
+#include <google/protobuf/util/json_util.h>
3
+
4
+#include "proto/alarm/v1/config.pb.h"
5
+#include "proto/alarm/v1/stream.pb.h"
6
+#include "proto/aclk/v1/lib.pb.h"
7
+#include "proto/chart/v1/config.pb.h"
8
+#include "proto/chart/v1/stream.pb.h"
9
+#include "proto/agent/v1/connection.pb.h"
10
+#include "proto/agent/v1/disconnect.pb.h"
11
+#include "proto/nodeinstance/connection/v1/connection.pb.h"
12
+#include "proto/nodeinstance/create/v1/creation.pb.h"
13
+#include "proto/nodeinstance/info/v1/info.pb.h"
14
+
15
+#include "libnetdata/libnetdata.h"
16
+
17
+#include "proto_2_json.h"
18
+
19
+using namespace google::protobuf::util;
20
+
21
+static google::protobuf::Message *msg_name_to_protomsg(const char *msgname)
22
+{
23
+//tx side
24
+ if (!strcmp(msgname, "UpdateAgentConnection"))
25
+ return new agent::v1::UpdateAgentConnection;
26
+ if (!strcmp(msgname, "UpdateNodeInstanceConnection"))
27
+ return new nodeinstance::v1::UpdateNodeInstanceConnection;
28
+ if (!strcmp(msgname, "CreateNodeInstance"))
29
+ return new nodeinstance::create::v1::CreateNodeInstance;
30
+ if (!strcmp(msgname, "ChartsAndDimensionsUpdated"))
31
+ return new chart::v1::ChartsAndDimensionsUpdated;
32
+ if (!strcmp(msgname, "ChartConfigsUpdated"))
33
+ return new chart::v1::ChartConfigsUpdated;
34
+ if (!strcmp(msgname, "ResetChartMessages"))
35
+ return new chart::v1::ResetChartMessages;
36
+ if (!strcmp(msgname, "RetentionUpdated"))
37
+ return new chart::v1::RetentionUpdated;
38
+ if (!strcmp(msgname, "UpdateNodeInfo"))
39
+ return new nodeinstance::info::v1::UpdateNodeInfo;
40
+ if (!strcmp(msgname, "AlarmLogHealth"))
41
+ return new alarms::v1::AlarmLogHealth;
42
+ if (!strcmp(msgname, "ProvideAlarmConfiguration"))
43
+ return new alarms::v1::ProvideAlarmConfiguration;
44
+ if (!strcmp(msgname, "AlarmSnapshot"))
45
+ return new alarms::v1::AlarmSnapshot;
46
+ if (!strcmp(msgname, "AlarmLogEntry"))
47
+ return new alarms::v1::AlarmLogEntry;
48
+
49
+//rx side
50
+ if (!strcmp(msgname, "CreateNodeInstanceResult"))
51
+ return new nodeinstance::create::v1::CreateNodeInstanceResult;
52
+ if (!strcmp(msgname, "SendNodeInstances"))
53
+ return new agent::v1::SendNodeInstances;
54
+ if (!strcmp(msgname, "StreamChartsAndDimensions"))
55
+ return new chart::v1::StreamChartsAndDimensions;
56
+ if (!strcmp(msgname, "ChartsAndDimensionsAck"))
57
+ return new chart::v1::ChartsAndDimensionsAck;
58
+ if (!strcmp(msgname, "UpdateChartConfigs"))
59
+ return new chart::v1::UpdateChartConfigs;
60
+ if (!strcmp(msgname, "StartAlarmStreaming"))
61
+ return new alarms::v1::StartAlarmStreaming;
62
+ if (!strcmp(msgname, "SendAlarmLogHealth"))
63
+ return new alarms::v1::SendAlarmLogHealth;
64
+ if (!strcmp(msgname, "SendAlarmConfiguration"))
65
+ return new alarms::v1::SendAlarmConfiguration;
66
+ if (!strcmp(msgname, "SendAlarmSnapshot"))
67
+ return new alarms::v1::SendAlarmSnapshot;
68
+ if (!strcmp(msgname, "DisconnectReq"))
69
+ return new agent::v1::DisconnectReq;
70
+
71
+ return NULL;
72
+}
73
+
74
+char *protomsg_to_json(const void *protobin, size_t len, const char *msgname)
75
+{
76
+ google::protobuf::Message *msg = msg_name_to_protomsg(msgname);
77
+ if (msg == NULL)
78
+ return strdupz("Don't know this message type by name.");
79
+
80
+ if (!msg->ParseFromArray(protobin, len))
81
+ return strdupz("Can't parse this message. Malformed or wrong parser used.");
82
+
83
+ JsonPrintOptions options;
84
+
85
+ std::string output;
86
+ google::protobuf::util::MessageToJsonString(*msg, &output, options);
87
+ delete msg;
88
+ return strdupz(output.c_str());
89
+}
aclk/schema-wrappers/proto_2_json.h
new
+18
@@ -0,0 +1,18 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef PROTO_2_JSON_H
4
+#define PROTO_2_JSON_H
5
+
6
+#include <sys/types.h>
7
+
8
+#ifdef __cplusplus
9
+extern "C" {
10
+#endif
11
+
12
+char *protomsg_to_json(const void *protobin, size_t len, const char *msgname);
13
+
14
+#ifdef __cplusplus
15
+}
16
+#endif
17
+
18
+#endif /* PROTO_2_JSON_H */
daemon/main.c
+8
@@ -401,6 +401,14 @@ static void log_init(void) {
401
snprintfz(filename, FILENAME_MAX, "%s/access.log", netdata_configured_log_dir);
402
stdaccess_filename = config_get(CONFIG_SECTION_LOGS, "access", filename);
403
404
+#ifdef ENABLE_ACLK
405
+ aclklog_enabled = config_get_boolean(CONFIG_SECTION_CLOUD, "conversation log", CONFIG_BOOLEAN_NO);
406
+ if (aclklog_enabled) {
407
+ snprintfz(filename, FILENAME_MAX, "%s/aclk.log", netdata_configured_log_dir);
408
+ aclklog_filename = config_get(CONFIG_SECTION_CLOUD, "conversation log file", filename);
409
+ }
410
+#endif
411
+
412
char deffacility[8];
413
snprintfz(deffacility,7,"%s","daemon");
414
facility_log = config_get(CONFIG_SECTION_LOGS, "facility", deffacility);
libnetdata/log/log.c
+38
@@ -20,6 +20,14 @@ const char *stderr_filename = NULL;
20
const char *stdout_filename = NULL;
21
const char *facility_log = NULL;
22
23
+#ifdef ENABLE_ACLK
24
+const char *aclklog_filename = NULL;
25
+int aclklog_fd = -1;
26
+FILE *aclklog = NULL;
27
+int aclklog_syslog = 1;
28
+int aclklog_enabled = 0;
29
+#endif
30
+
31
// ----------------------------------------------------------------------------
32
// Log facility(https://tools.ietf.org/html/rfc5424)
33
//
@@ -562,6 +570,11 @@ void reopen_all_log_files() {
570
if(stderr_filename)
571
open_log_file(STDERR_FILENO, stderr, stderr_filename, &error_log_syslog, 0, NULL);
572
573
+#ifdef ENABLE_ACLK
574
+ if (aclklog_enabled)
575
+ aclklog = open_log_file(aclklog_fd, aclklog, aclklog_filename, NULL, 0, &aclklog_fd);
576
+#endif
577
+
578
if(stdaccess_filename)
579
stdaccess = open_log_file(stdaccess_fd, stdaccess, stdaccess_filename, &access_log_syslog, 1, &stdaccess_fd);
580
}
@@ -572,6 +585,12 @@ void open_all_log_files() {
585
586
open_log_file(STDOUT_FILENO, stdout, stdout_filename, &output_log_syslog, 0, NULL);
587
open_log_file(STDERR_FILENO, stderr, stderr_filename, &error_log_syslog, 0, NULL);
588
+
589
+#ifdef ENABLE_ACLK
590
+ if(aclklog_enabled)
591
+ aclklog = open_log_file(aclklog_fd, aclklog, aclklog_filename, NULL, 0, &aclklog_fd);
592
+#endif
593
+
594
stdaccess = open_log_file(stdaccess_fd, stdaccess, stdaccess_filename, &access_log_syslog, 1, &stdaccess_fd);
595
}
596
@@ -912,3 +931,22 @@ void log_access( const char *fmt, ... ) {
931
netdata_mutex_unlock(&access_mutex);
932
}
933
}
934
+
935
+#ifdef ENABLE_ACLK
936
+void log_aclk_message_bin( const char *data, const size_t data_len, int tx, const char *mqtt_topic, const char *message_name) {
937
+ if (aclklog) {
938
+ static netdata_mutex_t aclklog_mutex = NETDATA_MUTEX_INITIALIZER;
939
+ netdata_mutex_lock(&aclklog_mutex);
940
+
941
+ char date[LOG_DATE_LENGTH];
942
+ log_date(date, LOG_DATE_LENGTH);
943
+ fprintf(aclklog, "%s: %s Msg:\"%s\", MQTT-topic:\"%s\": ", date, tx ? "OUTGOING" : "INCOMING", message_name, mqtt_topic);
944
+
945
+ fwrite(data, data_len, 1, aclklog);
946
+
947
+ fputc('\n', aclklog);
948
+
949
+ netdata_mutex_unlock(&aclklog_mutex);
950
+ }
951
+}
952
+#endif
libnetdata/log/log.h
+11
@@ -61,6 +61,13 @@ extern const char *stderr_filename;
61
extern const char *stdout_filename;
62
extern const char *facility_log;
63
64
+#ifdef ENABLE_ACLK
65
+extern const char *aclklog_filename;
66
+extern int aclklog_fd;
67
+extern FILE *aclklog;
68
+extern int aclklog_enabled;
69
+#endif
70
+
71
extern int access_log_syslog;
72
extern int error_log_syslog;
73
extern int output_log_syslog;
@@ -98,6 +105,10 @@ extern void error_int( const char *prefix, const char *file, const char *functio
105
extern void fatal_int( const char *file, const char *function, const unsigned long line, const char *fmt, ... ) NORETURN PRINTFLIKE(4, 5);
106
extern void log_access( const char *fmt, ... ) PRINTFLIKE(1, 2);
107
108
+#ifdef ENABLE_ACLK
109
+extern void log_aclk_message_bin( const char *data, const size_t data_len, int tx, const char *mqtt_topic, const char *message_name);
110
+#endif
111
+
112
# ifdef __cplusplus
113
}
114
# endif