Add a Google Cloud Pub/Sub connector to the exporting engine (#8855)
* Implement formatters * Add specific configuration options * Add the connector to the Autotools and CMake configuration * Initialize a connector instance * Publish netdata metrics * Fix internal stats * Add unit tests * Improve the documentation
Vladimir Kobal committed
May 14, 2020 at 10:54 UTC
b5f8c224a9636c071fc474634380b4a93ea21c28
17 files changed
+879
-7
CMakeLists.txt
+56
-1
@@ -252,6 +252,23 @@ find_library(HAVE_KINESIS aws-cpp-sdk-kinesis)
252
# later we use:
253
# ${HAVE_KINESIS}
254
255
+# -----------------------------------------------------------------------------
256
+# Detect libgrpc
257
+
258
+pkg_check_modules(GRPC grpc)
259
+# later we use:
260
+# ${GRPCF_LIBRARIES}
261
+# ${GRPC_CFLAGS_OTHER}
262
+# ${GRPC_INCLUDE_DIRS}
263
+
264
+# -----------------------------------------------------------------------------
265
+# Detect libgoogleapis_cpp_pubsub_protos
266
+
267
+pkg_check_modules(PUBSUB googleapis_cpp_pubsub_protos)
268
+# later we use:
269
+# ${PUBSUB_LIBRARIES}
270
+# ${PUBSUB_CFLAGS_OTHER}
271
+# ${PUBSUB_INCLUDE_DIRS}
272
273
# -----------------------------------------------------------------------------
274
# Detect libprotobuf
@@ -662,6 +679,13 @@ set(KINESIS_EXPORTING_FILES
679
exporting/aws_kinesis/aws_kinesis_put_record.h
680
)
681
682
+set(PUBSUB_EXPORTING_FILES
683
+ exporting/pubsub/pubsub.c
684
+ exporting/pubsub/pubsub.h
685
+ exporting/pubsub/pubsub_publish.cc
686
+ exporting/pubsub/pubsub_publish.h
687
+ )
688
+
689
set(MONGODB_EXPORTING_FILES
690
exporting/mongodb/mongodb.c
691
exporting/mongodb/mongodb.h
@@ -758,6 +782,25 @@ ELSE()
782
message(STATUS "kinesis backend: disabled (requires AWS SDK for C++)")
783
ENDIF()
784
785
+# -----------------------------------------------------------------------------
786
+# Pub/Sub exporting connector
787
+
788
+IF(GRPC_LIBRARIES AND PUBSUB_LIBRARIES)
789
+ SET(ENABLE_EXPORTING_PUBSUB True)
790
+ELSE()
791
+ SET(ENABLE_EXPORTING_PUBSUB False)
792
+ENDIF()
793
+
794
+IF(ENABLE_EXPORTING_PUBSUB)
795
+ message(STATUS "pubsub exporting connector: enabled")
796
+ list(APPEND NETDATA_FILES ${PUBSUB_EXPORTING_FILES})
797
+ list(APPEND NETDATA_COMMON_LIBRARIES ${GRPC_LIBRARIES} ${PUBSUB_LIBRARIES})
798
+ list(APPEND NETDATA_COMMON_INCLUDE_DIRS ${GRPC_INCLUDE_DIRS} ${PUBSUB_INCLUDE_DIRS})
799
+ list(APPEND NETDATA_COMMON_CFLAGS ${GRPC_CFLAGS_OTHER} ${PUBSUB_CFLAGS_OTHER})
800
+ELSE()
801
+ message(STATUS "pubsub exporting connector: disabled (requires grpc and googleapis)")
802
+ENDIF()
803
+
804
# -----------------------------------------------------------------------------
805
# prometheus remote write backend
806
@@ -867,7 +910,7 @@ ELSEIF(MACOS)
910
911
ENDIF()
912
870
-IF(ENABLE_BACKEND_KINESIS OR ENABLE_BACKEND_PROMETHEUS_REMOTE_WRITE)
913
+IF(ENABLE_BACKEND_KINESIS OR ENABLE_EXPORTING_PUBSUB OR ENABLE_BACKEND_PROMETHEUS_REMOTE_WRITE)
914
set_property(TARGET netdata PROPERTY CXX_STANDARD 11)
915
set_property(TARGET netdata PROPERTY CMAKE_CXX_STANDARD_REQUIRED ON)
916
ENDIF()
@@ -1040,6 +1083,7 @@ if(BUILD_TESTING)
1083
set(TEST_NAME exporting_engine)
1084
set(PROMETHEUS_REMOTE_WRITE_LINK_OPTIONS)
1085
set(KINESIS_LINK_OPTIONS)
1086
+ set(PUBSUB_LINK_OPTIONS)
1087
set(MONGODB_LINK_OPTIONS)
1088
if(ENABLE_BACKEND_PROMETHEUS_REMOTE_WRITE)
1089
list(APPEND EXPORTING_ENGINE_FILES ${PROMETHEUS_REMOTE_WRITE_EXPORTING_FILES} ${PROTO_SRCS} ${PROTO_HDRS})
@@ -1061,6 +1105,16 @@ if(ENABLE_BACKEND_KINESIS)
1105
-Wl,--wrap=kinesis_get_result
1106
)
1107
endif()
1108
+if(ENABLE_EXPORTING_PUBSUB)
1109
+ list(APPEND EXPORTING_ENGINE_FILES ${PUBSUB_EXPORTING_FILES})
1110
+ list(
1111
+ APPEND PUBSUB_LINK_OPTIONS
1112
+ -Wl,--wrap=pubsub_init
1113
+ -Wl,--wrap=pubsub_add_message
1114
+ -Wl,--wrap=pubsub_publish
1115
+ -Wl,--wrap=pubsub_get_result
1116
+ )
1117
+endif()
1118
if(MONGOC_LIBRARIES)
1119
list(APPEND EXPORTING_ENGINE_FILES ${MONGODB_EXPORTING_FILES})
1120
list(
@@ -1116,6 +1170,7 @@ endif()
1170
-Wl,--wrap=send_main_rusage
1171
${PROMETHEUS_REMOTE_WRITE_LINK_OPTIONS}
1172
${KINESIS_LINK_OPTIONS}
1173
+ ${PUBSUB_LINK_OPTIONS}
1174
${MONGODB_LINK_OPTIONS}
1175
)
1176
target_link_libraries(${TEST_NAME}_testdriver libnetdata ${NETDATA_COMMON_LIBRARIES} ${CMOCKA_LIBRARIES})
Makefile.am
+22
@@ -531,6 +531,13 @@ KINESIS_EXPORTING_FILES = \
531
exporting/aws_kinesis/aws_kinesis_put_record.h \
532
$(NULL)
533
534
+PUBSUB_EXPORTING_FILES = \
535
+ exporting/pubsub/pubsub.c \
536
+ exporting/pubsub/pubsub.h \
537
+ exporting/pubsub/pubsub_publish.cc \
538
+ exporting/pubsub/pubsub_publish.h \
539
+ $(NULL)
540
+
541
MONGODB_EXPORTING_FILES = \
542
exporting/mongodb/mongodb.c \
543
exporting/mongodb/mongodb.h \
@@ -752,6 +759,11 @@ if ENABLE_BACKEND_KINESIS
759
netdata_LDADD += $(OPTIONAL_KINESIS_LIBS)
760
endif
761
762
+if ENABLE_EXPORTING_PUBSUB
763
+ netdata_SOURCES += $(PUBSUB_EXPORTING_FILES)
764
+ netdata_LDADD += $(OPTIONAL_PUBSUB_LIBS)
765
+endif
766
+
767
if ENABLE_BACKEND_PROMETHEUS_REMOTE_WRITE
768
netdata_SOURCES += $(PROMETHEUS_REMOTE_WRITE_BACKEND_FILES) $(PROMETHEUS_REMOTE_WRITE_EXPORTING_FILES)
769
netdata_LDADD += $(OPTIONAL_PROMETHEUS_REMOTE_WRITE_LIBS)
@@ -906,6 +918,16 @@ if ENABLE_BACKEND_KINESIS
918
-Wl,--wrap=kinesis_get_result \
919
$(NULL)
920
endif
921
+if ENABLE_EXPORTING_PUBSUB
922
+ exporting_tests_exporting_engine_testdriver_SOURCES += $(PUBSUB_EXPORTING_FILES)
923
+ exporting_tests_exporting_engine_testdriver_LDADD += $(OPTIONAL_PUBSUB_LIBS)
924
+ exporting_tests_exporting_engine_testdriver_LDFLAGS += \
925
+ -Wl,--wrap=pubsub_init \
926
+ -Wl,--wrap=pubsub_add_message \
927
+ -Wl,--wrap=pubsub_publish \
928
+ -Wl,--wrap=pubsub_get_result \
929
+ $(NULL)
930
+endif
931
if ENABLE_BACKEND_MONGODB
932
exporting_tests_exporting_engine_testdriver_SOURCES += $(MONGODB_EXPORTING_FILES)
933
exporting_tests_exporting_engine_testdriver_LDADD += $(OPTIONAL_MONGOC_LIBS)
configure.ac
+62
-3
@@ -77,6 +77,12 @@ AC_ARG_ENABLE(
77
,
78
[enable_backend_kinesis="detect"]
79
)
80
+AC_ARG_ENABLE(
81
+ [exporting-pubsub],
82
+ [AS_HELP_STRING([--enable-exporting-pubsub], [enable pubsub exporting connector @<:@default autodetect@:>@])],
83
+ ,
84
+ [enable_exporting_pubsub="detect"]
85
+)
86
AC_ARG_ENABLE(
87
[backend-prometheus-remote-write],
88
[AS_HELP_STRING([--enable-backend-prometheus-remote-write], [enable prometheus remote write backend @<:@default autodetect@:>@])],
@@ -1064,6 +1070,54 @@ AC_MSG_RESULT([${enable_backend_kinesis}])
1070
AM_CONDITIONAL([ENABLE_BACKEND_KINESIS], [test "${enable_backend_kinesis}" = "yes"])
1071
1072
1073
+# -----------------------------------------------------------------------------
1074
+# Pub/Sub exporting connector - googleapis
1075
+
1076
+PKG_CHECK_MODULES(
1077
+ [GRPC],
1078
+ [grpc],
1079
+ [have_libgrpc=yes],
1080
+ [have_libgrpc=no]
1081
+)
1082
+
1083
+PKG_CHECK_MODULES(
1084
+ [PUBSUB],
1085
+ [googleapis_cpp_pubsub_protos],
1086
+ [have_pubsub_protos=yes],
1087
+ [have_pubsub_protos=no]
1088
+)
1089
+
1090
+AC_PATH_PROG([CXX_BINARY], [${CXX}], [no])
1091
+AS_IF(
1092
+ [test x"${CXX_BINARY}" == x"no"],
1093
+ [have_CXX_compiler=no],
1094
+ [have_CXX_compiler=yes]
1095
+)
1096
+
1097
+test "${enable_pubsub}" = "yes" -a "${have_grpc}" != "yes" && \
1098
+ AC_MSG_ERROR([libgrpc required but not found. try installing grpc])
1099
+
1100
+test "${enable_pubsub}" = "yes" -a "${have_pubsub_protos}" != "yes" && \
1101
+ AC_MSG_ERROR([libgoogleapis_cpp_pubsub_protos required but not found. try installing googleapis])
1102
+
1103
+test "${enable_backend_prometheus_remote_write}" = "yes" -a "${have_CXX_compiler}" != "yes" && \
1104
+ AC_MSG_ERROR([C++ compiler required but not found. try installing g++])
1105
+
1106
+AC_MSG_CHECKING([if pubsub exporting connector should be enabled])
1107
+if test "${enable_exporting_pubsub}" != "no" -a "${have_pubsub_protos}" = "yes" -a "${have_CXX_compiler}" = "yes"; then
1108
+ enable_exporting_pubsub="yes"
1109
+ AC_DEFINE([ENABLE_EXPORTING_PUBSUB], [1], [Pub/Sub API usability])
1110
+ OPTIONAL_PUBSUB_CFLAGS="${GRPC_CFLAGS} ${PUBSUB_CFLAGS}"
1111
+ CXX11FLAG="-std=c++11"
1112
+ OPTIONAL_PUBSUB_LIBS="${GRPC_LIBS} ${PUBSUB_LIBS}"
1113
+else
1114
+ enable_pubsub="no"
1115
+fi
1116
+
1117
+AC_MSG_RESULT([${enable_exporting_pubsub}])
1118
+AM_CONDITIONAL([ENABLE_EXPORTING_PUBSUB], [test "${enable_exporting_pubsub}" = "yes"])
1119
+
1120
+
1121
# -----------------------------------------------------------------------------
1122
# Prometheus remote write backend - libprotobuf, libsnappy, protoc
1123
@@ -1223,7 +1277,9 @@ AC_MSG_RESULT([${enable_lto}])
1277
1278
# -----------------------------------------------------------------------------
1279
1226
-AM_CONDITIONAL([ENABLE_CXX_LINKER], [test "${enable_backend_kinesis}" = "yes" -o "${enable_backend_prometheus_remote_write}" = "yes"])
1280
+AM_CONDITIONAL([ENABLE_CXX_LINKER], [test "${enable_backend_kinesis}" = "yes" \
1281
+ -o "${enable_exporting_pubsub}" = "yes" \
1282
+ -o "${enable_backend_prometheus_remote_write}" = "yes"])
1283
1284
AC_DEFINE_UNQUOTED([NETDATA_USER], ["${with_user}"], [use this user to drop privileged])
1285
@@ -1253,8 +1309,8 @@ AC_SUBST([webdir])
1309
1310
CFLAGS="${CFLAGS} ${OPTIONAL_MATH_CFLAGS} ${OPTIONAL_NFACCT_CFLAGS} ${OPTIONAL_ZLIB_CFLAGS} ${OPTIONAL_UUID_CFLAGS} \
1311
${OPTIONAL_LIBCAP_CFLAGS} ${OPTIONAL_IPMIMONITORING_CFLAGS} ${OPTIONAL_CUPS_CFLAGS} ${OPTIONAL_XENSTAT_FLAGS} \
1256
- ${OPTIONAL_KINESIS_CFLAGS} ${OPTIONAL_PROMETHEUS_REMOTE_WRITE_CFLAGS} ${OPTIONAL_MONGOC_CFLAGS} ${LWS_CFLAGS} \
1257
- ${OPTIONAL_JSONC_STATIC_CFLAGS}"
1312
+ ${OPTIONAL_KINESIS_CFLAGS} ${OPTIONAL_PUBSUB_CFLAGS} ${OPTIONAL_PROMETHEUS_REMOTE_WRITE_CFLAGS} \
1313
+ ${OPTIONAL_MONGOC_CFLAGS} ${LWS_CFLAGS} ${OPTIONAL_JSONC_STATIC_CFLAGS}"
1314
1315
CXXFLAGS="${CFLAGS} ${CXX11FLAG}"
1316
@@ -1295,6 +1351,8 @@ AC_SUBST([OPTIONAL_XENSTAT_CFLAGS])
1351
AC_SUBST([OPTIONAL_XENSTAT_LIBS])
1352
AC_SUBST([OPTIONAL_KINESIS_CFLAGS])
1353
AC_SUBST([OPTIONAL_KINESIS_LIBS])
1354
+AC_SUBST([OPTIONAL_PUBSUB_CFLAGS])
1355
+AC_SUBST([OPTIONAL_PUBSUB_LIBS])
1356
AC_SUBST([OPTIONAL_PROMETHEUS_REMOTE_WRITE_CFLAGS])
1357
AC_SUBST([OPTIONAL_PROMETHEUS_REMOTE_WRITE_LIBS])
1358
AC_SUBST([OPTIONAL_MONGOC_CFLAGS])
@@ -1373,6 +1431,7 @@ AC_CONFIG_FILES([
1431
exporting/prometheus/Makefile
1432
exporting/prometheus/remote_write/Makefile
1433
exporting/aws_kinesis/Makefile
1434
+ exporting/pubsub/Makefile
1435
exporting/mongodb/Makefile
1436
exporting/tests/Makefile
1437
health/Makefile
exporting/Makefile.am
+1
@@ -10,6 +10,7 @@ SUBDIRS = \
10
opentsdb \
11
prometheus \
12
aws_kinesis \
13
+ pubsub \
14
mongodb \
15
$(NULL)
16
exporting/exporting_engine.h
+7
@@ -50,6 +50,7 @@ typedef enum exporting_connector_types {
50
EXPORTING_CONNECTOR_TYPE_JSON, // Stores the data using JSON.
51
EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE, // The user selected to use Prometheus backend
52
EXPORTING_CONNECTOR_TYPE_KINESIS, // Send message to AWS Kinesis
53
+ EXPORTING_CONNECTOR_TYPE_PUBSUB, // Send message to Google Cloud Pub/Sub
54
EXPORTING_CONNECTOR_TYPE_MONGODB, // Send data to MongoDB collection
55
EXPORTING_CONNECTOR_TYPE_NUM // Number of backend types
56
} EXPORTING_CONNECTOR_TYPE;
@@ -87,6 +88,12 @@ struct aws_kinesis_specific_config {
88
char *secure_key;
89
};
90
91
+struct pubsub_specific_config {
92
+ char *credentials_file;
93
+ char *project_id;
94
+ char *topic_id;
95
+};
96
+
97
struct mongodb_specific_config {
98
char *database;
99
char *collection;
exporting/init_connectors.c
+10
@@ -13,6 +13,10 @@
13
#include "aws_kinesis/aws_kinesis.h"
14
#endif
15
16
+#if ENABLE_EXPORTING_PUBSUB
17
+#include "pubsub/pubsub.h"
18
+#endif
19
+
20
#if HAVE_MONGOC
21
#include "mongodb/mongodb.h"
22
#endif
@@ -58,6 +62,12 @@ int init_connectors(struct engine *engine)
62
#if HAVE_KINESIS
63
if (init_aws_kinesis_instance(instance) != 0)
64
return 1;
65
+#endif
66
+ break;
67
+ case EXPORTING_CONNECTOR_TYPE_PUBSUB:
68
+#if ENABLE_EXPORTING_PUBSUB
69
+ if (init_pubsub_instance(instance) != 0)
70
+ return 1;
71
#endif
72
break;
73
case EXPORTING_CONNECTOR_TYPE_MONGODB:
exporting/pubsub/Makefile.am
new
+8
@@ -0,0 +1,8 @@
1
+# SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+AUTOMAKE_OPTIONS = subdir-objects
4
+MAINTAINERCLEANFILES = $(srcdir)/Makefile.in
5
+
6
+dist_noinst_DATA = \
7
+ README.md \
8
+ $(NULL)
exporting/pubsub/README.md
new
+39
@@ -0,0 +1,39 @@
1
+<!--
2
+---
3
+title: "Export metrics to Google Cloud Pub/Sub Service"
4
+custom_edit_url: https://github.com/netdata/netdata/edit/master/exporting/pubsub/README.md
5
+---
6
+-->
7
+
8
+# Export metrics to Google Cloud Pub/Sub Service
9
+
10
+## Prerequisites
11
+
12
+To use the Pub/Sub service for metric collecting and processing, you should first
13
+[install](https://github.com/googleapis/cpp-cmakefiles) Google Cloud Platform C++ Proto Libraries.
14
+Pub/Sub support is also dependent on the dependencies of those libraries, like `protobuf` and `grpc`. Next, Netdata
15
+should be re-installed from the source. The installer will detect that the required libraries are now available.
16
+
17
+## Configuration
18
+
19
+To enable data sending to the Pub/Sub service, run `./edit-config exporting.conf` in the Netdata configuration directory
20
+and set the following options:
21
+
22
+```conf
23
+[pubsub:my_instance]
24
+ enabled = yes
25
+ destination = pubsub.googleapis.com
26
+ credentials file = /etc/netdata/google_cloud_credentials.json
27
+ project id = my_project
28
+ topic id = my_topic
29
+```
30
+
31
+Set the `destination` option to a Pub/Sub service endpoint. `pubsub.googleapis.com` is the default one.
32
+
33
+Next, create the credentials JSON file by following Google Cloud's [authentication guide]
34
+(https://cloud.google.com/docs/authentication/getting-started#creating_a_service_account). The user running the Agent
35
+(typically `netdata`) needs read access to `google_cloud_credentials.json`, which you can set with
36
+`chmod 400 google_cloud_credentials.json; chown netdata google_cloud_credentials.json`. Set the `credentials file`
37
+option to the full path of the file.
38
+
39
+[](<>)
exporting/pubsub/pubsub.c
new
+155
@@ -0,0 +1,155 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "pubsub.h"
4
+
5
+/**
6
+ * Initialize Pub/Sub connector instance
7
+ *
8
+ * @param instance an instance data structure.
9
+ * @return Returns 0 on success, 1 on failure.
10
+ */
11
+int init_pubsub_instance(struct instance *instance)
12
+{
13
+ instance->worker = pubsub_connector_worker;
14
+
15
+ instance->start_batch_formatting = NULL;
16
+ instance->start_host_formatting = format_host_labels_json_plaintext;
17
+ instance->start_chart_formatting = NULL;
18
+
19
+
20
+ if (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_AS_COLLECTED)
21
+ instance->metric_formatting = format_dimension_collected_json_plaintext;
22
+ else
23
+ instance->metric_formatting = format_dimension_stored_json_plaintext;
24
+
25
+ instance->end_chart_formatting = NULL;
26
+ instance->end_host_formatting = flush_host_labels;
27
+ instance->end_batch_formatting = NULL;
28
+
29
+ instance->send_header = NULL;
30
+ instance->check_response = NULL;
31
+
32
+ instance->buffer = (void *)buffer_create(0);
33
+ if (!instance->buffer) {
34
+ error("EXPORTING: cannot create buffer for Pub/Sub exporting connector instance %s", instance->config.name);
35
+ return 1;
36
+ }
37
+ uv_mutex_init(&instance->mutex);
38
+ uv_cond_init(&instance->cond_var);
39
+
40
+ struct pubsub_specific_data *connector_specific_data = callocz(1, sizeof(struct pubsub_specific_data));
41
+ instance->connector_specific_data = (void *)connector_specific_data;
42
+
43
+ struct pubsub_specific_config *connector_specific_config =
44
+ (struct pubsub_specific_config *)instance->config.connector_specific_config;
45
+ char error_message[ERROR_LINE_MAX + 1] = "";
46
+ if (pubsub_init(
47
+ (void *)connector_specific_data, error_message, instance->config.destination,
48
+ connector_specific_config->credentials_file, connector_specific_config->project_id,
49
+ connector_specific_config->topic_id)) {
50
+ error(
51
+ "EXPORTING: Cannot initialize a Pub/Sub publisher for instance %s: %s",
52
+ instance->config.name, error_message);
53
+ return 1;
54
+ }
55
+
56
+ return 0;
57
+}
58
+
59
+/**
60
+ * Pub/Sub connector worker
61
+ *
62
+ * Runs in a separate thread for every instance.
63
+ *
64
+ * @param instance_p an instance data structure.
65
+ */
66
+void pubsub_connector_worker(void *instance_p)
67
+{
68
+ struct instance *instance = (struct instance *)instance_p;
69
+ struct pubsub_specific_config *connector_specific_config = instance->config.connector_specific_config;
70
+ struct pubsub_specific_data *connector_specific_data = instance->connector_specific_data;
71
+
72
+ while (!netdata_exit) {
73
+ struct stats *stats = &instance->stats;
74
+ char error_message[ERROR_LINE_MAX + 1] = "";
75
+
76
+ uv_mutex_lock(&instance->mutex);
77
+ uv_cond_wait(&instance->cond_var, &instance->mutex);
78
+
79
+ // reset the monitoring chart counters
80
+ stats->received_bytes =
81
+ stats->sent_bytes =
82
+ stats->sent_metrics =
83
+ stats->lost_metrics =
84
+ stats->receptions =
85
+ stats->transmission_successes =
86
+ stats->transmission_failures =
87
+ stats->data_lost_events =
88
+ stats->lost_bytes =
89
+ stats->reconnects = 0;
90
+
91
+ BUFFER *buffer = (BUFFER *)instance->buffer;
92
+ size_t buffer_len = buffer_strlen(buffer);
93
+
94
+ stats->buffered_bytes = buffer_len;
95
+
96
+ if (pubsub_add_message(instance->connector_specific_data, (char *)buffer_tostring(buffer))) {
97
+ error("EXPORTING: Instance %s: Cannot add data to a message", instance->config.name);
98
+
99
+ stats->data_lost_events++;
100
+ stats->lost_metrics += stats->buffered_metrics;
101
+ stats->lost_bytes += buffer_len;
102
+
103
+ goto cleanup;
104
+ }
105
+
106
+ debug(
107
+ D_BACKEND, "EXPORTING: pubsub_publish(): project = %s, topic = %s, buffer = %zu",
108
+ connector_specific_config->project_id, connector_specific_config->topic_id, buffer_len);
109
+
110
+ if (pubsub_publish((void *)connector_specific_data, error_message, stats->buffered_metrics, buffer_len)) {
111
+ error("EXPORTING: Instance: %s: Cannot publish a message: %s", instance->config.name, error_message);
112
+
113
+ stats->transmission_failures++;
114
+ stats->data_lost_events++;
115
+ stats->lost_metrics += stats->buffered_metrics;
116
+ stats->lost_bytes += buffer_len;
117
+
118
+ goto cleanup;
119
+ }
120
+
121
+ stats->sent_bytes = buffer_len;
122
+ stats->transmission_successes++;
123
+
124
+ size_t sent_metrics = 0, lost_metrics = 0, sent_bytes = 0, lost_bytes = 0;
125
+
126
+ if (unlikely(pubsub_get_result(
127
+ connector_specific_data, error_message, &sent_metrics, &sent_bytes, &lost_metrics, &lost_bytes))) {
128
+ // oops! we couldn't send (all or some of the) data
129
+ error("EXPORTING: %s", error_message);
130
+ error(
131
+ "EXPORTING: failed to write data to service '%s'. Willing to write %zu bytes, wrote %zu bytes.",
132
+ instance->config.destination, lost_bytes, sent_bytes);
133
+
134
+ stats->transmission_failures++;
135
+ stats->data_lost_events++;
136
+ stats->lost_metrics += lost_metrics;
137
+ stats->lost_bytes += lost_bytes;
138
+ } else {
139
+ stats->receptions++;
140
+ stats->sent_metrics = sent_metrics;
141
+ }
142
+
143
+ cleanup:
144
+ send_internal_metrics(instance);
145
+
146
+ buffer_flush(buffer);
147
+ stats->buffered_metrics = 0;
148
+
149
+ uv_mutex_unlock(&instance->mutex);
150
+
151
+#ifdef UNIT_TESTING
152
+ break;
153
+#endif
154
+ }
155
+}
exporting/pubsub/pubsub.h
new
+13
@@ -0,0 +1,13 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_EXPORTING_PUBSUB_H
4
+#define NETDATA_EXPORTING_PUBSUB_H
5
+
6
+#include "exporting/exporting_engine.h"
7
+#include "exporting/json/json.h"
8
+#include "pubsub_publish.h"
9
+
10
+int init_pubsub_instance(struct instance *instance);
11
+void pubsub_connector_worker(void *instance_p);
12
+
13
+#endif //NETDATA_EXPORTING_PUBSUB_H
exporting/pubsub/pubsub_publish.cc
new
+229
@@ -0,0 +1,229 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include <google/pubsub/v1/pubsub.grpc.pb.h>
4
+#include <grpcpp/grpcpp.h>
5
+#include <stdexcept>
6
+#include "pubsub_publish.h"
7
+
8
+#define EVENT_CHECK_TIMEOUT 50
9
+
10
+struct response {
11
+ grpc::ClientContext *context;
12
+ google::pubsub::v1::PublishResponse *publish_response;
13
+ size_t tag;
14
+ grpc::Status *status;
15
+
16
+ size_t published_metrics;
17
+ size_t published_bytes;
18
+};
19
+
20
+static inline void copy_error_message(char *error_message_dst, const char *error_message_src)
21
+{
22
+ std::strncpy(error_message_dst, error_message_src, ERROR_LINE_MAX);
23
+ error_message_dst[ERROR_LINE_MAX] = '\0';
24
+}
25
+
26
+/**
27
+ * Initialize a Pub/Sub client and a data structure for responses.
28
+ *
29
+ * @param pubsub_specific_data_p a pointer to a structure with instance-wide data.
30
+ * @param error_message report error message to a caller.
31
+ * @param destination a Pub/Sub service endpoint.
32
+ * @param credentials_file a full path for a file with google application credentials.
33
+ * @param project_id a project ID.
34
+ * @param topic_id a topic ID.
35
+ * @return Returns 0 on success, 1 on failure.
36
+ */
37
+int pubsub_init(
38
+ void *pubsub_specific_data_p, char *error_message, const char *destination, const char *credentials_file,
39
+ const char *project_id, const char *topic_id)
40
+{
41
+ struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
42
+
43
+ try {
44
+ setenv("GOOGLE_APPLICATION_CREDENTIALS", credentials_file, 0);
45
+
46
+ std::shared_ptr<grpc::ChannelCredentials> credentials = grpc::GoogleDefaultCredentials();
47
+ if (credentials == nullptr) {
48
+ copy_error_message(error_message, "Can't load credentials");
49
+ return 1;
50
+ }
51
+
52
+ std::shared_ptr<grpc::Channel> channel = grpc::CreateChannel(destination, credentials);
53
+
54
+ google::pubsub::v1::Publisher::Stub *stub = new google::pubsub::v1::Publisher::Stub(channel);
55
+ if (!stub) {
56
+ copy_error_message(error_message, "Can't create a publisher stub");
57
+ return 1;
58
+ }
59
+
60
+ connector_specific_data->stub = stub;
61
+
62
+ google::pubsub::v1::PublishRequest *request = new google::pubsub::v1::PublishRequest;
63
+ connector_specific_data->request = request;
64
+ ((google::pubsub::v1::PublishRequest *)(connector_specific_data->request))
65
+ ->set_topic(std::string("projects/") + project_id + "/topics/" + topic_id);
66
+
67
+ grpc::CompletionQueue *cq = new grpc::CompletionQueue;
68
+ connector_specific_data->completion_queue = cq;
69
+
70
+ connector_specific_data->responses = new std::list<struct response>;
71
+
72
+ return 0;
73
+ } catch (std::exception const &ex) {
74
+ std::string em(std::string("Standard exception raised: ") + ex.what());
75
+ copy_error_message(error_message, em.c_str());
76
+ return 1;
77
+ }
78
+
79
+ return 0;
80
+}
81
+
82
+/**
83
+ * Add data to a Pub/Sub request message.
84
+ *
85
+ * @param pubsub_specific_data_p a pointer to a structure with instance-wide data.
86
+ * @param data a text buffer with metrics.
87
+ * @return Returns 0 on success, 1 on failure.
88
+ */
89
+int pubsub_add_message(void *pubsub_specific_data_p, char *data)
90
+{
91
+ struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
92
+
93
+ try {
94
+ google::pubsub::v1::PubsubMessage *message =
95
+ ((google::pubsub::v1::PublishRequest *)(connector_specific_data->request))->add_messages();
96
+ if (!message)
97
+ return 1;
98
+
99
+ message->set_data(data);
100
+ } catch (std::exception const &ex) {
101
+ return 1;
102
+ }
103
+
104
+ return 0;
105
+}
106
+
107
+/**
108
+ * Send data to the Pub/Sub service
109
+ *
110
+ * @param pubsub_specific_data_p a pointer to a structure with client and request outcome information.
111
+ * @param error_message report error message to a caller.
112
+ * @param buffered_metrics the number of metrics we are going to send.
113
+ * @param buffered_bytes the number of bytes we are going to send.
114
+ * @return Returns 0 on success, 1 on failure.
115
+ */
116
+int pubsub_publish(void *pubsub_specific_data_p, char *error_message, size_t buffered_metrics, size_t buffered_bytes)
117
+{
118
+ struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
119
+
120
+ try {
121
+ grpc::ClientContext *context = new grpc::ClientContext;
122
+
123
+ std::unique_ptr<grpc::ClientAsyncResponseReader<google::pubsub::v1::PublishResponse> > rpc(
124
+ ((google::pubsub::v1::Publisher::Stub *)(connector_specific_data->stub))
125
+ ->AsyncPublish(
126
+ context, (*(google::pubsub::v1::PublishRequest *)(connector_specific_data->request)),
127
+ ((grpc::CompletionQueue *)(connector_specific_data->completion_queue))));
128
+
129
+ struct response response;
130
+ response.context = context;
131
+ response.publish_response = new google::pubsub::v1::PublishResponse;
132
+ response.tag = connector_specific_data->last_tag++;
133
+ response.status = new grpc::Status;
134
+ response.published_metrics = buffered_metrics;
135
+ response.published_bytes = buffered_bytes;
136
+
137
+ rpc->Finish(response.publish_response, response.status, (void *)response.tag);
138
+
139
+ ((google::pubsub::v1::PublishRequest *)(connector_specific_data->request))->clear_messages();
140
+
141
+ ((std::list<struct response> *)(connector_specific_data->responses))->push_back(response);
142
+ } catch (std::exception const &ex) {
143
+ std::string em(std::string("Standard exception raised: ") + ex.what());
144
+ copy_error_message(error_message, em.c_str());
145
+ return 1;
146
+ }
147
+
148
+ return 0;
149
+}
150
+
151
+/**
152
+ * Get results from service responces
153
+ *
154
+ * @param pubsub_specific_data_p a pointer to a structure with instance-wide data.
155
+ * @param error_message report error message to a caller.
156
+ * @param sent_metrics report to a caller how many metrics was successfuly sent.
157
+ * @param sent_bytes report to a caller how many bytes was successfuly sent.
158
+ * @param lost_metrics report to a caller how many metrics was lost during transmission.
159
+ * @param lost_bytes report to a caller how many bytes was lost during transmission.
160
+ * @return Returns 0 if all data was sent successfully, 1 when data was lost on transmission.
161
+ */
162
+int pubsub_get_result(
163
+ void *pubsub_specific_data_p, char *error_message,
164
+ size_t *sent_metrics, size_t *sent_bytes, size_t *lost_metrics, size_t *lost_bytes)
165
+{
166
+ struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
167
+ std::list<struct response> *responses = (std::list<struct response> *)connector_specific_data->responses;
168
+ grpc_impl::CompletionQueue::NextStatus next_status;
169
+
170
+ *sent_metrics = 0;
171
+ *sent_bytes = 0;
172
+ *lost_metrics = 0;
173
+ *lost_bytes = 0;
174
+
175
+ try {
176
+ do {
177
+ std::list<struct response>::iterator response;
178
+ void *got_tag;
179
+ bool ok = false;
180
+
181
+ auto deadline = std::chrono::system_clock::now() + std::chrono::milliseconds(50);
182
+ next_status = (*(grpc::CompletionQueue *)(connector_specific_data->completion_queue))
183
+ .AsyncNext(&got_tag, &ok, deadline);
184
+
185
+ if (next_status == grpc::CompletionQueue::GOT_EVENT) {
186
+ for (response = responses->begin(); response != responses->end(); ++response) {
187
+ if ((void *)response->tag == got_tag)
188
+ break;
189
+ }
190
+
191
+ if (response == responses->end()) {
192
+ copy_error_message(error_message, "Cannot get Pub/Sub response");
193
+ return 1;
194
+ }
195
+
196
+ if (ok && response->publish_response->message_ids_size()) {
197
+ *sent_metrics += response->published_metrics;
198
+ *sent_bytes += response->published_bytes;
199
+ } else {
200
+ *lost_metrics += response->published_metrics;
201
+ *lost_bytes += response->published_bytes;
202
+ response->status->error_message().copy(error_message, ERROR_LINE_MAX);
203
+ error_message[ERROR_LINE_MAX] = '\0';
204
+ }
205
+
206
+ delete response->context;
207
+ delete response->publish_response;
208
+ delete response->status;
209
+ responses->erase(response);
210
+ }
211
+
212
+ if (next_status == grpc::CompletionQueue::SHUTDOWN) {
213
+ copy_error_message(error_message, "Completion queue shutdown");
214
+ return 1;
215
+ }
216
+
217
+ } while (next_status == grpc::CompletionQueue::GOT_EVENT);
218
+ } catch (std::exception const &ex) {
219
+ std::string em(std::string("Standard exception raised: ") + ex.what());
220
+ copy_error_message(error_message, em.c_str());
221
+ return 1;
222
+ }
223
+
224
+ if (*lost_metrics) {
225
+ return 1;
226
+ }
227
+
228
+ return 0;
229
+}
exporting/pubsub/pubsub_publish.h
new
+36
@@ -0,0 +1,36 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_EXPORTING_PUBSUB_PUBLISH_H
4
+#define NETDATA_EXPORTING_PUBSUB_PUBLISH_H
5
+
6
+#define ERROR_LINE_MAX 1023
7
+
8
+#ifdef __cplusplus
9
+extern "C" {
10
+#endif
11
+
12
+struct pubsub_specific_data {
13
+ void *stub;
14
+ void *request;
15
+ void *completion_queue;
16
+
17
+ void *responses;
18
+ size_t last_tag;
19
+};
20
+
21
+int pubsub_init(
22
+ void *pubsub_specific_data_p, char *error_message, const char *destination, const char *credentials_file,
23
+ const char *project_id, const char *topic_id);
24
+
25
+int pubsub_add_message(void *pubsub_specific_data_p, char *data);
26
+
27
+int pubsub_publish(void *pubsub_specific_data_p, char *error_message, size_t buffered_metrics, size_t buffered_bytes);
28
+int pubsub_get_result(
29
+ void *pubsub_specific_data_p, char *error_message,
30
+ size_t *sent_metrics, size_t *sent_bytes, size_t *lost_metrics, size_t *lost_bytes);
31
+
32
+#ifdef __cplusplus
33
+}
34
+#endif
35
+
36
+#endif //NETDATA_EXPORTING_PUBSUB_PUBLISH_H
exporting/read_config.c
+27
-3
@@ -145,6 +145,8 @@ EXPORTING_CONNECTOR_TYPE exporting_select_type(const char *type)
145
return EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE;
146
} else if (!strcmp(type, "kinesis") || !strcmp(type, "kinesis:plaintext")) {
147
return EXPORTING_CONNECTOR_TYPE_KINESIS;
148
+ } else if (!strcmp(type, "pubsub") || !strcmp(type, "pubsub:plaintext")) {
149
+ return EXPORTING_CONNECTOR_TYPE_PUBSUB;
150
} else if (!strcmp(type, "mongodb") || !strcmp(type, "mongodb:plaintext"))
151
return EXPORTING_CONNECTOR_TYPE_MONGODB;
152
@@ -288,6 +290,7 @@ struct engine *read_exporting_config()
290
while (tmp_ci_list) {
291
struct instance *tmp_instance;
292
char *instance_name;
293
+ char *default_destination = "localhost";
294
295
info("Instance %s on %s", tmp_ci_list->local_ci.instance_name, tmp_ci_list->local_ci.connector_name);
296
@@ -310,6 +313,13 @@ struct engine *read_exporting_config()
313
}
314
#endif
315
316
+#ifndef ENABLE_EXPORTING_PUBSUB
317
+ if (tmp_ci_list->backend_type == EXPORTING_CONNECTOR_TYPE_PUBSUB) {
318
+ error("Google Cloud Pub/Sub support isn't compiled");
319
+ goto next_connector_instance;
320
+ }
321
+#endif
322
+
323
#ifndef HAVE_MONGOC
324
if (tmp_ci_list->backend_type == EXPORTING_CONNECTOR_TYPE_MONGODB) {
325
error("MongoDB support isn't compiled");
@@ -328,7 +338,6 @@ struct engine *read_exporting_config()
338
339
tmp_instance->config.name = strdupz(tmp_ci_list->local_ci.instance_name);
340
331
- tmp_instance->config.destination = strdupz(exporter_get(instance_name, "destination", "localhost"));
341
342
tmp_instance->config.update_every =
343
exporter_get_number(instance_name, EXPORTING_UPDATE_EVERY_OPTION_NAME, EXPORTING_UPDATE_EVERY_DEFAULT);
@@ -376,15 +385,28 @@ struct engine *read_exporting_config()
385
struct aws_kinesis_specific_config *connector_specific_config =
386
callocz(1, sizeof(struct aws_kinesis_specific_config));
387
388
+ default_destination = "us-east-1";
389
+
390
tmp_instance->config.connector_specific_config = connector_specific_config;
391
392
connector_specific_config->stream_name = strdupz(exporter_get(instance_name, "stream name", "netdata"));
382
-
393
connector_specific_config->auth_key_id = strdupz(exporter_get(instance_name, "aws_access_key_id", ""));
384
-
394
connector_specific_config->secure_key = strdupz(exporter_get(instance_name, "aws_secret_access_key", ""));
395
}
396
397
+ if (tmp_instance->config.type == EXPORTING_CONNECTOR_TYPE_PUBSUB) {
398
+ struct pubsub_specific_config *connector_specific_config =
399
+ callocz(1, sizeof(struct pubsub_specific_config));
400
+
401
+ default_destination = "pubsub.googleapis.com";
402
+
403
+ tmp_instance->config.connector_specific_config = connector_specific_config;
404
+
405
+ connector_specific_config->credentials_file = strdupz(exporter_get(instance_name, "credentials file", ""));
406
+ connector_specific_config->project_id = strdupz(exporter_get(instance_name, "project id", ""));
407
+ connector_specific_config->topic_id = strdupz(exporter_get(instance_name, "topic id", ""));
408
+ }
409
+
410
if (tmp_instance->config.type == EXPORTING_CONNECTOR_TYPE_MONGODB) {
411
struct mongodb_specific_config *connector_specific_config =
412
callocz(1, sizeof(struct mongodb_specific_config));
@@ -398,6 +420,8 @@ struct engine *read_exporting_config()
420
instance_name, "collection", ""));
421
}
422
423
+ tmp_instance->config.destination = strdupz(exporter_get(instance_name, "destination", default_destination));
424
+
425
#ifdef NETDATA_INTERNAL_CHECKS
426
info(
427
" Dest=[%s], upd=[%d], buffer=[%d] timeout=[%ld] options=[%u]",
exporting/tests/exporting_doubles.c
+49
@@ -262,6 +262,55 @@ int __wrap_kinesis_get_result(void *request_outcomes_p, char *error_message, siz
262
}
263
#endif // HAVE_KINESIS
264
265
+#if ENABLE_EXPORTING_PUBSUB
266
+int __wrap_pubsub_init(
267
+ void *pubsub_specific_data_p, char *error_message, const char *destination, const char *credentials_file,
268
+ const char *project_id, const char *topic_id)
269
+{
270
+ function_called();
271
+ check_expected_ptr(pubsub_specific_data_p);
272
+ check_expected_ptr(error_message);
273
+ check_expected_ptr(destination);
274
+ check_expected_ptr(credentials_file);
275
+ check_expected_ptr(project_id);
276
+ check_expected_ptr(topic_id);
277
+ return mock_type(int);
278
+}
279
+
280
+int __wrap_pubsub_add_message(void *pubsub_specific_data_p, char *data)
281
+{
282
+ function_called();
283
+ check_expected_ptr(pubsub_specific_data_p);
284
+ check_expected_ptr(data);
285
+ return mock_type(int);
286
+}
287
+
288
+int __wrap_pubsub_publish(
289
+ void *pubsub_specific_data_p, char *error_message, size_t buffered_metrics, size_t buffered_bytes)
290
+{
291
+ function_called();
292
+ check_expected_ptr(pubsub_specific_data_p);
293
+ check_expected_ptr(error_message);
294
+ check_expected(buffered_metrics);
295
+ check_expected(buffered_bytes);
296
+ return mock_type(int);
297
+}
298
+
299
+int __wrap_pubsub_get_result(
300
+ void *pubsub_specific_data_p, char *error_message,
301
+ size_t *sent_metrics, size_t *sent_bytes, size_t *lost_metrics, size_t *lost_bytes)
302
+{
303
+ function_called();
304
+ check_expected_ptr(pubsub_specific_data_p);
305
+ check_expected_ptr(error_message);
306
+ check_expected_ptr(sent_metrics);
307
+ check_expected_ptr(sent_bytes);
308
+ check_expected_ptr(lost_metrics);
309
+ check_expected_ptr(lost_bytes);
310
+ return mock_type(int);
311
+}
312
+#endif // ENABLE_EXPORTING_PUBSUB
313
+
314
#if HAVE_MONGOC
315
void __wrap_mongoc_init()
316
{
exporting/tests/test_exporting_engine.c
+149
@@ -1470,6 +1470,144 @@ static void test_aws_kinesis_connector_worker(void **state)
1470
}
1471
#endif // HAVE_KINESIS
1472
1473
+#if ENABLE_EXPORTING_PUBSUB
1474
+static void test_init_pubsub_instance(void **state)
1475
+{
1476
+ struct engine *engine = *state;
1477
+ struct instance *instance = engine->instance_root;
1478
+
1479
+ instance->config.options = EXPORTING_SOURCE_DATA_AS_COLLECTED | EXPORTING_OPTION_SEND_NAMES;
1480
+
1481
+ struct pubsub_specific_config *connector_specific_config =
1482
+ callocz(1, sizeof(struct pubsub_specific_config));
1483
+ instance->config.connector_specific_config = connector_specific_config;
1484
+ connector_specific_config->credentials_file = strdupz("/test/credentials/file");
1485
+ connector_specific_config->project_id = strdupz("test_project_id");
1486
+ connector_specific_config->topic_id = strdupz("test_topic_id");
1487
+
1488
+ expect_function_call(__wrap_pubsub_init);
1489
+ expect_not_value(__wrap_pubsub_init, pubsub_specific_data_p, NULL);
1490
+ expect_string(__wrap_pubsub_init, destination, "localhost");
1491
+ expect_string(__wrap_pubsub_init, error_message, "");
1492
+ expect_string(__wrap_pubsub_init, credentials_file, "/test/credentials/file");
1493
+ expect_string(__wrap_pubsub_init, project_id, "test_project_id");
1494
+ expect_string(__wrap_pubsub_init, topic_id, "test_topic_id");
1495
+ will_return(__wrap_pubsub_init, 0);
1496
+
1497
+ assert_int_equal(init_pubsub_instance(instance), 0);
1498
+
1499
+ assert_ptr_equal(instance->worker, pubsub_connector_worker);
1500
+ assert_ptr_equal(instance->start_batch_formatting, NULL);
1501
+ assert_ptr_equal(instance->start_host_formatting, format_host_labels_json_plaintext);
1502
+ assert_ptr_equal(instance->start_chart_formatting, NULL);
1503
+ assert_ptr_equal(instance->metric_formatting, format_dimension_collected_json_plaintext);
1504
+ assert_ptr_equal(instance->end_chart_formatting, NULL);
1505
+ assert_ptr_equal(instance->end_host_formatting, flush_host_labels);
1506
+ assert_ptr_equal(instance->end_batch_formatting, NULL);
1507
+ assert_ptr_not_equal(instance->buffer, NULL);
1508
+ buffer_free(instance->buffer);
1509
+ assert_ptr_not_equal(instance->connector_specific_data, NULL);
1510
+ freez(instance->connector_specific_data);
1511
+
1512
+ instance->config.options = EXPORTING_SOURCE_DATA_AVERAGE | EXPORTING_OPTION_SEND_NAMES;
1513
+
1514
+ expect_function_call(__wrap_pubsub_init);
1515
+ expect_not_value(__wrap_pubsub_init, pubsub_specific_data_p, NULL);
1516
+ expect_string(__wrap_pubsub_init, destination, "localhost");
1517
+ expect_string(__wrap_pubsub_init, error_message, "");
1518
+ expect_string(__wrap_pubsub_init, credentials_file, "/test/credentials/file");
1519
+ expect_string(__wrap_pubsub_init, project_id, "test_project_id");
1520
+ expect_string(__wrap_pubsub_init, topic_id, "test_topic_id");
1521
+ will_return(__wrap_pubsub_init, 0);
1522
+
1523
+ assert_int_equal(init_pubsub_instance(instance), 0);
1524
+ assert_ptr_equal(instance->metric_formatting, format_dimension_stored_json_plaintext);
1525
+
1526
+ free(connector_specific_config->credentials_file);
1527
+ free(connector_specific_config->project_id);
1528
+ free(connector_specific_config->topic_id);
1529
+}
1530
+
1531
+static void test_pubsub_connector_worker(void **state)
1532
+{
1533
+ struct engine *engine = *state;
1534
+ struct instance *instance = engine->instance_root;
1535
+ struct stats *stats = &instance->stats;
1536
+
1537
+ __real_mark_scheduled_instances(engine);
1538
+
1539
+ expect_function_call(__wrap_rrdhost_is_exportable);
1540
+ expect_value(__wrap_rrdhost_is_exportable, instance, instance);
1541
+ expect_value(__wrap_rrdhost_is_exportable, host, localhost);
1542
+ will_return(__wrap_rrdhost_is_exportable, 1);
1543
+
1544
+ RRDSET *st = localhost->rrdset_root;
1545
+ expect_function_call(__wrap_rrdset_is_exportable);
1546
+ expect_value(__wrap_rrdset_is_exportable, instance, instance);
1547
+ expect_value(__wrap_rrdset_is_exportable, st, st);
1548
+ will_return(__wrap_rrdset_is_exportable, 1);
1549
+
1550
+ __real_prepare_buffers(engine);
1551
+
1552
+ struct pubsub_specific_config *connector_specific_config =
1553
+ callocz(1, sizeof(struct pubsub_specific_config));
1554
+ instance->config.connector_specific_config = connector_specific_config;
1555
+ connector_specific_config->credentials_file = strdupz("/test/credentials/file");
1556
+ connector_specific_config->project_id = strdupz("test_project_id");
1557
+ connector_specific_config->topic_id = strdupz("test_topic_id");
1558
+
1559
+ struct pubsub_specific_data *connector_specific_data = callocz(1, sizeof(struct pubsub_specific_data));
1560
+ instance->connector_specific_data = (void *)connector_specific_data;
1561
+
1562
+ expect_function_call(__wrap_pubsub_add_message);
1563
+ expect_not_value(__wrap_pubsub_add_message, pubsub_specific_data_p, NULL);
1564
+ // The buffer is prepared by Graphite exporting connector
1565
+ expect_string(
1566
+ __wrap_pubsub_add_message, data,
1567
+ "netdata.test-host.chart_name.dimension_name;TAG1=VALUE1 TAG2=VALUE2 123000321 15051\n");
1568
+ will_return(__wrap_pubsub_add_message, 0);
1569
+
1570
+ expect_function_call(__wrap_pubsub_publish);
1571
+ expect_not_value(__wrap_pubsub_publish, pubsub_specific_data_p, NULL);
1572
+ expect_string(__wrap_pubsub_publish, error_message, "");
1573
+ expect_value(__wrap_pubsub_publish, buffered_metrics, 1);
1574
+ expect_value(__wrap_pubsub_publish, buffered_bytes, 84);
1575
+ will_return(__wrap_pubsub_publish, 0);
1576
+
1577
+ expect_function_call(__wrap_pubsub_get_result);
1578
+ expect_not_value(__wrap_pubsub_get_result, pubsub_specific_data_p, NULL);
1579
+ expect_not_value(__wrap_pubsub_get_result, error_message, NULL);
1580
+ expect_not_value(__wrap_pubsub_get_result, sent_metrics, NULL);
1581
+ expect_not_value(__wrap_pubsub_get_result, sent_bytes, NULL);
1582
+ expect_not_value(__wrap_pubsub_get_result, lost_metrics, NULL);
1583
+ expect_not_value(__wrap_pubsub_get_result, lost_bytes, NULL);
1584
+ will_return(__wrap_pubsub_get_result, 0);
1585
+
1586
+ expect_function_call(__wrap_send_internal_metrics);
1587
+ expect_value(__wrap_send_internal_metrics, instance, instance);
1588
+ will_return(__wrap_send_internal_metrics, 0);
1589
+
1590
+ pubsub_connector_worker(instance);
1591
+
1592
+ assert_int_equal(stats->buffered_metrics, 0);
1593
+ assert_int_equal(stats->buffered_bytes, 84);
1594
+ assert_int_equal(stats->received_bytes, 0);
1595
+ assert_int_equal(stats->sent_bytes, 84);
1596
+ assert_int_equal(stats->sent_metrics, 0);
1597
+ assert_int_equal(stats->lost_metrics, 0);
1598
+ assert_int_equal(stats->receptions, 1);
1599
+ assert_int_equal(stats->transmission_successes, 1);
1600
+ assert_int_equal(stats->transmission_failures, 0);
1601
+ assert_int_equal(stats->data_lost_events, 0);
1602
+ assert_int_equal(stats->lost_bytes, 0);
1603
+ assert_int_equal(stats->reconnects, 0);
1604
+
1605
+ free(connector_specific_config->credentials_file);
1606
+ free(connector_specific_config->project_id);
1607
+ free(connector_specific_config->topic_id);
1608
+}
1609
+#endif // ENABLE_EXPORTING_PUBSUB
1610
+
1611
#if HAVE_MONGOC
1612
static void test_init_mongodb_instance(void **state)
1613
{
@@ -1776,6 +1914,17 @@ int main(void)
1914
test_res += cmocka_run_group_tests_name("kinesis_exporting_connector", kinesis_tests, NULL, NULL);
1915
#endif
1916
1917
+#if ENABLE_EXPORTING_PUBSUB
1918
+ const struct CMUnitTest pubsub_tests[] = {
1919
+ cmocka_unit_test_setup_teardown(
1920
+ test_init_pubsub_instance, setup_configured_engine, teardown_configured_engine),
1921
+ cmocka_unit_test_setup_teardown(
1922
+ test_pubsub_connector_worker, setup_initialized_engine, teardown_initialized_engine),
1923
+ };
1924
+
1925
+ test_res += cmocka_run_group_tests_name("pubsub_exporting_connector", pubsub_tests, NULL, NULL);
1926
+#endif
1927
+
1928
#if HAVE_MONGOC
1929
const struct CMUnitTest mongodb_tests[] = {
1930
cmocka_unit_test_setup_teardown(
exporting/tests/test_exporting_engine.h
+14
@@ -18,6 +18,10 @@
18
#include "exporting/aws_kinesis/aws_kinesis.h"
19
#endif
20
21
+#if ENABLE_EXPORTING_PUBSUB
22
+#include "exporting/pubsub/pubsub.h"
23
+#endif
24
+
25
#if HAVE_MONGOC
26
#include "exporting/mongodb/mongodb.h"
27
#endif
@@ -145,6 +149,16 @@ void __wrap_kinesis_put_record(
149
size_t data_len);
150
int __wrap_kinesis_get_result(void *request_outcomes_p, char *error_message, size_t *sent_bytes, size_t *lost_bytes);
151
152
+int __wrap_pubsub_init(
153
+ void *pubsub_specific_data_p, char *error_message, const char *destination, const char *credentials_file,
154
+ const char *project_id, const char *topic_id);
155
+int __wrap_pubsub_add_message(void *pubsub_specific_data_p, char *data);
156
+int __wrap_pubsub_publish(
157
+ void *pubsub_specific_data_p, char *error_message, size_t buffered_metrics, size_t buffered_bytes);
158
+int __wrap_pubsub_get_result(
159
+ void *pubsub_specific_data_p, char *error_message,
160
+ size_t *sent_metrics, size_t *sent_bytes, size_t *lost_metrics, size_t *lost_bytes);
161
+
162
void __wrap_mongoc_init();
163
mongoc_uri_t *__wrap_mongoc_uri_new_with_error(const char *uri_string, bson_error_t *error);
164
int32_t __wrap_mongoc_uri_get_option_as_int32(const mongoc_uri_t *uri, const char *option, int32_t fallback);
libnetdata/config/appconfig.c
+2
@@ -79,6 +79,8 @@ int is_valid_connector(char *type, int check_reserved)
79
return rc;
80
} else if (!strcmp(type, "kinesis") || !strcmp(type, "kinesis:plaintext")) {
81
return rc;
82
+ } else if (!strcmp(type, "pubsub") || !strcmp(type, "pubsub:plaintext")) {
83
+ return rc;
84
} else if (!strcmp(type, "mongodb") || !strcmp(type, "mongodb:plaintext")) {
85
return rc;
86
}